"""RPF-WP-0064-T02/T03 unit tests without Docker: admission, verifier, launcher. Adversarial receipt fixtures are derived from one valid synthetic receipt built here; nothing in this module touches Docker, OpenBao or custody material. """ from __future__ import annotations import copy import hashlib import json import os import shutil import subprocess import sys import tempfile import unittest from datetime import UTC, datetime, timedelta from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import openbao_restore_rehearsal as R # noqa: E402 A = R.adapter V = R.load("verify_restore_receipt", "scripts/openbao-isolated-restore/verify-restore-receipt.py") FAKE_IMAGE = "quay.io/openbao/openbao@sha256:" + hashlib.sha256(b"image").hexdigest() def digest(seed: str) -> str: return "sha256:" + hashlib.sha256(seed.encode()).hexdigest() class Base(unittest.TestCase): @classmethod def setUpClass(cls): cls.work = Path(tempfile.mkdtemp(prefix="rpf64-unit-")) cls.root, cls.commit = R.make_source_root(cls.work) @classmethod def tearDownClass(cls): shutil.rmtree(cls.work, ignore_errors=True) def admission(self, **override): now = datetime.now(UTC).replace(microsecond=0) adm = json.loads((R.ROOT / "docs/openbao-restore-admission.example.json").read_text()) adm.update({"attempt_id": "RPF-WP-0064-T04-unit-01", "material": "synthetic", "not_before": A.iso(now - timedelta(minutes=5)), "not_after": A.iso(now + timedelta(hours=1)), "image": FAKE_IMAGE, "adapter_commit": self.commit, "source_sha256": A.source_digests(self.root)}) adm["snapshot"].update({"ciphertext_sha256": digest("c"), "ciphertext_bytes": 10, "plaintext_sha256": digest("p"), "source_receipt_sha256": digest("r"), "age_recipient": R.Rehearsal.age_pair()[1], "source_cluster_id": "9b5aa97f-933c-bd94-095a-5b5d7f766da5", "source_applied_index": 50}) adm.update(override) return adm def receipt(self): rc = json.loads((R.ROOT / "docs/openbao-restore-receipt-v2.example.json").read_text()) adm = self.admission() now = datetime.now(UTC).replace(microsecond=0) rc.update({"attempt_id": adm["attempt_id"], "material": "synthetic", "admission_sha256": digest("adm"), "signer_principal": "founder@rehearsal", "anchor_protected": False, "signer_fingerprint": "SHA256:" + "AbCd" * 10 + "AbC", "adapter_commit": self.commit, "source_sha256": adm["source_sha256"], "started_at": A.iso(now - timedelta(seconds=40)), "finished_at": A.iso(now), "not_after": adm["not_after"], "elapsed_seconds": 40, "image": FAKE_IMAGE, "image_id": digest("id")}) rc["snapshot"].update({k: adm["snapshot"][k] for k in rc["snapshot"] if k in adm["snapshot"]}) rc["restored"].update({"cluster_id": adm["snapshot"]["source_cluster_id"], "applied_index": 50}) for row in rc["objects"]: row["observed_sha256"] = digest(row["object"]) return rc def verify(self, rc, **kw): return V.verify(copy.deepcopy(rc), source_root=self.root, **kw) class AdmissionTests(Base): def test_valid_synthetic_admission(self): A.validate_admission(self.admission()) def test_example_admission_refused_as_placeholder(self): with self.assertRaises(A.Refused): A.validate_admission(json.loads((R.ROOT / "docs/openbao-restore-admission.example.json").read_text())) def test_refusals(self): cases = [ {"phase": "B"}, {"phase_b_admitted": True}, {"image": "quay.io/openbao/openbao:2.5.4"}, {"image": "quay.io/openbao/openbao@sha256:" + "0" * 64}, {"image": "docker.io/x/openbao@" + digest("i")}, {"max_duration_seconds": 2701}, {"material": "original", "max_duration_seconds": 600}, {"namespace": "secrets-engine-recovery-attempt"}, {"readback_credential": "operator-oidc"}, {"unseal": {"key_threshold": 1, "key_shares": 1}}, {"extra": True}, {"adapter_commit": "HEAD"}, ] for override in cases: with self.subTest(override=override), self.assertRaises(A.Refused): A.validate_admission(self.admission(**override)) adm = self.admission() adm["source_sha256"].pop(next(iter(adm["source_sha256"]))) with self.assertRaises(A.Refused): A.validate_admission(adm) def test_window(self): adm = self.admission() A.check_window(adm) with self.assertRaises(A.Refused): A.check_window(adm, A.parse_iso(adm["not_after"])) def test_source_closure_detects_dirty_or_changed(self): adm = self.admission() A.check_source_closure(adm, self.root) with self.assertRaises(A.Refused): A.check_source_closure(dict(adm, adapter_commit="f" * 40), self.root) target = self.root / "openbao/auth/secrets-engine-jwt-role.json" original = target.read_bytes() try: target.write_bytes(original + b" ") with self.assertRaises(A.Refused): A.check_source_closure(adm, self.root) finally: target.write_bytes(original) def signed(self, adm, namespace=A.NAMESPACE): anchor = self.work / "anchor" anchor.mkdir(mode=0o700, exist_ok=True) key = anchor / "k" if not key.exists(): subprocess.run(["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-C", "f", "-f", str(key)], check=True) pub = (anchor / "k.pub").read_text().strip() (anchor / "allowed_signers").write_text(f'founder@rehearsal namespaces="{A.NAMESPACE}" {pub}\n') raw = json.dumps(adm).encode() path = Path(tempfile.mkstemp(dir=self.work)[1]) path.write_bytes(raw) subprocess.run(["ssh-keygen", "-q", "-Y", "sign", "-f", str(key), "-n", namespace, str(path)], check=True) sig = Path(str(path) + ".sig") os.chmod(sig, 0o600) return path, sig, A.sha256_bytes(raw), anchor / "allowed_signers" def test_signature_namespace_and_bytes_bound(self): path, sig, sha, signers = self.signed(self.admission()) _adm, _raw, signer, protected = A.load_admission(path, sig, sha, signers) self.assertEqual(signer["principal"], "founder@rehearsal") self.assertFalse(protected) # runtime-owned anchor: rehearsal only with self.assertRaises(A.Refused): A.load_admission(path, sig, digest("other"), signers) path2, sig2, sha2, _ = self.signed(self.admission(), namespace="secrets-engine-recovery-attempt") with self.assertRaises(A.Refused): A.load_admission(path2, sig2, sha2, signers) def test_original_material_needs_protected_anchor_and_t06_pin(self): path, sig, sha, signers = self.signed(self.admission(material="original")) with self.assertRaises(A.Refused): A.load_admission(path, sig, sha, signers) self.assertEqual(A.PINNED_IMAGE_DIGEST, "sha256:436eaf9778cad75507ff70ea26ace30dcbe15606e619ac3823495663d7f7c115") def test_target_takes_production_raft_node_id(self): # rapp-openbao runs the chart with setNodeId, so the snapshot's only voter is the pod name. self.assertEqual(A.RESTORE_NODE_ID, "openbao-0") self.assertEqual(json.loads(A.SERVER_CONFIG)["storage"]["raft"]["node_id"], "openbao-0") self.assertIn("leader_not_elected", A.ABORT_REASONS) def test_leader_wait_diagnostics_are_typed_and_redacted(self): line = A.sanitize_status_line(b"\nError: token hvs.CAESIabcdefghijklmnop local node not active\nmore") self.assertNotIn("hvs.CAES", line) self.assertIn("local node not active", line) self.assertLessEqual(len(A.sanitize_status_line(b"x" * 500) or ""), 200) self.assertIsNone(A.typed("5", int)) self.assertIsNone(A.typed(1, bool)) self.assertEqual(A.typed("http://openbao-0.openbao-internal:8200", str), "http://openbao-0.openbao-internal:8200") def test_normalise_and_project(self): source = {"token_ttl": "5m", "token_policies": ["b", "a"], "_note": "x"} ok, projected = A.project(source, {"token_ttl": 300, "token_policies": ["a", "b"], "extra": 1}, ("_note",)) self.assertTrue(ok) self.assertEqual(projected, {"token_policies": ["a", "b"], "token_ttl": 300}) self.assertFalse(A.project(source, {"token_ttl": 600, "token_policies": ["a", "b"]}, ("_note",))[0]) def test_reads_never_include_kv_data(self): self.assertFalse(any("/data/" in p for p in A.ALLOWED_READS)) self.assertIn(A.KV_METADATA_PATH, A.ALLOWED_READS) class VerifierTests(Base): def test_valid_receipts(self): self.assertEqual(self.verify(self.receipt()), "restored") rc = self.receipt() rc["objects"][-1].update({"present": False, "source_match": False, "observed_sha256": None}) rc.update(result="refused", reason="object_missing") self.assertEqual(self.verify(rc), "refused") rc = self.receipt() rc.update(result="aborted", reason="unseal_refused", objects=[]) self.assertEqual(self.verify(rc), "aborted") def test_example_keys_match_verifier_schema(self): rc = json.loads((R.ROOT / "docs/openbao-restore-receipt-v2.example.json").read_text()) self.assertEqual(set(rc), V.TOP) self.assertEqual(set(rc["teardown"]), V.TEARDOWN) self.assertEqual(tuple(o["object"] for o in rc["objects"]), V.REQUIRED_OBJECTS) self.assertEqual(tuple(o["object"] for o in A.OBJECTS), V.REQUIRED_OBJECTS) self.assertEqual(set(A.CLAIMS_NOT_MADE), V.CLAIMS) self.assertEqual(A.ABORT_REASONS, V.ABORT | V.DRILL_ABORT) self.assertEqual(A.DRILL_ABORT_REASONS, V.DRILL_ABORT) self.assertEqual(set(A.DRILL_CLAIMS_NOT_MADE), V.DRILL_CLAIMS) with self.assertRaises(V.Reject): self.verify(rc) # placeholders def reject(self, mutate, code=None, **kw): rc = self.receipt() mutate(rc) with self.assertRaises(V.Reject) as ctx: self.verify(rc, **kw) if code: self.assertEqual(str(ctx.exception), code) def test_adversarial_fixtures(self): share = "Kq3/" + "x9Z+" * 10 cases = { "unknown_field": (lambda r: r.update(notes="ok"), "schema_fields"), "missing_field": (lambda r: r.pop("teardown"), "schema_fields"), "missing_teardown": (lambda r: r["teardown"].update(container_removed=False), "teardown_missing"), "reaper_on_success": (lambda r: r["teardown"].update(performed_by="reaper"), "teardown_missing"), "unrevoked": (lambda r: r["teardown"].update(readback_credential_revoked=False), "teardown_missing"), "data_field": (lambda r: r["objects"][0].update(data={"x": 1}), "object_fields"), "kv_data_path": (lambda r: r["objects"][-1].update( path="platform/data/workloads/keycape/secrets-engine-openbao"), "kv_data_path"), "kv_data_read": (lambda r: r.update(kv_data_read=True), "secret_policy"), "credential_shaped": (lambda r: r.update(signer_principal="hvs.CAESIJ" + "a" * 30), None), "share_shaped": (lambda r: r["objects"][0].update(path=share), "secret_shaped_value"), "hex_key_shaped": (lambda r: r["objects"][0].update(path="f00d" * 16), "secret_shaped_value"), "age_identity": (lambda r: r["objects"][0].update(path="AGE-SECRET-KEY-1QQQ"), "secret_shaped_value"), "pem": (lambda r: r["objects"][0].update(path="-----BEGIN PRIVATE"), "secret_shaped_value"), "placeholder_digest": (lambda r: r["snapshot"].update(ciphertext_sha256="sha256:" + "1" * 64), "snapshot_digest"), "placeholder_text": (lambda r: r["objects"][0].update(path="replace-with-path"), "placeholder"), "untyped_digest": (lambda r: r.update(admission_sha256=hashlib.sha256(b"x").hexdigest()), "admission_digest"), "claims_missing": (lambda r: r["claims_not_made"].remove("consumer_continuity"), "claims_not_made"), "deadline_overrun": (lambda r: r.update(elapsed_seconds=4000, started_at=A.iso( A.parse_iso(r["finished_at"]) - timedelta(seconds=4000))), "deadline_overrun"), "original_short_deadline": (lambda r: r.update(material="original", anchor_protected=True, deadline_seconds=600), "deadline"), "original_unprotected_anchor": (lambda r: r.update(material="original"), "anchor_unprotected"), "restored_with_mismatch": (lambda r: r["objects"][3].update(source_match=False), "result_inconsistent"), "restored_network": (lambda r: r["network"].update(interfaces=["eth0", "lo"]), "network_not_contained"), "cluster_mismatch": (lambda r: r["restored"].update(cluster_id_match=False), "restore_inconsistent"), "objects_incomplete": (lambda r: r["objects"].pop(), "objects_incomplete"), "bad_reason": (lambda r: r.update(result="aborted", reason="because"), "result_inconsistent"), "phase_b": (lambda r: r.update(phase="B"), "phase"), "source_hash": (lambda r: r["source_sha256"].update( {"openbao/auth/secrets-engine-jwt-role.json": digest("zz")}), "source_hash_mismatch"), "source_commit": (lambda r: r.update(adapter_commit="a" * 40), "source_commit_mismatch"), } for name, (mutate, code) in cases.items(): with self.subTest(name): self.reject(mutate, code) def test_not_after_window(self): self.reject(lambda r: None, "verification_window_closed", not_after=datetime(2026, 1, 1, tzinfo=UTC)) def test_duplicate_keys_rejected_by_cli(self): path = self.work / "dup.json" path.write_text('{"schema": "a", "schema": "b"}') done = subprocess.run([sys.executable, str(R.VERIFIER), "--receipt", str(path), "--source-root", str(self.root)], capture_output=True) self.assertEqual(done.returncode, 1) self.assertIn(b"duplicate_key", done.stdout) def test_cli_pass_prints_digest_only(self): path = self.work / "good.json" path.write_text(json.dumps(self.receipt())) done = subprocess.run([sys.executable, str(R.VERIFIER), "--receipt", str(path), "--source-root", str(self.root), "--expect-result", "restored"], capture_output=True) self.assertEqual(done.returncode, 0, done.stdout) self.assertRegex(done.stdout.decode(), r"^PASS: restored receipt sha256:[a-f0-9]{64}\n$") class LauncherTests(Base): launcher = R.ROOT / "scripts/openbao-isolated-restore/run-isolated-restore.sh" def run_launcher(self, *args): return subprocess.run(["bash", str(self.launcher), *args], capture_output=True, stdin=subprocess.DEVNULL) def test_refusals_before_any_prompt(self): f = self.work / "f" f.write_text("x") self.assertIn(b"Usage", self.run_launcher().stdout) self.assertIn(b"exact sha256", self.run_launcher(str(f), str(f), "abc", str(f), str(self.work)).stdout) out = self.run_launcher(str(f), str(f), digest("x"), str(f), str(self.work)).stdout self.assertIn(b"interactive terminal", out) self.assertNotIn(b"hidden", out) def test_preflight_refuses_unsigned_admission_without_prompt(self): adm = self.work / "adm.json" raw = json.dumps(self.admission(material="original")).encode() adm.write_bytes(raw) os.chmod(adm, 0o600) done = subprocess.run([sys.executable, str(R.ROOT / "scripts/openbao-isolated-restore/preflight.py"), "--admission", str(adm), "--admission-sig", str(adm), "--admission-sha256", A.sha256_bytes(raw), "--ciphertext", str(adm), "--receipt", str(self.work / "r.json")], capture_output=True, stdin=subprocess.DEVNULL) self.assertEqual(done.returncode, 1) self.assertIn(b"admission_refused", done.stdout) self.assertNotIn(b"hidden", done.stdout + done.stderr) if __name__ == "__main__": unittest.main()