7e7ca8be51
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 306568@bnt-lap001 Assistant-Session: 8bdf8150-ea56-41e7-9132-13f04f6b0b91
296 lines
14 KiB
Python
296 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import importlib.util
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
REPO_DIR = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_DIR / "scripts"))
|
|
SPEC = importlib.util.spec_from_file_location(
|
|
"credential_change", REPO_DIR / "scripts/credential-change.py"
|
|
)
|
|
credential_change = importlib.util.module_from_spec(SPEC)
|
|
assert SPEC.loader is not None
|
|
sys.modules[SPEC.name] = credential_change
|
|
SPEC.loader.exec_module(credential_change)
|
|
|
|
import ccr_intake # noqa: E402
|
|
|
|
cc = credential_change
|
|
STANDARD = "CCR-2026-0051-mas-tem-relay-smtp.yaml"
|
|
POLICY = "openbao/policies/workload-kv-read-mas-tem-relay.hcl"
|
|
|
|
|
|
class CcrIntakeTests(unittest.TestCase):
|
|
"""Run against a scratch repo root so nothing is written into the real tree."""
|
|
|
|
def setUp(self) -> None:
|
|
self.root = Path(tempfile.mkdtemp())
|
|
self.addCleanup(shutil.rmtree, self.root, True)
|
|
self.ccrs = self.root / "credential-change-requests"
|
|
self.ccrs.mkdir()
|
|
(self.root / "openbao/policies").mkdir(parents=True)
|
|
shutil.copy2(REPO_DIR / "credential-change-requests" / STANDARD, self.ccrs / STANDARD)
|
|
shutil.copy2(REPO_DIR / POLICY, self.root / POLICY)
|
|
for target in (cc, ccr_intake):
|
|
patcher = mock.patch.object(target, "REPO_DIR", self.root)
|
|
patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
env = mock.patch.dict(os.environ, {"CCR_DIR": str(self.ccrs)})
|
|
env.start()
|
|
self.addCleanup(env.stop)
|
|
self.profile, _digest = ccr_intake.load_profile()
|
|
self.path = self.ccrs / STANDARD
|
|
# The live CCR moves through approval; the fixture must stay pre-approval.
|
|
self.edit(lambda d: (d.update(status="proposed"), d["review"].update(comments=[]),
|
|
d.pop("state_hub", None)))
|
|
|
|
def edit(self, change) -> None:
|
|
data = cc.load_yaml(self.path)
|
|
change(data)
|
|
cc.dump_yaml(self.path, data)
|
|
|
|
def violations(self) -> list[str]:
|
|
return ccr_intake.standard_violations(cc.load_yaml(self.path), self.profile, self.ccrs, cc)
|
|
|
|
def assert_violation(self, change, fragment: str) -> None:
|
|
self.edit(change)
|
|
found = self.violations()
|
|
self.assertTrue(any(fragment in item for item in found), found)
|
|
|
|
def test_standard_lane_passes_and_records_review_without_approval(self) -> None:
|
|
self.assertEqual(ccr_intake.review_standard(self.path, cc), [])
|
|
data = cc.load_yaml(self.path)
|
|
self.assertEqual(data["status"], "proposed")
|
|
comment = data["review"]["comments"][-1]
|
|
self.assertEqual(comment["decision"], "approved")
|
|
self.assertIn("ccr-standard-lane v1 sha256:", comment["comment"])
|
|
self.assertTrue(comment["reviewer"].startswith("railiance-platform-owner"))
|
|
self.assertIn("apply requires status approved, got proposed", cc.apply_blockers(data))
|
|
|
|
def test_check_mode_does_not_write(self) -> None:
|
|
before = self.path.read_text()
|
|
ccr_intake.review_standard(self.path, cc, write=False)
|
|
self.assertEqual(self.path.read_text(), before)
|
|
|
|
def test_out_of_profile_records_needs_platform_review(self) -> None:
|
|
self.edit(lambda d: d["risk"].update(classification="high"))
|
|
self.assertTrue(ccr_intake.review_standard(self.path, cc))
|
|
comment = cc.load_yaml(self.path)["review"]["comments"][-1]
|
|
self.assertEqual(comment["decision"], "needs_platform_review")
|
|
self.assertEqual(cc.load_yaml(self.path)["status"], "proposed")
|
|
|
|
def test_requires_proposed_status(self) -> None:
|
|
self.edit(lambda d: d.update(status="approved"))
|
|
with self.assertRaises(SystemExit):
|
|
ccr_intake.review_standard(self.path, cc)
|
|
|
|
def test_metadata_read_rejected(self) -> None:
|
|
self.assert_violation(lambda d: d["openbao"].update(metadata_read=True), "metadata_read")
|
|
|
|
def test_kv_path_outside_prefixes_rejected(self) -> None:
|
|
self.assert_violation(
|
|
lambda d: d["openbao"].update(kv_path="platform/operators/founder/x"), "outside the standard"
|
|
)
|
|
|
|
def test_supplier_path_needs_exactly_two_segments(self) -> None:
|
|
self.assert_violation(
|
|
lambda d: d["openbao"].update(kv_path="platform/suppliers/scaleway/a/b"), "outside the standard"
|
|
)
|
|
|
|
def test_policy_file_drift_rejected(self) -> None:
|
|
(self.root / POLICY).write_text('path "platform/data/suppliers/*" {\n capabilities = ["read"]\n}\n')
|
|
self.assertTrue(any("differs" in v for v in self.violations()))
|
|
|
|
def test_long_ttl_rejected(self) -> None:
|
|
self.assert_violation(lambda d: d["openbao"]["auth"].update(token_max_ttl="1h"), "token_max_ttl")
|
|
|
|
def test_default_policy_rejected(self) -> None:
|
|
self.assert_violation(
|
|
lambda d: d["openbao"]["auth"].update(token_no_default_policy=False), "token_no_default_policy"
|
|
)
|
|
|
|
def test_role_prefix_and_audience_enforced(self) -> None:
|
|
self.assert_violation(lambda d: d["openbao"]["auth"].update(role="mas-eso"), "role must start")
|
|
self.assert_violation(lambda d: d["openbao"]["auth"].update(audience="vault"), "audience")
|
|
|
|
def test_two_service_accounts_rejected(self) -> None:
|
|
self.assert_violation(
|
|
lambda d: d["openbao"]["auth"]["bound_claims"]["service_account_names"].append("other"),
|
|
"service account",
|
|
)
|
|
|
|
def test_reserved_namespace_rejected(self) -> None:
|
|
self.assert_violation(
|
|
lambda d: d["openbao"]["auth"]["bound_claims"].update(service_account_namespaces=["external-secrets"]),
|
|
"reserved",
|
|
)
|
|
|
|
def test_second_reader_of_path_rejected(self) -> None:
|
|
other = cc.load_yaml(self.path)
|
|
other.update(id="CCR-2026-0052", status="active")
|
|
cc.dump_yaml(self.ccrs / "CCR-2026-0052-other.yaml", other)
|
|
self.assertTrue(any("already has a reader: CCR-2026-0052" in v for v in self.violations()))
|
|
|
|
def test_cancelled_reader_is_not_a_conflict(self) -> None:
|
|
other = cc.load_yaml(self.path)
|
|
other.update(id="CCR-2026-0052", status="cancelled")
|
|
cc.dump_yaml(self.ccrs / "CCR-2026-0052-other.yaml", other)
|
|
self.assertEqual(self.violations(), [])
|
|
|
|
def test_next_id_and_duplicate_detection(self) -> None:
|
|
self.assertEqual(ccr_intake.next_ccr_id(self.ccrs, 2026), "CCR-2026-0052")
|
|
self.assertEqual(ccr_intake.next_ccr_id(self.ccrs, 2027), "CCR-2027-0001")
|
|
self.assertEqual(ccr_intake.duplicate_ids(self.ccrs, cc), {})
|
|
shutil.copy2(self.path, self.ccrs / "CCR-2026-0051-race.yaml")
|
|
self.assertIn("CCR-2026-0051", ccr_intake.duplicate_ids(self.ccrs, cc))
|
|
|
|
def submit_args(self, **overrides) -> argparse.Namespace:
|
|
draft_dir = Path(tempfile.mkdtemp())
|
|
self.addCleanup(shutil.rmtree, draft_dir, True)
|
|
draft = cc.load_yaml(self.path)
|
|
draft.update(id="CCR-DRAFT-MAS", status="draft")
|
|
draft["openbao"].update(
|
|
kv_path="platform/suppliers/scaleway/other-relay",
|
|
policy_name="workload-kv-read-other-relay",
|
|
policy_file="openbao/policies/workload-kv-read-other-relay.hcl",
|
|
)
|
|
draft["openbao"]["auth"]["policies"] = ["workload-kv-read-other-relay"]
|
|
draft["review"]["comments"] = [{"reviewer": "someone", "decision": "approved", "comment": "x"}]
|
|
cc.dump_yaml(draft_dir / "draft.yaml", draft)
|
|
(draft_dir / "policy.hcl").write_text(
|
|
(self.root / POLICY).read_text().replace("mas-tem-relay", "other-relay")
|
|
)
|
|
values = dict(ccr=str(draft_dir / "draft.yaml"), policy=str(draft_dir / "policy.hcl"),
|
|
slug="other-relay", requester="ops-mason", profile=str(ccr_intake.DEFAULT_PROFILE))
|
|
values.update(overrides)
|
|
return argparse.Namespace(**values)
|
|
|
|
def test_submit_allocates_id_copies_policy_and_reviews(self) -> None:
|
|
destination, violations = ccr_intake.submit(self.submit_args(), cc)
|
|
self.assertEqual(violations, [])
|
|
self.assertEqual(destination.name, "CCR-2026-0052-other-relay.yaml")
|
|
data = cc.load_yaml(destination)
|
|
self.assertEqual((data["id"], data["status"]), ("CCR-2026-0052", "proposed"))
|
|
self.assertEqual(len(data["review"]["comments"]), 1)
|
|
self.assertTrue(data["review"]["comments"][0]["reviewer"].startswith("railiance-platform-owner"))
|
|
self.assertTrue((self.root / "openbao/policies/workload-kv-read-other-relay.hcl").exists())
|
|
|
|
def test_submit_refuses_unlisted_requester(self) -> None:
|
|
with self.assertRaises(SystemExit):
|
|
ccr_intake.submit(self.submit_args(requester="random-agent"), cc)
|
|
|
|
def test_submit_refuses_policy_overwrite(self) -> None:
|
|
(self.root / "openbao/policies/workload-kv-read-other-relay.hcl").write_text("different\n")
|
|
with self.assertRaises(SystemExit):
|
|
ccr_intake.submit(self.submit_args(), cc)
|
|
|
|
def test_submit_refuses_secret_markers(self) -> None:
|
|
args = self.submit_args()
|
|
Path(args.ccr).write_text(Path(args.ccr).read_text() + "# hvs.notarealtoken\n")
|
|
with self.assertRaises(SystemExit):
|
|
ccr_intake.submit(args, cc)
|
|
|
|
def submitted_branch(self) -> tuple[Path, str]:
|
|
main_root = Path(tempfile.mkdtemp())
|
|
self.addCleanup(shutil.rmtree, main_root, True)
|
|
shutil.copytree(self.root, main_root, dirs_exist_ok=True)
|
|
destination, _violations = ccr_intake.submit(self.submit_args(), cc)
|
|
return main_root, destination.stem
|
|
|
|
def test_branch_gate_accepts_clean_submission(self) -> None:
|
|
main_root, stem = self.submitted_branch()
|
|
self.assertEqual(ccr_intake.branch_gate(main_root, stem, cc), [])
|
|
|
|
def test_branch_gate_refuses_extra_changes(self) -> None:
|
|
main_root, stem = self.submitted_branch()
|
|
(self.root / "registry").mkdir()
|
|
(self.root / "registry/ccr-standard-profile.yaml").write_text("submitters: [anyone]\n")
|
|
errors = ccr_intake.branch_gate(main_root, stem, cc)
|
|
self.assertIn("ccr branch may not change registry/ccr-standard-profile.yaml", errors)
|
|
|
|
def test_branch_gate_refuses_modified_existing_policy(self) -> None:
|
|
main_root, stem = self.submitted_branch()
|
|
path = self.ccrs / f"{stem}.yaml"
|
|
data = cc.load_yaml(path)
|
|
data["openbao"]["policy_file"] = POLICY
|
|
cc.dump_yaml(path, data)
|
|
(self.root / POLICY).write_text((self.root / POLICY).read_text() + "# widened\n")
|
|
errors = ccr_intake.branch_gate(main_root, stem, cc)
|
|
self.assertTrue(any("may not modify" in e for e in errors), errors)
|
|
|
|
def test_branch_gate_refuses_hand_edited_review(self) -> None:
|
|
main_root, stem = self.submitted_branch()
|
|
path = self.ccrs / f"{stem}.yaml"
|
|
data = cc.load_yaml(path)
|
|
data["review"]["comments"].append({"reviewer": "founder", "decision": "approved", "comment": "ok"})
|
|
cc.dump_yaml(path, data)
|
|
errors = ccr_intake.branch_gate(main_root, stem, cc)
|
|
self.assertTrue(any("policy-as-code approval" in e for e in errors), errors)
|
|
|
|
def test_branch_gate_refuses_unlisted_requester(self) -> None:
|
|
main_root, stem = self.submitted_branch()
|
|
path = self.ccrs / f"{stem}.yaml"
|
|
data = cc.load_yaml(path)
|
|
data["requester"]["agent"] = "someone-else"
|
|
cc.dump_yaml(path, data)
|
|
errors = ccr_intake.branch_gate(main_root, stem, cc)
|
|
self.assertTrue(any("listed submitter" in e for e in errors), errors)
|
|
|
|
def test_branch_gate_refuses_existing_ccr(self) -> None:
|
|
main_root = Path(tempfile.mkdtemp())
|
|
self.addCleanup(shutil.rmtree, main_root, True)
|
|
shutil.copytree(self.root, main_root, dirs_exist_ok=True)
|
|
errors = ccr_intake.branch_gate(main_root, Path(STANDARD).stem, cc)
|
|
self.assertTrue(any("already exists on main" in e for e in errors), errors)
|
|
|
|
def test_request_decision_records_id_after_platform_review(self) -> None:
|
|
ccr_intake.review_standard(self.path, cc)
|
|
with mock.patch.object(cc, "state_hub_post_json", return_value={"id": "dec-1", "status": "open"}) as post:
|
|
ccr_intake.request_decision(self.path, cc, "http://hub")
|
|
payload = post.call_args.args[2]
|
|
self.assertIn("APPROVE: CCR-2026-0051", payload["description"])
|
|
self.assertNotIn("status", payload)
|
|
self.assertNotIn("Decision link", payload["description"])
|
|
state_hub = cc.load_yaml(self.path)["state_hub"]
|
|
self.assertEqual(state_hub["decision_id"], "dec-1")
|
|
self.assertEqual(cc.load_yaml(self.path)["status"], "proposed")
|
|
with self.assertRaises(SystemExit):
|
|
ccr_intake.request_decision(self.path, cc, "http://hub")
|
|
|
|
def test_request_decision_requires_platform_approval(self) -> None:
|
|
self.edit(lambda d: d["risk"].update(classification="high"))
|
|
ccr_intake.review_standard(self.path, cc)
|
|
with mock.patch.object(cc, "state_hub_post_json") as post, self.assertRaises(SystemExit):
|
|
ccr_intake.request_decision(self.path, cc, "http://hub")
|
|
post.assert_not_called()
|
|
|
|
def test_request_decision_refuses_pre_resolved_decision(self) -> None:
|
|
ccr_intake.review_standard(self.path, cc)
|
|
resolved = {"id": "dec-2", "status": "resolved"}
|
|
with mock.patch.object(cc, "state_hub_post_json", return_value=resolved), self.assertRaises(SystemExit):
|
|
ccr_intake.request_decision(self.path, cc, "http://hub")
|
|
|
|
def test_render_policy_matches_reviewed_policy_file(self) -> None:
|
|
rendered = ccr_intake.standard_policy_hcl("platform/suppliers/scaleway/mas-tem-relay", cc)
|
|
self.assertEqual(rendered, (REPO_DIR / POLICY).read_text())
|
|
|
|
def test_plan_ref_passes_through_submit_and_is_shape_checked(self) -> None:
|
|
args = self.submit_args()
|
|
draft = cc.load_yaml(Path(args.ccr))
|
|
draft["plan_ref"] = {"repo": "ops-mason", "path": "plans/x.md", "commit": "d19e6b1"}
|
|
cc.dump_yaml(Path(args.ccr), draft)
|
|
destination, violations = ccr_intake.submit(args, cc)
|
|
self.assertEqual(violations, [])
|
|
self.assertEqual(cc.load_yaml(destination)["plan_ref"]["commit"], "d19e6b1")
|
|
self.assert_violation(lambda d: d.update(plan_ref={"repo": "ops-mason"}), "plan_ref")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|