"""Dependency-free WSGI transport for the user-engine portal. Authentication is deliberately delegated to KeyCape (or another OIDC-aware edge). The application accepts claims only when the edge presents a shared authentication marker configured at process start. This keeps passwords, MFA material, provider administration credentials, and browser sessions out of user-engine. """ from __future__ import annotations from dataclasses import asdict, is_dataclass from enum import Enum from html import escape import json import secrets from typing import Any, Callable, Iterable, Mapping from urllib.parse import parse_qs from user_engine.domain import AccountStatus from user_engine.errors import AuthorizationDenied, ConflictError, NotFoundError, ValidationError from user_engine.service import UserEngineService StartResponse = Callable[[str, list[tuple[str, str]]], Any] def _jsonable(value: Any) -> Any: if is_dataclass(value): return {key: _jsonable(item) for key, item in asdict(value).items()} if isinstance(value, Enum): return value.value if isinstance(value, Mapping): return {str(key): _jsonable(item) for key, item in value.items()} if isinstance(value, (tuple, list)): return [_jsonable(item) for item in value] if hasattr(value, "isoformat"): return value.isoformat() return value class PortalApplication: """Small, auditable HTTP adapter over :class:`UserEngineService`.""" def __init__( self, service: UserEngineService, *, trusted_proxy_secret: str, login_url: str, public_registration: bool = True, ) -> 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 def __call__(self, environ: Mapping[str, Any], start_response: StartResponse) -> Iterable[bytes]: correlation_id = environ.get("HTTP_X_REQUEST_ID") or f"corr_{secrets.token_hex(12)}" try: return self._dispatch(environ, start_response, str(correlation_id)) except (ValidationError, ConflictError) as exc: return self._error(start_response, "400 Bad Request", "invalid_request", str(exc), correlation_id) except AuthorizationDenied: return self._error(start_response, "403 Forbidden", "access_denied", "Access denied.", correlation_id) except NotFoundError: return self._error(start_response, "404 Not Found", "not_found", "Resource not found.", correlation_id) except (json.JSONDecodeError, UnicodeDecodeError): return self._error(start_response, "400 Bad Request", "invalid_json", "Malformed request body.", correlation_id) def _dispatch(self, environ: Mapping[str, Any], start_response: StartResponse, correlation_id: str) -> Iterable[bytes]: method = str(environ.get("REQUEST_METHOD", "GET")).upper() path = str(environ.get("PATH_INFO", "/")).rstrip("/") or "/" if path == "/healthz": return self._json(start_response, "200 OK", _jsonable(self.service.health()), correlation_id) if path == "/readyz": report = self.service.readiness() return self._json(start_response, "200 OK" if report.ready else "503 Service Unavailable", _jsonable(report), correlation_id) if path == "/login": start_response("303 See Other", [("Location", self.login_url), *self._security_headers(correlation_id)]) return [b""] if path == "/" and method == "GET": actor = self._optional_actor(environ) return self._html(start_response, self._home(actor), correlation_id) actor = self._actor(environ) if path == "/api/v1/me" and method == "GET": return self._json(start_response, "200 OK", _jsonable(self.service.me(self._claims(environ), correlation_id=correlation_id)), correlation_id) if path == "/api/v1/registrations" and method == "POST": if not self.public_registration: raise AuthorizationDenied("public registration disabled") body = self._body(environ) session = self.service.start_registration( actor, tenant=body.get("tenant"), correlation_id=correlation_id, ) return self._json(start_response, "201 Created", _jsonable(session), correlation_id) if path.startswith("/api/v1/registrations/") and path.endswith("/complete") and method == "POST": registration_id = path.split("/")[4] body = self._body(environ) result = self.service.complete_registration( actor, registration_id, display_name=body.get("display_name"), primary_email=body.get("primary_email"), correlation_id=correlation_id, ) return self._json(start_response, "200 OK", _jsonable(result), correlation_id) if path.startswith("/api/v1/tenants/") and path.endswith("/users") and method == "GET": tenant = path.split("/")[4] self.service.resolve_tenant_context(actor, tenant) memberships = self.service.store.memberships_for_tenant(tenant) offset, limit = self._page(environ) items = memberships[offset : offset + limit] payload = {"items": _jsonable(items), "offset": offset, "limit": limit, "total": len(memberships)} return self._json(start_response, "200 OK", payload, correlation_id) if path.startswith("/api/v1/tenants/") and "/users/" in path and method == "PATCH": parts = path.split("/") tenant, user_id = parts[4], parts[6] body = self._body(environ) status = AccountStatus(str(body["status"])) result = self.service.set_tenant_account_status( actor, user_id, status, tenant=tenant, correlation_id=correlation_id ) return self._json(start_response, "200 OK", _jsonable(result), correlation_id) if path.startswith("/admin/") and method == "GET": tenant = path.split("/")[2] self.service.resolve_tenant_context(actor, tenant) memberships = self.service.store.memberships_for_tenant(tenant) return self._html(start_response, self._admin(tenant, memberships), correlation_id) return self._error(start_response, "404 Not Found", "not_found", "Resource not found.", correlation_id) def _claims(self, environ: Mapping[str, Any]) -> Mapping[str, Any]: marker = str(environ.get("HTTP_X_USER_ENGINE_PROXY_SECRET", "")) if not secrets.compare_digest(marker, self.trusted_proxy_secret): raise AuthorizationDenied("untrusted identity source") raw = environ.get("HTTP_X_VERIFIED_OIDC_CLAIMS") if not raw: raise AuthorizationDenied("verified claims required") claims = json.loads(str(raw)) if not isinstance(claims, dict): raise AuthorizationDenied("verified claims must be an object") return claims def _actor(self, environ: Mapping[str, Any]) -> Any: return self.service.identity_adapter.normalize(self._claims(environ)) def _optional_actor(self, environ: Mapping[str, Any]) -> Any | None: try: return self._actor(environ) except (AuthorizationDenied, json.JSONDecodeError, ValidationError): return None @staticmethod def _body(environ: Mapping[str, Any]) -> Mapping[str, Any]: length = min(int(environ.get("CONTENT_LENGTH") or 0), 65536) payload = environ["wsgi.input"].read(length) if length else b"{}" value = json.loads(payload.decode("utf-8")) if not isinstance(value, dict): raise ValidationError("request body must be an object") return value @staticmethod def _page(environ: Mapping[str, Any]) -> tuple[int, int]: query = parse_qs(str(environ.get("QUERY_STRING", ""))) offset = max(0, int(query.get("offset", ["0"])[0])) limit = max(1, min(100, int(query.get("limit", ["25"])[0]))) return offset, limit def _home(self, actor: Any | None) -> str: identity = ( f"

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

" if actor is not None else f'

Sign in with KeyCape

' ) 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 _admin(self, tenant: str, memberships: tuple[Any, ...]) -> str: rows = "".join( f"{escape(item.user_id)}{escape(item.kind)}{escape(item.scope_id)}" for item in memberships ) or 'No members yet.' return self._page_html( f"{tenant} users", f"

{escape(tenant)} users

{rows}
UserRoleScope
", ) @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) -> list[bytes]: data = body.encode() start_response("200 OK", [("Content-Type", "text/html; charset=utf-8"), ("Content-Length", str(len(data))), *self._security_headers(correlation_id)]) return [data] def _json(self, start_response: StartResponse, status: str, payload: Any, correlation_id: str) -> list[bytes]: data = json.dumps(payload, separators=(",", ":"), default=str).encode() start_response(status, [("Content-Type", "application/json"), ("Content-Length", str(len(data))), *self._security_headers(correlation_id)]) return [data] def _error(self, start_response: StartResponse, status: str, code: str, message: str, correlation_id: str) -> list[bytes]: return self._json(start_response, status, {"error": {"code": code, "message": message, "correlation_id": correlation_id}}, correlation_id) @staticmethod def _security_headers(correlation_id: str) -> list[tuple[str, str]]: return [ ("X-Request-ID", correlation_id), ("Cache-Control", "no-store"), ("X-Content-Type-Options", "nosniff"), ("Referrer-Policy", "no-referrer"), ("Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'; form-action 'self'; frame-ancestors 'none'; base-uri 'none'"), ]