048362e3ef
First attended snapshot run failed closed at verify: the stored AGE_PRIVATE_KEY carries the standard '# created'/'# public key' comment lines. Nothing uploaded. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 680842@bnt-lap001 Assistant-Session: 54eb9e8e-7d6b-4f0c-b7a6-a6db62075883
455 lines
18 KiB
Python
Executable file
455 lines
18 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Attended OpenBao Raft snapshot lane for the railiance01 reboot hold point.
|
|
|
|
RPF-WP-0015-T03, docs/railiance01-coordinated-reboot.md step 6. Runs only as
|
|
the single, silent child of the contained ops-warden login::
|
|
|
|
python3 scripts/openbao-attended-exec.py -- scripts/openbao-reboot-snapshot.py \
|
|
--scenario-id <ID> --receipt-file /abs/new/file --private-dir /abs/0700/dir
|
|
|
|
Order: read-only ``bao status`` -> ``bao operator raft snapshot save`` into a
|
|
fresh 0700 temp dir -> ``bao status`` again (same cluster, index not behind) ->
|
|
age-encrypt to the declared offsite-lane recipient -> decrypt-stream verify
|
|
against the plaintext sha256 with the lane's escrow key (held only in memory)
|
|
-> upload ciphertext with ``If-None-Match: *`` to the governed Nextcloud lane
|
|
-> shred the temp dir (always) -> self-validate and write the receipt.
|
|
|
|
The process prints nothing: Warden fails closed on any child output. The only
|
|
output is the 0600 receipt file (or a value-safe failure record) at
|
|
``--receipt-file``. Only booleans, hashes, ids and status codes are recorded.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import hmac
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import signal
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SCRIPTS = ROOT / "scripts"
|
|
if str(SCRIPTS) not in sys.path:
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
|
|
RECIPIENT_SOURCE = ROOT / "lib" / "railiance-backup-common.sh"
|
|
LANE_PATH = "platform/workloads/railiance/backup/offsite-lane"
|
|
# Established offsite-lane folder used by every existing filesdrop upload.
|
|
UPLOAD_PREFIX = "openbao" # founder-created railiance-backups/openbao/ (2026-10-14)
|
|
SOURCE = {"source_cluster": "railiance01", "source_namespace": "openbao", "source_pod": "openbao-0"}
|
|
OPERATOR = "railiance-platform"
|
|
SCENARIO_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9-]{2,95}$")
|
|
AGE_RECIPIENT_RE = re.compile(r"^age1[023456789acdefghjklmnpqrstuvwxyz]{58}$")
|
|
AGE_HEADER = b"age-encryption.org/v1\n"
|
|
GZIP_MAGIC = b"\x1f\x8b"
|
|
CHUNK = 1024 * 1024
|
|
|
|
|
|
def _load(name: str, filename: str) -> Any:
|
|
spec = importlib.util.spec_from_file_location(name, SCRIPTS / filename)
|
|
if not spec or not spec.loader:
|
|
raise RuntimeError(f"cannot load {filename}")
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
# Reuse, do not copy: contained login env + safe_run from the lease harness,
|
|
# the reviewed receipt validator, and the governed Nextcloud request helper.
|
|
LEASE = _load("audit_core_database_lease_recovery", "audit-core-database-lease-recovery.py")
|
|
PREFLIGHT = LEASE.PREFLIGHT
|
|
import migrate_nextcloud_backup_account as NEXTCLOUD # noqa: E402
|
|
|
|
|
|
class SnapshotError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def fail(message: str) -> None:
|
|
raise SnapshotError(message)
|
|
|
|
|
|
def stamp(value: datetime) -> str:
|
|
return value.astimezone(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
def compact(value: datetime) -> str:
|
|
return value.astimezone(UTC).strftime("%Y%m%dT%H%M%SZ")
|
|
|
|
|
|
def utc_now() -> datetime:
|
|
return datetime.now(UTC)
|
|
|
|
|
|
def declared_recipient(path: Path = RECIPIENT_SOURCE) -> str:
|
|
"""The offsite lane's public recipient, as declared in the backup helper."""
|
|
try:
|
|
text = path.read_text(encoding="utf-8")
|
|
except OSError:
|
|
fail("offsite recipient declaration unavailable")
|
|
found = re.findall(r'RAILIANCE_BACKUP_AGE_PUBLIC_KEY:=(age1[0-9a-z]+)\}', text)
|
|
if len(found) != 1 or not AGE_RECIPIENT_RE.fullmatch(found[0]):
|
|
fail("offsite recipient declaration is not exactly one age recipient")
|
|
return found[0]
|
|
|
|
|
|
def inside_git_tree(path: Path) -> bool:
|
|
for candidate in (path, *path.parents):
|
|
if (candidate / ".git").exists():
|
|
return True
|
|
return False
|
|
|
|
|
|
def private_dir(value: str | None, environ: dict[str, str]) -> Path:
|
|
raw = value or environ.get("HOME", "")
|
|
path = Path(raw)
|
|
if not raw or not path.is_absolute():
|
|
fail("private dir must be an absolute path")
|
|
try:
|
|
info = path.lstat()
|
|
except OSError:
|
|
fail("private dir is missing")
|
|
if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o077:
|
|
fail("private dir must be an owner-only (0700) directory")
|
|
if inside_git_tree(path.resolve()):
|
|
fail("private dir must not be inside a Git tree")
|
|
return path
|
|
|
|
|
|
def open_receipt(value: str) -> int:
|
|
path = Path(value)
|
|
if not path.is_absolute() or inside_git_tree(path.parent.resolve()):
|
|
raise SnapshotError("receipt file must be an absolute path outside Git")
|
|
return os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(CHUNK), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def shred(path: Path) -> None:
|
|
"""Overwrite then unlink every regular file below ``path``; remove the tree."""
|
|
if not path.exists():
|
|
return
|
|
for item in sorted(path.rglob("*"), reverse=True):
|
|
try:
|
|
if item.is_file() and not item.is_symlink():
|
|
size = item.stat().st_size
|
|
with item.open("r+b") as handle:
|
|
remaining = size
|
|
while remaining > 0:
|
|
step = min(CHUNK, remaining)
|
|
handle.write(b"\0" * step)
|
|
remaining -= step
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
item.unlink()
|
|
elif item.is_dir() and not item.is_symlink():
|
|
item.rmdir()
|
|
else:
|
|
item.unlink()
|
|
except OSError:
|
|
pass
|
|
shutil.rmtree(path, ignore_errors=True)
|
|
|
|
|
|
class Lane:
|
|
"""All side effects of one attended run; tests replace the boundaries."""
|
|
|
|
def __init__(self, env: dict[str, str]) -> None:
|
|
self.env = env
|
|
self.tool_env = {"PATH": env.get("PATH", "/usr/bin:/bin"), "HOME": env.get("HOME", "/")}
|
|
|
|
def bao(self, args: list[str], *, label: str) -> str:
|
|
return LEASE.safe_run(["bao", *args], label=label, env=self.env).stdout
|
|
|
|
def status(self) -> dict[str, Any]:
|
|
try:
|
|
status = json.loads(self.bao(["status", "-format=json"], label="read OpenBao status"))
|
|
return {
|
|
"initialized": status["initialized"] is True,
|
|
"unsealed": status["sealed"] is False,
|
|
"storage_type": status.get("storage_type"),
|
|
"cluster_id": str(status["cluster_id"]),
|
|
"raft_applied_index": int(status["raft_applied_index"]),
|
|
}
|
|
except (json.JSONDecodeError, KeyError, TypeError, ValueError):
|
|
fail("OpenBao status is incomplete")
|
|
|
|
def snapshot(self, target: Path) -> None:
|
|
self.bao(["operator", "raft", "snapshot", "save", str(target)], label="save OpenBao Raft snapshot")
|
|
|
|
def lane_identity(self) -> str:
|
|
"""The escrow key may be stored as a standard age identity file
|
|
(``# created``/``# public key`` comments plus one key line); keep only that line."""
|
|
raw = self.bao(["kv", "get", "-field=AGE_PRIVATE_KEY", LANE_PATH], label="read offsite lane AGE_PRIVATE_KEY")
|
|
keys = []
|
|
for line in raw.replace("\r\n", "\n").split("\n"):
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
keys.append(line)
|
|
del raw
|
|
if len(keys) != 1 or not keys[0].startswith("AGE-SECRET-KEY-1"):
|
|
fail("offsite lane AGE_PRIVATE_KEY is unusable")
|
|
return keys[0]
|
|
|
|
def lane_field(self, field: str) -> str:
|
|
value = self.bao(["kv", "get", f"-field={field}", LANE_PATH], label=f"read offsite lane {field}")
|
|
value = value.strip()
|
|
if not value or "\n" in value or "\r" in value:
|
|
fail(f"offsite lane {field} is unusable")
|
|
return value
|
|
|
|
def tool(self, argv: list[str], *, label: str, payload: bytes = b"") -> bytes:
|
|
try:
|
|
result = subprocess.run(argv, input=payload, capture_output=True, env=self.tool_env, timeout=1800)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
fail(f"{label} unavailable")
|
|
if result.returncode != 0:
|
|
fail(f"{label} failed (exit {result.returncode})")
|
|
return result.stdout
|
|
|
|
def encrypt(self, recipients: list[str], source: Path, target: Path) -> None:
|
|
argv = ["age"]
|
|
for recipient in recipients:
|
|
argv += ["-r", recipient]
|
|
self.tool([*argv, "-o", str(target), str(source)], label="age encrypt")
|
|
|
|
def recipient_of(self, identity: bytes) -> str:
|
|
return self.tool(["age-keygen", "-y", "/dev/stdin"], label="derive escrow recipient", payload=identity).decode().strip()
|
|
|
|
def decrypt_digest(self, identity: bytes, ciphertext: Path) -> str:
|
|
"""Stream-decrypt into a hash; plaintext never touches disk again."""
|
|
digest = hashlib.sha256()
|
|
try:
|
|
proc = subprocess.Popen(
|
|
["age", "-d", "-i", "/dev/stdin", str(ciphertext)],
|
|
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, env=self.tool_env,
|
|
)
|
|
except OSError:
|
|
fail("age decrypt unavailable")
|
|
try:
|
|
assert proc.stdin and proc.stdout
|
|
proc.stdin.write(identity)
|
|
proc.stdin.close()
|
|
for chunk in iter(lambda: proc.stdout.read(CHUNK), b""):
|
|
digest.update(chunk)
|
|
code = proc.wait(timeout=1800)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
proc.kill()
|
|
proc.wait()
|
|
fail("age decrypt failed")
|
|
finally:
|
|
if proc.stdout:
|
|
proc.stdout.close()
|
|
if code != 0:
|
|
fail(f"age decrypt failed (exit {code})")
|
|
return digest.hexdigest()
|
|
|
|
def upload(self, url: str, token: str, body: bytes) -> int:
|
|
status, _ = NEXTCLOUD.request(
|
|
url, "PUT", body, (token, ""),
|
|
{"Content-Type": "application/octet-stream", "If-None-Match": "*",
|
|
"Content-Length": str(len(body))},
|
|
)
|
|
return status
|
|
|
|
|
|
def run(
|
|
args: argparse.Namespace, lane: Lane, receipt: dict[str, Any], now=utc_now
|
|
) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
recipient = declared_recipient()
|
|
base = private_dir(args.private_dir, lane.env)
|
|
stage = receipt.setdefault("stages", [])
|
|
|
|
stage.append("status_before")
|
|
before = lane.status()
|
|
if not (before["initialized"] and before["unsealed"] and before["storage_type"] == "raft"):
|
|
fail("OpenBao source is not an initialized, unsealed Raft node")
|
|
if before["raft_applied_index"] <= 0:
|
|
fail("OpenBao Raft applied index is not positive")
|
|
|
|
created = now()
|
|
work = Path(tempfile.mkdtemp(prefix="openbao-reboot-snapshot-", dir=base))
|
|
try:
|
|
os.chmod(work, 0o700)
|
|
plain = work / "openbao.snap"
|
|
cipher = work / "openbao.snap.age"
|
|
stage.append("snapshot_save")
|
|
lane.snapshot(plain)
|
|
if not plain.is_file() or plain.stat().st_size == 0:
|
|
fail("snapshot file was not written")
|
|
os.chmod(plain, 0o600)
|
|
with plain.open("rb") as handle:
|
|
if handle.read(2) != GZIP_MAGIC:
|
|
fail("snapshot is not a Raft snapshot archive")
|
|
|
|
stage.append("status_after")
|
|
after = lane.status()
|
|
if after["cluster_id"] != before["cluster_id"] or after["raft_applied_index"] < before["raft_applied_index"]:
|
|
fail("OpenBao identity or Raft index changed incoherently during the snapshot")
|
|
if not (after["initialized"] and after["unsealed"]):
|
|
fail("OpenBao source sealed during the snapshot")
|
|
plain_sha = sha256_file(plain)
|
|
|
|
stage.append("encrypt")
|
|
recipients = [recipient, *([args.custodian_recipient] if args.custodian_recipient else [])]
|
|
lane.encrypt(recipients, plain, cipher)
|
|
os.chmod(cipher, 0o600)
|
|
with cipher.open("rb") as handle:
|
|
if handle.read(len(AGE_HEADER)) != AGE_HEADER:
|
|
fail("ciphertext is not an age file")
|
|
cipher_sha = sha256_file(cipher)
|
|
if cipher_sha == plain_sha:
|
|
fail("ciphertext equals plaintext")
|
|
|
|
stage.append("verify")
|
|
url = lane.lane_field("NC_WEBDAV_URL")
|
|
token = lane.lane_field("NC_WEBDAV_TOKEN")
|
|
identity = bytearray((lane.lane_identity() + "\n").encode())
|
|
try:
|
|
escrow_recipient = lane.recipient_of(bytes(identity))
|
|
if not hmac.compare_digest(escrow_recipient, recipient):
|
|
fail("escrow key does not match the declared offsite recipient")
|
|
decrypted_sha = lane.decrypt_digest(bytes(identity), cipher)
|
|
finally:
|
|
identity[:] = b"\0" * len(identity)
|
|
if not hmac.compare_digest(decrypted_sha, plain_sha):
|
|
fail("decrypted snapshot hash does not match the plaintext")
|
|
|
|
stage.append("upload")
|
|
if url not in (
|
|
f"{NEXTCLOUD.HOST}/public.php/dav/filesdrop/{token}",
|
|
f"{NEXTCLOUD.HOST}/public.php/dav/files/{token}",
|
|
):
|
|
fail("offsite lane URL does not match the governed provider contract")
|
|
uploaded = now()
|
|
name = f"openbao-raft-snapshot-{args.scenario_id}-{compact(uploaded)}.snap.age"
|
|
status = lane.upload(f"{url}/{UPLOAD_PREFIX}/{name}", token, cipher.read_bytes())
|
|
if status != 201:
|
|
fail("offsite upload did not create a new object")
|
|
# Only a 201 on a no-overwrite PUT names a location; the drop cannot read back.
|
|
receipt["upload_http_status"] = status
|
|
receipt["encrypted_location_ref"] = f"offhost-custody:{args.scenario_id}:{compact(uploaded)}"
|
|
finally:
|
|
stage.append("shred")
|
|
shred(work)
|
|
receipt["temp_dir_removed"] = not work.exists()
|
|
|
|
if not receipt["temp_dir_removed"]:
|
|
fail("private temp dir could not be removed")
|
|
return {
|
|
"receipt_version": 1,
|
|
"receipt_id": f"{args.scenario_id}-SNAPSHOT-{compact(created)}",
|
|
"created_at": stamp(created),
|
|
"operator": OPERATOR,
|
|
**SOURCE,
|
|
"cluster_id": before["cluster_id"],
|
|
# Lower bound: the snapshot contains at least the pre-save index.
|
|
"raft_applied_index": before["raft_applied_index"],
|
|
"snapshot_created": True,
|
|
"source_initialized": True,
|
|
"source_unsealed": True,
|
|
"snapshot_sha256": f"sha256:{plain_sha}",
|
|
"snapshot_encrypted": True,
|
|
"encrypted_snapshot_sha256": f"sha256:{cipher_sha}",
|
|
"encrypted_location_ref": f"offhost-custody:{args.scenario_id}:{compact(uploaded)}",
|
|
"encrypted_copy_off_host": True,
|
|
"off_host_evidence": "upload acknowledged (HTTP 201, If-None-Match: *); remote hash not read back",
|
|
"upload_folder": UPLOAD_PREFIX,
|
|
"recipients": [recipient, *([args.custodian_recipient] if args.custodian_recipient else [])],
|
|
"custodian_escrow_ref": args.custodian_escrow_ref,
|
|
"escrow_key_read_decision": "reviews/RPF-WP-0015-T03-founder-go-20261014.json#escrow_key_read",
|
|
"encryption_verified": True,
|
|
"hash_verified": True,
|
|
"no_secret_material_recorded": True,
|
|
}, after
|
|
|
|
|
|
class QuietParser(argparse.ArgumentParser):
|
|
def error(self, message: str) -> None: # never print usage
|
|
raise SnapshotError("invalid arguments")
|
|
|
|
|
|
def parser() -> argparse.ArgumentParser:
|
|
p = QuietParser(description=__doc__, add_help=False)
|
|
p.add_argument("--scenario-id", required=True)
|
|
p.add_argument("--receipt-file", required=True)
|
|
p.add_argument("--private-dir")
|
|
p.add_argument("--remote", default="railiance01", choices=["railiance01"])
|
|
p.add_argument("--custodian-recipient", help="extra age recipient whose key is held outside OpenBao")
|
|
p.add_argument("--custodian-escrow-ref", required=True,
|
|
help="value-safe reference to where a decryption key is held outside OpenBao")
|
|
return p
|
|
|
|
|
|
def main(argv: list[str] | None = None, *, lane_factory=Lane, now=utc_now) -> int:
|
|
try:
|
|
args = parser().parse_args(argv)
|
|
if not SCENARIO_RE.fullmatch(args.scenario_id):
|
|
return 2
|
|
# Always required: where a decryption key is held outside OpenBao (alone, when the
|
|
# offline key is the offsite recipient; with --custodian-recipient otherwise).
|
|
if not (args.custodian_escrow_ref or "").strip() or (
|
|
args.custodian_recipient and not AGE_RECIPIENT_RE.fullmatch(args.custodian_recipient)
|
|
):
|
|
return 2
|
|
fd = open_receipt(args.receipt_file)
|
|
except (SnapshotError, OSError):
|
|
return 2
|
|
record: dict[str, Any] = {"status": "failed", "scenario_id": args.scenario_id, "error": "internal_error"}
|
|
result: dict[str, Any] | None = None
|
|
def _abort(signum, _frame): # warden timeout/abort: still run every finally/shred
|
|
raise SnapshotError(f"interrupted by signal {signum}")
|
|
|
|
signal.signal(signal.SIGTERM, _abort)
|
|
signal.signal(signal.SIGHUP, _abort)
|
|
try:
|
|
os.umask(0o077)
|
|
env = LEASE.contained_bao_env()
|
|
receipt, live = run(args, lane_factory(env), record, now=now)
|
|
# Self-check with the exact preflight validator before claiming success.
|
|
check = Path(tempfile.mkdtemp(prefix="openbao-receipt-check-", dir=private_dir(args.private_dir, env)))
|
|
try:
|
|
probe = check / "receipt.json"
|
|
probe.write_text(json.dumps(receipt), encoding="utf-8")
|
|
PREFLIGHT.validate_snapshot_receipt(probe, live_openbao=live, now=now(), max_age_hours=1.0)
|
|
finally:
|
|
shutil.rmtree(check, ignore_errors=True)
|
|
result = receipt
|
|
except (SnapshotError, LEASE.ProcedureError, PREFLIGHT.PreflightError) as exc:
|
|
record["error"] = str(exc)
|
|
except Exception: # noqa: BLE001 - value-safe: never record exception text
|
|
record["error"] = "internal_error"
|
|
finally:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
|
json.dump(result if result is not None else record, handle, indent=2, sort_keys=False)
|
|
handle.write("\n")
|
|
return 0 if result is not None else 1
|
|
|
|
|
|
def silence() -> None:
|
|
devnull = os.open(os.devnull, os.O_WRONLY)
|
|
os.dup2(devnull, 1)
|
|
os.dup2(devnull, 2)
|
|
os.close(devnull)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
silence()
|
|
raise SystemExit(main())
|