approval-engine/approval_engine/execution_confirmation.py

"""Exact consumption evidence for an independently authenticated mediator.

This client supplies PIP facts, never an execution admission or PDP decision.
The mediator still owns claim/PDP validation, deadlines and a one-shot journal.
"""

from datetime import datetime
from typing import Any
from urllib.parse import quote

from .binding import DIGEST_RE, binding_digest
from .pep import ApprovalHTTPClient, ApprovalProtocolError, _ID_RE


def _inputs(approval_id: str, request_digest: str, decision_id: str) -> None:
    if not isinstance(approval_id, str) or not _ID_RE.fullmatch(approval_id):
        raise ApprovalProtocolError("approval id is invalid")
    if not isinstance(request_digest, str) or not DIGEST_RE.fullmatch(request_digest):
        raise ApprovalProtocolError("canonical request digest is required")
    if not isinstance(decision_id, str) or not decision_id.strip():
        raise ApprovalProtocolError("exact decision id is required")


def _evidence(payload: dict[str, Any], request_digest: str, decision_id: str) -> str:
    if payload.get("request_digest") != request_digest:
        raise ApprovalProtocolError("consumed request digest does not match")
    if payload.get("decision_id") != decision_id:
        raise ApprovalProtocolError("consumed decision id does not match")
    value = payload.get("consumed_at")
    try:
        stamp = datetime.fromisoformat(value) if isinstance(value, str) else None
    except ValueError:
        stamp = None
    if stamp is None or stamp.tzinfo is None or stamp.utcoffset() is None:
        raise ApprovalProtocolError("consumption time is missing or invalid")
    if any(key in payload for key in ("allow", "deny", "effect")):
        raise ApprovalProtocolError("consumption evidence is not a decision")
    return value


class ExecutionConfirmationClient(ApprovalHTTPClient):
    def consume_exact(
        self, approval_id: str, request_digest: str, decision_id: str
    ) -> dict[str, Any]:
        """One CAS call, with exact response validation and no automatic retry."""
        _inputs(approval_id, request_digest, decision_id)
        result = self._request(
            "POST", f"/v1/approvals/{quote(approval_id, safe='')}/consume",
            {"request_digest": request_digest, "decision_id": decision_id},
        )
        if result.get("approval_id") != approval_id or result.get("status") != "consumed":
            raise ApprovalProtocolError("exact consumption was not confirmed")
        _evidence(result, request_digest, decision_id)
        if type(result.get("idempotent")) is not bool:
            raise ApprovalProtocolError("idempotent must be an explicit boolean")
        return {"approval_id": approval_id, "status": "consumed",
                "request_digest": request_digest, "decision_id": decision_id,
                "consumed_at": result["consumed_at"], "idempotent": result["idempotent"]}

    def confirm_consumption(
        self, approval_id: str, request_digest: str, decision_id: str, *,
        expected_binding: dict[str, Any], expected_pdp_digest: str,
    ) -> dict[str, str]:
        """Read the authoritative object; return historical facts only.

        Approval entries may be present on the wire but are not republished.
        This method is safe for reconciliation after an unknown CAS outcome;
        reconciliation cannot authorize retrying protected backend work.
        """
        _inputs(approval_id, request_digest, decision_id)
        if not isinstance(expected_pdp_digest, str) or not DIGEST_RE.fullmatch(expected_pdp_digest):
            raise ApprovalProtocolError("exact approval-free PDP digest is required")
        try:
            expected = binding_digest(expected_binding)
        except (TypeError, ValueError, KeyError):
            raise ApprovalProtocolError("expected act binding is invalid") from None
        obj = self._request("GET", f"/v1/approvals/{quote(approval_id, safe='')}")
        if obj.get("id") != approval_id or obj.get("status") != "consumed":
            raise ApprovalProtocolError("durable consumption was not confirmed")
        if any(key in obj for key in ("allow", "deny", "effect")):
            raise ApprovalProtocolError("consumption object is not a decision")
        served = obj.get("binding")
        try:
            same_act = (
                isinstance(served, dict)
                and binding_digest(served) == expected
                and served.get("digest") == expected
            )
        except (TypeError, ValueError, KeyError):
            same_act = False
        if not same_act or served.get("pdp_digest") != expected_pdp_digest:
            raise ApprovalProtocolError("consumed approval binding does not match")
        if served.get("pdp_path") is not True or served.get("human_control") is not True:
            raise ApprovalProtocolError("production execution requires declared PDP and human control")
        consumption = obj.get("consumption")
        if not isinstance(consumption, dict):
            raise ApprovalProtocolError("durable consumption evidence is missing")
        consumed_at = _evidence(consumption, request_digest, decision_id)
        return {"approval_id": approval_id, "request_digest": request_digest,
                "decision_id": decision_id, "consumed_at": consumed_at}