a2da4085e0
Adds --mode break-glass-drill with its own signed admission (drill: "break-glass", attempt id RPF-WP-0064-T10-*, snapshot_label). After the phase A restore and unseal it checks the held root token with lookup-self only, runs a fresh generate-root ceremony with re-entered shares, does one read-only sys/auth action, revokes the generated token (204 then 403) and drills a cancelled attempt. Phase A is unchanged. The verifier types the receipt extension and rejects token, accessor, OTP, nonce, encoded-token and share shapes. Offline 34 and native 16 pass. Founder runbooks for the drill and the read-only production SSO-down walk are in the T10 section. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 4515@bnt-lap001 Assistant-Session: 0bf96127-6e04-41be-bfda-d11717c3f354
306 lines
16 KiB
Python
306 lines
16 KiB
Python
"""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()
|