"""CCR-2026-0038 service-jwt-login lane: validator, invariant (ii), fake-bao applier.""" from __future__ import annotations import copy import importlib.util import json import os import re import stat import sys import tempfile import unittest from pathlib import Path import yaml ROOT = Path(__file__).resolve().parents[1] CCR_PATH = ROOT / "credential-change-requests/CCR-2026-0038-secrets-engine-service-jwt-login.yaml" GUARDED = "platform/data/workloads/keycape/secrets-engine-openbao" def load(name: str, rel: str): spec = importlib.util.spec_from_file_location(name, ROOT / rel) module = importlib.util.module_from_spec(spec) sys.modules[name] = module spec.loader.exec_module(module) return module lane = load("service_jwt_lane", "scripts/service_jwt_lane.py") cc = load("credential_change_cli", "scripts/credential-change.py") def ccr() -> dict: return yaml.safe_load(CCR_PATH.read_text()) FAKE_BAO = r'''#!/usr/bin/env python3 import json, os, sys state = json.load(open(os.environ["FAKE_BAO_STATE"])) open(os.environ["FAKE_BAO_LOG"], "a").write(json.dumps(sys.argv[1:]) + "\n") args = [a for a in sys.argv[1:] if a != "-format=json"] if args[:2] == ["auth", "list"]: print(json.dumps({k: {"type": v} for k, v in state["mounts"].items()})) elif args[:2] == ["policy", "read"]: print(state["policy"]) elif args[0] == "read" and args[1].endswith("/config"): print(json.dumps({"data": state["config"]})) elif args[0] == "read": print(json.dumps({"data": state["role"]})) ''' class ServiceJwtLaneTests(unittest.TestCase): def test_ccr_validates_and_records_the_delegated_apply(self): _ccr, errors, _w = cc.validate_ccr(CCR_PATH) self.assertEqual(errors, []) self.assertEqual(_ccr["status"], "applied") self.assertTrue(_ccr["openbao"]["auth"]["bound_claims_confirmed"]) evidence = _ccr["verification"]["evidence"] self.assertTrue( any(e["kind"] == "delegated_metadata_apply" and e["result"] == "passed" for e in evidence), evidence, ) def test_invariant_ii_policy_has_no_capability_on_the_client_secret_path(self): policy = (ROOT / "openbao/policies/secrets-engine-login-self.hcl").read_text() stanzas = re.findall(r'path "([^"]+)"\s*{\s*capabilities = \[([^\]]*)\]', policy) for path, caps in stanzas: if path.startswith("platform/"): self.assertEqual(caps.strip(), '"deny"', path) self.assertIn((GUARDED, '"deny"'), stanzas) granted = {p for p, c in stanzas if c.strip() != '"deny"'} self.assertEqual(granted, {p for p, _ in lane.SELF_GRANTS}) self.assertNotIn("*", policy) def test_validator_rejects_widening(self): cases = [] c = ccr(); c["openbao"]["guarded_paths"] = ["platform/data/workloads/*"]; cases.append(c) c = ccr(); c["openbao"]["auth"]["mount"] = "netkingdom"; cases.append(c) c = ccr(); c["openbao"]["policy_name"] = "platform-admin"; cases.append(c) c = ccr(); c["target"]["environment"] = "staging"; cases.append(c) for case in cases: self.assertTrue(lane.validate(case, ROOT), case["openbao"]) def test_validator_rejects_bad_role_file(self): with tempfile.TemporaryDirectory() as tmp: repo = Path(tmp) (repo / "openbao/auth").mkdir(parents=True) (repo / "openbao/policies").mkdir(parents=True) c = ccr() (repo / c["openbao"]["policy_file"]).write_text(lane.generated_policy_hcl(c)) (repo / c["openbao"]["auth"]["config_file"]).write_text( (ROOT / c["openbao"]["auth"]["config_file"]).read_text()) base = json.loads((ROOT / c["openbao"]["auth"]["role_file"]).read_text()) self.assertEqual(self._role_errors(repo, c, base), []) for mutate in ( lambda r: r.update(token_ttl="1h"), lambda r: r.update(token_no_default_policy=False), lambda r: r.update(token_policies=["secrets-engine-login-self", "default"]), lambda r: r.update(token_period="5m"), lambda r: r["bound_claims"].update(roles="*"), lambda r: r.update(bound_audiences=["other"]), lambda r: r.update(token_num_uses=0), ): role = copy.deepcopy(base); mutate(role) self.assertTrue(self._role_errors(repo, c, role), role) def _role_errors(self, repo, c, role): (repo / c["openbao"]["auth"]["role_file"]).write_text(json.dumps(role)) return lane.validate(c, repo) def _fake(self, tmp: Path, mounts: dict, policy: str | None = None, role_ttl=300): c = ccr() role = lane.role_payload(c, ROOT); role["token_ttl"] = role_ttl role["token_max_ttl"] = role["token_explicit_max_ttl"] = 300 state = {"mounts": mounts, "policy": policy if policy is not None else lane.generated_policy_hcl(c), "config": lane.config_payload(c, ROOT), "role": role} (tmp / "state.json").write_text(json.dumps(state)) bao = tmp / "bao"; bao.write_text(FAKE_BAO); bao.chmod(bao.stat().st_mode | stat.S_IEXEC) os.environ["FAKE_BAO_STATE"] = str(tmp / "state.json") os.environ["FAKE_BAO_LOG"] = str(tmp / "log") return c, str(bao) def _log(self, tmp: Path): return [json.loads(l) for l in (tmp / "log").read_text().splitlines()] def test_apply_enables_absent_mount_writes_in_order_and_reads_back(self): with tempfile.TemporaryDirectory() as t: tmp = Path(t) c, bao = self._fake(tmp, {"keycape-services/": "jwt"}) # first auth list sees no mount; later readback sees jwt state = json.loads((tmp / "state.json").read_text()) state["mounts"] = {} (tmp / "state.json").write_text(json.dumps(state)) with self.assertRaises(lane.BaoError): # readback: mount absent in fake lane.run_apply(c, ROOT, bao, quiet=True) verbs = [a[:2] for a in self._log(tmp)] self.assertEqual(verbs[:5], [["auth", "list"], ["policy", "write"], ["auth", "enable"], ["write", "auth/keycape-services/config"], ["write", "auth/keycape-services/role/secrets-engine-openbao"]]) self.assertIn("jwt", self._log(tmp)[2]) def test_apply_on_existing_jwt_mount_skips_enable_and_passes_readback(self): with tempfile.TemporaryDirectory() as t: tmp = Path(t) c, bao = self._fake(tmp, {"keycape-services/": "jwt"}) lane.run_apply(c, ROOT, bao, quiet=True) verbs = [a[:2] for a in self._log(tmp)] self.assertNotIn(["auth", "enable"], verbs) for argv in self._log(tmp): self.assertFalse(any("token" in a.lower() and "=" in a for a in argv), argv) def test_apply_refuses_foreign_mount_type_before_any_write(self): with tempfile.TemporaryDirectory() as t: tmp = Path(t) c, bao = self._fake(tmp, {"keycape-services/": "oidc"}) with self.assertRaises(lane.BaoError): lane.run_apply(c, ROOT, bao, quiet=True) self.assertEqual([a[:2] for a in self._log(tmp)], [["auth", "list"]]) def test_readback_detects_policy_drift(self): with tempfile.TemporaryDirectory() as t: tmp = Path(t) c, bao = self._fake(tmp, {"keycape-services/": "jwt"}, policy='path "platform/data/*" {}') self.assertIn("policy text differs from source", lane.readback_mismatches(c, ROOT, bao)) if __name__ == "__main__": unittest.main()