secrets-engine/tests/test_decision_replay.py

"""Digest join verified against real flex-auth DecisionEnvelopes (FLEX-WP-0021-T03).

These replace a hand-maintained pin that was computed *including* the request
`id`. docs/canonical-request-digest.md excludes `id`, `policy_version` and
`caring_context` from the hashed material, and both real envelopes confirm it:
a digest computed over the old material matches no issued decision, which would
have failed closed against every correct allow.

Both fixtures are required. `provenance.input_claim_digests.context` appears
only when the request carries a non-empty context, so a validator asserting it
is always present passes `destroy` and fails `rotate`.
"""
import json
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest

from types import SimpleNamespace

from secrets_engine.authorization import (
    REQUEST_TENANT,
    approval_binding_digest,
    build_action_request,
    digest_material,
    digest_material,
    request_digest,
    validate_decision_envelope,
)
from secrets_engine.errors import DecisionError

FIXTURES = Path(__file__).parent / "fixtures" / "flex-auth-replay"

PACKAGE_DIGEST = "sha256:bd11c5fe77ce6439c65fea225ad6b71d2110efc5e7b5bc9b499c59cd0a53b8b4"
SNAPSHOT_DIGEST = "sha256:f5a309bc0b36721fd6d9ad7f53eb21222162bc2eac62a0ab0802a9a1d51340bb"

CASES = {
    "rotate": {
        "file": "decision_rotate.json",
        "digest": "sha256:de67324f54187055307a833235f83ced9fcd3a20952a27b3d19493ed39734345",
        "action": "rotate",
        "context_claim_digest": None,
    },
    "destroy": {
        "file": "decision_destroy_dual_control.json",
        "digest": "sha256:c749ee2dc3cdf927a70a3e5b27cff4d97a438d3264153b4b2e3bcacbaf82091a",
        "action": "destroy",
        "context_claim_digest": "sha256:8b73d29ecef286d42e03d2420531d6c45219f325a7ae004c1ecfc781203a2800",
    },
}


def _envelope(name):
    return json.loads((FIXTURES / CASES[name]["file"]).read_text())


def _request_from(envelope):
    """Rebuild the normalized tuple the binding carries.

    Per the contract, a consumer re-hashing the *original unenriched* request
    will not match a decision that turned on registry attributes; the binding is
    the evaluator's statement of what it hashed.
    """
    binding = envelope["binding"]
    request = {"id": envelope["request_id"]}
    for key in ("tenant", "subject", "action", "resource"):
        if binding.get(key) is not None:
            request[key] = binding[key]
    if binding.get("context") is not None:
        request["context"] = binding["context"]
    return request


def _refresh_lifetime(envelope):
    """Lifetime moves with the clock and must never be pinned."""
    now = datetime.now(timezone.utc)
    envelope["lifetime"]["not_before"] = (now - timedelta(minutes=1)).strftime(
        "%Y-%m-%dT%H:%M:%SZ"
    )
    envelope["lifetime"]["expires_at"] = (now + timedelta(minutes=14)).strftime(
        "%Y-%m-%dT%H:%M:%SZ"
    )
    return envelope


@pytest.mark.parametrize("name", list(CASES))
def test_request_digest_matches_the_issued_decision(name):
    envelope = _envelope(name)
    expected = CASES[name]["digest"]
    assert envelope["binding"]["request_digest"] == expected, "fixture drifted"
    assert request_digest(_request_from(envelope)) == expected


@pytest.mark.parametrize("name", list(CASES))
def test_correlation_fields_are_not_hashed(name):
    """id, policy_version and caring_context must not move the digest."""
    request = _request_from(envelope := _envelope(name))
    baseline = request_digest(request)
    assert baseline == envelope["binding"]["request_digest"]
    for field, value in (
        ("id", "check:some-other-correlation-id"),
        ("policy_version", "v99"),
        ("caring_context", {"anything": "here"}),
    ):
        assert request_digest({**request, field: value}) == baseline, field
    stripped = {k: v for k, v in request.items() if k != "id"}
    assert request_digest(stripped) == baseline


@pytest.mark.parametrize("name", list(CASES))
def test_digest_material_is_exactly_the_published_tuple(name):
    material = digest_material(_request_from(_envelope(name)))
    assert set(material) <= {"tenant", "subject", "action", "resource", "context"}
    assert "id" not in material


@pytest.mark.parametrize("name", list(CASES))
def test_real_envelope_validates_against_the_published_package(name):
    envelope = _refresh_lifetime(_envelope(name))
    result = validate_decision_envelope(
        envelope,
        _request_from(envelope),
        accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"},
        accepted_policy_versions={"v2"},
    )
    assert result.action == CASES[name]["action"]
    assert result.subject_id == "secrets-engine"


@pytest.mark.parametrize("name", list(CASES))
def test_provenance_digests_are_pinned(name):
    provenance = _envelope(name)["provenance"]
    assert provenance["policy_package_digest"] == PACKAGE_DIGEST
    assert provenance["registry_snapshot_digest"] == SNAPSHOT_DIGEST
    assert provenance["evaluator"] == "flex-auth/local"
    assert provenance["mode"] == "standalone"


@pytest.mark.parametrize("name", list(CASES))
def test_input_claim_digest_is_present_only_with_a_context(name):
    """The reason two fixtures exist: this field is conditional."""
    provenance = _envelope(name)["provenance"]
    expected = CASES[name]["context_claim_digest"]
    actual = (provenance.get("input_claim_digests") or {}).get("context")
    assert actual == expected


@pytest.mark.parametrize("name", list(CASES))
def test_a_tampered_binding_field_breaks_the_digest(name):
    envelope = _envelope(name)
    request = _request_from(envelope)
    request["action"] = "handoff"
    assert request_digest(request) != envelope["binding"]["request_digest"]


def test_expired_real_envelope_fails_closed():
    """The shipped lifetime is 15m from allow_ttl and has long since passed."""
    with pytest.raises(DecisionError, match="lifetime has expired"):
        envelope = _envelope("rotate")
        validate_decision_envelope(
            envelope,
            _request_from(envelope),
            accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"},
            accepted_policy_versions={"v2"},
        )


def test_embedded_claim_uses_approval_engine_vocabulary_not_ours():
    """The published vocabulary gap, asserted rather than described.

    The dual-control request carries a complete approval-claim in
    context.approval. Its binding speaks approval-engine's language while the
    decision speaks ours, and no mapping between them is published -- which is
    why this engine compares pdp_digest and refuses to derive a native digest
    from its own vocabulary.
    """
    envelope = _envelope("destroy")
    claim = envelope["binding"]["context"]["approval"]
    assert claim["kind"] == "approval-claim"
    assert claim["issuer"] == "approval-engine"
    assert claim["valid_now"] is True
    assert claim["binding"]["action"] == "secrets.kv.destroy"
    assert envelope["binding"]["action"] == "destroy"
    assert claim["binding"]["action"] != envelope["binding"]["action"]
    assert set(claim["binding"]["target"]) == {"id", "stage"}
    assert claim["binding"]["target"]["id"] != envelope["resource"]["id"]


APPROVAL_BINDING_DIGEST = (
    "sha256:fa07becfaa471394d06aee5fa3cd66352bf0cc69ef24900240489684cda8cd56"
)


def test_embedded_claim_pdp_digest_cannot_equal_the_carrying_request_digest():
    """Carrying the claim inside a hashed context makes the two unequal.

    context is part of the digest material, so embedding an approval-claim
    changes the request digest of the very request that carries it. A
    pdp_digest recorded at issue time therefore cannot equal the final digest
    of the dual-control request. This is why comparing pdp_digest against
    request_digest can never pass and would fail destroy closed forever.
    """
    envelope = _envelope("destroy")
    binding = envelope["binding"]
    pdp = binding["context"]["approval"]["binding"]["pdp_digest"]
    request = _request_from(envelope)
    assert request_digest(request) == binding["request_digest"]
    assert pdp != binding["request_digest"]


def test_pdp_digest_equals_the_published_approval_binding_digest():
    """FLEX-DEC-2026-007 closed the circularity, and we reproduce the value.

    ``approval_binding_digest`` is the canonical digest with context.approval
    removed. Recomputing it here from our own canonical implementation is the
    hermetic proof that this engine hashes the same material flex-auth does --
    a pin alone would only assert the constant, not that we can derive it.
    """
    envelope = _envelope("destroy")
    binding = envelope["binding"]
    pdp = binding["context"]["approval"]["binding"]["pdp_digest"]
    assert binding["approval_binding_digest"] == APPROVAL_BINDING_DIGEST
    assert pdp == APPROVAL_BINDING_DIGEST

    assert approval_binding_digest(_request_from(envelope)) == APPROVAL_BINDING_DIGEST


def test_approval_binding_digest_is_not_a_replay_identity():
    """It must not collapse into request_digest, or an allow becomes replayable.

    Two requests differing only in which approval was presented share an
    approval_binding_digest while their decisions differ -- one allows, one
    denies dual_control_required. The fixture asserts the two digests disagree
    on a claim-bearing request so the distinction stays real.
    """
    envelope = _envelope("destroy")
    binding = envelope["binding"]
    assert binding["approval_binding_digest"] != binding["request_digest"]


def test_ordinary_decision_carries_no_approval_binding_digest():
    """The field is omitted, not duplicated, on a claim-free decision."""
    binding = _envelope("rotate")["binding"]
    assert "approval" not in binding.get("context", {})
    assert "approval_binding_digest" not in binding


def test_embedded_claim_declares_the_pdp_path():
    """pdp_path is the issuer's declaration, and it is what our PEP requires.

    Path intent is never inferred from a pdp_digest that happens to be present;
    approvals issued before approval-engine schema v3 carry pdp_path false
    regardless of any digest they hold.
    """
    claim = _envelope("destroy")["binding"]["context"]["approval"]
    assert claim["binding"]["pdp_path"] is True


def test_decision_validation_ties_the_claim_to_the_approval_binding_digest():
    """Step 2 checks the identity against the real envelope, not a local guess.

    The PEP knows the claim's pdp_digest from step 1. When the request it sent
    carried that claim, the decision must name the same claim-free envelope, or
    the approval was issued against some other request.
    """
    envelope = _refresh_lifetime(_envelope("destroy"))
    request = _request_from(envelope)
    result = validate_decision_envelope(
        envelope,
        request,
        accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"},
        accepted_policy_versions={"v2"},
        expected_approval_binding_digest=APPROVAL_BINDING_DIGEST,
    )
    assert result.action == "destroy"

    with pytest.raises(DecisionError, match="approval binding digest"):
        validate_decision_envelope(
            envelope,
            request,
            accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"},
            accepted_policy_versions={"v2"},
            expected_approval_binding_digest="sha256:" + "c" * 64,
        )


def test_claim_bearing_request_without_a_binding_digest_fails_closed():
    """A decision that records no binding digest cannot tie the claim to itself.

    Comparing against request_digest instead would be the fail-open direction
    the whole field exists to prevent, so the absence is refused outright.
    """
    envelope = _refresh_lifetime(_envelope("destroy"))
    request = _request_from(envelope)
    del envelope["binding"]["approval_binding_digest"]
    with pytest.raises(DecisionError, match="records no approval_binding_digest"):
        validate_decision_envelope(
            envelope,
            request,
            accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"},
            accepted_policy_versions={"v2"},
            expected_approval_binding_digest=APPROVAL_BINDING_DIGEST,
        )


def test_a_forged_binding_digest_is_recomputed_not_trusted():
    """The field is verified against our own canonical digest, never taken on faith."""
    envelope = _refresh_lifetime(_envelope("destroy"))
    request = _request_from(envelope)
    envelope["binding"]["approval_binding_digest"] = "sha256:" + "d" * 64
    with pytest.raises(DecisionError, match="does not match this request"):
        validate_decision_envelope(
            envelope,
            request,
            accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"},
            accepted_policy_versions={"v2"},
        )


# --- tenant scoping (v2) -----------------------------------------------------
#
# GLAS-WP-0015 asked for the exact CheckRequest tenant and wrong-tenant denial
# evidence. These are that evidence, taken from a real deny envelope rather than
# asserted, plus the pin that stops our request tenant drifting from the package
# it is scoped by.

WRONG_TENANT_FIXTURE = "decision_wrong_tenant_deny.json"


def test_our_request_tenant_is_the_package_known_tenant():
    """The tenant we send must be the one the package allows.

    v2 reads request_tenant := object.get(input, "tenant", "") and allows only
    known_tenant := "tenant:platform". Pinning our constant against the real
    allow envelopes means a package retenanting shows up here rather than as a
    wrong_tenant denial in production.
    """
    for name in ("rotate", "destroy"):
        assert _envelope(name)["binding"]["tenant"] == REQUEST_TENANT


def test_built_request_carries_the_tenant_and_hashes_it():
    """An absent tenant is a denial, not an ignored field, so it must be sent.

    tenant is part of the digest material, so omitting it also produces a digest
    that matches no correctly issued decision -- the same class of defect as
    hashing excluded fields, arriving from the other direction.
    """
    entry = SimpleNamespace(id="glas-primary", stage="prod")
    request = build_action_request(
        entry,
        "rotate",
        subject_id="secrets-engine",
        subject_type="service",
        purpose="rotate-exposed-key",
        fields=["password"],
    )
    assert request["tenant"] == REQUEST_TENANT
    assert "tenant" in digest_material(request)

    untenanted = dict(request)
    del untenanted["tenant"]
    assert request_digest(untenanted) != request_digest(request)


def test_build_action_request_refuses_an_empty_tenant():
    entry = SimpleNamespace(id="glas-primary", stage="prod")
    with pytest.raises(DecisionError, match="requires a tenant"):
        build_action_request(
            entry,
            "rotate",
            subject_id="secrets-engine",
            subject_type="service",
            purpose="rotate-exposed-key",
            tenant="",
        )


def test_wrong_tenant_deny_envelope_fails_closed():
    """Wrong-tenant denial evidence, from a real v2 deny envelope.

    flex-auth sent tenant:coulomb on an otherwise-valid rotate and the package
    denied it with matched_rule wrong_tenant, so the tenant alone carried the
    denial. Our consumer must refuse it on effect before anything else -- a deny
    is not a decision we may act on, whatever else it validates.
    """
    # Not lifetime-refreshed: a deny carries no lifetime at all, which is the
    # property test_wrong_tenant_deny_carries_no_lifetime pins.
    envelope = json.loads((FIXTURES / WRONG_TENANT_FIXTURE).read_text())
    assert envelope["effect"] == "deny"
    assert envelope["reason"] == "wrong_tenant"
    assert envelope["matched_rule"] == "wrong_tenant"
    assert envelope["binding"]["tenant"] == "tenant:coulomb"
    assert envelope["binding"]["tenant"] != REQUEST_TENANT

    with pytest.raises(DecisionError, match="effect is not allow"):
        validate_decision_envelope(
            envelope,
            _request_from(envelope),
            accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"},
            accepted_policy_versions={"v2"},
        )


def test_wrong_tenant_deny_carries_no_lifetime():
    """A deny has no lifetime, which is why effect must be checked first.

    DecisionEnvelope requires lifetime only when effect is allow. A consumer
    that validated lifetime before effect would raise a confusing missing-field
    error on a perfectly well-formed denial.
    """
    envelope = json.loads((FIXTURES / WRONG_TENANT_FIXTURE).read_text())
    assert envelope.get("lifetime") is None


def test_the_superseded_v1_package_is_not_accepted():
    """v1 had no tenant rule and failed open; pinning it must not be possible.

    A consumer still pinned to v1 would keep getting allows it should never
    have had and could not tell from the version string that the rule changed
    underneath it, which is exactly why flex-auth superseded v1 rather than
    amending it.
    """
    envelope = _refresh_lifetime(_envelope("rotate"))
    with pytest.raises(DecisionError, match="policy version is not accepted"):
        validate_decision_envelope(
            envelope,
            _request_from(envelope),
            accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"},
            accepted_policy_versions={"v1"},
        )