"""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 collections import deque from threading import Lock from time import monotonic 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, ...] = (), registration_rate_limit: int = 10, registration_rate_window_seconds: int = 60, ) -> 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 ) if registration_rate_limit < 1 or registration_rate_window_seconds < 1: raise ValueError("registration rate limit and window must be positive") self.registration_rate_limit = registration_rate_limit self.registration_rate_window_seconds = registration_rate_window_seconds self._registration_attempts: dict[str, deque[float]] = {} self._registration_attempts_lock = Lock() 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 method == "POST" and path in { "/register", "/registration/verify", "/registration/resume", "/registration/cancel", "/api/v1/public/registrations", "/api/v1/public/registrations/verify", "/api/v1/public/registrations/resume", "/api/v1/public/registrations/cancel", } and not self._accept_registration_attempt(environ): return self._error( start_response, "429 Too Many Requests", "rate_limited", "Too many registration attempts. Try again later.", correlation_id, ) 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 == "/registration/cancel" and method == "GET": query = parse_qs(str(environ.get("QUERY_STRING", ""))) handle = str(query.get("handle", [""])[0]) if len(handle) < 16: raise ValidationError("cancellation handle is invalid") token = secrets.token_urlsafe(32) return self._html( start_response, self._registration_cancel_form(token, handle), correlation_id, extra_headers=[("Set-Cookie", self._registration_csrf_cookie(token))], ) if path == "/registration/cancel" and method == "POST": body = self._form_body(environ) self._require_registration_csrf(environ, str(body.get("csrf_token", ""))) return self._cancel_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 ) if path == "/api/v1/public/registrations/cancel" and method == "POST": return self._cancel_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", "

Check your email.

" "

If the address can be registered, a verification link is on its way. " "The link expires after 30 minutes.

", ), 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 _cancel_public_registration( self, environ, start_response, correlation_id, *, body: Mapping[str, Any] | None = None, browser: bool = False, ): 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.cancel( 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("cancellation 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, ) self.service.abandon_registration( actor, session.registration_id, correlation_id=correlation_id ) if browser: return self._html( start_response, self._page_html( "Registration canceled", "

Registration canceled.

" "

The request can no longer be verified. You may start again at any time.

" '

Start again

', ), correlation_id, ) return self._json( start_response, "200 OK", {"status": "registration_canceled"}, correlation_id, ) 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", "

Your identity is ready.

" "

Continue to the protected identity service to create your password.

" f'

Create password

', ), 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") def _accept_registration_attempt(self, environ: Mapping[str, Any]) -> bool: """Apply a bounded per-source sliding-window limit to public writes. Only ``REMOTE_ADDR`` is trusted. Forwarded headers are deliberately ignored because the ingress must normalize the peer address before the request reaches this process. """ source = str(environ.get("REMOTE_ADDR") or "unknown")[:128] now = monotonic() cutoff = now - self.registration_rate_window_seconds with self._registration_attempts_lock: attempts = self._registration_attempts.setdefault(source, deque()) while attempts and attempts[0] <= cutoff: attempts.popleft() if len(attempts) >= self.registration_rate_limit: return False attempts.append(now) if len(self._registration_attempts) > 4096: for key in tuple(self._registration_attempts): values = self._registration_attempts[key] while values and values[0] <= cutoff: values.popleft() if not values: del self._registration_attempts[key] return True @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"

Signed in as {escape(actor.preferred_username)}.

" '

Continue onboarding

' if actor is not None else ( f'

Sign in with KeyCape

' + ( '

New here? Create an account.

' if self.public_registration and self.registration_verification is not None else "" ) ) ) return self._page_html( "Identity & access", "

Your account, on your terms.

" "

Join a tenant, complete onboarding, and manage access without exposing credentials to applications.

" + identity, ) def _registration_form(self, csrf_token: str) -> str: client_options = "".join( f'' for item in sorted(self.registration_clients) ) tenant_options = "".join( f'' for item in sorted(self.registration_tenants) ) return self._page_html( "Create account", f"""

Create your account.

We will verify your email before creating an identity.

Already have an account? Sign in

""", ) def _registration_verification_form(self, csrf_token: str, handle: str) -> str: return self._page_html( "Verify email", f"""

Verify your email.

Confirm to finish creating your identity. This verification link can be used once.

""", ) def _registration_resume_form( self, csrf_token: str, registration_id: str, resume_handle: str ) -> str: return self._page_html( "Finish account setup", f"""

Your email is verified.

The identity service is temporarily unavailable. Try again without repeating verification.

""", ) def _registration_cancel_form(self, csrf_token: str, handle: str) -> str: return self._page_html( "Cancel registration", f"""

Cancel this registration?

This verification request will stop working. No login identity will be created.

Keep registration

""", ) 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 'No members yet.' invitation_rows = "".join( self._invitation_admin_row(tenant, item, csrf_token) for item in invitations ) or 'No invitations yet.' diagnostic_items = "".join( f"
  • {escape(item.replace('_', ' '))}
    • " for item in diagnostics.issues ) or "
  • No lifecycle gaps detected.
    • " return self._page_html( f"{tenant} users", f"""

    {escape(tenant)} users

    Add a user

    Invite a user

    Invitations

    {invitation_rows}
    EmailRoleStatusExpiresAction

    Lifecycle diagnostics

    Diagnostics contain machine-readable gap categories only; credentials and factor evidence are never displayed.

    Members

    {rows}
    UserEmailRoleStatusDirectoryAction
    """, ) def _invitation_admin_row(self, tenant: str, invitation: Any, csrf_token: str) -> str: actions = "" if invitation.status.value == "pending": actions = f"""
    """ expires = invitation.expires_at.isoformat() if invitation.expires_at else "โ€”" return ( f"{escape(invitation.primary_email)}{escape(invitation.role)}" f"{escape(invitation.status.value)}{escape(expires)}{actions}" ) 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"""Linked as {escape(directory.subject)}
    """ if directory else f"""
    """ ) action += f"""
    """ if platform_operator: action += f"""
    """ return ( f"{escape(user.display_name or membership.user_id) if user else escape(membership.user_id)}" f"{escape(user.primary_email or '') if user else ''}" f"{escape(membership.kind)}" f"{escape(status.value)}" f"{'linked' if directory else 'pending'}{action}" ) def _invitation_acceptance(self, invitation: Any, csrf_token: str) -> str: return self._page_html( "Accept invitation", f"""

    Join {escape(invitation.tenant)}

    You were invited as {escape(invitation.role)}. The invitation expires at {escape(invitation.expires_at.isoformat() if invitation.expires_at else 'the tenant policy deadline')}.

    Your password and MFA remain on the identity-provider surface.

    """, ) def _platform(self, csrf_token: str) -> str: return self._page_html( "Platform administration", f"""

    Platform administration

    Create tenant

    First administrator (optional)
    """, ) def _platform_result(self, result: Any, tenant: str, admin_prepared: bool) -> str: return self._page_html( "Tenant created", f"""

    Tenant {escape(result.status)}

    {escape(tenant)} was processed by the tenant authority.

    {'The first administrator is prepared and awaiting onboarding.' if admin_prepared else 'No first administrator was requested.'}

    Open tenant administration

    Return to platform administration

    """, ) def _onboarding( self, session: Any, memberships: tuple[Any, ...], journeys: tuple[Any, ...], selected_tenant: str, csrf_token: str, ) -> str: membership_items = "".join( f"
  • {escape(item.tenant)} โ€” {escape(item.kind)}
    • " for item in memberships ) or "
  • No tenant memberships yet.
    • " journey_items = "".join( self._onboarding_journey_item(item, csrf_token) for item in journeys ) or "
  • No additional onboarding steps are required.
    • " 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"""

    Welcome, {escape(session.user.display_name or session.actor.preferred_username or session.user.user_id)}

    Email and sign-in

    {escape(verification)}

    Passwords and MFA are managed by your identity provider.

    Profile and consent

    Tenant access

    Viewing {escape(selected_tenant)}.

    Reauthenticate in this tenant to change the authoritative login context.

    Onboarding progress

    """, ) @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"
  • {escape(journey.status.value)}" f"
      {steps}
    • " ) @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"""
    """ elif step.handoff is not None or step.subsystem != "user-engine": action = "Continue on the provider-owned surface; this page will resume after its callback." gap = f" Support category: {escape(step.lifecycle_gap)}" if step.lifecycle_gap else "" return ( f"
  • {escape(step.title)} โ€” {escape(step.status.value)}{gap}{action}
    • " ) 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", "

    Login identity created

    " "

    The password is handled only by the NetKingdom identity " "surface. This short-lived link is single use.

    " f'

    ' "Continue to password setup

    " f'

    ' "Return to tenant administration

    ", ) 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""" {escape(title)} ยท Railiance
    Railiance identity
    {body}
    """ 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'"), ]