103 lines
2.9 KiB
Python
103 lines
2.9 KiB
Python
from io import DEFAULT_BUFFER_SIZE
|
|
import os
|
|
import time
|
|
import json
|
|
from psycopg2.sql import NULL
|
|
import requests
|
|
import psycopg2 as ps
|
|
|
|
def read_file(path):
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
return f.read()
|
|
|
|
def getenv(name):
|
|
try:
|
|
return os.environ[name]
|
|
except KeyError:
|
|
print(f"[ERROR] Missing environment variable {name}. You should NOT run this script by hand, please use systemd mastodon-kanidm-sync.service.")
|
|
exit(1)
|
|
|
|
|
|
# Import configuration
|
|
KANIDM_URL = getenv("KANIDM_URL")
|
|
KANIDM_TOKEN = read_file(getenv("KANIDM_TOKEN_PATH")).strip()
|
|
USERDATA = read_file(getenv("USERDATA_FILE_PATH")).strip()
|
|
|
|
# Fetch configuration from userdata file
|
|
# Userdata file is json list with information about what users are configured by kanidm
|
|
userdata = json.loads(USERDATA)
|
|
|
|
# Load database
|
|
conn = ps.connect(
|
|
dbname=getenv("POSTGRES_DBNAME"),
|
|
user=getenv("POSTGRES_USER"),
|
|
host=getenv("POSTGRES_HOST")
|
|
)
|
|
|
|
# Fetch current userdata from database
|
|
cur = conn.cursor()
|
|
cur.execute('''
|
|
SELECT identities.uid, users.id, user_roles.name
|
|
FROM users
|
|
JOIN identities
|
|
ON users.id = identities.id
|
|
LEFT JOIN user_roles
|
|
ON users.role_id = user_roles.id;
|
|
'''
|
|
)
|
|
|
|
state = cur.fetchall()
|
|
|
|
users = {}
|
|
for i in state:
|
|
users[i[0]] = {
|
|
"id": i[1],
|
|
"role": i[2],
|
|
"isKanidmUser": False
|
|
}
|
|
|
|
# Fetch Kanidm userdata
|
|
kanidm_users_raw = requests.get(
|
|
f"{KANIDM_URL}/v1/person",
|
|
headers={
|
|
"Authorization": f"Bearer {KANIDM_TOKEN}",
|
|
"Content-Type": "application/json",
|
|
},
|
|
timeout=5,
|
|
).json()
|
|
|
|
for i in kanidm_users_raw:
|
|
i = i["attrs"]
|
|
for uid in i["name"]: # [user].attrs.name is a list
|
|
if uid in users: # Don't apply anything for users who have no mastodon access (sp.matrix.users) or didn't register
|
|
for group in i["attrs"]["memberof"]:
|
|
if group.startswith("sp.matrix.admins@"):
|
|
if uid not in userdata:
|
|
userdata.append(uid)
|
|
users[uid]["isKanidmUser"] = True
|
|
users[uid]["role"] = "Admin"
|
|
print(f"[INFO] {uid} got role Admin")
|
|
break
|
|
elif group.startswith("sp.matrix.moderators@"):
|
|
if uid not in userdata:
|
|
userdata.append(uid)
|
|
users[uid]["isKanidmUser"] = True
|
|
users[uid]["role"] = "Moderator"
|
|
print(f"[INFO] {uid} got role Moderator")
|
|
break
|
|
elif uid in userdata:
|
|
# If user, who previously had a role, has no roles set by Kanidm, delete them from userdata list so allow setting roles directly by mastodon
|
|
users[uid]["isKanidmUser"] = True
|
|
users[uid]["role"] = None
|
|
userdata.remove(uid)
|
|
print(f"[INFO] {uid} has no roles")
|
|
|
|
print("[DEBUG] ", users) # DEBUG
|
|
|
|
cur.close()
|
|
conn.close()
|
|
|
|
print("[INFO] Final userdata file: ", userdata)
|
|
with open(getenv("USERDATA_FILE_PATH"), "w") as f:
|
|
f.write(json.dumps(userdata))
|