weircon-front/mcpkeys.py

"""Central MCP-nøgleregister/vault for portalen.

Ét sted for homelab'ens MCP-servere:
- **external**-entries (crawlkit, Gitea, HA, Paperless): vi gemmer deres token i
  klartekst, så `claude mcp add`-kommandoen altid kan vises. Portalen autentificerer
  dem IKKE — det gør den enkelte downstream-server.
- **portal_tokens**: bearer-tokens til portalens EGEN MCP (hash til auth + klartekst
  til visning).

Filen skrives mode 600 (samme niveau som /etc/portal.env).
"""
import hashlib
import hmac
import json
import os
import secrets
from datetime import datetime, timezone

import httpx

import config

REG = config.MCP_REGISTRY
SELF_URL = config.MCP_SELF_URL


def _now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def _hash(t: str) -> str:
    return hashlib.sha256(t.encode("utf-8")).hexdigest()


def _load() -> dict:
    try:
        d = json.loads(REG.read_text("utf-8"))
    except Exception:
        d = {}
    d.setdefault("external", [])
    d.setdefault("portal_tokens", [])
    d.setdefault("oauth_access", {"default": None, "users": {}})
    _backfill_allowed(d)
    return d


def _namespaces(d: dict) -> list:
    """Every backend the gateway can expose: the portal itself + each external."""
    return ["portal"] + [e["id"] for e in d.get("external", []) if e.get("id")]


def _backfill_allowed(d: dict) -> None:
    """Give pre-scoping tokens an explicit allow-list of what exists TODAY.

    Deliberately not a wildcard: a token minted before scoping keeps exactly the
    access it already had, and any backend added later (mail, say) is denied
    until someone ticks it. Default-deny for new backends is the whole point.
    """
    current = _namespaces(d)
    for t in d.get("portal_tokens", []):
        if t.get("allowed") is None:
            t["allowed"] = list(current)
    oa = d.get("oauth_access") or {}
    if oa.get("default") is None:
        oa["default"] = list(current)
        d["oauth_access"] = oa


def _save(d: dict) -> None:
    REG.write_text(json.dumps(d, ensure_ascii=False, indent=2), "utf-8")
    try:
        os.chmod(REG, 0o600)
    except OSError:
        pass


# ---------------- external MCP-entries ----------------
def list_external() -> list:
    return _load()["external"]


def upsert_external(entry: dict) -> None:
    d = _load()
    d["external"] = [e for e in d["external"] if e.get("id") != entry["id"]] + [entry]
    _save(d)


def delete_external(eid: str) -> None:
    d = _load()
    d["external"] = [e for e in d["external"] if e.get("id") != eid]
    _save(d)


# ---------------- portalens egne MCP-tokens ----------------
def list_portal_tokens() -> list:
    return _load()["portal_tokens"]


def mint_portal_token(name: str, allowed: list | None = None) -> str:
    token = secrets.token_urlsafe(32)
    d = _load()
    d["portal_tokens"].append({
        "name": name, "token_hash": _hash(token),
        "allowed": list(allowed) if allowed is not None else [],
        "created_at": _now(), "last_used_at": None, "revoked": False})
    _save(d)
    return token


def namespaces() -> list:
    """Backends the gateway can expose (for the admin UI's checkboxes)."""
    return _namespaces(_load())


def set_token_allowed(name: str, allowed: list) -> None:
    d = _load()
    for t in d["portal_tokens"]:
        if t.get("name") == name:
            t["allowed"] = list(allowed)
    _save(d)


def get_oauth_access() -> dict:
    return _load()["oauth_access"]


def set_oauth_access(default: list, users: dict) -> None:
    d = _load()
    d["oauth_access"] = {"default": list(default),
                         "users": {k: list(v) for k, v in users.items()}}
    _save(d)


def set_oauth_default(allowed: list) -> None:
    d = _load()
    oa = d.get("oauth_access") or {}
    oa["default"] = list(allowed)
    d["oauth_access"] = oa
    _save(d)


def set_oauth_user(username: str, allowed: list) -> None:
    d = _load()
    oa = d.get("oauth_access") or {}
    users = oa.get("users") or {}
    users[username] = list(allowed)
    oa["users"] = users
    d["oauth_access"] = oa
    _save(d)


def del_oauth_user(username: str) -> None:
    d = _load()
    oa = d.get("oauth_access") or {}
    (oa.get("users") or {}).pop(username, None)
    d["oauth_access"] = oa
    _save(d)


def allowed_for_oauth(subject: str | None) -> list:
    """What an Authentik-authenticated caller may reach.

    Falls back to the configured default when the user has no explicit entry —
    the default is itself a list, so it can be narrowed to nothing.
    """
    oa = _load()["oauth_access"]
    users = oa.get("users") or {}
    for key in filter(None, [subject, (subject or "").lower()]):
        if key in users:
            return list(users[key])
    return list(oa.get("default") or [])


def allowed_for_oauth_any(candidates) -> list:
    """What an Authentik caller may reach, matched on ANY of its identifiers.

    Authentik's ``sub`` is a UUID, but the admin UI stores entries by *username*
    — so matching on the subject alone never hits and everyone silently gets the
    default. Try every identity claim the token carries, case-insensitively.
    """
    oa = _load()["oauth_access"]
    users = oa.get("users") or {}
    lowered = {str(k).lower(): v for k, v in users.items()}
    for c in candidates:
        if not c:
            continue
        if c in users:
            return list(users[c])
        hit = lowered.get(str(c).lower())
        if hit is not None:
            return list(hit)
    return list(oa.get("default") or [])


def resolve_portal_token(token: str) -> dict | None:
    """Return the token record (incl. its allow-list), or None. Bumps
    last_used_at at most ~1/min, like verify_portal_token."""
    if not token:
        return None
    h = _hash(token)
    d = _load()
    now = _now()
    for t in d["portal_tokens"]:
        if not t.get("revoked") and hmac.compare_digest(t.get("token_hash", ""), h):
            last = t.get("last_used_at") or ""
            if last < now[:16]:
                t["last_used_at"] = now
                _save(d)
            return dict(t)
    return None


def revoke_portal_token(name: str) -> None:
    d = _load()
    for t in d["portal_tokens"]:
        if t.get("name") == name:
            t["revoked"] = True
    _save(d)


def verify_portal_token(token: str) -> bool:
    """True for a valid, non-revoked portal token."""
    return resolve_portal_token(token) is not None


# ---------------- helpers ----------------
def build_command(*, mcp_id, transport="http", url=None, token=None,
                  header="Authorization", command=None) -> str:
    if transport == "stdio":
        return f"claude mcp add --scope user {mcp_id} -- {(command or '').strip()}".strip()
    hdr = f' --header "{header}: Bearer {token}"' if token else ""
    return f"claude mcp add --scope user --transport {transport} {mcp_id} {url}{hdr}"


class PortalMCPAuthMiddleware:
    """Pure-ASGI gate: require a valid portal bearer token on the /mcp endpoint;
    everything else (incl. lifespan + the web UI) passes through untouched."""

    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope.get("type") == "http" and scope.get("path", "").rstrip("/") == "/mcp":
            headers = dict(scope.get("headers") or [])
            raw = headers.get(b"authorization", b"").decode("latin-1")
            token = raw[7:].strip() if raw[:7].lower() == "bearer " else ""
            if not verify_portal_token(token):
                body = b'{"error":"unauthorized"}'
                await send({"type": "http.response.start", "status": 401,
                            "headers": [(b"content-type", b"application/json"),
                                        (b"content-length", str(len(body)).encode())]})
                await send({"type": "http.response.body", "body": body})
                return
        await self.app(scope, receive, send)


async def health_check(entry: dict) -> dict:
    """Ping an external MCP endpoint. Any 2xx/4xx = reachable (401 = auth required)."""
    res = {"at": _now(), "ok": None, "detail": ""}
    url = entry.get("url")
    if entry.get("transport") == "stdio" or not url:
        res["detail"] = "stdio (lokal) — ikke pinget"
        return res
    try:
        async with httpx.AsyncClient(timeout=8) as c:
            r = await c.post(url, json={"jsonrpc": "2.0", "id": 1, "method": "initialize",
                                        "params": {}},
                             headers={"Content-Type": "application/json",
                                      "Accept": "application/json, text/event-stream"})
        res["ok"] = 200 <= r.status_code < 500
        res["detail"] = f"HTTP {r.status_code}" + (" — nåbar (auth kræves)"
                                                   if r.status_code == 401 else "")
    except Exception as e:
        res["ok"] = False
        res["detail"] = str(e)[:160]
    return res