axion1337.chat-gitops/apps/production/concierge-bot.py

#!/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 <name> - 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()