#!/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", "leader_wait"} LEADER_WAIT = {"waited_seconds", "polls", "elected", "last_leader", "last_seal", "last_error"} LEADER_FIELDS = {"ha_enabled": bool, "is_self": bool, "leader_address": str, "leader_cluster_address": str, "raft_committed_index": int, "raft_applied_index": int, "active_time": str} SEAL_FIELDS = {"sealed": bool, "initialized": bool, "storage_type": str, "version": str} def typed_or_none(value: object, kind: type) -> bool: if value is None: return True if kind is str: return isinstance(value, str) and len(value) <= 200 return type(value) is kind 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", "leader_not_elected", "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") wait = restored["leader_wait"] if wait is not None: need(isinstance(wait, dict) and set(wait) == LEADER_WAIT, "leader_wait") need(type(wait["waited_seconds"]) is int and type(wait["polls"]) is int and type(wait["elected"]) is bool, "leader_wait") for name, spec in (("last_leader", LEADER_FIELDS), ("last_seal", SEAL_FIELDS)): part = wait[name] need(isinstance(part, dict) and set(part) == set(spec) and all(typed_or_none(part[k], t) for k, t in spec.items()), "leader_wait") need(wait["last_error"] is None or (isinstance(wait["last_error"], str) and len(wait["last_error"]) <= 200), "leader_wait") 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())