"""HTTP surface. Introspection and lifecycle mutation; never a decision.""" from __future__ import annotations import json from typing import Any, Callable from .auth import Authenticator, DenyAllAuthenticator, Identity from .errors import ApprovalError, Forbidden from .store import Engine FORBIDDEN_DECISION_KEYS = frozenset({"effect", "decision", "allow", "deny"}) def _read_json(environ: dict[str, Any]) -> dict[str, Any]: try: length = int(environ.get("CONTENT_LENGTH") or 0) except (TypeError, ValueError) as exc: raise ApprovalError("invalid content length") from exc if length < 0 or length > 256 * 1024: raise ApprovalError("request body is too large") if length == 0: return {} raw = environ["wsgi.input"].read(length) if len(raw) != length: raise ApprovalError("truncated request body") if not raw: return {} try: data = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise ApprovalError("invalid json") from exc if not isinstance(data, dict): raise ApprovalError("json object required") return data def _assert_not_decision(payload: Any) -> None: if isinstance(payload, dict): bad = FORBIDDEN_DECISION_KEYS & set(payload) if bad: raise RuntimeError(f"decision-shaped keys leaked: {sorted(bad)}") for value in payload.values(): _assert_not_decision(value) elif isinstance(payload, list): for item in payload: _assert_not_decision(item) class App: def __init__( self, engine: Engine, authenticator: Authenticator | None = None, *, require_persistent: bool = False, ) -> None: self.engine = engine self.authenticator = authenticator or DenyAllAuthenticator() self.require_persistent = require_persistent def identity(self, environ: dict[str, Any], scope: str) -> Identity: identity = self.authenticator.authenticate( environ.get("HTTP_AUTHORIZATION") ).require(scope) if identity.tenant != self.engine.tenant: raise Forbidden("caller tenant does not match this approval store") return identity def __call__(self, environ: dict[str, Any], start_response: Callable) -> list[bytes]: method = environ.get("REQUEST_METHOD", "GET").upper() path = environ.get("PATH_INFO") or "/" try: status, body = self.dispatch(method, path, environ) except ApprovalError as exc: status, body = exc.http_status, {"error": exc.reason_code, "message": str(exc)} except ValueError as exc: status, body = 422, {"error": "unprocessable", "message": str(exc)} _assert_not_decision(body) payload = json.dumps(body, sort_keys=True).encode("utf-8") start_response( f"{status} {'OK' if status < 400 else 'ERROR'}", [ ("Content-Type", "application/json"), ("Content-Length", str(len(payload))), ], ) return [payload] def dispatch(self, method: str, path: str, environ: dict[str, Any]) -> tuple[int, dict[str, Any]]: if path in ("/healthz", "/v1/healthz") and method == "GET": return 200, {"status": "ok"} if path in ("/readyz", "/v1/readyz") and method == "GET": storage = self.engine.storage_status() ready = storage["schema_current"] and ( storage["persistent"] or not self.require_persistent ) self.engine.outbox_stats() return (200 if ready else 503), { "status": "ok" if ready else "unavailable", "store": "ok" if ready else "not-production-ready", } if path == "/v1/storage/status" and method == "GET": self.identity(environ, "approval:observe") return 200, self.engine.storage_status(integrity=True) if path == "/v1/cadence" and method == "GET": self.identity(environ, "approval:observe") return 200, self.engine.transition_counts() | {"form": "heartbeat-or-reconciliation"} if path == "/v1/outbox/stats" and method == "GET": self.identity(environ, "approval:observe") return 200, self.engine.outbox_stats() if path == "/v1/heartbeat" and method == "POST": self.identity(environ, "approval:emit") return 200, self.engine.emit_heartbeat() if path == "/v1/approvals" and method == "POST": identity = self.identity(environ, "approval:create") data = _read_json(environ) binding = data.get("binding") or {} if binding.get("actor") != identity.subject: raise Forbidden("binding.actor must match the authenticated subject") obj = self.engine.create( binding, data.get("validity") or {}, int(data.get("required_count") or 1), pdp_digest=data.get("pdp_digest"), approval_id=data.get("id"), ) return 201, obj.as_dict() parts = path.strip("/").split("/") if len(parts) >= 3 and parts[0] == "v1" and parts[1] == "approvals": approval_id = parts[2] rest = parts[3:] if not rest and method == "GET": self.identity(environ, "approval:read") return 200, self.engine.get(approval_id).as_dict() if rest == ["claim"] and method == "GET": self.identity(environ, "approval:read") return 200, self.engine.claim(approval_id) if rest == ["entries"] and method == "POST": identity = self.identity(environ, "approval:approve") _read_json(environ) obj = self.engine.add_entry( approval_id, identity.subject, assurance=json.dumps(identity.assurance, sort_keys=True), evidence_ref=identity.evidence_ref, ) return 200, obj.as_dict() if rest == ["revoke"] and method == "POST": self.identity(environ, "approval:revoke") return 200, self.engine.revoke(approval_id).as_dict() if rest == ["supersede"] and method == "POST": self.identity(environ, "approval:supersede") data = _read_json(environ) return 200, self.engine.supersede(approval_id, data.get("successor_id")) if rest == ["consume"] and method == "POST": identity = self.identity(environ, "approval:consume") if identity.principal_type not in {"service", "agent"}: raise Forbidden("consume requires a service or agent principal") data = _read_json(environ) return 200, self.engine.consume( approval_id, data.get("request_digest"), decision_id=data.get("decision_id"), ) if "check" in path or path.endswith("/authorize"): return 404, {"error": "not_found", "message": "no such surface"} return 404, {"error": "not_found", "message": path} def call( app: App, method: str, path: str, body: dict[str, Any] | None = None, *, authorization: str | None = "Bearer test-token", ) -> tuple[int, dict[str, Any]]: """In-process WSGI helper for tests.""" raw = json.dumps(body or {}).encode("utf-8") if body is not None else b"" environ = { "REQUEST_METHOD": method, "PATH_INFO": path, "wsgi.input": _Bytes(raw), "CONTENT_LENGTH": str(len(raw)) if body is not None else "0", "QUERY_STRING": "", } if authorization is not None: environ["HTTP_AUTHORIZATION"] = authorization status_headers: list[tuple[str, list]] = [] def start_response(status: str, headers: list[tuple[str, str]]) -> None: status_headers.append((status, headers)) result = b"".join(app(environ, start_response)) code = int(status_headers[0][0].split()[0]) return code, json.loads(result.decode("utf-8")) class _Bytes: def __init__(self, data: bytes) -> None: self._data = data def read(self, n: int = -1) -> bytes: if n < 0: out, self._data = self._data, b"" return out out, self._data = self._data[:n], self._data[n:] return out