coulomb/railiance-platform
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

RPF-WP-0064-T03: v2 restore receipt and independent verifier

Browse source 
Receipt schema railiance-openbao-restore-receipt/v2 extending the drill
template, with typed sha256 digests only. The verifier does not import the
adapter: exact schema, result consistency, required teardown, claims_not_made,
deadline and --not-after, secret-shape/placeholder/data refusal, and source
hash comparison against a clean checkout at adapter_commit. 17 offline tests
including 28 adversarial receipt fixtures.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 4384@bnt-lap001
Assistant-Session: 80d1f3da-7a91-4519-9737-9ba128b4da64
This commit is contained in:
codex 2026-10-23 20:49:35 +02:00
parent 0cc3281ab3
commit 7f67c9c244
4 changed files with 983 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

159
docs/openbao-restore-receipt-v2.example.json Normal file
View file

@ -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"
]
}

273
scripts/openbao-isolated-restore/verify-restore-receipt.py Normal file
View file

@ -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"(?<![a-f0-9])[a-f0-9]{32,}(?![a-f0-9])"),
re.compile(r"eyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\."),
)
BASE64_BLOB = re.compile(r"(?<![:A-Za-z0-9+/])[A-Za-z0-9+/]{40,}={0,2}(?![A-Za-z0-9+/=])")
def base64_blob(text: str) -> 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())

263
tests/openbao_restore_rehearsal.py Normal file
View file

@ -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]

288
tests/test_openbao_isolated_restore.py Normal file
View file

@ -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()