secrets-engine/src/secrets_engine/approval_consume.py

"""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, field
from pathlib import Path
from typing import Any, Callable
from urllib.error import HTTPError, URLError
from urllib.request import Request

from secrets_engine.approval_auth import approval_auth_configured, approval_token, credential_urlopen
from secrets_engine.execution_contract import ExecutionContract, require_execution_deadline
from secrets_engine.approval_claim import observe_pdp_approval_claim, validate_approval_claim
from secrets_engine.catalog import human_control_required
from secrets_engine.decision_check import check_decision, require_supported_pdp_address
from secrets_engine.decision_signature import load_trust, strict_json
from secrets_engine.authorization import (
    build_action_request,
    validate_decision_envelope,
    wait_for_decision_start,
)
from secrets_engine.errors import DecisionError
from secrets_engine.application_time import read_window, recheck_binding
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 = ""
    human_control: bool = False
    decision_not_before: str = ""
    decision_expires_at: str = ""


@dataclass(frozen=True)
class ApprovalObservation:
    """Fresh approval input awaiting PDP correspondence; never a consume binding."""

    approval_id: str
    request: dict[str, Any] = field(repr=False)
    claim: dict[str, Any] = field(repr=False)


@dataclass(frozen=True)
class AuthorizedAction:
    """A validated claim and decision for one exact proposed action.

    Holding this is not authority to act: GH-DEC-2026-003 still requires a
    successful CAS consume before the OpenBao call.
    """

    binding: ConsumeBinding
    decision_id: str
    expires_at: str
    # Detached original served artifacts for the narrow task-execution handoff.
    artifacts: str = field(default="", repr=False, compare=False)

    @property
    def id(self) -> str:
        """The actual PDP decision id for shared CLI evidence/plan callers."""
        return self.decision_id

    def as_evidence(self) -> dict[str, object]:
        return {
            "authorization_decision_id": self.decision_id,
            "authorization_expires_at": self.expires_at,
            "request_digest": self.binding.request_digest,
            "approval_human_control": self.binding.human_control,
        }


@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 _authorization_id(entry: Any, decision: Any) -> str:
    """Non-secret approval-engine object id for this lane, or "" if unbound.

    flex-auth: ``ActionAuthorization.id`` is the approval-engine object UUID.
    It is never the State Hub decision UUID, so it is not inferred from one.
    """
    approval = getattr(entry, "approval", None) or {}
    if isinstance(approval, dict):
        declared = str(approval.get("authorization_id", "") or "").strip()
        if declared:
            return declared
    for attr in ("authorization_id", "action_authorization_id"):
        served = str(getattr(decision, attr, "") or "").strip()
        if served:
            return served
    return ""


def _request_purpose(entry: Any) -> str:
    """Declared purpose for the request context. Never invented at call time."""
    approval = getattr(entry, "approval", None) or {}
    if isinstance(approval, dict):
        declared = str(approval.get("purpose", "") or "").strip()
        if declared:
            return declared
    for consumer in getattr(entry, "consumers", None) or []:
        if isinstance(consumer, dict):
            declared = str(consumer.get("purpose", "") or "").strip()
            if declared:
                return declared
    return ""


def _expected_request(
    cfg: Any,
    entry: Any,
    action: str,
    *,
    fields: tuple[str, ...] = (),
    policy_targets: tuple[str, ...] = (),
    auth_targets: tuple[str, ...] = (),
    execution_contract: ExecutionContract | None = None,
) -> dict[str, Any]:
    """Build the exact CheckRequest both steps must agree on.

    Steps 1 and 2 must describe the same proposed action or the digests cannot
    correspond, so neither builds its own.
    """
    subject_id = str(getattr(cfg, "authorization_subject_id", "") or "")
    subject_type = str(getattr(cfg, "authorization_subject_type", "") or "")
    if not subject_id or not subject_type:
        raise DecisionError(
            "authorization join requires SECRETS_ENGINE_AUTHORIZATION_SUBJECT_ID "
            "and _SUBJECT_TYPE; the PEP must not assert an unnamed subject"
        )
    purpose = _request_purpose(entry)
    if not purpose:
        raise DecisionError(
            "authorization join requires a declared approval/consumer purpose"
        )
    return build_action_request(
        entry,
        action,
        subject_id=subject_id,
        subject_type=subject_type,
        purpose=purpose,
        fields=fields,
        policy_targets=policy_targets,
        auth_targets=auth_targets,
        execution_contract=execution_contract,
    )


def fetch_approval_claim(
    *,
    base_url: str,
    token_file: Path | None = None,
    token_provider: Callable[[], str] | None = None,
    authorization_id: str,
    timeout_seconds: float = 3,
    opener: Callable[..., Any] = credential_urlopen,
) -> dict[str, Any]:
    """GET /v1/approvals/{id}/claim (PIP). Any non-200 fails closed.

    The body is approval-engine's approval-claim, not a flex-auth
    ActionAuthorization -- that object is deferred and was never ratified
    (GH-DEC-2026-005 / FLEX-DEC-2026-006).
    """
    if not base_url or not base_url.startswith(("http://", "https://")):
        raise DecisionError("approval-engine claim URL is missing or invalid")
    ident = authorization_id.strip()
    if not ident or "/" in ident or any(ch.isspace() for ch in ident):
        raise DecisionError("approval claim requires a concrete authorization id")
    token = _request_token(token_file, token_provider)
    request = Request(
        base_url.rstrip("/") + f"/v1/approvals/{ident}/claim",
        method="GET",
    )
    request.add_header("Authorization", f"Bearer {token}")
    request.add_header("Accept", "application/json")
    try:
        with opener(request, timeout=timeout_seconds) as response:
            if getattr(response, "status", 200) != 200:
                raise DecisionError("approval claim did not return a claim")
            payload = json.loads(response.read(_MAX_BODY).decode("utf-8"))
    except HTTPError as e:
        raise DecisionError(f"approval claim refused: {_status_message(e.code)}") from e
    except URLError as e:
        raise DecisionError("approval-engine is unreachable for claim") from e
    except json.JSONDecodeError as e:
        raise DecisionError("approval claim returned a non-JSON body") from e
    if not isinstance(payload, dict):
        raise DecisionError("approval claim returned a non-object body")
    return payload


def resolve_approval_observation(
    cfg: Any,
    entry: Any,
    action: str,
    decision: Any,
    *,
    fields: tuple[str, ...] = (),
    policy_targets: tuple[str, ...] = (),
    auth_targets: tuple[str, ...] = (),
    execution_contract: ExecutionContract | None = None,
    opener: Callable[..., Any] | None = None,
) -> ApprovalObservation | None:
    """Observe a fresh fact and prepare the claim-bearing request (step 1).

    Missing serving coordinates return None for undeclared lanes. A declared
    human control raises instead, so no stage/demo fallback can discharge it.
    Correspondence and a consume binding require authorize_action's PDP step.
    """
    execution_request = None
    if action == "execute-ccr" or execution_contract is not None:
        execution_request = _expected_request(
            cfg, entry, action, fields=fields, policy_targets=policy_targets,
            auth_targets=auth_targets, execution_contract=execution_contract,
        )
        # The caller's descriptor is not authenticated principal/owner evidence.
        # This requirement cannot disappear on an unserved/demo lane.
    base_url = str(getattr(cfg, "approval_url", "") or "")
    auth_configured = approval_auth_configured(cfg)
    authorization_id = _authorization_id(entry, decision)
    required_human_control = action == "execute-ccr" or human_control_required(entry)
    if not base_url or not auth_configured or not authorization_id:
        if required_human_control:
            raise DecisionError("human_control requires a served approval claim and exact action binding")
        return None

    expected_request = execution_request or _expected_request(
        cfg, entry, action,
        fields=fields, policy_targets=policy_targets, auth_targets=auth_targets,
        execution_contract=execution_contract,
    )

    require_execution_deadline(expected_request, time_window=read_window(cfg))
    claim = fetch_approval_claim(
        base_url=base_url,
        token_provider=lambda: approval_token(cfg, scope="approval:read"),
        authorization_id=authorization_id,
        opener=opener or credential_urlopen,
    )
    observe_pdp_approval_claim(claim, approval_id=authorization_id,
                               require_human_control=required_human_control, time_window=read_window(cfg))
    require_execution_deadline(expected_request, time_window=read_window(cfg),
                               bounds=(claim["validity"]["expires_at"],))
    expected_request["context"]["approval"] = claim
    return ApprovalObservation(authorization_id, expected_request, claim)


def authorize_action(
    cfg: Any,
    entry: Any,
    action: str,
    decision: Any = None,
    *,
    fields: tuple[str, ...] = (),
    policy_targets: tuple[str, ...] = (),
    auth_targets: tuple[str, ...] = (),
    execution_contract: ExecutionContract | None = None,
    opener: Callable[..., Any] | None = None,
    pdp_opener: Callable[..., Any] | None = None,
) -> AuthorizedAction | None:
    """Run steps 1 and 2 for one proposed action, or return None if unserved.

    Step 1 validates the approval-claim; step 2 obtains and validates the
    flex-auth DecisionEnvelope. Returning None means no serving path is
    configured at all, which leaves production fail-closed. A configured but
    failing path raises: a partial deployment must not read as an absent one.
    """
    serving = (getattr(cfg, "approval_url", "") and approval_auth_configured(cfg)
               and _authorization_id(entry, decision))
    if serving or (getattr(cfg, "pdp_url", "") and getattr(cfg, "pdp_token_file", None)):
        if not getattr(cfg, "pdp_url", "") or not getattr(cfg, "pdp_token_file", None):
            raise DecisionError("production action requires configured access-engine URL/token and responder trust")
        require_supported_pdp_address(str(cfg.pdp_url))
        load_trust(getattr(cfg, "pdp_trust_file", None))
        verifier = getattr(cfg, "pdp_verifier", None)
        if verifier is None or not Path(verifier).is_absolute() or not Path(verifier).is_file():
            raise DecisionError("access-engine signature verifier is not installed")
    observation = resolve_approval_observation(
        cfg, entry, action, decision,
        fields=fields,
        policy_targets=policy_targets,
        auth_targets=auth_targets,
        execution_contract=execution_contract,
        opener=opener,
    )
    if observation is None:
        return None

    pdp_url = str(getattr(cfg, "pdp_url", "") or "")
    pdp_token = getattr(cfg, "pdp_token_file", None)
    if not pdp_url or not pdp_token:
        raise DecisionError(
            "production action requires an access-engine decision; "
            "SECRETS_ENGINE_PDP_URL / _PDP_TOKEN_FILE are unset"
        )
    package = str(getattr(cfg, "authorization_policy_package", "") or "")
    version = str(getattr(cfg, "authorization_policy_version", "") or "")
    if not package or not version:
        raise DecisionError(
            "authorization join requires an explicitly configured policy "
            "package/version pin; the reserved coordinate is a reservation, "
            "not a publication, and must not be used as a default"
        )

    expected_request = observation.request
    envelope = check_decision(
        base_url=pdp_url,
        token_file=Path(pdp_token),
        request=expected_request,
        opener=pdp_opener,
        trust_file=getattr(cfg, "pdp_trust_file", None),
        verifier=getattr(cfg, "pdp_verifier", None),
    )
    if not getattr(cfg, "clock_trust_file", None):
        wait_for_decision_start(envelope)
    window = read_window(cfg)
    validated = validate_decision_envelope(
        envelope,
        expected_request,
        accepted_policy_packages={package},
        accepted_policy_versions={version},
        expected_approval_binding_digest=observation.claim["binding"]["pdp_digest"],
        time_window=window,
    )
    # Recheck freshness after the network call before issuing a consume binding.
    validate_approval_claim(
        observation.claim, approval_id=observation.approval_id,
        expected_pdp_digest=envelope["binding"]["approval_binding_digest"],
        require_human_control=expected_request["context"].get("human_control") is True,
        time_window=window,
    )
    require_execution_deadline(expected_request, time_window=window,
                               bounds=(observation.claim["validity"]["expires_at"], validated.expires_at))
    if validated.action != action:
        raise DecisionError("access-engine decision does not bind this action")
    return AuthorizedAction(
        binding=ConsumeBinding(
            approval_id=observation.approval_id,
            request_digest=validated.request_digest,
            decision_id=validated.decision_id,
            human_control=observation.claim["binding"].get("human_control") is True,
            decision_not_before=envelope["lifetime"].get("not_before") or "",
            decision_expires_at=validated.expires_at,
        ),
        decision_id=validated.decision_id,
        expires_at=validated.expires_at,
        artifacts=json.dumps({"request":expected_request,"claim":observation.claim,"envelope":envelope,
                              "authenticated_envelope_json":getattr(envelope,"authenticated_json","")},
                             sort_keys=True,separators=(",",":"),allow_nan=False),
    )


def consume_approval(
    *,
    base_url: str,
    token_file: Path | None = None,
    token_provider: Callable[[], str] | None = None,
    binding: ConsumeBinding,
    timeout_seconds: float = 3,
    opener: Callable[..., Any] = credential_urlopen,
    strict_response: bool = False,
) -> 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 = _request_token(token_file, token_provider)
    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 = (strict_json if strict_response else json.loads)(raw.decode("utf-8"))
    except (UnicodeDecodeError, ValueError, RecursionError) 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")
    if strict_response:
        from datetime import datetime
        if (payload.get("approval_id") != approval_id
                or payload.get("decision_id") != binding.decision_id
                or payload.get("idempotent") is not False
                or any(key in payload for key in ("effect","allow","deny","decision"))):
            raise DecisionError("strict task consumption correspondence refused")
        try:
            stamp = datetime.fromisoformat(payload["consumed_at"].replace("Z","+00:00"))
            if stamp.tzinfo is None or stamp.utcoffset() is None:
                raise ValueError("timezone required")
        except (KeyError,TypeError,AttributeError,ValueError):
            raise DecisionError("strict task consumption timestamp refused") from None
    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 production or an explicitly human-controlled action.

    Undeclared build/test and the unsafe-demo exception retain their existing
    behavior. Declared human controls require a verified binding and consume
    regardless of stage/demo; missing inputs cannot fall back to a lane review.
    """
    required_human_control = human_control_required(entry)
    if required_human_control:
        if binding is None or binding.human_control is not True:
            raise DecisionError("human_control requires an observed declared-human consume binding")
    else:
        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 "")
    auth_configured = approval_auth_configured(cfg)
    if not base_url:
        raise DecisionError(
            "production OpenBao call requires approval-engine consume; "
            "SECRETS_ENGINE_APPROVAL_URL is unset"
        )
    if not auth_configured:
        raise DecisionError(
            "production OpenBao call requires approval-engine consume; "
            "SECRETS_ENGINE_APPROVAL_TOKEN_FILE or _CLIENT_SECRET_FILE is unset"
        )
    recheck_binding(cfg, binding)
    consumed = consume_approval(
        base_url=base_url,
        token_provider=lambda: approval_token(cfg, scope="approval:consume"),
        binding=binding,
        opener=opener or credential_urlopen,
    )
    if evidence is not None and hasattr(evidence, "mark_consumed"):
        evidence.mark_consumed(consumed)
    recheck_binding(cfg, binding)
    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"


def _request_token(
    token_file: Path | None, token_provider: Callable[[], str] | None,
) -> str:
    if (token_file is None) == (token_provider is None):
        raise DecisionError("approval request requires exactly one credential provider")
    token = (
        token_provider() if token_provider is not None
        else read_strict_token_file(Path(token_file), purpose="approval credential")
    )
    if not isinstance(token, str) or not token or any(ch.isspace() for ch in token):
        raise DecisionError("approval credential is invalid")
    return token