"""approval-engine approval-claim consumer (GH-DEC-2026-005, step 1). The claim is a *fact about an approval object*, never a decision. It carries no `effect`, `allow`, or `deny`, and holding one with ``valid_now: true`` is not authority to act -- it is one input the decision point weighs. Contract: ``approval-engine/docs/approval-claim.md`` (schema 0.1). The native binding digest here is NOT the flex-auth CheckRequest digest: it is taken over ``{action, actor, principal, purpose, target}``. The two are deliberately different functions and must not be compared to each other. """ from __future__ import annotations import hashlib import json from datetime import datetime, timezone from typing import Any from secrets_engine.errors import DecisionError SCHEMA_VERSION = "0.1" KIND = "approval-claim" ISSUER = "approval-engine" VALID_REASON = "ok" def claim_binding_digest( *, action: str, actor: str, principal: str, purpose: str, target: dict[str, Any], ) -> str: """sha256 over canonical JSON of the approval-engine binding. Sorted keys at every level, no insignificant whitespace. Wrong action, target, or scope changes this JSON and therefore the digest, which is what stops a decision rendered for request R being replayed for request R'. """ canonical = json.dumps( { "action": action, "actor": actor, "principal": principal, "purpose": purpose, "target": target, }, sort_keys=True, separators=(",", ":"), ensure_ascii=False, ) return "sha256:" + hashlib.sha256(canonical.encode("utf-8")).hexdigest() def binding_from_check_request(request: dict[str, Any]) -> dict[str, Any]: """Map a flex-auth CheckRequest onto the claim binding fields. Mapping is published in ``approval-claim.md``: target is the resource object, actor is ``subject.id``, principal is ``subject.attributes.principal`` when present and ``subject.id`` otherwise, purpose is ``context.purpose``. """ subject = request.get("subject") or {} if not isinstance(subject, dict): raise DecisionError("check request subject must be an object") attributes = subject.get("attributes") or {} principal = "" if isinstance(attributes, dict): principal = str(attributes.get("principal", "") or "") actor = str(subject.get("id", "") or "") context = request.get("context") or {} purpose = "" if isinstance(context, dict): purpose = str(context.get("purpose", "") or "") resource = request.get("resource") if not isinstance(resource, dict): raise DecisionError("check request resource must be an object") return { "action": str(request.get("action", "") or ""), "actor": actor, "principal": principal or actor, "purpose": purpose, "target": resource, } def _parse_time(value: object, name: str) -> datetime: if not isinstance(value, str) or not value: raise DecisionError(f"approval claim {name} must be a timestamp") text = value.strip().replace("Z", "+00:00") try: parsed = datetime.fromisoformat(text) except ValueError as e: raise DecisionError(f"approval claim {name} is not a valid timestamp") from e if parsed.tzinfo is None: raise DecisionError(f"approval claim {name} must carry a timezone") return parsed.astimezone(timezone.utc) def validate_approval_claim( claim: object, *, approval_id: str, expected_binding_digest: str = "", expected_pdp_digest: str = "", now: datetime | None = None, ) -> dict[str, Any]: """Run the published consumer checks. Any failure means do not act. Implements ``approval-claim.md`` "Required verification": issuer, valid_now, not consumed, binding digest match (native or PDP), freshness, reason_code. The distinct-approver threshold is folded into ``valid_now`` by the issuer; the claim does not expose approver entries, so it cannot be re-checked here. """ if not isinstance(claim, dict): raise DecisionError("approval claim must be an object") if claim.get("schema_version") != SCHEMA_VERSION: raise DecisionError("unsupported approval claim schema version") if claim.get("kind") != KIND: raise DecisionError("approval claim kind is not approval-claim") if claim.get("issuer") != ISSUER: raise DecisionError("approval claim issuer is not approval-engine") for forbidden in ("effect", "decision", "allow", "deny"): if forbidden in claim: raise DecisionError( f"approval claim carries '{forbidden}'; a claim is not a decision" ) served_id = str(claim.get("approval_id", "") or "") if not served_id or served_id != approval_id: raise DecisionError("approval claim is for a different approval object") if claim.get("valid_now") is not True: raise DecisionError( "approval claim is not valid now " f"(reason_code={claim.get('reason_code', 'unknown')})" ) if claim.get("consumed") is not False: raise DecisionError("approval claim is already consumed") if claim.get("reason_code") != VALID_REASON: raise DecisionError("approval claim reason code is not ok") binding = claim.get("binding") if not isinstance(binding, dict): raise DecisionError("approval claim binding must be an object") native = str(binding.get("digest", "") or "") pdp = str(binding.get("pdp_digest", "") or "") # Prefer the PDP digest when the issuer recorded one at issue time. if expected_pdp_digest and pdp: if pdp != expected_pdp_digest: raise DecisionError("approval claim pdp digest does not match the request") elif expected_binding_digest: if native != expected_binding_digest: raise DecisionError("approval claim binding digest does not match the request") else: raise DecisionError("approval claim comparison requires an expected digest") current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) freshness = claim.get("freshness") if not isinstance(freshness, dict): raise DecisionError("approval claim freshness must be an object") if _parse_time(freshness.get("not_after"), "freshness.not_after") <= current: raise DecisionError("approval claim observation is stale; re-fetch") validity = claim.get("validity") if not isinstance(validity, dict): raise DecisionError("approval claim validity must be an object") if _parse_time(validity.get("expires_at"), "validity.expires_at") <= current: raise DecisionError("approval claim validity window has expired") if validity.get("not_before") is not None: if _parse_time(validity.get("not_before"), "validity.not_before") > current: raise DecisionError("approval claim is not yet valid") return claim