tegwick
eece796b8b
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
100 lines
5.2 KiB
Python
100 lines
5.2 KiB
Python
"""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}
|