titan-iac/services/communication/guest-register-configmap.yaml

# services/communication/guest-register-configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: matrix-guest-register
data:
  server.py: |
    import json
    import os
    import random
    import secrets
    from http.server import BaseHTTPRequestHandler, HTTPServer
    from urllib import error, parse, request

    MATRIX_BASE = os.environ.get("MATRIX_BASE", "http://othrys-synapse-matrix-synapse:8008").rstrip("/")
    SERVER_NAME = os.environ.get("MATRIX_SERVER_NAME", "live.bstein.dev")

    AS_TOKEN = os.environ["AS_TOKEN"]
    HS_TOKEN = os.environ["HS_TOKEN"]

    RATE_WINDOW_SEC = int(os.environ.get("RATE_WINDOW_SEC", "60"))
    RATE_MAX = int(os.environ.get("RATE_MAX", "30"))
    _rate = {}  # ip -> [window_start, count]

    ADJ = ["brisk", "calm", "eager", "gentle", "merry", "nifty", "rapid", "sunny", "witty", "zesty"]
    NOUN = ["otter", "falcon", "comet", "ember", "grove", "harbor", "meadow", "raven", "river", "summit"]

    def _json(method, url, *, token=None, body=None, timeout=20):
        headers = {"Content-Type": "application/json"}
        if token:
            headers["Authorization"] = f"Bearer {token}"
        data = None
        if body is not None:
            data = json.dumps(body).encode()
        req = request.Request(url, data=data, headers=headers, method=method)
        try:
            with request.urlopen(req, timeout=timeout) as resp:
                raw = resp.read()
                payload = json.loads(raw.decode()) if raw else {}
                return resp.status, payload
        except error.HTTPError as e:
            raw = e.read()
            try:
                payload = json.loads(raw.decode()) if raw else {}
            except Exception:
                payload = {}
            return e.code, payload

    def _generate_localpart():
        return "guest-" + secrets.token_hex(6)

    def _generate_displayname():
        return f"{random.choice(ADJ)}-{random.choice(NOUN)}"

    def _set_displayname(access_token, user_id, displayname):
        try:
            _json(
                "PUT",
                f"{MATRIX_BASE}/_matrix/client/v3/profile/{parse.quote(user_id)}/displayname",
                token=access_token,
                body={"displayname": displayname},
                timeout=15,
            )
        except Exception:
            return

    def _register_user(localpart):
        url = f"{MATRIX_BASE}/_matrix/client/v3/register?access_token={parse.quote(AS_TOKEN)}"
        status, payload = _json(
            "POST",
            url,
            body={
                "type": "m.login.application_service",
                "username": localpart,
                "inhibit_login": False,
                "initial_device_display_name": "Guest session",
            },
            timeout=25,
        )
        if status != 200 or "access_token" not in payload:
            raise RuntimeError("register_failed")
        return payload

    def _rate_check(ip, now):
        win, cnt = _rate.get(ip, (now, 0))
        if now - win > RATE_WINDOW_SEC:
            _rate[ip] = (now, 1)
            return True
        if cnt >= RATE_MAX:
            return False
        _rate[ip] = (win, cnt + 1)
        return True

    def _is_appservice_auth(auth_header):
        if not auth_header:
            return False
        parts = auth_header.split(" ", 1)
        if len(parts) != 2:
            return False
        scheme, token = parts
        if scheme.lower() != "bearer":
            return False
        return secrets.compare_digest(token, HS_TOKEN)

    class Handler(BaseHTTPRequestHandler):
        server_version = "matrix-guest-register"

        def _send_json(self, code, payload):
            body = json.dumps(payload).encode()
            self.send_response(code)
            self.send_header("Content-Type", "application/json")
            self.send_header("Access-Control-Allow-Origin", "*")
            self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
            self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization, X-Requested-With")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

        def do_OPTIONS(self):  # noqa: N802
            self.send_response(204)
            self.send_header("Access-Control-Allow-Origin", "*")
            self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
            self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization, X-Requested-With")
            self.end_headers()

        def do_GET(self):  # noqa: N802
            parsed = parse.urlparse(self.path)

            if parsed.path in ("/healthz", "/"):
                return self._send_json(200, {"ok": True})

            if parsed.path.startswith("/_matrix/app/v1/users/"):
                if not _is_appservice_auth(self.headers.get("authorization", "")):
                    return self._send_json(401, {"errcode": "M_UNAUTHORIZED", "error": "unauthorized"})
                return self._send_json(200, {})

            if parsed.path.startswith("/_matrix/app/v1/rooms/"):
                if not _is_appservice_auth(self.headers.get("authorization", "")):
                    return self._send_json(401, {"errcode": "M_UNAUTHORIZED", "error": "unauthorized"})
                return self._send_json(200, {})

            return self._send_json(404, {"errcode": "M_NOT_FOUND", "error": "not_found"})

        def do_POST(self):  # noqa: N802
            parsed = parse.urlparse(self.path)

            if parsed.path.startswith("/_matrix/app/v1/transactions/"):
                if not _is_appservice_auth(self.headers.get("authorization", "")):
                    return self._send_json(401, {"errcode": "M_UNAUTHORIZED", "error": "unauthorized"})
                return self._send_json(200, {})

            if parsed.path not in ("/_matrix/client/v3/register", "/_matrix/client/r0/register"):
                return self._send_json(404, {"errcode": "M_NOT_FOUND", "error": "not_found"})

            qs = parse.parse_qs(parsed.query)
            kind = (qs.get("kind") or ["user"])[0]
            if kind != "guest":
                return self._send_json(
                    403,
                    {
                        "errcode": "M_FORBIDDEN",
                        "error": "Registration is disabled; use https://bstein.dev/request-access for accounts.",
                    },
                )

            xfwd = self.headers.get("x-forwarded-for", "")
            ip = (xfwd.split(",")[0].strip() if xfwd else "") or self.client_address[0]
            now = __import__("time").time()
            if not _rate_check(ip, now):
                return self._send_json(429, {"errcode": "M_LIMIT_EXCEEDED", "error": "rate_limited"})

            length = int(self.headers.get("content-length", "0") or "0")
            _ = self.rfile.read(length) if length else b"{}"

            try:
                displayname = _generate_displayname()
                localpart = _generate_localpart()
                login_payload = _register_user(localpart)
                _set_displayname(login_payload.get("access_token"), login_payload.get("user_id"), displayname)
            except Exception:
                return self._send_json(502, {"errcode": "M_UNKNOWN", "error": "guest_provision_failed"})

            resp = {
                "user_id": login_payload.get("user_id") or f"@{localpart}:{SERVER_NAME}",
                "access_token": login_payload.get("access_token"),
                "device_id": login_payload.get("device_id"),
                "home_server": SERVER_NAME,
            }
            return self._send_json(200, resp)

    def main():
        port = int(os.environ.get("PORT", "8080"))
        HTTPServer(("0.0.0.0", port), Handler).serve_forever()

    if __name__ == "__main__":
        main()