diff --git a/docs/openbao-restore-receipt-v2.example.json b/docs/openbao-restore-receipt-v2.example.json new file mode 100644 index 0000000..e3ebc85 --- /dev/null +++ b/docs/openbao-restore-receipt-v2.example.json @@ -0,0 +1,159 @@ +{ + "schema": "railiance-openbao-restore-receipt/v2", + "extends": "docs/openbao-restore-drill-evidence.example.json", + "attempt_id": "RPF-WP-0064-T07-YYYYMMDD-01", + "material": "original", + "phase": "A", + "admission_sha256": "sha256:5555555555555555555555555555555555555555555555555555555555555555", + "admission_namespace": "railiance-openbao-restore-attempt", + "signer_principal": "founder", + "signer_fingerprint": "replace-with-ssh-signer-fingerprint", + "anchor_protected": true, + "abort_operator": "founder", + "adapter_commit": "replace-with-40-hex-adapter-commit", + "source_sha256": { + "docs/contracts/secrets-engine-openbao-jwt-login.yaml": "sha256:0000000000000000000000000000000000000000000000000000000000000000", + "openbao/auth/keycape-services-config.json": "sha256:0000000000000000000000000000000000000000000000000000000000000000", + "openbao/auth/netkingdom-oidc-config.json": "sha256:0000000000000000000000000000000000000000000000000000000000000000", + "openbao/auth/netkingdom-platform-admin-role.json": "sha256:0000000000000000000000000000000000000000000000000000000000000000", + "openbao/auth/secrets-engine-jwt-role.json": "sha256:0000000000000000000000000000000000000000000000000000000000000000", + "openbao/policies/secrets-engine-login-self.hcl": "sha256:0000000000000000000000000000000000000000000000000000000000000000", + "scripts/openbao-isolated-restore/preflight.py": "sha256:0000000000000000000000000000000000000000000000000000000000000000", + "scripts/openbao-isolated-restore/restore_adapter.py": "sha256:0000000000000000000000000000000000000000000000000000000000000000", + "scripts/openbao-isolated-restore/run-isolated-restore.sh": "sha256:0000000000000000000000000000000000000000000000000000000000000000", + "scripts/openbao-isolated-restore/verify-restore-receipt.py": "sha256:0000000000000000000000000000000000000000000000000000000000000000" + }, + "started_at": "YYYY-MM-DDTHH:MM:SSZ", + "finished_at": "YYYY-MM-DDTHH:MM:SSZ", + "not_after": "YYYY-MM-DDTHH:MM:SSZ", + "deadline_seconds": 2700, + "elapsed_seconds": 0, + "image": "quay.io/openbao/openbao@sha256:0000000000000000000000000000000000000000000000000000000000000000", + "image_id": "sha256:6666666666666666666666666666666666666666666666666666666666666666", + "snapshot": { + "ciphertext_sha256": "sha256:1111111111111111111111111111111111111111111111111111111111111111", + "ciphertext_bytes": 0, + "plaintext_sha256": "sha256:2222222222222222222222222222222222222222222222222222222222222222", + "source_cluster_id": "replace-with-source-cluster-id", + "source_applied_index": 0, + "source_receipt_sha256": "sha256:3333333333333333333333333333333333333333333333333333333333333333", + "ciphertext_verified": true, + "plaintext_verified": true + }, + "restored": { + "cluster_id": "replace-with-restored-cluster-id", + "applied_index": 0, + "sealed_after_restore": true, + "unsealed_with_original_shares": true, + "cluster_id_match": true, + "applied_index_not_behind": true, + "throwaway_init_invalidated": true + }, + "network": { + "network_mode": "none", + "docker_networks": [ + "none" + ], + "interfaces": [ + "lo" + ], + "published_ports": [], + "mounts": [], + "egress_refused": true + }, + "objects": [ + { + "object": "auth_mount_keycape_services", + "path": "sys/auth", + "present": true, + "source_match": true, + "source_path": null, + "observed_sha256": "sha256:4444444444444444444444444444444444444444444444444444444444444444" + }, + { + "object": "auth_mount_netkingdom", + "path": "sys/auth", + "present": true, + "source_match": true, + "source_path": null, + "observed_sha256": "sha256:4444444444444444444444444444444444444444444444444444444444444444" + }, + { + "object": "kv_mount_platform", + "path": "sys/mounts", + "present": true, + "source_match": true, + "source_path": null, + "observed_sha256": "sha256:4444444444444444444444444444444444444444444444444444444444444444" + }, + { + "object": "keycape_services_config", + "path": "auth/keycape-services/config", + "present": true, + "source_match": true, + "source_path": "openbao/auth/keycape-services-config.json", + "observed_sha256": "sha256:4444444444444444444444444444444444444444444444444444444444444444" + }, + { + "object": "secrets_engine_jwt_role", + "path": "auth/keycape-services/role/secrets-engine-openbao", + "present": true, + "source_match": true, + "source_path": "openbao/auth/secrets-engine-jwt-role.json", + "observed_sha256": "sha256:4444444444444444444444444444444444444444444444444444444444444444" + }, + { + "object": "netkingdom_oidc_config", + "path": "auth/netkingdom/config", + "present": true, + "source_match": true, + "source_path": "openbao/auth/netkingdom-oidc-config.json", + "observed_sha256": "sha256:4444444444444444444444444444444444444444444444444444444444444444" + }, + { + "object": "platform_admin_role", + "path": "auth/netkingdom/role/platform-admin", + "present": true, + "source_match": true, + "source_path": "openbao/auth/netkingdom-platform-admin-role.json", + "observed_sha256": "sha256:4444444444444444444444444444444444444444444444444444444444444444" + }, + { + "object": "policy_secrets_engine_login_self", + "path": "sys/policies/acl/secrets-engine-login-self", + "present": true, + "source_match": true, + "source_path": "openbao/policies/secrets-engine-login-self.hcl", + "observed_sha256": "sha256:4444444444444444444444444444444444444444444444444444444444444444" + }, + { + "object": "kv_registration_metadata", + "path": "platform/metadata/workloads/keycape/secrets-engine-openbao", + "present": true, + "source_match": true, + "source_path": null, + "observed_sha256": "sha256:4444444444444444444444444444444444444444444444444444444444444444" + } + ], + "result": "restored", + "reason": "all_required_objects_match", + "teardown": { + "container_removed": true, + "container_absent_verified": true, + "plaintext_memfd_closed": true, + "custody_memfd_closed": true, + "readback_credential_revoked": true, + "docker_volumes_created": 0, + "performed_by": "session" + }, + "kv_data_read": false, + "no_secret_material_recorded": true, + "claims_not_made": [ + "consumer_continuity", + "live_jwt_login", + "phase_b_network", + "owner_acceptance", + "recurring_restore_assurance", + "production_redirect" + ] +} diff --git a/scripts/openbao-isolated-restore/verify-restore-receipt.py b/scripts/openbao-isolated-restore/verify-restore-receipt.py new file mode 100644 index 0000000..f5b7045 --- /dev/null +++ b/scripts/openbao-isolated-restore/verify-restore-receipt.py @@ -0,0 +1,273 @@ +#!/usr/bin/env python3 +"""Independent verifier for v2 isolated OpenBao restore receipts (RPF-WP-0064-T03). + +Does not import the adapter. Checks the exact schema, typed digests, result +consistency, deadline, required teardown evidence, claims not made, refuses +placeholders and anything shaped like a credential, share, key or KV value, +and compares ``source_sha256`` against a clean checkout at ``adapter_commit``. + +Exit 0 = receipt accepted. Prints only PASS/FAIL and a reason code. +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import subprocess +import sys +from datetime import UTC, datetime +from pathlib import Path + +SCHEMA = "railiance-openbao-restore-receipt/v2" +NAMESPACE = "railiance-openbao-restore-attempt" +MAX_DURATION = 2700 +TOP = {"schema", "extends", "attempt_id", "material", "phase", "admission_sha256", "admission_namespace", + "signer_principal", "signer_fingerprint", "anchor_protected", "abort_operator", "adapter_commit", + "source_sha256", "started_at", "finished_at", "not_after", "deadline_seconds", "elapsed_seconds", + "image", "image_id", "snapshot", "restored", "network", "objects", "result", "reason", "teardown", + "kv_data_read", "no_secret_material_recorded", "claims_not_made"} +SNAPSHOT = {"ciphertext_sha256", "ciphertext_bytes", "plaintext_sha256", "source_cluster_id", + "source_applied_index", "source_receipt_sha256", "ciphertext_verified", "plaintext_verified"} +RESTORED = {"cluster_id", "applied_index", "sealed_after_restore", "unsealed_with_original_shares", + "cluster_id_match", "applied_index_not_behind", "throwaway_init_invalidated"} +NETWORK = {"network_mode", "docker_networks", "interfaces", "published_ports", "mounts", "egress_refused"} +TEARDOWN = {"container_removed", "container_absent_verified", "plaintext_memfd_closed", "custody_memfd_closed", + "readback_credential_revoked", "docker_volumes_created", "performed_by"} +OBJECT = {"object", "path", "present", "source_match", "source_path", "observed_sha256"} +REQUIRED_OBJECTS = ( + "auth_mount_keycape_services", "auth_mount_netkingdom", "kv_mount_platform", "keycape_services_config", + "secrets_engine_jwt_role", "netkingdom_oidc_config", "platform_admin_role", + "policy_secrets_engine_login_self", "kv_registration_metadata", +) +CLAIMS = {"consumer_continuity", "live_jwt_login", "phase_b_network", "owner_acceptance", + "recurring_restore_assurance", "production_redirect"} +ABORT = {"admission_refused", "preflight_refused", "custody_identity_refused", "ciphertext_mismatch", + "decrypt_failed", "plaintext_mismatch", "target_start_failed", "network_containment_failed", + "throwaway_init_failed", "restore_failed", "unseal_refused", "cluster_mismatch", "applied_index_behind", + "readback_credential_failed", "readback_failed", "deadline_exceeded", "operator_abort", "parent_lost", + "internal_error"} +DIGEST = re.compile(r"^sha256:[a-f0-9]{64}$") +TIME = re.compile(r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ$") +UUID = re.compile(r"^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$") +# Secret-shaped values: OpenBao/Vault credentials, age identities, PEM blocks, +# base64 or hex blobs the size of unseal shares/keys, JWT-like triples. +SECRET_SHAPES = ( + re.compile(r"\b(?:hv[sbr]|s|b|r)\.[A-Za-z0-9_-]{20,}"), + re.compile(r"AGE-SECRET-KEY-1", re.I), + re.compile(r"-----BEGIN"), + re.compile(r"(? bool: + """Random base64 (shares, keys) mixes cases and digits; repo paths do not.""" + return any(re.search(r"[A-Z]", m) and re.search(r"[a-z]", m) and re.search(r"[0-9+]", m) + for m in BASE64_BLOB.findall(text)) +PLACEHOLDERS = re.compile(r"(?i)replace-with|placeholder|todo|changeme|<[^>]*>|example\.invalid|YYYY") +FORBIDDEN_KEYS = re.compile(r"(?i)^(?:data|value|values|secret|secrets|token|root_token|unseal_keys.*|keys?|" + r"otp|nonce|encoded_.*|password|plaintext)$") + + +class Reject(Exception): + pass + + +def need(condition: object, code: str) -> None: + if not condition: + raise Reject(code) + + +def strict(raw: bytes) -> object: + def unique(pairs): + out = {} + for key, value in pairs: + need(key not in out, "duplicate_key") + out[key] = value + return out + return json.loads(raw, object_pairs_hook=unique) + + +def walk(value: object, path: str = "$"): + if isinstance(value, dict): + for key, item in value.items(): + yield path, key, None + yield from walk(item, f"{path}.{key}") + elif isinstance(value, list): + for index, item in enumerate(value): + yield from walk(item, f"{path}[{index}]") + else: + yield path, None, value + + +def digest_ok(value: object) -> bool: + return isinstance(value, str) and bool(DIGEST.fullmatch(value)) and len(set(value[7:])) > 2 + + +def parse_time(value: object) -> datetime: + need(isinstance(value, str) and TIME.fullmatch(value), "bad_time") + return datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=UTC) + + +def check_content(receipt: dict) -> None: + for path, key, value in walk(receipt): + if key is not None: + need(not FORBIDDEN_KEYS.match(key), "forbidden_field") + continue + if isinstance(value, str): + need(not PLACEHOLDERS.search(value), "placeholder") + if path == "$.adapter_commit": + continue + stripped = re.sub(r"sha256:[a-f0-9]{64}", "", value) + for shape in SECRET_SHAPES: + need(not shape.search(stripped), "secret_shaped_value") + need(not base64_blob(stripped), "secret_shaped_value") + + +def check_sources(receipt: dict, root: Path) -> None: + sources = receipt["source_sha256"] + need(isinstance(sources, dict) and sources, "sources") + head = subprocess.run(["git", "-C", str(root), "rev-parse", "HEAD"], capture_output=True, timeout=10) + need(head.returncode == 0 and head.stdout.decode().strip() == receipt["adapter_commit"], "source_commit_mismatch") + dirty = subprocess.run(["git", "-C", str(root), "status", "--porcelain", "--", *sources], + capture_output=True, timeout=10) + need(dirty.returncode == 0 and not dirty.stdout.strip(), "source_checkout_dirty") + for name, expected in sources.items(): + need(digest_ok(expected) and not name.startswith("/") and ".." not in Path(name).parts, "sources") + path = root / name + need(path.is_file() and not path.is_symlink(), "source_missing") + need("sha256:" + hashlib.sha256(path.read_bytes()).hexdigest() == expected, "source_hash_mismatch") + + +def verify(receipt: object, *, source_root: Path, not_after: datetime | None = None, + now: datetime | None = None) -> str: + need(isinstance(receipt, dict) and set(receipt) == TOP, "schema_fields") + need(receipt["schema"] == SCHEMA and receipt["admission_namespace"] == NAMESPACE, "schema") + need(receipt["extends"] == "docs/openbao-restore-drill-evidence.example.json", "schema") + need(receipt["phase"] == "A" and receipt["material"] in ("original", "synthetic"), "phase") + need(isinstance(receipt["attempt_id"], str) and re.fullmatch(r"RPF-WP-0064-T0[47]-[A-Za-z0-9-]{3,64}", + receipt["attempt_id"]), "attempt_id") + need(digest_ok(receipt["admission_sha256"]), "admission_digest") + need(isinstance(receipt["signer_principal"], str) and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._@-]{0,127}", + receipt["signer_principal"]), "signer") + need(isinstance(receipt["signer_fingerprint"], str) + and re.fullmatch(r"SHA256:[A-Za-z0-9+/]{43}", receipt["signer_fingerprint"]), "signer") + need(receipt["anchor_protected"] is True or receipt["material"] == "synthetic", "anchor_unprotected") + need(isinstance(receipt["adapter_commit"], str) and re.fullmatch(r"[a-f0-9]{40}", receipt["adapter_commit"]), + "adapter_commit") + image = receipt["image"] + need(isinstance(image, str) and re.fullmatch(r"quay\.io/openbao/openbao@sha256:[a-f0-9]{64}", image) + and digest_ok(image.split("@", 1)[1]), "image") + need(receipt["image_id"] is None or digest_ok(receipt["image_id"]), "image") + need(receipt["no_secret_material_recorded"] is True and receipt["kv_data_read"] is False, "secret_policy") + need(isinstance(receipt["claims_not_made"], list) and set(receipt["claims_not_made"]) == CLAIMS + and len(receipt["claims_not_made"]) == len(CLAIMS), "claims_not_made") + + started, finished = parse_time(receipt["started_at"]), parse_time(receipt["finished_at"]) + deadline = receipt["deadline_seconds"] + need(type(deadline) is int and 60 <= deadline <= MAX_DURATION, "deadline") + need(receipt["material"] == "synthetic" or deadline == MAX_DURATION, "deadline") + elapsed = receipt["elapsed_seconds"] + need(type(elapsed) is int and 0 <= elapsed == int((finished - started).total_seconds()), "elapsed") + need(elapsed <= deadline + 120, "deadline_overrun") + need(finished <= parse_time(receipt["not_after"]) or receipt["reason"] == "deadline_exceeded", "window") + if not_after is not None: + need((now or datetime.now(UTC)) <= not_after, "verification_window_closed") + + snap = receipt["snapshot"] + need(isinstance(snap, dict) and set(snap) == SNAPSHOT, "snapshot_fields") + for field in ("ciphertext_sha256", "plaintext_sha256", "source_receipt_sha256"): + need(digest_ok(snap[field]), "snapshot_digest") + need(type(snap["ciphertext_bytes"]) is int and snap["ciphertext_bytes"] > 0, "snapshot") + need(isinstance(snap["source_cluster_id"], str) and UUID.fullmatch(snap["source_cluster_id"]), "snapshot") + need(type(snap["source_applied_index"]) is int and snap["source_applied_index"] > 0, "snapshot") + need(all(type(snap[f]) is bool for f in ("ciphertext_verified", "plaintext_verified")), "snapshot") + + restored = receipt["restored"] + need(isinstance(restored, dict) and set(restored) == RESTORED, "restored_fields") + need(restored["cluster_id"] is None or (isinstance(restored["cluster_id"], str) + and UUID.fullmatch(restored["cluster_id"])), "restored") + need(restored["applied_index"] is None or type(restored["applied_index"]) is int, "restored") + network = receipt["network"] + need(isinstance(network, dict) and set(network) == NETWORK, "network_fields") + teardown = receipt["teardown"] + need(isinstance(teardown, dict) and set(teardown) == TEARDOWN, "teardown_fields") + need(teardown["container_removed"] is True and teardown["container_absent_verified"] is True + and teardown["plaintext_memfd_closed"] is True and teardown["custody_memfd_closed"] is True + and teardown["docker_volumes_created"] == 0 and teardown["performed_by"] in ("session", "reaper"), + "teardown_missing") + need(teardown["readback_credential_revoked"] in (True, False, None), "teardown_fields") + + objects = receipt["objects"] + need(isinstance(objects, list), "objects") + for row in objects: + need(isinstance(row, dict) and set(row) == OBJECT, "object_fields") + need(type(row["present"]) is bool and type(row["source_match"]) is bool, "object_fields") + need(row["observed_sha256"] is None or digest_ok(row["observed_sha256"]), "object_digest") + need(not row["present"] or row["observed_sha256"] is not None, "object_digest") + need(not row["source_match"] or row["present"], "object_fields") + need("/data/" not in row["path"], "kv_data_path") + need(row["source_path"] is None or row["source_path"] in receipt["source_sha256"], "object_source") + names = [row["object"] for row in objects] + + result, reason = receipt["result"], receipt["reason"] + if result in ("restored", "refused"): + need(tuple(names) == REQUIRED_OBJECTS, "objects_incomplete") + need(snap["ciphertext_verified"] and snap["plaintext_verified"], "restore_inconsistent") + need(restored["sealed_after_restore"] is True and restored["unsealed_with_original_shares"] is True + and restored["cluster_id_match"] is True and restored["applied_index_not_behind"] is True + and restored["throwaway_init_invalidated"] is True + and restored["cluster_id"] == snap["source_cluster_id"] + and restored["applied_index"] >= snap["source_applied_index"], "restore_inconsistent") + need(network["network_mode"] == "none" and network["docker_networks"] == ["none"] + and network["interfaces"] == ["lo"] and network["published_ports"] == [] and network["mounts"] == [] + and network["egress_refused"] is True, "network_not_contained") + need(teardown["readback_credential_revoked"] is True and teardown["performed_by"] == "session", + "teardown_missing") + need(digest_ok(receipt["image_id"]), "image") + all_ok = all(r["present"] and r["source_match"] for r in objects) + if result == "restored": + need(all_ok and reason == "all_required_objects_match", "result_inconsistent") + else: + missing = any(not r["present"] for r in objects) + need(not all_ok and reason == ("object_missing" if missing else "object_mismatch"), "result_inconsistent") + elif result == "aborted": + need(reason in ABORT, "result_inconsistent") + need(not objects or len(objects) == len(REQUIRED_OBJECTS), "objects") + if network["docker_networks"]: + need(network["docker_networks"] == ["none"] and network["published_ports"] == [], "network_not_contained") + else: + raise Reject("result") + check_content(receipt) + check_sources(receipt, source_root) + return result + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) + parser.add_argument("--receipt", required=True) + parser.add_argument("--source-root", required=True, help="clean checkout at the receipt's adapter_commit") + parser.add_argument("--not-after", help="refuse after this UTC time (YYYY-MM-DDTHH:MM:SSZ)") + parser.add_argument("--expect-result", choices=("restored", "refused", "aborted")) + opts = parser.parse_args(argv) + try: + path = Path(opts.receipt) + need(path.is_file() and not path.is_symlink() and path.stat().st_size <= 256 * 1024, "receipt_file") + raw = path.read_bytes() + result = verify(strict(raw), source_root=Path(opts.source_root), + not_after=parse_time(opts.not_after) if opts.not_after else None) + need(opts.expect_result in (None, result), "unexpected_result") + except Reject as exc: + print(f"FAIL: {exc}") + return 1 + except (ValueError, OSError, subprocess.SubprocessError): + print("FAIL: unreadable_receipt") + return 1 + print(f"PASS: {result} receipt sha256:{hashlib.sha256(raw).hexdigest()}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/openbao_restore_rehearsal.py b/tests/openbao_restore_rehearsal.py new file mode 100644 index 0000000..bdd8d82 --- /dev/null +++ b/tests/openbao_restore_rehearsal.py @@ -0,0 +1,263 @@ +"""Synthetic-material rehearsal fixture for RPF-WP-0064-T04. + +Everything here is generated per run: SSH admission key, age identity, Shamir +shares, root credentials and the KV value. Source containers run the pinned +OpenBao image with ``--network none`` and tmpfs storage only and are removed. +No real snapshot, key, share or cluster is touched. +""" +from __future__ import annotations + +import importlib.util +import json +import os +import secrets +import shutil +import subprocess +import tempfile +import time +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +REHEARSAL_IMAGE = "quay.io/openbao/openbao:2.5.4" + + +def load(name: str, rel: str): + spec = importlib.util.spec_from_file_location(name, ROOT / rel) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +adapter = load("restore_adapter", "scripts/openbao-isolated-restore/restore_adapter.py") +VERIFIER = ROOT / "scripts/openbao-isolated-restore/verify-restore-receipt.py" + + +def sh(argv, **kw) -> subprocess.CompletedProcess: + kw.setdefault("capture_output", True) + kw.setdefault("timeout", 120) + return subprocess.run(argv, **kw) + + +def docker_status() -> tuple[bool, str]: + if not shutil.which("docker") or not shutil.which("age") or not shutil.which("ssh-keygen"): + return False, "docker/age/ssh-keygen unavailable" + if sh(["docker", "version", "--format", "{{.Server.Version}}"], timeout=20).returncode != 0: + return False, "docker daemon unavailable" + inspect = sh(["docker", "image", "inspect", REHEARSAL_IMAGE, "--format", "{{json .RepoDigests}}"], timeout=20) + if inspect.returncode != 0: + return False, f"{REHEARSAL_IMAGE} not pulled locally (make openbao-isolated-restore-image)" + digests = [d for d in json.loads(inspect.stdout) if d.startswith(adapter.IMAGE_REPOSITORY + "@")] + return (True, digests[0]) if digests else (False, "image has no repository digest") + + +def make_source_root(base: Path, mutate: dict[str, bytes] | None = None) -> tuple[Path, str]: + """A clean git checkout holding exactly the adapter source closure.""" + root = base / ("src-" + secrets.token_hex(3)) + for name in adapter.SOURCE_PATHS: + target = root / name + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes((mutate or {}).get(name, (ROOT / name).read_bytes())) + env = {**os.environ, "GIT_AUTHOR_NAME": "rehearsal", "GIT_AUTHOR_EMAIL": "rehearsal@invalid", + "GIT_COMMITTER_NAME": "rehearsal", "GIT_COMMITTER_EMAIL": "rehearsal@invalid"} + for argv in (["git", "init", "-q"], ["git", "add", "-A"], + ["git", "-c", "core.hooksPath=/dev/null", "commit", "-q", "-m", "rehearsal closure"]): + done = sh(argv, cwd=root, env=env) + assert done.returncode == 0, done.stderr + commit = sh(["git", "rev-parse", "HEAD"], cwd=root).stdout.decode().strip() + return root, commit + + +class SyntheticBao: + """A generated-material OpenBao in a networkless container (source side).""" + + def __init__(self, image: str): + self.image = image + self.name = "rpf64-rehearsal-src-" + secrets.token_hex(4) + self.credential = "" + self.shares: list[str] = [] + + def __enter__(self): + argv = ["docker", "run", "--detach", "--pull=never", "--name", self.name, "--network", "none", + "--user", "100:1000", "--read-only", "--cap-drop", "ALL", "--security-opt", "no-new-privileges"] + for target, options in adapter.TMPFS.items(): + argv += ["--tmpfs", f"{target}:{options}"] + argv += ["-e", "BAO_LOCAL_CONFIG=" + adapter.SERVER_CONFIG, "-e", "BAO_ADDR=http://127.0.0.1:8200", + self.image, "server"] + assert sh(argv).returncode == 0 + try: + return self._start() + except BaseException: + sh(["docker", "rm", "-f", "-v", self.name]) + raise + + def _start(self): + for _ in range(60): + if self.bao(["status", "-format=json"]).returncode in (0, 2): + break + time.sleep(0.5) + init = json.loads(self.bao(["operator", "init", "-key-shares=3", "-key-threshold=2", + "-format=json"], check=True).stdout) + self.shares, self.credential = init["unseal_keys_b64"], init["root_token"] + for share in self.shares[:2]: + self.bao(["write", "sys/unseal", "-"], body=json.dumps({"key": share}).encode(), check=True) + for _ in range(60): + leader = self.bao(["read", "-format=json", "sys/leader"]) + if leader.returncode == 0 and json.loads(leader.stdout)["data"].get("is_self"): + break + time.sleep(0.5) + return self + + def __exit__(self, *exc): + sh(["docker", "rm", "-f", "-v", self.name]) + + def bao(self, args, *, body: bytes = b"", auth: bool = False, check: bool = False): + script = 'IFS= read -r t; BAO_TOKEN="$t"; export BAO_TOKEN; exec "$@"' if auth else 'exec "$@"' + data = (self.credential.encode() + b"\n" + body) if auth else body + done = sh(["docker", "exec", "-i", self.name, "sh", "-c", script, "sh", "bao", *args], input=data) + if check: + assert done.returncode == 0, done.stderr.decode()[-400:] + return done + + def put(self, path: str, payload: dict): + self.bao(["write", path, "-"], body=json.dumps(payload).encode(), auth=True, check=True) + + def seed(self, *, kv_registration: bool = True, jwt_role: bool = True) -> None: + """Mirror the real source shapes (T01) with generated values only.""" + src = lambda rel: json.loads((ROOT / rel).read_bytes()) # noqa: E731 + self.bao(["auth", "enable", "-path=keycape-services", "jwt"], auth=True, check=True) + self.bao(["auth", "enable", "-path=netkingdom", "oidc"], auth=True, check=True) + self.bao(["secrets", "enable", "-path=platform", "kv-v2"], auth=True, check=True) + self.put("auth/keycape-services/config", + {**src("openbao/auth/keycape-services-config.json"), "skip_jwks_validation": True}) + if jwt_role: + self.put("auth/keycape-services/role/secrets-engine-openbao", + src("openbao/auth/secrets-engine-jwt-role.json")) + oidc = {k: v for k, v in src("openbao/auth/netkingdom-oidc-config.json").items() if k != "_note"} + self.put("auth/netkingdom/config", {**oidc, "oidc_client_secret": "synthetic-" + secrets.token_hex(8), + "skip_jwks_validation": True}) + self.put("auth/netkingdom/role/platform-admin", src("openbao/auth/netkingdom-platform-admin-role.json")) + policy = (ROOT / "openbao/policies/secrets-engine-login-self.hcl").read_bytes() + self.bao(["policy", "write", "secrets-engine-login-self", "-"], body=policy, auth=True, check=True) + if kv_registration: + self.kv_value = "synthetic-kv-" + secrets.token_urlsafe(24) + self.put("platform/data/workloads/keycape/secrets-engine-openbao", + {"data": {"client_credential": self.kv_value}}) + + def facts(self) -> tuple[str, int]: + status = json.loads(self.bao(["read", "-format=json", "sys/seal-status"], check=True).stdout)["data"] + leader = json.loads(self.bao(["read", "-format=json", "sys/leader"], check=True).stdout)["data"] + return status["cluster_id"], leader["raft_applied_index"] + + def snapshot(self) -> bytes: + done = self.bao(["operator", "raft", "snapshot", "save", "/dev/stdout"], auth=True, check=True) + assert done.stdout[:2] == b"\x1f\x8b" + return done.stdout + + +class Rehearsal: + """Generated custody, admission signing and session plumbing.""" + + def __init__(self, image: str): + self.image = image + self.work = Path(tempfile.mkdtemp(prefix="rpf64-rehearsal-", dir="/dev/shm")) + os.chmod(self.work, 0o700) + self.anchor = self.work / "anchor" + self.anchor.mkdir(mode=0o700) + key = self.work / "founder_ed25519" + assert sh(["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-C", "founder@rehearsal", "-f", + str(key)]).returncode == 0 + self.ssh_key = key + pub = (self.work / "founder_ed25519.pub").read_text().strip() + (self.anchor / "allowed_signers").write_text( + f'founder@rehearsal namespaces="{adapter.NAMESPACE}" {pub}\n') + os.environ[adapter.ANCHOR_ENV] = str(self.anchor / "allowed_signers") + self.age_identity, self.age_recipient = self.age_pair() + self.secrets_seen: list[str] = [self.age_identity] + + @staticmethod + def age_pair() -> tuple[str, str]: + out = sh(["age-keygen"]).stdout.decode().splitlines() + identity = next(line for line in out if line.startswith("AGE-SECRET-KEY-1")) + recipient = sh(["age-keygen", "-y"], input=identity.encode()).stdout.decode().strip() + return identity, recipient + + def close(self) -> None: + os.environ.pop(adapter.ANCHOR_ENV, None) + shutil.rmtree(self.work, ignore_errors=True) + + def snapshot_set(self, *, kv_registration=True, jwt_role=True) -> dict: + with SyntheticBao(self.image) as bao: + bao.seed(kv_registration=kv_registration, jwt_role=jwt_role) + cluster_id, index = bao.facts() + plain = bao.snapshot() + shares = list(bao.shares) + self.secrets_seen += [bao.credential, *shares, getattr(bao, "kv_value", "")] + cipher = sh(["age", "-r", self.age_recipient], input=plain).stdout + path = self.work / ("snapshot-" + secrets.token_hex(3) + ".age") + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "wb") as out: + out.write(cipher) + return {"path": path, "cipher": cipher, "plain_sha": adapter.sha256_bytes(plain), "cluster_id": cluster_id, + "index": index, "shares": shares} + + def admission(self, snap: dict, source_root: Path, commit: str, **override) -> dict: + now = adapter.now_utc() + adm = { + "schema": adapter.ADMISSION_SCHEMA, "namespace": adapter.NAMESPACE, + "attempt_id": "RPF-WP-0064-T04-rehearsal-" + secrets.token_hex(3), "phase": "A", + "phase_b_admitted": False, "material": "synthetic", + "not_before": adapter.iso(now.replace(second=0) - __import__("datetime").timedelta(minutes=1)), + "not_after": adapter.iso(now + __import__("datetime").timedelta(hours=1)), + "max_duration_seconds": adapter.MAX_DURATION_SECONDS, "abort_operator": "founder", + "image": self.image, "adapter_commit": commit, + "snapshot": {"ciphertext_sha256": adapter.sha256_bytes(snap["cipher"]), + "ciphertext_bytes": len(snap["cipher"]), "plaintext_sha256": snap["plain_sha"], + "age_recipient": self.age_recipient, "source_cluster_id": snap["cluster_id"], + "source_applied_index": snap["index"], + "source_receipt_sha256": adapter.sha256_bytes(b"synthetic-source-receipt" + os.urandom(8))}, + "unseal": {"key_threshold": 2, "key_shares": 3}, + "readback_credential": "generate-root-from-entered-shares", + "source_sha256": adapter.source_digests(source_root), + } + for key, value in override.items(): + if key in adm["snapshot"]: + adm["snapshot"][key] = value + else: + adm[key] = value + return adm + + def signed(self, adm: dict) -> tuple[Path, Path, str]: + raw = json.dumps(adm, indent=2, sort_keys=True).encode() + path = self.work / (adm["attempt_id"] + ".json") + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "wb") as out: + out.write(raw) + done = sh(["ssh-keygen", "-q", "-Y", "sign", "-f", str(self.ssh_key), "-n", adapter.NAMESPACE, str(path)]) + assert done.returncode == 0, done.stderr + sig = Path(str(path) + ".sig") + os.chmod(sig, 0o600) + return path, sig, adapter.sha256_bytes(raw) + + def session(self, adm: dict, snap: dict, source_root: Path, answers: list, **kw): + path, sig, digest = self.signed(adm) + loaded, _raw, signer, protected = adapter.load_admission(path, sig, digest, Path(os.environ[adapter.ANCHOR_ENV])) + queue = list(answers) + + def prompt(_label: str) -> bytearray: + item = queue.pop(0) + value = item() if callable(item) else item + return bytearray(value.encode()) + receipt = self.work / (adm["attempt_id"] + "-receipt.json") + session = adapter.Session(admission=loaded, admission_sha256=digest, signer=signer, + anchor_is_protected=protected, ciphertext=snap["path"], receipt_path=receipt, + root=source_root, prompt=prompt, say=lambda _m: None, **kw) + return session, receipt + + @staticmethod + def verify(receipt: Path, source_root: Path, *extra: str) -> subprocess.CompletedProcess: + return sh(["python3", str(VERIFIER), "--receipt", str(receipt), "--source-root", str(source_root), *extra]) + + +def leftover_containers() -> list[str]: + done = sh(["docker", "ps", "-a", "--format", "{{.Names}}", "--filter", "name=rpf64-"]) + return [n for n in done.stdout.decode().split() if n] diff --git a/tests/test_openbao_isolated_restore.py b/tests/test_openbao_isolated_restore.py new file mode 100644 index 0000000..0efe817 --- /dev/null +++ b/tests/test_openbao_isolated_restore.py @@ -0,0 +1,288 @@ +"""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.assertIsNone(A.PINNED_IMAGE_DIGEST, "T06 fills the pin; update this test then") + + 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) + 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()