approval-engine/approval_engine/api.py

"""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"),
                pdp_path=bool(data.get("pdp_path", False)),
                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