a2da4085e0
Adds --mode break-glass-drill with its own signed admission (drill: "break-glass", attempt id RPF-WP-0064-T10-*, snapshot_label). After the phase A restore and unseal it checks the held root token with lookup-self only, runs a fresh generate-root ceremony with re-entered shares, does one read-only sys/auth action, revokes the generated token (204 then 403) and drills a cancelled attempt. Phase A is unchanged. The verifier types the receipt extension and rejects token, accessor, OTP, nonce, encoded-token and share shapes. Offline 34 and native 16 pass. Founder runbooks for the drill and the read-only production SSO-down walk are in the T10 section. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 4515@bnt-lap001 Assistant-Session: 0bf96127-6e04-41be-bfda-d11717c3f354
72 lines
3.5 KiB
Python
72 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Metadata-only preflight for the isolated original-OpenBao restore (RPF-WP-0064-T02).
|
|
|
|
Runs before any custody prompt. Never asks for, opens or decrypts custody
|
|
material. Checks: signed admission (exact bytes, sha256, fixed SSH namespace,
|
|
protected allowed_signers anchor, window), exact committed source closure,
|
|
pinned image digest present locally (``--pull=never``), ciphertext size and
|
|
sha256, required tools, ``/dev/shm`` tmpfs and that no prior container of this
|
|
adapter exists. Prints only OK/ERROR, never values.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import importlib.util
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
_spec = importlib.util.spec_from_file_location("restore_adapter", HERE / "restore_adapter.py")
|
|
adapter = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(adapter)
|
|
require, Refused = adapter.require, adapter.Refused
|
|
|
|
|
|
def check_local(adm: dict, ciphertext: Path, docker: str = "docker") -> None:
|
|
for tool in ("age", "age-keygen", "ssh-keygen", "git", docker):
|
|
require(shutil.which(tool), "preflight_refused")
|
|
mounts = Path("/proc/mounts").read_text().splitlines()
|
|
require(any(line.split()[1:3] == ["/dev/shm", "tmpfs"] for line in mounts), "preflight_refused")
|
|
inspect = subprocess.run([docker, "image", "inspect", adm["image"], "--format", "{{json .RepoDigests}}"],
|
|
capture_output=True, timeout=30)
|
|
require(inspect.returncode == 0 and adm["image"] in json.loads(inspect.stdout or b"[]"), "preflight_refused")
|
|
stale = subprocess.run([docker, "ps", "-a", "-q", "--filter", "name=rpf64-restore-"],
|
|
capture_output=True, timeout=30)
|
|
require(stale.returncode == 0 and not stale.stdout.strip(), "preflight_refused")
|
|
raw = adapter.read_owner_file(ciphertext, 512 * 1024 * 1024)
|
|
snap = adm["snapshot"]
|
|
require(len(raw) == snap["ciphertext_bytes"] and adapter.sha256_bytes(raw) == snap["ciphertext_sha256"],
|
|
"ciphertext_mismatch")
|
|
require(raw.startswith(b"age-encryption.org/v1\n"), "ciphertext_mismatch")
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0])
|
|
for name in ("admission", "admission-sig", "admission-sha256", "ciphertext", "receipt"):
|
|
parser.add_argument("--" + name, required=True)
|
|
parser.add_argument("--mode", choices=(adapter.PHASE_A, adapter.DRILL_MODE), default=adapter.PHASE_A)
|
|
opts = parser.parse_args(argv)
|
|
try:
|
|
adm, _raw, _signer, _protected = adapter.load_admission(
|
|
Path(opts.admission), Path(opts.admission_sig), opts.admission_sha256, adapter.anchor_path(), opts.mode)
|
|
require(adm["material"] == "original", "admission_refused")
|
|
adapter.check_window(adm)
|
|
adapter.check_source_closure(adm, adapter.ROOT)
|
|
check_local(adm, Path(opts.ciphertext))
|
|
receipt = Path(opts.receipt)
|
|
require(receipt.is_absolute() and not receipt.exists() and receipt.parent.is_dir(), "preflight_refused")
|
|
except Refused as exc:
|
|
print(f"ERROR: preflight refused ({exc.reason}); no custody input requested.")
|
|
return 1
|
|
except Exception:
|
|
print("ERROR: preflight refused (internal_error); no custody input requested.")
|
|
return 1
|
|
print("OK: signed admission, source closure, pinned image, ciphertext and local target verified.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|