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, replace
from enum import Enum
from html import escape
import hashlib
import hmac
import json
import re
import secrets
from typing import Any, Callable, Iterable, Mapping
from urllib.parse import parse_qs, urlencode, urlsplit

from user_engine.domain import (
    AccountStatus,
    Actor,
    FactorVerification,
    FamilyMemberSpec,
    IdentityFactorType,
    PrincipalType,
)
from user_engine.errors import AuthorizationDenied, ConflictError, NotFoundError, ValidationError
from user_engine.oidc import OIDCClient, cookie_value
from user_engine.ports import (
    IdentityProvisioningPort,
    ProvisioningRequest,
    RegistrationVerificationPort,
    RegistrationVerificationRequest,
    TenantManagementPort,
)
from user_engine.service import PLATFORM_TENANT, 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,
        provisioning: IdentityProvisioningPort | None = None,
        tenant_management: TenantManagementPort | None = None,
        outbox_delivery: Callable[[Any], None] | None = None,
        registration_verification: RegistrationVerificationPort | None = None,
        registration_clients: tuple[str, ...] = (),
        registration_tenants: tuple[str, ...] = (),
        registration_oidc_issuer: str = "",
        registration_password_setup_origins: tuple[str, ...] = (),
    ) -> 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
        self.provisioning = provisioning
        self.tenant_management = tenant_management
        self.outbox_delivery = outbox_delivery
        self.registration_verification = registration_verification
        self.registration_clients = frozenset(registration_clients)
        self.registration_tenants = frozenset(registration_tenants)
        self.registration_oidc_issuer = registration_oidc_issuer.rstrip("/")
        self.registration_password_setup_origins = frozenset(
            origin.rstrip("/") for origin in registration_password_setup_origins
        )

    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 ConflictError as exc:
            return self._error(start_response, "409 Conflict", "conflict", str(exc), correlation_id)
        except (ValidationError, ValueError) as exc:
            return self._error(start_response, "400 Bad Request", "invalid_request", str(exc), correlation_id)
        except RuntimeError:
            return self._error(
                start_response,
                "502 Bad Gateway",
                "provisioning_unavailable",
                "Identity provisioning is temporarily unavailable.",
                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 == "/metrics":
            supplied = str(environ.get("HTTP_X_USER_ENGINE_PROXY_SECRET", ""))
            if not secrets.compare_digest(supplied, self.trusted_proxy_secret):
                raise AuthorizationDenied("metrics require the trusted workload marker")
            return self._metrics(start_response, correlation_id)
        if path in {"/login", "/oidc/start"}:
            query = parse_qs(str(environ.get("QUERY_STRING", "")))
            tenant_hint = query.get("tenant_hint", [None])[0]
            if tenant_hint is not None and not str(tenant_hint).startswith("tenant:"):
                raise ValidationError("tenant_hint must be a tenant identifier")
            location = (
                self.oidc_client.begin(tenant_hint=str(tenant_hint) if tenant_hint else None)
                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)

        if path == "/register" and method == "GET":
            if not self.public_registration or self.registration_verification is None:
                raise NotFoundError("public registration is unavailable")
            token = secrets.token_urlsafe(32)
            return self._html(
                start_response, self._registration_form(token), correlation_id,
                extra_headers=[("Set-Cookie", self._registration_csrf_cookie(token))],
            )
        if path == "/register" and method == "POST":
            body = self._form_body(environ)
            self._require_registration_csrf(environ, str(body.get("csrf_token", "")))
            return self._start_public_registration(
                environ, start_response, correlation_id, body=body, browser=True
            )
        if path == "/registration/verify" and method == "GET":
            query = parse_qs(str(environ.get("QUERY_STRING", "")))
            handle = str(query.get("handle", [""])[0])
            if len(handle) < 16:
                raise ValidationError("verification handle is invalid")
            token = secrets.token_urlsafe(32)
            return self._html(
                start_response, self._registration_verification_form(token, handle),
                correlation_id,
                extra_headers=[("Set-Cookie", self._registration_csrf_cookie(token))],
            )
        if path == "/registration/verify" and method == "POST":
            body = self._form_body(environ)
            self._require_registration_csrf(environ, str(body.get("csrf_token", "")))
            return self._verify_public_registration(
                environ, start_response, correlation_id, body=body, browser=True
            )
        if path == "/registration/resume" and method == "POST":
            body = self._form_body(environ)
            self._require_registration_csrf(environ, str(body.get("csrf_token", "")))
            return self._resume_public_registration(
                environ, start_response, correlation_id, body=body, browser=True
            )

        if path == "/api/v1/public/registrations" and method == "POST":
            return self._start_public_registration(
                environ, start_response, correlation_id
            )
        if path == "/api/v1/public/registrations/verify" and method == "POST":
            return self._verify_public_registration(
                environ, start_response, correlation_id
            )
        if path == "/api/v1/public/registrations/resume" and method == "POST":
            return self._resume_public_registration(
                environ, start_response, 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/me/profile" and method == "PATCH":
            self._idempotency_key(environ)
            self.service.me(self._claims(environ), correlation_id=correlation_id)
            body = self._body(environ)
            updated = self.service.update_self_service_profile(
                actor, display_name=str(body.get("display_name", "")),
                consent_accepted=bool(body.get("consent_accepted", False)),
                consent_version=str(body.get("consent_version") or "portal-terms-v1"),
                correlation_id=correlation_id,
            )
            return self._json(start_response, "200 OK", _jsonable(updated), correlation_id)
        if path == "/onboarding" and method == "GET":
            session = self.service.me(self._claims(environ), correlation_id=correlation_id)
            memberships = self.service.store.memberships_for_user(session.user.user_id)
            journeys = self.service.store.onboarding_journeys_for_user(session.user.user_id)
            query = parse_qs(str(environ.get("QUERY_STRING", "")))
            selected_tenant = query.get("tenant", [session.actor.tenant])[0]
            allowed_tenants = {item.tenant for item in memberships} | {session.actor.tenant}
            if selected_tenant not in allowed_tenants:
                raise AuthorizationDenied("tenant selection is not a membership")
            return self._html(
                start_response,
                self._onboarding(
                    session, memberships, journeys, str(selected_tenant),
                    self._csrf_token(environ),
                ),
                correlation_id,
            )
        if path == "/onboarding/profile" and method == "POST":
            body = self._form_body(environ)
            self._require_csrf(environ, str(body.get("csrf_token", "")))
            self.service.me(self._claims(environ), correlation_id=correlation_id)
            self.service.update_self_service_profile(
                actor, display_name=str(body.get("display_name", "")),
                consent_accepted=body.get("consent_accepted") == "yes",
                consent_version="portal-terms-v1",
                correlation_id=correlation_id,
            )
            return self._redirect(start_response, "/onboarding", correlation_id)
        if path.startswith("/onboarding/") and "/steps/" in path and path.endswith("/complete") and method == "POST":
            body = self._form_body(environ)
            self._require_csrf(environ, str(body.get("csrf_token", "")))
            parts = path.split("/")
            journey_id, step_key = parts[2], parts[4]
            session = self.service.me(self._claims(environ), correlation_id=correlation_id)
            journey = self.service.store.onboarding_journey(journey_id)
            if journey is None or journey.user_id != session.user.user_id:
                raise NotFoundError("onboarding journey not found")
            step = next((item for item in journey.steps if item.step_key == step_key), None)
            if step is None:
                raise NotFoundError("onboarding step not found")
            if step.subsystem != "user-engine" or step.handoff is not None:
                raise AuthorizationDenied("subsystem-owned steps require their handoff")
            self.service.complete_onboarding_step(
                actor, journey_id, step_key, correlation_id=correlation_id
            )
            return self._redirect(start_response, "/onboarding", correlation_id)
        if path.startswith("/invitations/") and method == "GET":
            invitation = self.service.store.family_invitation(path.split("/")[2])
            if invitation is None:
                raise NotFoundError("invitation not found")
            self.service.resolve_tenant_context(actor, invitation.tenant)
            return self._html(
                start_response,
                self._invitation_acceptance(invitation, self._csrf_token(environ)),
                correlation_id,
            )
        if path.startswith("/invitations/") and method == "POST":
            body = self._form_body(environ)
            self._require_csrf(environ, str(body.get("csrf_token", "")))
            self.service.accept_family_invitation(
                self._claims(environ), path.split("/")[2], correlation_id=correlation_id
            )
            return self._redirect(start_response, "/onboarding", correlation_id)
        if path == "/api/v1/platform/tenants" and method == "POST":
            self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
            if self.tenant_management is None:
                raise ValidationError("tenant management is unavailable")
            idempotency_key = self._idempotency_key(environ)
            body = self._body(environ)
            tenant = str(body.get("tenant") or "")
            if not tenant.startswith("tenant:") or tenant == PLATFORM_TENANT:
                raise ValidationError("a non-platform tenant identifier is required")
            result = self.tenant_management.create_tenant(
                tenant=tenant,
                display_name=str(body.get("display_name") or tenant),
                idempotency_key=idempotency_key,
                correlation_id=correlation_id,
            )
            admin = body.get("first_admin")
            bootstrap = None
            if admin is not None:
                if not isinstance(admin, Mapping):
                    raise ValidationError("first_admin must be an object")
                user = self.service.create_user(
                    actor, display_name=admin.get("display_name"),
                    primary_email=admin.get("primary_email"),
                    correlation_id=correlation_id,
                )
                account = self.service.set_tenant_account_status(
                    actor, user.user_id, AccountStatus.INVITED,
                    tenant=tenant, correlation_id=correlation_id,
                )
                membership = self.service.add_membership(
                    actor, user.user_id, tenant=tenant, scope_type="tenant",
                    scope_id=tenant, kind="tenant-admin",
                    correlation_id=correlation_id,
                )
                bootstrap = {"user": user, "tenant_account": account, "membership": membership}
            return self._json(start_response, "201 Created", {
                "tenant": _jsonable(result), "first_admin": _jsonable(bootstrap),
            }, correlation_id)
        if path == "/api/v1/platform/outbox/deliver" and method == "POST":
            self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
            if self.outbox_delivery is None:
                raise ValidationError("outbox delivery is unavailable")
            body = self._body(environ)
            events = self.service.deliver_outbox(
                actor, self.outbox_delivery,
                worker_id=str(body.get("worker_id") or "portal-operator"),
                max_attempts=int(body.get("max_attempts") or 3),
                correlation_id=correlation_id,
            )
            return self._json(start_response, "200 OK", {"items": _jsonable(events)}, correlation_id)
        if path.startswith("/api/v1/platform/outbox/") and path.endswith("/replay") and method == "POST":
            self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
            event = self.service.replay_outbox(
                actor, path.split("/")[5], correlation_id=correlation_id
            )
            return self._json(start_response, "200 OK", _jsonable(event), correlation_id)
        if path.startswith("/api/v1/platform/tenants/") and path.endswith("/recover") and method == "POST":
            self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
            if self.provisioning is None:
                raise ValidationError("identity provisioning is unavailable")
            parts = path.split("/")
            tenant, user_id = parts[5], parts[7]
            idempotency_key = self._idempotency_key(environ)
            user = self.service.store.user(user_id)
            if user is None:
                raise NotFoundError("user not found")
            request = ProvisioningRequest(
                user_id=user_id, tenant=tenant, primary_email=user.primary_email,
                display_name=user.display_name, idempotency_key=idempotency_key,
                correlation_id=correlation_id,
                roles=tuple(item.kind for item in self.service.store.memberships_for_user(user_id, tenant=tenant)),
            )
            identity = next(iter(self.service.store.identities_for_user(user_id)), None)
            if identity is None:
                provisioned = self.provisioning.provision(request)
                self.service.link_identity(
                    actor, user_id, issuer="urn:netkingdom:directory",
                    subject=provisioned.external_subject, provider=provisioned.provider,
                    correlation_id=correlation_id,
                )
                recovery = {"status": provisioned.status, "changed": ("identity",)}
            else:
                reconciled = self.provisioning.reconcile(
                    request, external_subject=identity.subject, desired_status="active"
                )
                recovery = {"status": reconciled.status, "drift": reconciled.drift, "changed": reconciled.changed}
            account = self.service.set_tenant_account_status(
                actor, user_id, AccountStatus.ACTIVE, tenant=tenant,
                correlation_id=correlation_id,
            )
            return self._json(start_response, "200 OK", {
                "recovery": _jsonable(recovery), "tenant_account": _jsonable(account),
            }, correlation_id)
        if path.startswith("/api/v1/invitations/") and path.endswith("/claim") and method == "POST":
            invitation_id = path.split("/")[4]
            accepted = self.service.accept_family_invitation(
                self._claims(environ), invitation_id, correlation_id=correlation_id
            )
            return self._json(start_response, "200 OK", _jsonable(accepted), correlation_id)
        if path.startswith("/api/v1/onboarding/") and "/steps/" in path and path.endswith("/complete") and method == "POST":
            self._idempotency_key(environ)
            parts = path.split("/")
            journey_id, step_key = parts[4], parts[6]
            session = self.service.me(self._claims(environ), correlation_id=correlation_id)
            journey = self.service.store.onboarding_journey(journey_id)
            if journey is None or journey.user_id != session.user.user_id:
                raise NotFoundError("onboarding journey not found")
            step = next((item for item in journey.steps if item.step_key == step_key), None)
            if step is None:
                raise NotFoundError("onboarding step not found")
            if step.subsystem != "user-engine" or step.handoff is not None:
                raise AuthorizationDenied("subsystem-owned steps require their handoff")
            updated = self.service.complete_onboarding_step(
                actor, journey_id, step_key, correlation_id=correlation_id
            )
            return self._json(start_response, "200 OK", _jsonable(updated), 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 path.endswith("/users") and method == "POST":
            tenant = path.split("/")[4]
            self.service.resolve_tenant_context(actor, tenant)
            body = self._body(environ)
            user = self.service.create_user(
                actor,
                display_name=body.get("display_name"),
                primary_email=body.get("primary_email"),
                correlation_id=correlation_id,
            )
            # Platform operators may create an identity for a tenant other than
            # their own.  Ensure the lifecycle record follows the requested
            # tenant instead of only retaining the actor tenant created by the
            # generic domain operation.
            tenant_account = self.service.set_tenant_account_status(
                actor,
                user.user_id,
                AccountStatus.ACTIVE,
                tenant=tenant,
                correlation_id=correlation_id,
            )
            membership = self.service.add_membership(
                actor,
                user.user_id,
                tenant=tenant,
                scope_type="tenant",
                scope_id=tenant,
                kind=str(body.get("role", "user")),
                correlation_id=correlation_id,
            )
            return self._json(start_response, "201 Created", {
                "user": _jsonable(user),
                "tenant_account": _jsonable(tenant_account),
                "membership": _jsonable(membership),
                "provisioning_status": "pending",
            }, correlation_id)
        if path.startswith("/api/v1/tenants/") and path.endswith("/invitations"):
            tenant = path.split("/")[4]
            self.service.resolve_tenant_context(actor, tenant)
            if method == "GET":
                items = self.service.store.family_invitations_for_tenant(tenant)
                return self._json(start_response, "200 OK", {"items": _jsonable(items)}, correlation_id)
            if method == "POST":
                body = self._body(environ)
                invited = self.service.invite_family_member(
                    actor,
                    tenant=tenant,
                    family_scope_id=str(body.get("scope_id") or tenant),
                    application_id=str(body.get("application_id") or "app.user-portal"),
                    member=FamilyMemberSpec(
                        primary_email=str(body.get("primary_email") or ""),
                        display_name=body.get("display_name"),
                        role=str(body.get("role") or "user"),
                    ),
                    correlation_id=correlation_id,
                )
                return self._json(start_response, "201 Created", _jsonable(invited), correlation_id)
        if path.startswith("/api/v1/tenants/") and "/invitations/" in path and method == "POST":
            parts = path.split("/")
            tenant, invitation_id, action = parts[4], parts[6], parts[7]
            self.service.resolve_tenant_context(actor, tenant)
            expected = self._expected_version(environ)
            if action == "resend":
                value = self.service.resend_family_invitation(
                    actor, invitation_id, correlation_id=correlation_id,
                    expected_version=expected,
                )
            elif action == "expire":
                value = self.service.revoke_family_invitation(
                    actor, invitation_id, correlation_id=correlation_id,
                    expected_version=expected,
                )
            else:
                raise NotFoundError("invitation action not found")
            return self._json(start_response, "200 OK", _jsonable(value), correlation_id)
        if path.startswith("/api/v1/tenants/") and path.endswith("/provision") and method == "POST":
            if self.provisioning is None:
                raise ValidationError("identity provisioning is unavailable")
            parts = path.split("/")
            tenant, user_id = parts[4], parts[6]
            self.service.resolve_tenant_context(actor, tenant)
            user = self.service.store.user(user_id)
            if user is None:
                raise NotFoundError("user not found")
            idempotency_key = str(environ.get("HTTP_IDEMPOTENCY_KEY", ""))
            if len(idempotency_key) < 16:
                raise ValidationError("Idempotency-Key must contain at least 16 characters")
            result = self.provisioning.provision(ProvisioningRequest(
                user_id=user.user_id,
                tenant=tenant,
                primary_email=user.primary_email,
                display_name=user.display_name,
                idempotency_key=idempotency_key,
                correlation_id=correlation_id,
                roles=tuple(
                    membership.kind
                    for membership in self.service.store.memberships_for_user(
                        user.user_id, tenant=tenant
                    )
                ),
            ))
            identity = self.service.link_identity(
                actor,
                user.user_id,
                issuer="urn:netkingdom:directory",
                subject=result.external_subject,
                provider=result.provider,
                correlation_id=correlation_id,
            )
            return self._json(start_response, "200 OK", {
                "provisioning": _jsonable(result),
                "identity": _jsonable(identity),
            }, 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"]))
            idempotency_key = str(environ.get("HTTP_IDEMPOTENCY_KEY", ""))
            if len(idempotency_key) < 16:
                raise ValidationError("Idempotency-Key must contain at least 16 characters")
            result = self._change_status(
                actor, tenant, user_id, status,
                idempotency_key=idempotency_key,
                correlation_id=correlation_id,
            )
            return self._json(start_response, "200 OK", _jsonable(result), correlation_id)
        if path.startswith("/api/v1/tenants/") and "/users/" in path and method == "DELETE":
            if self.provisioning is None:
                raise ValidationError("identity provisioning is unavailable")
            parts = path.split("/")
            tenant, user_id = parts[4], parts[6]
            self.service.resolve_tenant_context(actor, tenant)
            idempotency_key = str(environ.get("HTTP_IDEMPOTENCY_KEY", ""))
            if len(idempotency_key) < 16:
                raise ValidationError("Idempotency-Key must contain at least 16 characters")
            identity = next(iter(self.service.store.identities_for_user(user_id)), None)
            if identity is not None:
                self.provisioning.deprovision(
                    external_subject=identity.subject,
                    idempotency_key=idempotency_key,
                    correlation_id=correlation_id,
                )
            account = self.service.set_tenant_account_status(
                actor, user_id, AccountStatus.DISABLED,
                tenant=tenant, correlation_id=correlation_id,
            )
            return self._json(start_response, "200 OK", {
                "status": "removed", "tenant_account": _jsonable(account),
                "provider_identity_removed": identity is not None,
            }, correlation_id)
        if path == "/platform" and method == "GET":
            self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
            return self._html(
                start_response, self._platform(self._csrf_token(environ)), correlation_id
            )
        if path == "/platform/tenants" and method == "POST":
            self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
            if self.tenant_management is None:
                raise ValidationError("tenant management is unavailable")
            body = self._form_body(environ)
            self._require_csrf(environ, str(body.get("csrf_token", "")))
            tenant = str(body.get("tenant", ""))
            if not tenant.startswith("tenant:") or tenant == PLATFORM_TENANT:
                raise ValidationError("a non-platform tenant identifier is required")
            result = self.tenant_management.create_tenant(
                tenant=tenant, display_name=str(body.get("display_name") or tenant),
                idempotency_key=f"portal-tenant-{tenant}", correlation_id=correlation_id,
            )
            email = str(body.get("admin_email", ""))
            if email:
                user = self.service.create_user(
                    actor, display_name=body.get("admin_display_name"),
                    primary_email=email, correlation_id=correlation_id,
                )
                self.service.set_tenant_account_status(
                    actor, user.user_id, AccountStatus.INVITED,
                    tenant=tenant, correlation_id=correlation_id,
                )
                self.service.add_membership(
                    actor, user.user_id, tenant=tenant, scope_type="tenant",
                    scope_id=tenant, kind="tenant-admin", correlation_id=correlation_id,
                )
            return self._html(
                start_response,
                self._platform_result(result, tenant, bool(email)), 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)
            invitations = self.service.store.family_invitations_for_tenant(tenant)
            diagnostics = self.service.tenant_diagnostics(
                actor, tenant=tenant, correlation_id=correlation_id
            )
            return self._html(
                start_response,
                self._admin(
                    tenant, memberships, invitations, diagnostics,
                    "platform-operator" in actor.roles, self._csrf_token(environ),
                ),
                correlation_id,
            )
        if path.startswith("/admin/") and method == "POST":
            parts = path.split("/")
            tenant = parts[2]
            self.service.resolve_tenant_context(actor, tenant)
            body = self._form_body(environ)
            self._require_csrf(environ, str(body.get("csrf_token", "")))
            if len(parts) == 4 and parts[3] == "users":
                user = self.service.create_user(
                    actor,
                    display_name=body.get("display_name"),
                    primary_email=body.get("primary_email"),
                    correlation_id=correlation_id,
                )
                self.service.set_tenant_account_status(
                    actor, user.user_id, AccountStatus.ACTIVE,
                    tenant=tenant, correlation_id=correlation_id,
                )
                self.service.add_membership(
                    actor, user.user_id, tenant=tenant, scope_type="tenant",
                    scope_id=tenant, kind=str(body.get("role", "user")),
                    correlation_id=correlation_id,
                )
                return self._redirect(start_response, f"/admin/{tenant}", correlation_id)
            if len(parts) == 4 and parts[3] == "invitations":
                self.service.invite_family_member(
                    actor, tenant=tenant, family_scope_id=tenant,
                    application_id="app.user-portal",
                    member=FamilyMemberSpec(
                        primary_email=str(body.get("primary_email", "")),
                        display_name=body.get("display_name"),
                        role=str(body.get("role", "user")),
                    ), correlation_id=correlation_id,
                )
                return self._redirect(start_response, f"/admin/{tenant}", correlation_id)
            if len(parts) == 6 and parts[3] == "invitations":
                invitation_id, action = parts[4], parts[5]
                version = int(body.get("version", "0"))
                if action == "resend":
                    self.service.resend_family_invitation(
                        actor, invitation_id, expected_version=version,
                        correlation_id=correlation_id,
                    )
                elif action == "expire":
                    self.service.revoke_family_invitation(
                        actor, invitation_id, expected_version=version,
                        correlation_id=correlation_id,
                    )
                else:
                    raise NotFoundError("invitation action not found")
                return self._redirect(start_response, f"/admin/{tenant}", correlation_id)
            if len(parts) == 6 and parts[3] == "users" and parts[5] == "provision":
                if self.provisioning is None:
                    raise ValidationError("identity provisioning is unavailable")
                user_id = parts[4]
                user = self.service.store.user(user_id)
                if user is None:
                    raise NotFoundError("user not found")
                result = self.provisioning.provision(ProvisioningRequest(
                    user_id=user.user_id,
                    tenant=tenant,
                    primary_email=user.primary_email,
                    display_name=user.display_name,
                    idempotency_key=f"portal-{user.user_id}-{tenant}",
                    correlation_id=correlation_id,
                    roles=tuple(
                        item.kind for item in self.service.store.memberships_for_user(
                            user.user_id, tenant=tenant
                        )
                    ),
                ))
                self.service.link_identity(
                    actor, user.user_id, issuer="urn:netkingdom:directory",
                    subject=result.external_subject, provider=result.provider,
                    correlation_id=correlation_id,
                )
                if result.password_setup_url:
                    return self._html(
                        start_response,
                        self._password_setup_handoff(
                            result.password_setup_url, tenant
                        ),
                        correlation_id,
                    )
                query = urlencode({"provisioned": user.user_id, "status": result.status})
                return self._redirect(start_response, f"/admin/{tenant}?{query}", correlation_id)
            if len(parts) == 6 and parts[3] == "users" and parts[5] == "status":
                status = AccountStatus(str(body.get("status", "")))
                if status not in {AccountStatus.ACTIVE, AccountStatus.SUSPENDED}:
                    raise ValidationError("browser lifecycle supports active or suspended")
                self._change_status(
                    actor, tenant, parts[4], status,
                    idempotency_key=f"portal-status-{parts[4]}-{status.value}",
                    correlation_id=correlation_id,
                )
                return self._redirect(start_response, f"/admin/{tenant}", correlation_id)
            if len(parts) == 6 and parts[3] == "users" and parts[5] == "remove":
                if self.provisioning is None:
                    raise ValidationError("identity provisioning is unavailable")
                user_id = parts[4]
                identity = next(iter(self.service.store.identities_for_user(user_id)), None)
                if identity is not None:
                    self.provisioning.deprovision(
                        external_subject=identity.subject,
                        idempotency_key=f"portal-remove-{tenant}-{user_id}",
                        correlation_id=correlation_id,
                    )
                self.service.set_tenant_account_status(
                    actor, user_id, AccountStatus.DISABLED,
                    tenant=tenant, correlation_id=correlation_id,
                )
                return self._redirect(start_response, f"/admin/{tenant}", correlation_id)
            if len(parts) == 6 and parts[3] == "users" and parts[5] == "recover":
                self.service.resolve_tenant_context(actor, PLATFORM_TENANT)
                if self.provisioning is None:
                    raise ValidationError("identity provisioning is unavailable")
                user_id = parts[4]
                user = self.service.store.user(user_id)
                if user is None:
                    raise NotFoundError("user not found")
                request = ProvisioningRequest(
                    user_id=user_id, tenant=tenant, primary_email=user.primary_email,
                    display_name=user.display_name,
                    idempotency_key=f"portal-recover-{tenant}-{user_id}",
                    correlation_id=correlation_id,
                    roles=tuple(item.kind for item in self.service.store.memberships_for_user(user_id, tenant=tenant)),
                )
                identity = next(iter(self.service.store.identities_for_user(user_id)), None)
                if identity is None:
                    result = self.provisioning.provision(request)
                    self.service.link_identity(
                        actor, user_id, issuer="urn:netkingdom:directory",
                        subject=result.external_subject, provider=result.provider,
                        correlation_id=correlation_id,
                    )
                else:
                    self.provisioning.reconcile(
                        request, external_subject=identity.subject, desired_status="active"
                    )
                self.service.set_tenant_account_status(
                    actor, user_id, AccountStatus.ACTIVE,
                    tenant=tenant, correlation_id=correlation_id,
                )
                return self._redirect(start_response, f"/admin/{tenant}?recovered={user_id}", correlation_id)
        return self._error(start_response, "404 Not Found", "not_found", "Resource not found.", correlation_id)

    def _change_status(
        self,
        actor: Any,
        tenant: str,
        user_id: str,
        status: AccountStatus,
        *,
        idempotency_key: str,
        correlation_id: str,
    ) -> Any:
        if self.provisioning is None:
            raise ValidationError("identity provisioning is unavailable")
        self.service.resolve_tenant_context(actor, tenant)
        identity = next(
            (
                item for item in self.service.store.identities_for_user(user_id)
                if item.provider == "netkingdom-lldap"
            ),
            None,
        )
        if identity is None:
            raise ValidationError("user has no managed login identity")
        if status == AccountStatus.SUSPENDED:
            self.provisioning.suspend(
                external_subject=identity.subject,
                idempotency_key=idempotency_key,
                correlation_id=correlation_id,
            )
        elif status == AccountStatus.ACTIVE:
            self.provisioning.reactivate(
                external_subject=identity.subject,
                idempotency_key=idempotency_key,
                correlation_id=correlation_id,
            )
        else:
            raise ValidationError("provider lifecycle supports active or suspended")
        return self.service.set_tenant_account_status(
            actor, user_id, status, tenant=tenant, correlation_id=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 _start_public_registration(
        self,
        environ: Mapping[str, Any],
        start_response: StartResponse,
        correlation_id: str,
        *, body: Mapping[str, Any] | None = None, browser: bool = False,
    ) -> Iterable[bytes]:
        if not self.public_registration or self.registration_verification is None:
            raise NotFoundError("public registration is unavailable")
        body = body if body is not None else self._body(environ)
        username = self._registration_username(body.get("username"))
        email = self._registration_email(body.get("email"))
        display_name = str(body.get("display_name") or "").strip() or None
        if display_name is not None and len(display_name) > 200:
            raise ValidationError("display_name is too long")
        client_id = str(body.get("client_id") or "")
        tenant = str(body.get("tenant") or "")
        if client_id not in self.registration_clients:
            raise ValidationError("client_id is not eligible for registration")
        if tenant not in self.registration_tenants:
            raise ValidationError("tenant is not eligible for registration")

        applicant_subject = f"applicant_{secrets.token_hex(16)}"
        actor = Actor(
            issuer="urn:netkingdom:public-registration",
            subject=applicant_subject,
            tenant=tenant,
            principal_type=PrincipalType.HUMAN,
            audience=("user-engine",),
            roles=("registration-applicant",),
            authorized_party=client_id,
            preferred_username=username,
        )
        session = self.service.start_registration(
            actor,
            tenant=tenant,
            correlation_id=correlation_id,
            applicant_username=username,
            client_id=client_id,
        )
        self.registration_verification.request(
            RegistrationVerificationRequest(
                registration_id=session.registration_id,
                normalized_email=email,
                preferred_username=username,
                client_id=client_id,
                tenant=tenant,
                correlation_id=correlation_id,
                display_name=display_name,
            )
        )
        if browser:
            return self._html(
                start_response,
                self._page_html(
                    "Check your email",
                    "<h1>Check your email.</h1>"
                    "<p>If the address can be registered, a verification link is on its way. "
                    "The link expires after 30 minutes.</p>",
                ),
                correlation_id,
            )
        return self._json(
            start_response,
            "202 Accepted",
            {"status": "verification_requested"},
            correlation_id,
        )

    def _verify_public_registration(
        self,
        environ: Mapping[str, Any],
        start_response: StartResponse,
        correlation_id: str,
        *, body: Mapping[str, Any] | None = None, browser: bool = False,
    ) -> Iterable[bytes]:
        if not self.public_registration or self.registration_verification is None:
            raise NotFoundError("public registration is unavailable")
        body = body if body is not None else self._body(environ)
        evidence = self.registration_verification.consume(str(body.get("handle") or ""))
        session = self.service.store.registration_session(evidence.registration_id)
        if session is None:
            raise NotFoundError("registration not found")
        if (
            evidence.client_id != session.client_id
            or evidence.tenant != session.tenant
            or evidence.preferred_username != session.applicant_username
        ):
            raise AuthorizationDenied("verification evidence does not match registration")
        actor = Actor(
            issuer="urn:netkingdom:public-registration",
            subject=str(session.started_by_subject),
            tenant=session.tenant,
            principal_type=PrincipalType.HUMAN,
            audience=("user-engine",),
            roles=("registration-applicant",),
            authorized_party=session.client_id,
            preferred_username=session.applicant_username,
        )
        updated = self.service.attach_registration_factor(
            actor,
            session.registration_id,
            FactorVerification(
                factor_type=IdentityFactorType.EMAIL,
                normalized_value=evidence.normalized_email,
                verification_id=evidence.verification_id,
                source_system=evidence.source_system,
                assurance=dict(evidence.assurance),
            ),
            correlation_id=correlation_id,
        )
        resume_handle = secrets.token_urlsafe(32)
        updated = replace(
            updated,
            provisioning_resume_hash=hashlib.sha256(resume_handle.encode()).hexdigest(),
        )
        self.service.store.save_registration_session(updated)
        if self.provisioning is None or not self.registration_oidc_issuer:
            raise RuntimeError("public registration provisioning is unavailable")
        completion = self.service.complete_registration(
            actor,
            updated.registration_id,
            display_name=evidence.display_name,
            primary_email=evidence.normalized_email,
            correlation_id=correlation_id,
        )
        try:
            return self._provision_public_registration(
                start_response, actor, completion.user, completion.session,
                evidence.normalized_email, evidence.display_name,
                evidence.preferred_username, correlation_id, browser=browser,
            )
        except RuntimeError:
            if browser:
                token = secrets.token_urlsafe(32)
                return self._html(
                    start_response,
                    self._registration_resume_form(
                        token, updated.registration_id, resume_handle
                    ),
                    correlation_id,
                    extra_headers=[("Set-Cookie", self._registration_csrf_cookie(token))],
                )
            return self._json(start_response, "202 Accepted", {
                "status": "provisioning_pending",
                "registration_id": updated.registration_id,
                "resume_handle": resume_handle,
            }, correlation_id)

    def _resume_public_registration(
        self, environ, start_response, correlation_id, *,
        body: Mapping[str, Any] | None = None, browser: bool = False,
    ):
        if not self.public_registration:
            raise NotFoundError("public registration is unavailable")
        body = body if body is not None else self._body(environ)
        session = self.service.store.registration_session(str(body.get("registration_id") or ""))
        handle = str(body.get("resume_handle") or "")
        digest = hashlib.sha256(handle.encode()).hexdigest()
        if (
            session is None or session.status.value != "completed"
            or not session.provisioning_resume_hash
            or not hmac.compare_digest(digest, session.provisioning_resume_hash)
        ):
            raise AuthorizationDenied("registration resume is invalid")
        factors = self.service.store.factors_for_registration(session.registration_id)
        email_factor = next((item for item in factors if item.factor_type == IdentityFactorType.EMAIL), None)
        if email_factor is None or not session.user_id:
            raise ValidationError("registration is not recoverable")
        actor = Actor(
            issuer="urn:netkingdom:public-registration", subject=str(session.started_by_subject),
            tenant=session.tenant, principal_type=PrincipalType.HUMAN,
            audience=("user-engine",), roles=("registration-applicant",),
            authorized_party=session.client_id, preferred_username=session.applicant_username,
        )
        user = self.service.store.user(session.user_id)
        if user is None:
            raise ValidationError("registration user is unavailable")
        return self._provision_public_registration(
            start_response, actor, user, session, email_factor.normalized_value,
            user.display_name, str(session.applicant_username), correlation_id,
            browser=browser,
        )

    def _provision_public_registration(
        self, start_response, actor, user, session, email, display_name,
        preferred_username, correlation_id,
        *, browser: bool = False,
    ):
        provisioned = self.provisioning.provision(
            ProvisioningRequest(
                user_id=user.user_id,
                tenant=session.tenant,
                primary_email=email,
                display_name=display_name,
                preferred_username=preferred_username,
                idempotency_key=f"public-registration-{session.registration_id}",
                correlation_id=correlation_id,
                roles=("user",),
            )
        )
        self.service.link_identity(
            actor,
            user.user_id,
            issuer=self.registration_oidc_issuer,
            subject=provisioned.external_subject,
            provider=provisioned.provider,
            correlation_id=correlation_id,
        )
        self.service.store.save_registration_session(
            replace(session, provisioning_resume_hash=None)
        )
        if provisioned.password_setup_url:
            self._validate_registration_handoff(provisioned.password_setup_url)
            if browser:
                return self._html(
                    start_response,
                    self._page_html(
                        "Create your password",
                        "<h1>Your identity is ready.</h1>"
                        "<p>Continue to the protected identity service to create your password.</p>"
                        f'<p><a class="button" rel="noreferrer" href="{escape(provisioned.password_setup_url)}">Create password</a></p>',
                    ),
                    correlation_id,
                )
            return self._redirect(
                start_response, provisioned.password_setup_url, correlation_id
            )
        return self._json(
            start_response,
            "200 OK",
            {"status": "identity_ready", "login_required": True},
            correlation_id,
        )

    def _validate_registration_handoff(self, setup_url: str) -> None:
        parsed = urlsplit(setup_url)
        origin = f"{parsed.scheme}://{parsed.netloc}"
        if (
            parsed.scheme != "https"
            or not parsed.netloc
            or origin not in self.registration_password_setup_origins
        ):
            raise ValidationError("password setup handoff is not allow-listed")

    @staticmethod
    def _registration_username(value: Any) -> str:
        username = str(value or "").strip().lower()
        if not re.fullmatch(r"[a-z][a-z0-9._-]{2,31}", username):
            raise ValidationError("username is invalid")
        if username in {"admin", "administrator", "platform-root", "root", "system"}:
            raise ValidationError("username is reserved")
        return username

    @staticmethod
    def _registration_email(value: Any) -> str:
        email = str(value or "").strip().lower()
        if len(email) > 254 or not re.fullmatch(r"[^@\s]+@[^@\s]+\.[^@\s]+", email):
            raise ValidationError("email is invalid")
        return email

    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 _form_body(environ: Mapping[str, Any]) -> Mapping[str, str]:
        content_type = str(environ.get("CONTENT_TYPE", "")).partition(";")[0]
        if content_type != "application/x-www-form-urlencoded":
            raise ValidationError("form content type is required")
        length = min(int(environ.get("CONTENT_LENGTH") or 0), 65536)
        payload = environ["wsgi.input"].read(length).decode("utf-8")
        return {key: values[0] for key, values in parse_qs(payload).items()}

    def _csrf_token(self, environ: Mapping[str, Any]) -> str:
        if self.oidc_client is None:
            raise AuthorizationDenied("browser session required")
        session_id = cookie_value(str(environ.get("HTTP_COOKIE", "")), "ue_session")
        token = self.oidc_client.csrf_token(session_id or "")
        if token is None:
            raise AuthorizationDenied("browser session required")
        return token

    def _require_csrf(self, environ: Mapping[str, Any], supplied: str) -> None:
        expected = self._csrf_token(environ)
        if not supplied or not secrets.compare_digest(supplied, expected):
            raise AuthorizationDenied("invalid CSRF token")

    @staticmethod
    def _registration_csrf_cookie(token: str) -> str:
        return (
            f"ue_registration_csrf={token}; Path=/; HttpOnly; Secure; "
            "SameSite=Strict; Max-Age=1800"
        )

    def _require_registration_csrf(
        self, environ: Mapping[str, Any], supplied: str
    ) -> None:
        expected = cookie_value(
            str(environ.get("HTTP_COOKIE", "")), "ue_registration_csrf"
        )
        if not supplied or not expected or not secrets.compare_digest(supplied, expected):
            raise AuthorizationDenied("invalid registration CSRF token")

    @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

    @staticmethod
    def _expected_version(environ: Mapping[str, Any]) -> int:
        value = str(environ.get("HTTP_IF_MATCH", "")).strip().strip('"')
        if not value.isdigit():
            raise ValidationError("If-Match invitation version is required")
        return int(value)

    @staticmethod
    def _idempotency_key(environ: Mapping[str, Any]) -> str:
        value = str(environ.get("HTTP_IDEMPOTENCY_KEY", ""))
        if len(value) < 16:
            raise ValidationError("Idempotency-Key must contain at least 16 characters")
        return value

    def _home(self, actor: Any | None) -> str:
        identity = (
            f"<p>Signed in as <strong>{escape(actor.preferred_username)}</strong>.</p>"
            '<p><a class="button" href="/onboarding">Continue onboarding</a></p>'
            if actor is not None
            else (
                f'<p><a class="button" href="/login">Sign in with KeyCape</a></p>'
                + (
                    '<p>New here? <a href="/register">Create an account</a>.</p>'
                    if self.public_registration and self.registration_verification is not None
                    else ""
                )
            )
        )
        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 _registration_form(self, csrf_token: str) -> str:
        client_options = "".join(
            f'<option value="{escape(item)}">{escape(item)}</option>'
            for item in sorted(self.registration_clients)
        )
        tenant_options = "".join(
            f'<option value="{escape(item)}">{escape(item.removeprefix("tenant:"))}</option>'
            for item in sorted(self.registration_tenants)
        )
        return self._page_html(
            "Create account",
            f"""<h1>Create your account.</h1>
<p>We will verify your email before creating an identity.</p>
<form method="post" action="/register">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<label>Username <input name="username" required minlength="3" maxlength="32" pattern="[A-Za-z][A-Za-z0-9._-]+" autocomplete="username"></label>
<label>Email <input name="email" type="email" required autocomplete="email"></label>
<label>Display name <input name="display_name" maxlength="200" autocomplete="name"></label>
<label>Application <select name="client_id" required>{client_options}</select></label>
<label>Community <select name="tenant" required>{tenant_options}</select></label>
<button type="submit">Send verification email</button>
</form><p><a href="/login">Already have an account? Sign in</a></p>""",
        )

    def _registration_verification_form(self, csrf_token: str, handle: str) -> str:
        return self._page_html(
            "Verify email",
            f"""<h1>Verify your email.</h1>
<p>Confirm to finish creating your identity. This verification link can be used once.</p>
<form method="post" action="/registration/verify">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<input type="hidden" name="handle" value="{escape(handle)}">
<button type="submit">Verify and continue</button></form>""",
        )

    def _registration_resume_form(
        self, csrf_token: str, registration_id: str, resume_handle: str
    ) -> str:
        return self._page_html(
            "Finish account setup",
            f"""<h1>Your email is verified.</h1>
<p>The identity service is temporarily unavailable. Try again without repeating verification.</p>
<form method="post" action="/registration/resume">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<input type="hidden" name="registration_id" value="{escape(registration_id)}">
<input type="hidden" name="resume_handle" value="{escape(resume_handle)}">
<button type="submit">Try identity setup again</button></form>""",
        )
    def _admin(
        self, tenant: str, memberships: tuple[Any, ...],
        invitations: tuple[Any, ...], diagnostics: Any,
        platform_operator: bool, csrf_token: str,
    ) -> str:
        rows = "".join(
            self._admin_row(tenant, item, platform_operator, csrf_token)
            for item in memberships
        ) or '<tr><td colspan="6">No members yet.</td></tr>'
        invitation_rows = "".join(
            self._invitation_admin_row(tenant, item, csrf_token)
            for item in invitations
        ) or '<tr><td colspan="5">No invitations yet.</td></tr>'
        diagnostic_items = "".join(
            f"<li>{escape(item.replace('_', ' '))}</li>" for item in diagnostics.issues
        ) or "<li>No lifecycle gaps detected.</li>"
        return self._page_html(
            f"{tenant} users",
            f"""<h1>{escape(tenant)} users</h1>
<section><h2>Add a user</h2>
<form method="post" action="/admin/{escape(tenant)}/users">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<label>Display name <input name="display_name" required autocomplete="name"></label>
<label>Email <input name="primary_email" type="email" required autocomplete="email"></label>
<label>Role <select name="role"><option value="user">User</option><option value="tenant-admin">Tenant administrator</option></select></label>
<button type="submit">Add user</button></form></section>
<section><h2>Invite a user</h2>
<form method="post" action="/admin/{escape(tenant)}/invitations">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<label>Display name <input name="display_name" autocomplete="name"></label>
<label>Email <input name="primary_email" type="email" required autocomplete="email"></label>
<label>Role <select name="role"><option value="user">User</option><option value="tenant-admin">Tenant administrator</option></select></label>
<button type="submit">Send invitation</button></form></section>
<section><h2>Invitations</h2><table><thead><tr><th>Email</th><th>Role</th><th>Status</th><th>Expires</th><th>Action</th></tr></thead><tbody>{invitation_rows}</tbody></table></section>
<section aria-labelledby="lifecycle-gaps"><h2 id="lifecycle-gaps">Lifecycle diagnostics</h2><p>Diagnostics contain machine-readable gap categories only; credentials and factor evidence are never displayed.</p><ul>{diagnostic_items}</ul></section>
<section><h2>Members</h2><table><thead><tr><th>User</th><th>Email</th><th>Role</th><th>Status</th><th>Directory</th><th>Action</th></tr></thead><tbody>{rows}</tbody></table></section>""",
        )

    def _invitation_admin_row(self, tenant: str, invitation: Any, csrf_token: str) -> str:
        actions = ""
        if invitation.status.value == "pending":
            actions = f"""<form method="post" action="/admin/{escape(tenant)}/invitations/{escape(invitation.invitation_id)}/resend">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}"><input type="hidden" name="version" value="{invitation.version}"><button type="submit">Resend</button></form>
<form method="post" action="/admin/{escape(tenant)}/invitations/{escape(invitation.invitation_id)}/expire">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}"><input type="hidden" name="version" value="{invitation.version}"><button type="submit">Expire</button></form>"""
        expires = invitation.expires_at.isoformat() if invitation.expires_at else "—"
        return (
            f"<tr><td>{escape(invitation.primary_email)}</td><td>{escape(invitation.role)}</td>"
            f"<td>{escape(invitation.status.value)}</td><td>{escape(expires)}</td><td>{actions}</td></tr>"
        )

    def _admin_row(
        self, tenant: str, membership: Any,
        platform_operator: bool, csrf_token: str,
    ) -> str:
        user = self.service.store.user(membership.user_id)
        identities = self.service.store.identities_for_user(membership.user_id)
        directory = next(
            (item for item in identities if item.provider == "netkingdom-lldap"),
            None,
        )
        tenant_account = self.service.store.tenant_account(tenant, membership.user_id)
        status = tenant_account.status if tenant_account else AccountStatus.INVITED
        action = (
            f"""<span>Linked as {escape(directory.subject)}</span>
<form method="post" action="/admin/{escape(tenant)}/users/{escape(membership.user_id)}/provision">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<button type="submit">Create password setup link</button></form>
<form method="post" action="/admin/{escape(tenant)}/users/{escape(membership.user_id)}/status">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<input type="hidden" name="status" value="{'active' if status == AccountStatus.SUSPENDED else 'suspended'}">
<button type="submit">{'Reactivate' if status == AccountStatus.SUSPENDED else 'Suspend'}</button></form>"""
            if directory
            else f"""<form method="post" action="/admin/{escape(tenant)}/users/{escape(membership.user_id)}/provision">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<button type="submit">Create login</button></form>"""
        )
        action += f"""<form method="post" action="/admin/{escape(tenant)}/users/{escape(membership.user_id)}/remove">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<button type="submit">Remove account</button></form>"""
        if platform_operator:
            action += f"""<form method="post" action="/admin/{escape(tenant)}/users/{escape(membership.user_id)}/recover">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<button type="submit">Recover identity</button></form>"""
        return (
            f"<tr><td>{escape(user.display_name or membership.user_id) if user else escape(membership.user_id)}</td>"
            f"<td>{escape(user.primary_email or '') if user else ''}</td>"
            f"<td>{escape(membership.kind)}</td>"
            f"<td>{escape(status.value)}</td>"
            f"<td>{'linked' if directory else 'pending'}</td><td>{action}</td></tr>"
        )

    def _invitation_acceptance(self, invitation: Any, csrf_token: str) -> str:
        return self._page_html(
            "Accept invitation",
            f"""<h1>Join {escape(invitation.tenant)}</h1>
<p>You were invited as <strong>{escape(invitation.role)}</strong>. The invitation expires at {escape(invitation.expires_at.isoformat() if invitation.expires_at else 'the tenant policy deadline')}.</p>
<form method="post" action="/invitations/{escape(invitation.invitation_id)}">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<button type="submit">Accept invitation</button></form>
<p>Your password and MFA remain on the identity-provider surface.</p>""",
        )

    def _platform(self, csrf_token: str) -> str:
        return self._page_html(
            "Platform administration",
            f"""<h1>Platform administration</h1>
<section aria-labelledby="create-tenant"><h2 id="create-tenant">Create tenant</h2>
<form method="post" action="/platform/tenants">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<label>Tenant identifier <input name="tenant" required pattern="tenant:.+" placeholder="tenant:friendly:example"></label>
<label>Display name <input name="display_name" required></label>
<fieldset><legend>First administrator (optional)</legend>
<label>Name <input name="admin_display_name" autocomplete="name"></label>
<label>Email <input name="admin_email" type="email" autocomplete="email"></label></fieldset>
<button type="submit">Create tenant</button></form></section>""",
        )

    def _platform_result(self, result: Any, tenant: str, admin_prepared: bool) -> str:
        return self._page_html(
            "Tenant created",
            f"""<h1>Tenant {escape(result.status)}</h1>
<p><strong>{escape(tenant)}</strong> was processed by the tenant authority.</p>
<p>{'The first administrator is prepared and awaiting onboarding.' if admin_prepared else 'No first administrator was requested.'}</p>
<p><a class="button" href="/admin/{escape(tenant)}">Open tenant administration</a></p>
<p><a href="/platform">Return to platform administration</a></p>""",
        )

    def _onboarding(
        self, session: Any, memberships: tuple[Any, ...], journeys: tuple[Any, ...],
        selected_tenant: str, csrf_token: str,
    ) -> str:
        membership_items = "".join(
            f"<li><a href=\"/onboarding?{urlencode({'tenant': item.tenant})}\">{escape(item.tenant)}</a> — {escape(item.kind)}</li>"
            for item in memberships
        ) or "<li>No tenant memberships yet.</li>"
        journey_items = "".join(
            self._onboarding_journey_item(item, csrf_token) for item in journeys
        ) or "<li>No additional onboarding steps are required.</li>"
        verification = "Verified by your identity provider" if session.actor.assurance else "Verification pending"
        consent_checked = " checked" if session.user.consented_at else ""
        return self._page_html(
            "Onboarding",
            f"""<h1>Welcome, {escape(session.user.display_name or session.actor.preferred_username or session.user.user_id)}</h1>
<section aria-labelledby="verification"><h2 id="verification">Email and sign-in</h2><p>{escape(verification)}</p><p>Passwords and MFA are managed by your identity provider.</p></section>
<section aria-labelledby="profile"><h2 id="profile">Profile and consent</h2>
<form method="post" action="/onboarding/profile"><input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<label>Display name <input name="display_name" required maxlength="200" autocomplete="name" value="{escape(session.user.display_name or '')}"></label>
<label><input name="consent_accepted" type="checkbox" value="yes"{consent_checked}> I accept portal terms version 1</label>
<button type="submit">Save profile</button></form></section>
<section aria-labelledby="tenants"><h2 id="tenants">Tenant access</h2><p>Viewing <strong>{escape(selected_tenant)}</strong>.</p><ul>{membership_items}</ul>
<p><a href="/login?{urlencode({'tenant_hint': selected_tenant})}">Reauthenticate in this tenant</a> to change the authoritative login context.</p></section>
<section aria-labelledby="steps"><h2 id="steps">Onboarding progress</h2><ul>{journey_items}</ul></section>""",
        )

    @staticmethod
    def _onboarding_journey_item(journey: Any, csrf_token: str) -> str:
        steps = "".join(
            PortalApplication._onboarding_step_item(journey.journey_id, step, csrf_token)
            for step in journey.steps
        )
        return (
            f"<li><strong>{escape(journey.status.value)}</strong>"
            f"<ol>{steps}</ol></li>"
        )

    @staticmethod
    def _onboarding_step_item(journey_id: str, step: Any, csrf_token: str) -> str:
        action = ""
        if (
            step.status.value == "in_progress"
            and step.subsystem == "user-engine"
            and step.handoff is None
        ):
            action = f"""<form method="post" action="/onboarding/{escape(journey_id)}/steps/{escape(step.step_key)}/complete">
<input type="hidden" name="csrf_token" value="{escape(csrf_token)}">
<button type="submit">Mark complete</button></form>"""
        elif step.handoff is not None or step.subsystem != "user-engine":
            action = "<span>Continue on the provider-owned surface; this page will resume after its callback.</span>"
        gap = f" <span>Support category: {escape(step.lifecycle_gap)}</span>" if step.lifecycle_gap else ""
        return (
            f"<li><span>{escape(step.title)} — {escape(step.status.value)}</span>{gap}{action}</li>"
        )

    def _password_setup_handoff(self, setup_url: str, tenant: str) -> str:
        if not setup_url.startswith("https://"):
            raise ValidationError("password setup handoff must use HTTPS")
        return self._page_html(
            "Password setup",
            "<h1>Login identity created</h1>"
            "<p>The password is handled only by the NetKingdom identity "
            "surface. This short-lived link is single use.</p>"
            f'<p><a class="button" rel="noreferrer" href="{escape(setup_url)}">'
            "Continue to password setup</a></p>"
            f'<p><a href="/admin/{escape(tenant)}">'
            "Return to tenant administration</a></p>",
        )

    def _redirect(self, start_response: StartResponse, location: str, correlation_id: str) -> list[bytes]:
        start_response("303 See Other", [("Location", location), *self._security_headers(correlation_id)])
        return [b""]

    @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)}}
section{{margin:2rem 0}}form{{display:grid;gap:.8rem;max-width:42rem}}label{{display:grid;gap:.25rem}}
input,select,button{{font:inherit;padding:.65rem}}button{{background:var(--accent);color:white;border:0;border-radius:.3rem;cursor:pointer}}
a:focus-visible,input:focus-visible,select:focus-visible,button:focus-visible{{outline:3px solid #e59f24;outline-offset:3px}}@media(max-width:640px){{body{{font-size:16px}}table{{display:block;overflow-x:auto}}}}
</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,
        *, extra_headers: list[tuple[str, str]] | None = None,
    ) -> list[bytes]:
        data = body.encode()
        start_response("200 OK", [
            ("Content-Type", "text/html; charset=utf-8"),
            ("Content-Length", str(len(data))),
            *(extra_headers or []),
            *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 _metrics(
        self, start_response: StartResponse, correlation_id: str
    ) -> list[bytes]:
        report = self.service.readiness()
        counts = self.service.operability_snapshot().metrics
        lines = [
            "# HELP user_engine_ready Whether runtime dependencies are ready.",
            "# TYPE user_engine_ready gauge",
            f"user_engine_ready {1 if report.ready else 0}",
            "# HELP user_engine_records Durable logical record counts by kind.",
            "# TYPE user_engine_records gauge",
        ]
        for kind, count in sorted(counts.items()):
            safe_kind = "".join(
                character for character in str(kind)
                if character.isalnum() or character in "_-"
            )
            lines.append(f'user_engine_records{{kind="{safe_kind}"}} {int(count)}')
        data = ("\n".join(lines) + "\n").encode()
        start_response(
            "200 OK",
            [
                ("Content-Type", "text/plain; version=0.0.4; charset=utf-8"),
                ("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'"),
        ]