secrets-engine/src/secrets_engine/approval_claim.py

"""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: it is expressed in flex-auth's vocabulary, which is
    # the one the caller actually proposed. The native digest is over
    # approval-engine's own vocabulary and is only comparable when the caller
    # supplies a binding built in that vocabulary (see the note in
    # resolve_consume_binding about the missing mapping).
    if expected_pdp_digest:
        if not pdp:
            raise DecisionError(
                "approval claim records no pdp_digest, and no published mapping "
                "exists between approval-engine and secrets-engine action/target "
                "vocabularies; the claim cannot be tied to this exact action"
            )
        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