diff --git a/src/ops_mason/ccr_gate.py b/src/ops_mason/ccr_gate.py index 6930311..cc2729e 100644 --- a/src/ops_mason/ccr_gate.py +++ b/src/ops_mason/ccr_gate.py @@ -103,6 +103,11 @@ def generate_policy(kv_path: str, *, platform: Path | None = None) -> str: return policy +# One decision. sync-decision also records when it resolved and the API url +# (credential-change.py sync_state_hub_decision). Those are not a second decision. +_STATE_HUB_KEYS = {"decision_id", "decision_api_url", "decision_resolved_at"} + + def _decision_id(ccr: dict[str, Any]) -> str: state = ccr.get("state_hub") if not isinstance(state, dict): @@ -110,12 +115,51 @@ def _decision_id(ccr: dict[str, Any]) -> str: decision_id = state.get("decision_id") if not isinstance(decision_id, str) or _UUID.fullmatch(decision_id) is None: raise EvidenceRefused("plan approval must be the single hub decision") - extra = [key for key in state if key not in {"decision_id", "decision_api_url"}] + extra = [key for key in state if key not in _STATE_HUB_KEYS] if extra: raise EvidenceRefused("plan approval must be the single hub decision") return decision_id +def _standard_lane_auth(platform: Path | None) -> dict[str, Any]: + """Token bounds the standard-lane profile requires on a Kubernetes auth block.""" + root = platform or PLATFORM + profile_path = root / "registry" / "ccr-standard-profile.yaml" + if not profile_path.is_file(): + raise EvidenceRefused(f"standard-lane profile not found at {profile_path}") + profile = yaml.safe_load(profile_path.read_text(encoding="utf-8")) + auth = profile.get("auth") if isinstance(profile, dict) else None + if not isinstance(auth, dict): + raise EvidenceRefused("standard-lane profile has no auth block") + seconds = auth.get("max_ttl_seconds") + no_default = auth.get("token_no_default_policy") + audience = auth.get("audience") + if ( + isinstance(seconds, bool) + or not isinstance(seconds, int) + or seconds < 1 + or not isinstance(no_default, bool) + or not isinstance(audience, str) + or not audience + ): + raise EvidenceRefused("standard-lane profile auth bounds are incomplete") + if seconds % 86400 == 0: + ttl = f"{seconds // 86400}d" + elif seconds % 3600 == 0: + ttl = f"{seconds // 3600}h" + elif seconds % 60 == 0: + ttl = f"{seconds // 60}m" + else: + ttl = f"{seconds}s" + return { + "audience": audience, + "ttl": ttl, + "token_max_ttl": ttl, + "token_explicit_max_ttl": ttl, + "token_no_default_policy": no_default, + } + + def draft_from_plan( *, kv_path: str, @@ -134,6 +178,7 @@ def draft_from_plan( if _COMMIT.fullmatch(plan_ref["commit"]) is None: raise EvidenceRefused("plan_ref commit must be a hex git commit") policy = generate_policy(kv_path, platform=platform) + lane = _standard_lane_auth(platform) out_dir.mkdir(parents=True, exist_ok=True) policy_path = out_dir / "policy.hcl" policy_path.write_text(policy if policy.endswith("\n") else policy + "\n", encoding="utf-8") @@ -170,8 +215,11 @@ def draft_from_plan( "method": "kubernetes", "mount": "kubernetes", "role": f"external-secrets-{slug}", - "audience": "openbao", - "ttl": "15m", + "audience": lane["audience"], + "ttl": lane["ttl"], + "token_max_ttl": lane["token_max_ttl"], + "token_explicit_max_ttl": lane["token_explicit_max_ttl"], + "token_no_default_policy": lane["token_no_default_policy"], "policies": [policy_name], "bound_claims": { "service_account_names": [slug], diff --git a/tests/test_ccr_gate.py b/tests/test_ccr_gate.py index f7cf412..91f9f10 100644 --- a/tests/test_ccr_gate.py +++ b/tests/test_ccr_gate.py @@ -1,5 +1,6 @@ """The CCR record-evidence gate matches a policy and refuses a different one.""" +import importlib.util from pathlib import Path import pytest @@ -9,6 +10,7 @@ from ops_mason.ccr_gate import EvidenceRefused, draft_from_plan, record_evidence DECISION_ID = "00000000-0000-4000-8000-000000000001" KV_PATH = "platform/workloads/fixture/reader" +PLATFORM = Path.home() / "railiance-platform" def _draft(tmp_path: Path) -> dict: @@ -29,6 +31,37 @@ def _evidence(draft: Path) -> list: return (document.get("verification") or {}).get("evidence") or [] +def _platform_modules(): + def load(name: str, path: Path): + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + return ( + load("credential_change_for_ccr_gate", PLATFORM / "scripts" / "credential-change.py"), + load("ccr_intake_for_ccr_gate", PLATFORM / "scripts" / "ccr_intake.py"), + ) + + +def _record(drafted: dict) -> None: + record_evidence( + drafted["draft"], + built_policy=drafted["built_policy"], + actor="ops-mason", + kind="metadata_apply", + result="passed", + detail=["built policy matched the CCR policy file"], + ) + + +def _differ(policy_path: Path) -> None: + policy_path.write_text( + policy_path.read_text(encoding="utf-8").replace('["read"]', '["create"]', 1), + encoding="utf-8", + ) + + def test_record_evidence_accepts_a_trailing_newline_and_refuses_a_different_policy(tmp_path): drafted = _draft(tmp_path) policy_path = Path(drafted["policy_file"]) @@ -53,20 +86,42 @@ def test_record_evidence_accepts_a_trailing_newline_and_refuses_a_different_poli assert recorded[0]["result"] == "passed" assert "matched the CCR policy file" in recorded[0]["details"][0] - policy_path.write_text( - policy_path.read_text(encoding="utf-8").replace('["read"]', '["create"]', 1), - encoding="utf-8", - ) + _differ(policy_path) with pytest.raises(EvidenceRefused, match="record-evidence refused"): - record_evidence( - drafted["draft"], - built_policy=drafted["built_policy"], - actor="ops-mason", - kind="metadata_apply", - result="passed", - detail=["built policy matched the CCR policy file"], - ) + _record(drafted) assert len(_evidence(drafted["draft"])) == 1 stored = drafted["draft"].read_text(encoding="utf-8") assert "sync-decision" not in stored assert DECISION_ID in stored + + +def test_draft_carries_the_standard_lane_token_bounds(tmp_path): + drafted = _draft(tmp_path) + cc, intake = _platform_modules() + profile, _digest = intake.load_profile(PLATFORM / "registry" / "ccr-standard-profile.yaml") + document = cc.load_yaml(drafted["draft"]) + auth = document["openbao"]["auth"] + max_ttl = profile["auth"]["max_ttl_seconds"] + for key in ("ttl", "token_max_ttl", "token_explicit_max_ttl"): + assert intake.ttl_seconds(auth[key]) == max_ttl + assert auth["token_no_default_policy"] is profile["auth"]["token_no_default_policy"] + assert intake.standard_violations(document, profile, drafted["draft"].parent, cc) == [] + + +def test_synced_hub_decision_accepts_a_newline_and_refuses_a_different_policy(tmp_path): + drafted = _draft(tmp_path) + document = yaml.safe_load(drafted["draft"].read_text(encoding="utf-8")) + decision_id = document["state_hub"]["decision_id"] + document["state_hub"]["decision_api_url"] = f"http://127.0.0.1:8000/decisions/{decision_id}" + document["state_hub"]["decision_resolved_at"] = "2026-10-18T10:46:17.864665Z" + drafted["draft"].write_text(yaml.safe_dump(document, sort_keys=False), encoding="utf-8") + policy_path = Path(drafted["policy_file"]) + policy_path.write_text(policy_path.read_text(encoding="utf-8") + "\n", encoding="utf-8") + + _record(drafted) + assert len(_evidence(drafted["draft"])) == 1 + + _differ(policy_path) + with pytest.raises(EvidenceRefused, match="record-evidence refused"): + _record(drafted) + assert len(_evidence(drafted["draft"])) == 1