user-engine/src/user_engine/web.py

"""Dependency-free WSGI transport for the user-engine portal.

Authentication is deliberately delegated to KeyCape (or another OIDC-aware
edge).  The application accepts claims only when the edge presents a shared
authentication marker configured at process start.  This keeps passwords,
MFA material, provider administration credentials, and browser sessions out
of user-engine.
"""

from __future__ import annotations

from dataclasses import asdict, is_dataclass
from enum import Enum
from html import escape
import json
import secrets
from typing import Any, Callable, Iterable, Mapping
from urllib.parse import parse_qs

from user_engine.domain import AccountStatus
from user_engine.errors import AuthorizationDenied, ConflictError, NotFoundError, ValidationError
from user_engine.oidc import OIDCClient, cookie_value
from user_engine.service import UserEngineService

StartResponse = Callable[[str, list[tuple[str, str]]], Any]


def _jsonable(value: Any) -> Any:
    if is_dataclass(value):
        return {key: _jsonable(item) for key, item in asdict(value).items()}
    if isinstance(value, Enum):
        return value.value
    if isinstance(value, Mapping):
        return {str(key): _jsonable(item) for key, item in value.items()}
    if isinstance(value, (tuple, list)):
        return [_jsonable(item) for item in value]
    if hasattr(value, "isoformat"):
        return value.isoformat()
    return value


class PortalApplication:
    """Small, auditable HTTP adapter over :class:`UserEngineService`."""

    def __init__(
        self,
        service: UserEngineService,
        *,
        trusted_proxy_secret: str,
        login_url: str,
        public_registration: bool = True,
        oidc_client: OIDCClient | None = None,
    ) -> None:
        if len(trusted_proxy_secret) < 24:
            raise ValueError("trusted proxy secret must contain at least 24 characters")
        self.service = service
        self.trusted_proxy_secret = trusted_proxy_secret
        self.login_url = login_url
        self.public_registration = public_registration
        self.oidc_client = oidc_client

    def __call__(self, environ: Mapping[str, Any], start_response: StartResponse) -> Iterable[bytes]:
        correlation_id = environ.get("HTTP_X_REQUEST_ID") or f"corr_{secrets.token_hex(12)}"
        try:
            return self._dispatch(environ, start_response, str(correlation_id))
        except (ValidationError, ConflictError, ValueError) as exc:
            return self._error(start_response, "400 Bad Request", "invalid_request", str(exc), correlation_id)
        except AuthorizationDenied:
            return self._error(start_response, "403 Forbidden", "access_denied", "Access denied.", correlation_id)
        except NotFoundError:
            return self._error(start_response, "404 Not Found", "not_found", "Resource not found.", correlation_id)
        except (json.JSONDecodeError, UnicodeDecodeError):
            return self._error(start_response, "400 Bad Request", "invalid_json", "Malformed request body.", correlation_id)

    def _dispatch(self, environ: Mapping[str, Any], start_response: StartResponse, correlation_id: str) -> Iterable[bytes]:
        method = str(environ.get("REQUEST_METHOD", "GET")).upper()
        path = str(environ.get("PATH_INFO", "/")).rstrip("/") or "/"
        if path == "/healthz":
            return self._json(start_response, "200 OK", _jsonable(self.service.health()), correlation_id)
        if path == "/readyz":
            report = self.service.readiness()
            return self._json(start_response, "200 OK" if report.ready else "503 Service Unavailable", _jsonable(report), correlation_id)
        if path in {"/login", "/oidc/start"}:
            location = self.oidc_client.begin() if self.oidc_client else self.login_url
            start_response("303 See Other", [("Location", location), *self._security_headers(correlation_id)])
            return [b""]
        if path == "/oidc/callback":
            if self.oidc_client is None:
                raise NotFoundError("OIDC login is not configured")
            query = parse_qs(str(environ.get("QUERY_STRING", "")))
            if query.get("error"):
                raise AuthorizationDenied("OIDC login failed")
            session_id = self.oidc_client.complete(
                code=query.get("code", [""])[0],
                state=query.get("state", [""])[0],
            )
            headers = [
                ("Location", "/"),
                ("Set-Cookie", f"ue_session={session_id}; Path=/; HttpOnly; Secure; SameSite=Lax; Max-Age=3600"),
                *self._security_headers(correlation_id),
            ]
            start_response("303 See Other", headers)
            return [b""]
        if path == "/logout" and method == "POST":
            session_id = cookie_value(str(environ.get("HTTP_COOKIE", "")), "ue_session")
            if session_id and self.oidc_client:
                self.oidc_client.logout(session_id)
            start_response(
                "303 See Other",
                [("Location", "/"), ("Set-Cookie", "ue_session=; Path=/; HttpOnly; Secure; SameSite=Lax; Max-Age=0"), *self._security_headers(correlation_id)],
            )
            return [b""]
        if path == "/" and method == "GET":
            actor = self._optional_actor(environ)
            return self._html(start_response, self._home(actor), correlation_id)

        actor = self._actor(environ)
        if path == "/api/v1/me" and method == "GET":
            return self._json(start_response, "200 OK", _jsonable(self.service.me(self._claims(environ), correlation_id=correlation_id)), correlation_id)
        if path == "/api/v1/registrations" and method == "POST":
            if not self.public_registration:
                raise AuthorizationDenied("public registration disabled")
            body = self._body(environ)
            session = self.service.start_registration(
                actor,
                tenant=body.get("tenant"),
                correlation_id=correlation_id,
            )
            return self._json(start_response, "201 Created", _jsonable(session), correlation_id)
        if path.startswith("/api/v1/registrations/") and path.endswith("/complete") and method == "POST":
            registration_id = path.split("/")[4]
            body = self._body(environ)
            result = self.service.complete_registration(
                actor,
                registration_id,
                display_name=body.get("display_name"),
                primary_email=body.get("primary_email"),
                correlation_id=correlation_id,
            )
            return self._json(start_response, "200 OK", _jsonable(result), correlation_id)
        if path.startswith("/api/v1/tenants/") and path.endswith("/users") and method == "GET":
            tenant = path.split("/")[4]
            self.service.resolve_tenant_context(actor, tenant)
            memberships = self.service.store.memberships_for_tenant(tenant)
            offset, limit = self._page(environ)
            items = memberships[offset : offset + limit]
            payload = {"items": _jsonable(items), "offset": offset, "limit": limit, "total": len(memberships)}
            return self._json(start_response, "200 OK", payload, correlation_id)
        if path.startswith("/api/v1/tenants/") and "/users/" in path and method == "PATCH":
            parts = path.split("/")
            tenant, user_id = parts[4], parts[6]
            body = self._body(environ)
            status = AccountStatus(str(body["status"]))
            result = self.service.set_tenant_account_status(
                actor, user_id, status, tenant=tenant, correlation_id=correlation_id
            )
            return self._json(start_response, "200 OK", _jsonable(result), correlation_id)
        if path.startswith("/admin/") and method == "GET":
            tenant = path.split("/")[2]
            self.service.resolve_tenant_context(actor, tenant)
            memberships = self.service.store.memberships_for_tenant(tenant)
            return self._html(start_response, self._admin(tenant, memberships), correlation_id)
        return self._error(start_response, "404 Not Found", "not_found", "Resource not found.", correlation_id)

    def _claims(self, environ: Mapping[str, Any]) -> Mapping[str, Any]:
        if self.oidc_client is not None:
            session_id = cookie_value(str(environ.get("HTTP_COOKIE", "")), "ue_session")
            if session_id:
                claims = self.oidc_client.claims(session_id)
                if claims is not None:
                    return claims
        marker = str(environ.get("HTTP_X_USER_ENGINE_PROXY_SECRET", ""))
        if not secrets.compare_digest(marker, self.trusted_proxy_secret):
            raise AuthorizationDenied("untrusted identity source")
        raw = environ.get("HTTP_X_VERIFIED_OIDC_CLAIMS")
        if not raw:
            raise AuthorizationDenied("verified claims required")
        claims = json.loads(str(raw))
        if not isinstance(claims, dict):
            raise AuthorizationDenied("verified claims must be an object")
        return claims

    def _actor(self, environ: Mapping[str, Any]) -> Any:
        return self.service.identity_adapter.normalize(self._claims(environ))

    def _optional_actor(self, environ: Mapping[str, Any]) -> Any | None:
        try:
            return self._actor(environ)
        except (AuthorizationDenied, json.JSONDecodeError, ValidationError):
            return None

    @staticmethod
    def _body(environ: Mapping[str, Any]) -> Mapping[str, Any]:
        length = min(int(environ.get("CONTENT_LENGTH") or 0), 65536)
        payload = environ["wsgi.input"].read(length) if length else b"{}"
        value = json.loads(payload.decode("utf-8"))
        if not isinstance(value, dict):
            raise ValidationError("request body must be an object")
        return value

    @staticmethod
    def _page(environ: Mapping[str, Any]) -> tuple[int, int]:
        query = parse_qs(str(environ.get("QUERY_STRING", "")))
        offset = max(0, int(query.get("offset", ["0"])[0]))
        limit = max(1, min(100, int(query.get("limit", ["25"])[0])))
        return offset, limit

    def _home(self, actor: Any | None) -> str:
        identity = (
            f"<p>Signed in as <strong>{escape(actor.preferred_username)}</strong>.</p>"
            if actor is not None
            else f'<p><a class="button" href="/login">Sign in with KeyCape</a></p>'
        )
        return self._page_html(
            "Identity & access",
            "<h1>Your account, on your terms.</h1>"
            "<p>Join a tenant, complete onboarding, and manage access without exposing credentials to applications.</p>"
            + identity,
        )

    def _admin(self, tenant: str, memberships: tuple[Any, ...]) -> str:
        rows = "".join(
            f"<tr><td>{escape(item.user_id)}</td><td>{escape(item.kind)}</td><td>{escape(item.scope_id)}</td></tr>"
            for item in memberships
        ) or '<tr><td colspan="3">No members yet.</td></tr>'
        return self._page_html(
            f"{tenant} users",
            f"<h1>{escape(tenant)} users</h1><table><thead><tr><th>User</th><th>Role</th><th>Scope</th></tr></thead><tbody>{rows}</tbody></table>",
        )

    @staticmethod
    def _page_html(title: str, body: str) -> str:
        return f"""<!doctype html><html lang="en"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>{escape(title)} · Railiance</title><style>
:root{{--ink:#17201c;--paper:#f5f1e8;--accent:#195b47;--line:#c8c1b3}}
*{{box-sizing:border-box}}body{{margin:0;background:var(--paper);color:var(--ink);font:18px/1.55 system-ui,sans-serif}}
header,main{{max-width:68rem;margin:auto;padding:1.25rem}}header{{border-bottom:1px solid var(--line)}}
h1{{font:clamp(2.2rem,7vw,5.5rem)/.98 Georgia,serif;max-width:13ch}}a{{color:var(--accent)}}
.button{{display:inline-block;background:var(--accent);color:white;padding:.8rem 1.15rem;border-radius:.3rem;text-decoration:none}}
table{{width:100%;border-collapse:collapse;background:#fff}}th,td{{padding:.75rem;text-align:left;border-bottom:1px solid var(--line)}}
a:focus-visible{{outline:3px solid #e59f24;outline-offset:3px}}@media(max-width:640px){{body{{font-size:16px}}}}
</style></head><body><header><strong>Railiance identity</strong></header><main>{body}</main></body></html>"""

    def _html(self, start_response: StartResponse, body: str, correlation_id: str) -> list[bytes]:
        data = body.encode()
        start_response("200 OK", [("Content-Type", "text/html; charset=utf-8"), ("Content-Length", str(len(data))), *self._security_headers(correlation_id)])
        return [data]

    def _json(self, start_response: StartResponse, status: str, payload: Any, correlation_id: str) -> list[bytes]:
        data = json.dumps(payload, separators=(",", ":"), default=str).encode()
        start_response(status, [("Content-Type", "application/json"), ("Content-Length", str(len(data))), *self._security_headers(correlation_id)])
        return [data]

    def _error(self, start_response: StartResponse, status: str, code: str, message: str, correlation_id: str) -> list[bytes]:
        return self._json(start_response, status, {"error": {"code": code, "message": message, "correlation_id": correlation_id}}, correlation_id)

    @staticmethod
    def _security_headers(correlation_id: str) -> list[tuple[str, str]]:
        return [
            ("X-Request-ID", correlation_id),
            ("Cache-Control", "no-store"),
            ("X-Content-Type-Options", "nosniff"),
            ("Referrer-Policy", "no-referrer"),
            ("Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'; form-action 'self'; frame-ancestors 'none'; base-uri 'none'"),
        ]