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 .errors import ApprovalError
from .store import Engine

FORBIDDEN_DECISION_KEYS = frozenset({"effect", "decision", "allow", "deny"})


def _read_json(environ: dict[str, Any]) -> dict[str, Any]:
    length = int(environ.get("CONTENT_LENGTH") or 0)
    if length == 0:
        return {}
    raw = environ["wsgi.input"].read(length)
    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) -> None:
        self.engine = engine

    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 ("/readyz", "/v1/readyz") and method == "GET":
            self.engine.outbox_stats()
            return 200, {"status": "ok", "store": "ok"}
        if path == "/v1/cadence" and method == "GET":
            return 200, self.engine.transition_counts() | {"form": "heartbeat-or-reconciliation"}
        if path == "/v1/outbox/stats" and method == "GET":
            return 200, self.engine.outbox_stats()
        if path == "/v1/heartbeat" and method == "POST":
            return 200, self.engine.emit_heartbeat()
        if path == "/v1/approvals" and method == "POST":
            data = _read_json(environ)
            obj = self.engine.create(
                data.get("binding") or {},
                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":
                return 200, self.engine.get(approval_id).as_dict()
            if rest == ["claim"] and method == "GET":
                return 200, self.engine.claim(approval_id)
            if rest == ["entries"] and method == "POST":
                data = _read_json(environ)
                obj = self.engine.add_entry(
                    approval_id,
                    data.get("subject_id") or "",
                    assurance=data.get("assurance"),
                    evidence_ref=data.get("evidence_ref"),
                )
                return 200, obj.as_dict()
            if rest == ["revoke"] and method == "POST":
                return 200, self.engine.revoke(approval_id).as_dict()
            if rest == ["supersede"] and method == "POST":
                data = _read_json(environ)
                return 200, self.engine.supersede(approval_id, data.get("successor_id"))
            if rest == ["consume"] and method == "POST":
                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) -> 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": "",
    }
    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