secrets-engine/tests/test_approval_claim.py

"""approval-engine approval-claim consumer (step 1 of GH-DEC-2026-003).

Covers the published "Required verification" list in
approval-engine/docs/approval-claim.md. The claim is a fact, never a decision.
"""
import copy
from datetime import datetime, timedelta, timezone

import pytest

from secrets_engine.approval_claim import (
    binding_from_check_request,
    claim_binding_digest,
    validate_approval_claim,
)
from secrets_engine.errors import DecisionError

APPROVAL_ID = "3d1c0a8e-6b7f-4c21-9a0e-1f2b3c4d5e6f"


def _binding():
    return {
        "action": "secrets.kv.destroy",
        "target": {"id": "lane-openbao-root", "stage": "prod"},
        "actor": "agt-secrets-engine",
        "principal": "bernd",
        "purpose": "rotate-exposed-key",
    }


def _claim(**over):
    now = datetime.now(timezone.utc)
    binding = dict(_binding())
    binding["digest"] = claim_binding_digest(**_binding())
    claim = {
        "schema_version": "0.1",
        "kind": "approval-claim",
        "issuer": "approval-engine",
        "approval_id": APPROVAL_ID,
        "state": "valid",
        "valid_now": True,
        "consumed": False,
        "binding": binding,
        "freshness": {
            "observed_at": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
            "ttl_seconds": 30,
            "not_after": (now + timedelta(seconds=30)).strftime("%Y-%m-%dT%H:%M:%SZ"),
        },
        "validity": {
            "not_before": (now - timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ"),
            "expires_at": (now + timedelta(hours=3)).strftime("%Y-%m-%dT%H:%M:%SZ"),
        },
        "reason_code": "ok",
    }
    claim.update(over)
    return claim


def _validate(claim, **kw):
    kw.setdefault("approval_id", APPROVAL_ID)
    kw.setdefault("expected_binding_digest", claim_binding_digest(**_binding()))
    return validate_approval_claim(claim, **kw)


def test_valid_claim_passes():
    assert _validate(_claim())["approval_id"] == APPROVAL_ID


def test_binding_digest_is_canonical_sorted_json():
    """Sorted keys at every level, no insignificant whitespace."""
    digest = claim_binding_digest(**_binding())
    assert digest.startswith("sha256:") and len(digest) == 71
    shuffled = {
        "purpose": "rotate-exposed-key",
        "actor": "agt-secrets-engine",
        "target": {"stage": "prod", "id": "lane-openbao-root"},
        "action": "secrets.kv.destroy",
        "principal": "bernd",
    }
    assert claim_binding_digest(**shuffled) == digest


def test_claim_digest_is_not_the_flex_auth_request_digest():
    """The two digest functions are different by contract and must not be mixed."""
    from secrets_engine.authorization import request_digest

    request = {
        "subject": {"id": "agt-secrets-engine"},
        "action": "secrets.kv.destroy",
        "resource": {"id": "lane-openbao-root", "type": "t", "system": "s"},
        "context": {"purpose": "rotate-exposed-key"},
    }
    assert claim_binding_digest(**binding_from_check_request(request)) != request_digest(
        request
    )


def test_check_request_maps_onto_claim_binding():
    request = {
        "subject": {"id": "user:alice", "attributes": {"principal": "bernd"}},
        "action": "deactivate",
        "resource": {"id": "catalog:x", "type": "secret-catalog-lane"},
        "context": {"purpose": "contract-test"},
    }
    mapped = binding_from_check_request(request)
    assert mapped["actor"] == "user:alice"
    assert mapped["principal"] == "bernd"
    assert mapped["purpose"] == "contract-test"
    assert mapped["target"] == request["resource"]


def test_principal_falls_back_to_actor_when_absent():
    request = {
        "subject": {"id": "user:alice"},
        "action": "deactivate",
        "resource": {"id": "catalog:x"},
        "context": {"purpose": "p"},
    }
    assert binding_from_check_request(request)["principal"] == "user:alice"


def test_pdp_digest_is_preferred_when_the_issuer_recorded_one():
    claim = _claim()
    claim["binding"]["pdp_digest"] = "sha256:" + "a" * 64
    claim["binding"]["pdp_path"] = True
    claim["binding"]["digest"] = "sha256:" + "b" * 64  # native no longer matches
    assert _validate(claim, expected_pdp_digest="sha256:" + "a" * 64)


@pytest.mark.parametrize(
    ("mutation", "match"),
    [
        (lambda c: c.update(issuer="state-hub"), "issuer is not approval-engine"),
        (lambda c: c.update(kind="decision"), "kind is not approval-claim"),
        (lambda c: c.update(schema_version="0.2"), "schema version"),
        (lambda c: c.update(valid_now=False, reason_code="revoked"), "not valid now"),
        (lambda c: c.update(consumed=True), "already consumed"),
        (lambda c: c.update(reason_code="superseded"), "reason code is not ok"),
        (lambda c: c.update(effect="allow"), "a claim is not a decision"),
        (lambda c: c.update(approval_id="00000000-0000-0000-0000-000000000000"),
         "different approval object"),
        (lambda c: c["binding"].update(digest="sha256:" + "f" * 64),
         "binding digest does not match"),
    ],
)
def test_invalid_claims_fail_closed(mutation, match):
    claim = _claim()
    mutation(claim)
    with pytest.raises(DecisionError, match=match):
        _validate(claim)


def test_stale_observation_is_rejected_even_when_still_valid():
    """Stale is not the same as valid_now false; the object may still be good."""
    claim = _claim()
    past = datetime.now(timezone.utc) - timedelta(seconds=1)
    claim["freshness"]["not_after"] = past.strftime("%Y-%m-%dT%H:%M:%SZ")
    assert claim["valid_now"] is True
    with pytest.raises(DecisionError, match="stale"):
        _validate(claim)


def test_expired_validity_window_fails_closed():
    claim = _claim()
    past = datetime.now(timezone.utc) - timedelta(minutes=1)
    claim["validity"]["expires_at"] = past.strftime("%Y-%m-%dT%H:%M:%SZ")
    with pytest.raises(DecisionError, match="validity window has expired"):
        _validate(claim)


def test_not_yet_valid_fails_closed():
    claim = _claim()
    future = datetime.now(timezone.utc) + timedelta(minutes=5)
    claim["validity"]["not_before"] = future.strftime("%Y-%m-%dT%H:%M:%SZ")
    with pytest.raises(DecisionError, match="not yet valid"):
        _validate(claim)


def test_comparison_requires_an_expected_digest():
    with pytest.raises(DecisionError, match="requires an expected digest"):
        validate_approval_claim(_claim(), approval_id=APPROVAL_ID)