b6148da443
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
199 lines
9 KiB
Python
199 lines
9 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
|
|
|
|
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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|