"""FLEX-DEC-2026-012 against independent producer request/output fixtures.""" import copy import json from datetime import datetime, timezone from pathlib import Path from types import SimpleNamespace import pytest from secrets_engine.authorization import REQUEST_TENANT, build_action_request, request_digest, validate_decision_envelope from secrets_engine.decision_check import check_decision, require_supported_pdp_address from secrets_engine.errors import DecisionError FIXTURES = Path(__file__).parent / "fixtures" / "flex-auth-replay" CASES = { "rotate": ("check_request_allow_rotate.json", "decision_rotate.json", "sha256:de67324f54187055307a833235f83ced9fcd3a20952a27b3d19493ed39734345"), "destroy": ("check_request_allow_destroy_dual_control.json", "decision_destroy_dual_control.json", "sha256:c749ee2dc3cdf927a70a3e5b27cff4d97a438d3264153b4b2e3bcacbaf82091a"), "deny": ("check_request_deny_wrong_tenant.json", "decision_wrong_tenant_deny.json", None), } def pair(name): inp, out, _ = CASES[name] return json.loads((FIXTURES / inp).read_text()), json.loads((FIXTURES / out).read_text()) def validate(request, decision, **over): # Validate at the actual producer decision time; do not rewrite evidence. now = datetime.fromisoformat(decision["provenance"]["decision_time"].replace("Z", "+00:00")) return validate_decision_envelope(decision, request, accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"}, accepted_policy_versions={"v2"}, now=now, **over) @pytest.mark.parametrize("name", ["rotate", "destroy"]) def test_real_producer_request_validates(name): request, decision = pair(name) assert request_digest(request) == decision["binding"]["submitted_request_digest"] assert request_digest(request) != decision["binding"]["request_digest"] assert decision["binding"]["request_digest"] == CASES[name][2] assert decision["provenance"]["registry_snapshot_digest"] == "sha256:f5a309bc0b36721fd6d9ad7f53eb21222162bc2eac62a0ab0802a9a1d51340bb" assert decision["provenance"]["policy_package_digest"] == "sha256:bd11c5fe77ce6439c65fea225ad6b71d2110efc5e7b5bc9b499c59cd0a53b8b4" assert validate(request, decision).action == name @pytest.mark.parametrize("name", ["rotate", "destroy"]) def test_correlation_fields_are_excluded(name): request, _ = pair(name) baseline = request_digest(request) for key, value in (("id", "other"), ("policy_version", "v99"), ("caring_context", {"extra": "context"})): assert request_digest({**request, key: value}) == baseline def test_approval_join_is_between_evaluator_origin_values(): request, decision = pair("destroy") pdp = request["context"]["approval"]["binding"]["pdp_digest"] assert pdp == decision["binding"]["approval_binding_digest"] == "sha256:fa07becfaa471394d06aee5fa3cd66352bf0cc69ef24900240489684cda8cd56" without = copy.deepcopy(request) del without["context"]["approval"] assert request_digest(without) != pdp assert validate(request, decision, expected_approval_binding_digest=pdp).action == "destroy" @pytest.mark.parametrize("change", ["claim_id", "claim_absent", "purpose", "action", "tenant", "field", "target"]) def test_allow_cannot_be_replayed_for_changed_submitted_material(change): request, decision = pair("destroy") if change == "claim_id": request["context"]["approval"]["approval_id"] = "other-approval" elif change == "claim_absent": del request["context"]["approval"] elif change == "purpose": request["context"]["purpose"] = "other-purpose" elif change in ("action", "tenant"): request[change] = "other" elif change == "field": request["resource"]["attributes"]["fields"] = ["other"] else: request["resource"]["id"] = "catalog:other" with pytest.raises(DecisionError, match="submitted request digest"): validate(request, decision) @pytest.mark.parametrize("value", [None, "", "sha256:" + "f" * 64]) def test_missing_or_wrong_submitted_binding_refused(value): request, decision = pair("rotate") if value is None: del decision["binding"]["submitted_request_digest"] else: decision["binding"]["submitted_request_digest"] = value with pytest.raises(DecisionError, match="submitted request digest"): validate(request, decision) @pytest.mark.parametrize("value", [None, "bad", "sha256:" + "f" * 64]) def test_approval_binding_missing_malformed_or_wrong_refused(value): request, decision = pair("destroy") decision["binding"]["approval_binding_digest"] = value with pytest.raises(DecisionError, match="approval[_ ]binding_digest|approval binding digest"): validate(request, decision) def test_claim_free_shortcut_refused(): request, decision = pair("rotate") with pytest.raises(DecisionError, match="carried approval"): validate(request, decision, expected_approval_binding_digest=decision["binding"]["request_digest"]) def test_wrong_tenant_and_superseded_policy_refused(): request, decision = pair("deny") assert decision["reason"] == "wrong_tenant" assert decision.get("lifetime") is None with pytest.raises(DecisionError, match="effect is not allow"): validate(request, decision) request, decision = pair("rotate") decision["provenance"]["policy_version"] = "v1" with pytest.raises(DecisionError, match="policy version"): validate(request, decision) def test_real_expired_allow_refused(): request, decision = pair("rotate") with pytest.raises(DecisionError, match="lifetime has expired"): validate_decision_envelope(decision, request, accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"}, accepted_policy_versions={"v2"}, now=datetime(2100, 1, 1, tzinfo=timezone.utc)) def test_request_builder_tenant_boundary(): entry = SimpleNamespace(id="glas-primary", stage="prod") args = dict(subject_id="secrets-engine", subject_type="service", purpose="rotation") request = build_action_request(entry, "rotate", **args) assert request["tenant"] == REQUEST_TENANT with pytest.raises(DecisionError, match="requires a tenant"): build_action_request(entry, "rotate", tenant="", **args) # --- supported PDP address (FLEX-DEC-2026-010) ------------------------------- @pytest.mark.parametrize( "url,fragment", [ # The exact address flex-auth originally handed over. It does not fail # to resolve from a workstation, it resolves to an unrelated public host. ("http://flex-auth-secrets-engine.flex-auth.svc.cluster.local:8080", "in-cluster Service name"), ("http://flex-auth-secrets-engine.flex-auth.svc.cluster.local.:8080", "in-cluster Service name"), ("http://flex-auth-secrets-engine.flex-auth.svc:8080", "in-cluster Service name"), # A public address is refused even though it would "work": nothing # authenticates the responder, so reaching something is not reaching # the pin. ("http://80.158.43.29:8080", "is not loopback"), ("https://flex-auth.example.com", "is not loopback"), ], ) def test_unsupported_pdp_addresses_are_refused(url, fragment): with pytest.raises(DecisionError, match=fragment): require_supported_pdp_address(url) @pytest.mark.parametrize( "url", [ "http://127.0.0.1:18080", "http://localhost:18080", "http://[::1]:18080", ], ) def test_loopback_forward_addresses_are_accepted(url): require_supported_pdp_address(url) def test_the_guard_runs_before_the_token_is_read(tmp_path): """A bad address must not cause the bearer token to be read, let alone sent. The token is the thing a misdirected request would leak, so the address check has to come first rather than alongside. """ missing = tmp_path / "never-read.token" with pytest.raises(DecisionError, match="in-cluster Service name"): check_decision( base_url="http://flex-auth-secrets-engine.flex-auth.svc.cluster.local:8080", token_file=missing, request={}, ) assert not missing.exists()