"""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 secrets_engine.authorization import ( approval_binding_digest, digest_material, request_digest, validate_decision_envelope, ) from secrets_engine.errors import DecisionError FIXTURES = Path(__file__).parent / "fixtures" / "flex-auth-replay" PACKAGE_DIGEST = "sha256:fe0070b79f66442ae6c218697a49c470c6c8f670aa57a30c078a5284d097bd8c" 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={"v1"}, ) 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={"v1"}, ) 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={"v1"}, 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={"v1"}, 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={"v1"}, 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={"v1"}, )