"""PEP consume-before-side-effect client (GH-DEC-2026-003). The PEP that is about to cause a protected OpenBao write MUST obtain a successful approval-engine CAS consume first. Holding a claim or an ALLOW is not authority to act. Conflict, unavailability, or a missing binding means do not call OpenBao. This module does not render an authorization decision. The consume response is mutation evidence, never a permission. """ from __future__ import annotations import json import re from dataclasses import dataclass from pathlib import Path from typing import Any, Callable from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen from secrets_engine.errors import DecisionError from secrets_engine.openbao import read_strict_token_file from secrets_engine.pep_stance import demo_exception_enabled DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$") _MAX_BODY = 256 * 1024 @dataclass(frozen=True) class ConsumeBinding: """Inputs the PEP presents to approval-engine consume.""" approval_id: str request_digest: str decision_id: str = "" @dataclass(frozen=True) class ConsumedApproval: """Non-secret confirmation that consume succeeded for this request.""" approval_id: str request_digest: str decision_id: str = "" idempotent: bool = False consumed_at: str = "" def as_evidence(self) -> dict[str, object]: payload: dict[str, object] = { "approval_consumed": True, "approval_id": self.approval_id, "request_digest": self.request_digest, "approval_consume_idempotent": self.idempotent, } if self.decision_id: payload["decision_id"] = self.decision_id if self.consumed_at: payload["approval_consumed_at"] = self.consumed_at return payload def resolve_consume_binding( _cfg: Any, _entry: Any, _action: str, _decision: Any, ) -> ConsumeBinding | None: """Return the served consume binding, or None if it is not available. The durable ActionAuthorization / approval serving path is still external (SECRETS-WP-0007-T04 / SECRETS-WP-0008-T02). Tests may replace this hook. """ return None def consume_approval( *, base_url: str, token_file: Path, binding: ConsumeBinding, timeout_seconds: float = 3, opener: Callable[..., Any] = urlopen, ) -> ConsumedApproval: """POST /v1/approvals/{id}/consume. Fail closed on anything but confirmed use.""" if not base_url or not base_url.startswith(("http://", "https://")): raise DecisionError("approval-engine consume URL is missing or invalid") approval_id = binding.approval_id.strip() if not approval_id or "/" in approval_id or any(ch.isspace() for ch in approval_id): raise DecisionError("approval consume requires a concrete approval id") if not DIGEST_RE.fullmatch(binding.request_digest): raise DecisionError("approval consume requires the canonical request digest") token = read_strict_token_file(Path(token_file), purpose="approval consume credential") body: dict[str, str] = {"request_digest": binding.request_digest} if binding.decision_id: body["decision_id"] = binding.decision_id encoded = json.dumps(body).encode("utf-8") request = Request( base_url.rstrip("/") + f"/v1/approvals/{approval_id}/consume", data=encoded, method="POST", headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Accept": "application/json", }, ) try: response = opener(request, timeout=timeout_seconds) try: status = int(response.getcode()) raw = response.read(_MAX_BODY + 1) finally: response.close() except HTTPError as exc: status = int(getattr(exc, "code", 0) or 0) try: exc.read(_MAX_BODY) except Exception: pass raise DecisionError(_status_message(status)) from None except (URLError, TimeoutError, OSError): raise DecisionError( "approval-engine unreachable; OpenBao must not be called" ) from None if status != 200 or len(raw) > _MAX_BODY: raise DecisionError(_status_message(status if status != 200 else 502)) try: payload = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise DecisionError("approval consume returned invalid JSON") from exc if not isinstance(payload, dict): raise DecisionError("approval consume returned invalid payload") if payload.get("status") != "consumed": raise DecisionError("approval consumption was not confirmed") if payload.get("request_digest") != binding.request_digest: raise DecisionError("approval consume digest does not match the request") decision_id = payload.get("decision_id") if decision_id is not None and ( not isinstance(decision_id, str) or not decision_id ): raise DecisionError("approval consume returned an invalid decision id") consumed_at = payload.get("consumed_at") if consumed_at is not None and not isinstance(consumed_at, str): raise DecisionError("approval consume returned an invalid consumed_at") return ConsumedApproval( approval_id=str(payload.get("approval_id") or approval_id), request_digest=binding.request_digest, decision_id=decision_id or binding.decision_id, idempotent=bool(payload.get("idempotent")), consumed_at=consumed_at or "", ) def require_production_consume( cfg: Any, entry: Any, *, binding: ConsumeBinding | None, evidence: Any = None, opener: Callable[..., Any] | None = None, ) -> ConsumedApproval | None: """CAS-consume before a production OpenBao call. No-op off the prod path. Build/test remain fail-open relative to approval-engine. The three-factor unsafe-demo exception is not a consume path. Missing binding, URL, or credential fail closed so a stance bypass cannot reach OpenBao. """ if getattr(entry, "stage", "") != "prod": return None if demo_exception_enabled(cfg): return None if binding is None: raise DecisionError( "production OpenBao call requires CAS consume of an approval " "after an access-engine ALLOW; no durable consume binding is served" ) base_url = str(getattr(cfg, "approval_url", "") or "") token_file = getattr(cfg, "approval_token_file", None) if not base_url: raise DecisionError( "production OpenBao call requires approval-engine consume; " "SECRETS_ENGINE_APPROVAL_URL is unset" ) if not token_file: raise DecisionError( "production OpenBao call requires approval-engine consume; " "SECRETS_ENGINE_APPROVAL_TOKEN_FILE is unset" ) consumed = consume_approval( base_url=base_url, token_file=Path(token_file), binding=binding, opener=opener or urlopen, ) if evidence is not None and hasattr(evidence, "mark_consumed"): evidence.mark_consumed(consumed) return consumed def _status_message(status: int) -> str: if status == 409: return "approval consume conflict; OpenBao must not be called" if status == 404: return "approval not found; OpenBao must not be called" if status in {401, 403}: return "approval consume unauthorized; OpenBao must not be called" if status == 503: return "approval-engine unavailable; OpenBao must not be called" if status == 0: return "approval-engine unreachable; OpenBao must not be called" return "approval consume failed; OpenBao must not be called"