#!/usr/bin/env python3 # @concierge - Gaeste-Einladungen mit Ablauf, Freischaltung und begrenzter # Verlaengerung (gitops#48, Design von sorb am 2026-08-01 festgezurrt). # # WARUM EIN EIGENER BOT UND NICHT DRAUPNIR # Draupnir ist ein Moderationsbot ohne Lebenszyklus-Funktionen. Er kann einen # Gast policy-seitig einschraenken, aber Links erzeugen, Ablaeufe verwalten und # Konten deaktivieren kann er nicht. Ihn dafuer zu verbiegen hiesse, Upstream- # Code zu forken, den wir sonst unveraendert mitziehen. # # WARUM AUTHENTIK UND NICHT SYNAPSE-REGISTRATION-TOKENS # In diesem Stack laeuft Registrierung ausschliesslich ueber Authentik (MAS-OIDC). # Synapse kennt gar keinen offenen Registrierungsweg - ein Registration-Token # waere wirkungslos. Der natuerliche Einladungslink ist deshalb ein # Authentik-Invitation-Token: einmalig verwendbar, mit eigenem Ablaufdatum. # # BERECHTIGUNG = GRUPPE **UND** RAUM # Autoritativ ist die Mitgliedschaft in der Authentik-Gruppe (INVITE_GROUP). # Zusaetzlich nimmt der Bot Kommandos nur im Einladungsraum an. Die Gruppe ist # die Kontrolle, der Raum die Transparenz: Jede Einladung hinterlaesst einen # nachlesbaren Eintrag, wer wen eingeladen hat. Beides zusammen, weil eine # Gruppe allein unsichtbar ist und ein Raum allein nicht autorisiert. # # ⚠️ ZUORDNUNG MATRIX -> AUTHENTIK # Der Bot nimmt an, dass der Matrix-Localpart dem Authentik-Benutzernamen # entspricht (@gast:axion1337.chat -> "gast"). Das gilt in diesem Stack, weil # MAS die Konten aus Authentik provisioniert. Stimmt das einmal nicht, findet # der Bot den Nutzer nicht und sagt das - er raet nicht. # # FEHLERVERHALTEN, BEWUSST ASYMMETRISCH # - Einladen/Freischalten scheitert LAUT: lieber keine Einladung als eine, von # der niemand weiss. # - Die Ablaufpruefung deaktiviert NUR, wenn Authentik sauber geantwortet hat. # Ein API-Fehler darf nicht dazu fuehren, dass Konten reihenweise abgeschaltet # werden - im Zweifel bleibt ein Gast einen Durchlauf laenger aktiv. # # Stdlib only, wie die uebrigen Bots dieses Verbunds. import json import logging import os import time import urllib.error import urllib.parse import urllib.request from datetime import datetime, timedelta, timezone log = logging.getLogger("concierge") MATRIX = os.environ["MATRIX_HOMESERVER"].rstrip("/") ROOM = os.environ["MATRIX_ROOM_ID"] AUTHENTIK = os.environ["AUTHENTIK_URL"].rstrip("/") INVITE_GROUP = os.environ.get("INVITE_GROUP", "invite-berechtigt") MEMBER_GROUP = os.environ.get("MEMBER_GROUP", "members") ADMIN_GROUP = os.environ.get("ADMIN_GROUP", "authentik Admins") INVITE_FLOW = os.environ.get("INVITE_FLOW_SLUG", "matrix-invitation") GUEST_DAYS = int(os.environ.get("GUEST_DAYS", "3")) MAX_RENEWALS = int(os.environ.get("MAX_RENEWALS", "2")) SWEEP_SECONDS = int(os.environ.get("SWEEP_SECONDS", "900")) # Attribute am Authentik-Nutzer. Praefix, damit sie nicht mit Feldern anderer # Werkzeuge kollidieren, die sich denselben attributes-Topf teilen. ATTR_EXPIRES = "threadnet_guest_expires_at" ATTR_RENEWALS = "threadnet_guest_renewals" ATTR_INVITED_BY = "threadnet_invited_by" def _read(path_env, direct_env): """Token entweder aus einer Datei (Secret-Mount) oder direkt. Dateien sind der Normalfall - ein Wert in der Umgebung steht in jedem Prozess-Dump.""" p = os.environ.get(path_env) if p: with open(p) as f: return f.read().strip() return os.environ[direct_env] MATRIX_TOKEN = _read("MATRIX_TOKEN_FILE", "MATRIX_TOKEN") AUTHENTIK_TOKEN = _read("AUTHENTIK_TOKEN_FILE", "AUTHENTIK_TOKEN") def _call(url, token, method="GET", body=None, scheme="Bearer"): data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, method=method) req.add_header("Authorization", f"{scheme} {token}") if data: req.add_header("Content-Type", "application/json") with urllib.request.urlopen(req, timeout=60) as r: raw = r.read() return json.loads(raw) if raw else {} def ak(path, method="GET", body=None): return _call(f"{AUTHENTIK}/api/v3{path}", AUTHENTIK_TOKEN, method, body) def mx(path, method="GET", body=None): return _call(f"{MATRIX}/_matrix/client/v3{path}", MATRIX_TOKEN, method, body) def say(text): txn = str(int(time.time() * 1000)) room = urllib.parse.quote(ROOM) mx(f"/rooms/{room}/send/m.room.message/{txn}", "PUT", {"msgtype": "m.notice", "body": text}) # --- Authentik --------------------------------------------------------------- def find_user(username): r = ak(f"/core/users/?username={urllib.parse.quote(username)}") for u in r.get("results", []): if u["username"] == username: return u return None def group_uuid(name): r = ak(f"/core/groups/?name={urllib.parse.quote(name)}") for g in r.get("results", []): if g["name"] == name: return g["pk"] return None def in_group(user, name): return any(g.get("name") == name for g in user.get("groups_obj", [])) def set_attrs(user, **changes): """attributes ist ein einzelnes JSON-Feld: PATCH ersetzt es komplett. Wer nur einen Schluessel schickt, loescht alle anderen - deshalb immer mischen.""" attrs = dict(user.get("attributes") or {}) for k, v in changes.items(): if v is None: attrs.pop(k, None) else: attrs[k] = v return ak(f"/core/users/{user['pk']}/", "PATCH", {"attributes": attrs}) def localpart(mxid): return mxid.lstrip("@").split(":")[0] # --- Kommandos --------------------------------------------------------------- def darf_einladen(sender): u = find_user(localpart(sender)) return u is not None and in_group(u, INVITE_GROUP), u def ist_admin(sender): u = find_user(localpart(sender)) return u is not None and in_group(u, ADMIN_GROUP) def cmd_einladen(sender, args): ok, _ = darf_einladen(sender) if not ok: say(f"{sender}: du bist nicht in der Gruppe '{INVITE_GROUP}'.") return name = (args or "gast").strip().replace(" ", "-")[:40] expires = datetime.now(timezone.utc) + timedelta(days=GUEST_DAYS) inv = ak("/stages/invitation/invitations/", "POST", { "name": f"gast-{name}-{int(time.time())}", "expires": expires.isoformat(), "single_use": True, "fixed_data": {ATTR_INVITED_BY: sender}, }) link = f"{AUTHENTIK}/if/flow/{INVITE_FLOW}/?itoken={inv['pk']}" say(f"Einladung von {sender} fuer '{name}':\n{link}\n" f"Einmalig verwendbar, verfaellt {expires:%d.%m.%Y %H:%M} UTC.") def cmd_freischalten(sender, args): if not ist_admin(sender): say(f"{sender}: Freischalten darf nur die Gruppe '{ADMIN_GROUP}'.") return u = find_user(localpart(args.strip())) if not u: say(f"Kein Authentik-Konto zu '{args.strip()}' gefunden.") return set_attrs(u, **{ATTR_EXPIRES: None, ATTR_RENEWALS: None}) gid = group_uuid(MEMBER_GROUP) if gid: ak(f"/core/groups/{gid}/add_user/", "POST", {"pk": u["pk"]}) say(f"{u['username']} ist dauerhaft freigeschaltet (von {sender}).") def cmd_verlaengern(sender, args): ok, _ = darf_einladen(sender) if not ok: say(f"{sender}: du bist nicht in der Gruppe '{INVITE_GROUP}'.") return u = find_user(localpart(args.strip())) if not u: say(f"Kein Authentik-Konto zu '{args.strip()}' gefunden.") return used = int((u.get("attributes") or {}).get(ATTR_RENEWALS, 0)) if used >= MAX_RENEWALS: say(f"{u['username']}: {MAX_RENEWALS} Verlaengerungen sind aufgebraucht. " f"Jetzt muss ein Admin freischalten.") return neu = datetime.now(timezone.utc) + timedelta(days=1) set_attrs(u, **{ATTR_EXPIRES: neu.isoformat(), ATTR_RENEWALS: used + 1}) if not u.get("is_active"): ak(f"/core/users/{u['pk']}/", "PATCH", {"is_active": True}) say(f"{u['username']} um einen Tag verlaengert ({used + 1}/{MAX_RENEWALS}), " f"laeuft {neu:%d.%m.%Y %H:%M} UTC ab.") def cmd_status(_sender, _args): r = ak("/core/users/?page_size=200") zeilen = [] for u in r.get("results", []): exp = (u.get("attributes") or {}).get(ATTR_EXPIRES) if exp: used = (u.get("attributes") or {}).get(ATTR_RENEWALS, 0) zustand = "aktiv" if u.get("is_active") else "deaktiviert" zeilen.append(f" {u['username']}: laeuft {exp[:16]} ab, " f"{used}/{MAX_RENEWALS} verlaengert, {zustand}") say("Gaeste:\n" + ("\n".join(zeilen) if zeilen else " keine offenen Gastkonten")) def cmd_hilfe(_sender, _args): say("!einladen - Einladungslink erzeugen\n" "!verlaengern @nutzer - um einen Tag verlaengern (begrenzt)\n" "!freischalten @nutzer - dauerhaft freischalten (nur Admins)\n" "!status - offene Gastkonten anzeigen") BEFEHLE = { "!einladen": cmd_einladen, "!verlaengern": cmd_verlaengern, "!freischalten": cmd_freischalten, "!status": cmd_status, "!hilfe": cmd_hilfe, } # --- Ablaufpruefung ---------------------------------------------------------- def sweep(): try: r = ak("/core/users/?page_size=200") except Exception as e: # KEIN Deaktivieren bei API-Fehlern - siehe Kopfkommentar. log.warning("Ablaufpruefung uebersprungen, Authentik nicht erreichbar: %s", e) return jetzt = datetime.now(timezone.utc) for u in r.get("results", []): exp = (u.get("attributes") or {}).get(ATTR_EXPIRES) if not exp or not u.get("is_active"): continue try: faellig = datetime.fromisoformat(exp) except ValueError: log.warning("%s: unlesbares Ablaufdatum %r", u["username"], exp) continue if faellig.tzinfo is None: faellig = faellig.replace(tzinfo=timezone.utc) if faellig <= jetzt: ak(f"/core/users/{u['pk']}/", "PATCH", {"is_active": False}) say(f"Gastkonto {u['username']} ist abgelaufen und wurde deaktiviert. " f"'!verlaengern @{u['username']}' oder Admin-Freischaltung.") # --- Hauptschleife ----------------------------------------------------------- def main(): logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") mx(f"/rooms/{urllib.parse.quote(ROOM)}/join", "POST", {}) # Ab jetzt, nicht die Raumhistorie: ein Neustart soll keine alten Kommandos # erneut ausfuehren. since = mx("/sync?timeout=0").get("next_batch") log.info("bereit, Raum %s", ROOM) letzter_sweep = 0.0 while True: try: if time.time() - letzter_sweep > SWEEP_SECONDS: sweep() letzter_sweep = time.time() r = mx(f"/sync?since={urllib.parse.quote(since)}&timeout=30000") since = r.get("next_batch", since) raum = r.get("rooms", {}).get("join", {}).get(ROOM, {}) for ev in raum.get("timeline", {}).get("events", []): if ev.get("type") != "m.room.message": continue c = ev.get("content", {}) if c.get("msgtype") != "m.text": continue text = (c.get("body") or "").strip() wort = text.split(" ", 1)[0].lower() if wort not in BEFEHLE: continue rest = text[len(wort):].strip() try: BEFEHLE[wort](ev["sender"], rest) except Exception as e: log.exception("Kommando %s fehlgeschlagen", wort) say(f"'{wort}' fehlgeschlagen: {e}") except urllib.error.HTTPError as e: log.warning("HTTP %s bei /sync - warte", e.code) time.sleep(10) except Exception: log.exception("Schleifenfehler") time.sleep(10) if __name__ == "__main__": main()