tegwick
ee4e901611
Assistant: codex Assistant-Model: gpt-5.6-luna Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
184 lines
8 KiB
Python
184 lines
8 KiB
Python
"""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()
|