"""Offline tests for scripts/openbao-reboot-snapshot.py (no OpenBao, no network). ``bao`` and the Nextcloud upload are faked; ``age`` runs for real against a throwaway key when installed (otherwise those cases are skipped). """ from __future__ import annotations import gzip import importlib.util import io import json import os import shutil import subprocess import sys import tempfile import unittest from contextlib import redirect_stderr, redirect_stdout from datetime import UTC, datetime from pathlib import Path from unittest import mock ROOT = Path(__file__).resolve().parents[1] SCRIPT = ROOT / "scripts" / "openbao-reboot-snapshot.py" SPEC = importlib.util.spec_from_file_location("openbao_reboot_snapshot", SCRIPT) assert SPEC and SPEC.loader snap = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(snap) HAVE_AGE = bool(shutil.which("age") and shutil.which("age-keygen")) CLUSTER = "fd28df5d-98ec-57dd-42ec-9b3e4f4e53bf" SCENARIO = "RPF-WP-0015-T03-REBOOT-20261014-01" TOKEN = "tok3nValue" URL = f"{snap.NEXTCLOUD.HOST}/public.php/dav/filesdrop/{TOKEN}" def throwaway_key() -> tuple[str, str]: out = subprocess.run(["age-keygen"], capture_output=True, text=True, check=True).stdout identity = next(line for line in out.splitlines() if line.startswith("AGE-SECRET-KEY-1")) recipient = subprocess.run( ["age-keygen", "-y"], input=identity + "\n", capture_output=True, text=True, check=True ).stdout.strip() return identity, recipient class FakeLane(snap.Lane): """Replaces only the bao CLI and the HTTP boundary.""" def __init__(self, env, *, identity, status_seq=None, upload_status=201, url=URL, payload=None): super().__init__(env) self.calls: list[str] = [] self.identity = identity self.status_seq = list(status_seq or [self.status_json(100), self.status_json(105)]) self.upload_status = upload_status self.url = url self.payload = payload if payload is not None else gzip.compress(b"raft-state" * 1000) self.uploads: list[tuple[str, bytes]] = [] self.snapshot_path: Path | None = None @staticmethod def status_json(index, sealed=False, initialized=True, cluster=CLUSTER): return json.dumps({"initialized": initialized, "sealed": sealed, "storage_type": "raft", "cluster_id": cluster, "raft_applied_index": index, "t": 2, "n": 3}) def bao(self, args, *, label): assert self.env.get("WARDEN_ROUTING_CATALOG") and "BAO_TOKEN" not in self.env if args[:1] == ["status"]: self.calls.append("status") return self.status_seq.pop(0) if args[:4] == ["operator", "raft", "snapshot", "save"]: self.calls.append("snapshot") self.snapshot_path = Path(args[4]) assert not self.snapshot_path.exists() self.snapshot_path.write_bytes(self.payload) return "" if args[:2] == ["kv", "get"]: field = args[2].removeprefix("-field=") assert args[3] == snap.LANE_PATH self.calls.append(f"kv:{field}") return {"NC_WEBDAV_URL": self.url, "NC_WEBDAV_TOKEN": TOKEN, "AGE_PRIVATE_KEY": self.identity}[field] + "\n" raise AssertionError(f"unexpected bao call {args}") def upload(self, url, token, body): self.calls.append("upload") self.uploads.append((url, body)) return self.upload_status @unittest.skipUnless(HAVE_AGE, "age/age-keygen not installed") class SnapshotLaneTest(unittest.TestCase): @classmethod def setUpClass(cls): cls.identity, cls.recipient = throwaway_key() def setUp(self): self.tmp = Path(tempfile.mkdtemp(prefix="snap-test-")) self.home = self.tmp / "home" self.home.mkdir(mode=0o700) self.receipt = self.tmp / "receipt.json" self.env = {"WARDEN_ROUTING_CATALOG": "/catalog.yaml", "BAO_ADDR": "http://127.0.0.1:18200", "HOME": str(self.home), "PATH": os.environ.get("PATH", "/usr/bin:/bin")} self.lanes: list[FakeLane] = [] patcher = mock.patch.object(snap, "declared_recipient", return_value=self.recipient) patcher.start() self.addCleanup(patcher.stop) self.addCleanup(shutil.rmtree, self.tmp, True) def factory(self, **kwargs): def make(env): lane = FakeLane(env, identity=kwargs.pop("identity", self.identity), **kwargs) self.lanes.append(lane) return lane return make def run_main(self, env=None, extra=(), **kwargs): out, err = io.StringIO(), io.StringIO() escrow = () if "--custodian-escrow-ref" in extra else ("--custodian-escrow-ref", "founder-offline:test") argv = ["--scenario-id", SCENARIO, "--receipt-file", str(self.receipt), *escrow, *extra] with mock.patch.dict(os.environ, env or self.env, clear=True), redirect_stdout(out), redirect_stderr(err): code = snap.main(argv, lane_factory=self.factory(**kwargs)) self.assertEqual(out.getvalue() + err.getvalue(), "", "child must stay silent") return code, json.loads(self.receipt.read_text()) def assert_clean(self): self.assertEqual(list(self.home.iterdir()), [], "private temp dir must be shredded") def test_happy_path_order_and_receipt_validates(self): code, receipt = self.run_main() self.assertEqual(code, 0) lane = self.lanes[0] self.assertEqual(lane.calls, ["status", "snapshot", "status", "kv:NC_WEBDAV_URL", "kv:NC_WEBDAV_TOKEN", "kv:AGE_PRIVATE_KEY", "upload"]) self.assertEqual(self.receipt.stat().st_mode & 0o777, 0o600) live = {"cluster_id": CLUSTER, "raft_applied_index": 105} checked = snap.PREFLIGHT.validate_snapshot_receipt( self.receipt, live_openbao=live, now=datetime.now(UTC), max_age_hours=24) self.assertTrue(checked["verified"]) self.assertEqual(receipt["raft_applied_index"], 100) self.assertTrue(receipt["encrypted_location_ref"].startswith(f"offhost-custody:{SCENARIO}:")) url, body = lane.uploads[0] self.assertTrue(url.startswith(f"{URL}/openbao/openbao-raft-snapshot-{SCENARIO}-")) self.assertTrue(body.startswith(snap.AGE_HEADER)) self.assertNotIn(lane.payload, body) text = self.receipt.read_text() for value in (self.identity, TOKEN, "AGE-SECRET-KEY"): self.assertNotIn(value, text) self.assertFalse(lane.snapshot_path.exists()) self.assert_clean() def test_upload_conflict_fails_and_cleans_up(self): code, record = self.run_main(upload_status=412) self.assertEqual(code, 1) self.assertEqual(record["status"], "failed") self.assertIn("did not create", record["error"]) self.assertTrue(record["temp_dir_removed"]) self.assertNotIn("receipt_version", record) self.assertNotIn("encrypted_location_ref", record) # no location without a 201 self.assertNotIn("upload_http_status", record) self.assert_clean() def test_custodian_recipient_alone_decrypts_upload(self): identity, recipient = throwaway_key() code, receipt = self.run_main( extra=("--custodian-recipient", recipient, "--custodian-escrow-ref", "founder-offline:test")) self.assertEqual(code, 0) self.assertEqual(receipt["recipients"][1], recipient) self.assertEqual(receipt["custodian_escrow_ref"], "founder-offline:test") self.assertEqual(receipt["upload_folder"], "openbao") _, body = self.lanes[0].uploads[0] with tempfile.TemporaryDirectory() as tmp: key, ct = Path(tmp) / "k", Path(tmp) / "c" key.write_bytes(identity if isinstance(identity, bytes) else identity.encode()) ct.write_bytes(body) out = subprocess.run(["age", "-d", "-i", str(key), str(ct)], capture_output=True, check=False) self.assertEqual(out.returncode, 0) self.assertEqual(out.stdout, self.lanes[0].payload) def test_escrow_ref_alone_suffices_and_is_cited(self): code, receipt = self.run_main() self.assertEqual(code, 0) self.assertEqual(len(receipt["recipients"]), 1) self.assertEqual(receipt["custodian_escrow_ref"], "founder-offline:test") self.assertIn("escrow_key_read", receipt["escrow_key_read_decision"]) def test_preflight_rejects_receipt_without_escrow_ref(self): code, receipt = self.run_main() self.assertEqual(code, 0) receipt["custodian_escrow_ref"] = None bad = self.home / "bad.json" bad.write_text(json.dumps(receipt)) with self.assertRaises(snap.PREFLIGHT.PreflightError): snap.PREFLIGHT.validate_snapshot_receipt( bad, live_openbao={"cluster_id": CLUSTER, "raft_applied_index": 105}, now=datetime.now(UTC), max_age_hours=24) bad.unlink() def test_escrow_key_in_age_identity_file_format_is_accepted(self): recipient = snap.declared_recipient() identity_file = f"# created: 2026-01-01T00:00:00Z\n# public key: {recipient}\n{self.identity}" code, receipt = self.run_main(identity=identity_file) self.assertEqual(code, 0) self.assertNotIn("AGE-SECRET-KEY", self.receipt.read_text()) def test_escrow_key_with_two_keys_or_junk_is_unusable(self): for bad in (f"{self.identity}\n{self.identity}", f"junk\n{self.identity}"): code, record = self.run_main(identity=bad) self.assertEqual(code, 1) self.assertIn("AGE_PRIVATE_KEY is unusable", record["error"]) self.assertNotIn("upload", self.lanes[-1].calls) self.receipt.unlink() self.assert_clean() def test_custodian_recipient_requires_escrow_ref(self): _, recipient = throwaway_key() argv = ["--scenario-id", SCENARIO, "--receipt-file", str(self.receipt), "--custodian-recipient", recipient] with mock.patch.dict(os.environ, self.env, clear=True): self.assertEqual(snap.main(argv, lane_factory=self.factory()), 2) self.assertFalse(self.receipt.exists()) def test_escrow_key_mismatch_never_uploads(self): other, _ = throwaway_key() code, record = self.run_main(identity=other) self.assertEqual(code, 1) self.assertIn("escrow key does not match", record["error"]) self.assertNotIn("upload", self.lanes[0].calls) self.assertNotIn(other, json.dumps(record)) self.assert_clean() def test_decrypt_hash_mismatch_never_uploads(self): with mock.patch.object(snap.Lane, "decrypt_digest", return_value="0" * 64): code, record = self.run_main() self.assertEqual(code, 1) self.assertIn("decrypted snapshot hash", record["error"]) self.assertNotIn("upload", self.lanes[0].calls) self.assert_clean() def test_sealed_source_never_snapshots(self): code, record = self.run_main(status_seq=[FakeLane.status_json(100, sealed=True)]) self.assertEqual(code, 1) self.assertEqual(self.lanes[0].calls, ["status"]) self.assert_clean() def test_cluster_change_during_snapshot_fails(self): seq = [FakeLane.status_json(100), FakeLane.status_json(105, cluster="other")] code, record = self.run_main(status_seq=seq) self.assertEqual(code, 1) self.assertNotIn("kv:AGE_PRIVATE_KEY", self.lanes[0].calls) self.assert_clean() def test_non_snapshot_payload_refused(self): code, record = self.run_main(payload=b"plain text, not a raft archive") self.assertEqual(code, 1) self.assertIn("not a Raft snapshot", record["error"]) self.assert_clean() def test_unexpected_provider_url_never_uploads(self): code, record = self.run_main(url="https://elsewhere.example/dav") self.assertEqual(code, 1) self.assertNotIn("upload", self.lanes[0].calls) self.assert_clean() def test_internal_error_is_value_safe(self): with mock.patch.object(snap, "sha256_file", side_effect=ValueError(TOKEN)): code, record = self.run_main() self.assertEqual(record["error"], "internal_error") self.assertNotIn(TOKEN, json.dumps(record)) self.assert_clean() def test_refused_outside_contained_login(self): for env in ({"HOME": str(self.home)}, {**self.env, "BAO_TOKEN": "x"}, {**self.env, "BAO_ADDR": "https://openbao.example"}): self.receipt.unlink(missing_ok=True) self.lanes.clear() code, record = self.run_main(env=env) self.assertEqual(code, 1) self.assertEqual(record["status"], "failed") self.assertEqual(self.lanes, [], "no bao call may happen outside containment") def test_private_dir_must_be_owner_only_and_outside_git(self): loose = self.tmp / "loose" loose.mkdir(mode=0o755) loose.chmod(0o755) code, record = self.run_main(extra=["--private-dir", str(loose)]) self.assertIn("owner-only", record["error"]) self.receipt.unlink() code, record = self.run_main(extra=["--private-dir", str(ROOT / "scripts")]) self.assertEqual(code, 1) self.assertEqual(self.lanes[-1].calls if self.lanes else [], []) def test_existing_receipt_or_bad_scenario_refused(self): self.receipt.write_text("keep") with mock.patch.dict(os.environ, self.env, clear=True): self.assertEqual(snap.main(["--scenario-id", SCENARIO, "--receipt-file", str(self.receipt)]), 2) self.assertEqual(snap.main(["--scenario-id", "../x", "--receipt-file", str(self.tmp / "n.json")]), 2) self.assertEqual(snap.main(["--bogus"]), 2) self.assertEqual(self.receipt.read_text(), "keep") class BoundaryTest(unittest.TestCase): def test_declared_recipient_comes_from_offsite_lane_helper(self): recipient = snap.declared_recipient() self.assertRegex(recipient, snap.AGE_RECIPIENT_RE) self.assertIn(recipient, (ROOT / "lib" / "railiance-backup-common.sh").read_text()) def test_upload_uses_no_overwrite_governed_request(self): lane = snap.Lane({"PATH": "/usr/bin"}) with mock.patch.object(snap.NEXTCLOUD, "request", return_value=(201, b"")) as request: self.assertEqual(lane.upload(URL + "/forgejo/x.age", TOKEN, b"cipher"), 201) url, method, body, auth, headers = request.call_args.args self.assertEqual((method, body, auth), ("PUT", b"cipher", (TOKEN, ""))) self.assertEqual(headers["If-None-Match"], "*") def test_script_is_silent_and_refuses_as_process_outside_login(self): with tempfile.TemporaryDirectory() as tmp: receipt = Path(tmp) / "r.json" env = {k: v for k, v in os.environ.items() if k not in ("WARDEN_ROUTING_CATALOG", "BAO_TOKEN", "VAULT_TOKEN", "OPENBAO_TOKEN")} env["HOME"] = tmp result = subprocess.run( [sys.executable, str(SCRIPT), "--scenario-id", SCENARIO, "--receipt-file", str(receipt), "--custodian-escrow-ref", "founder-offline:test"], capture_output=True, env=env, timeout=60) self.assertEqual((result.stdout, result.stderr), (b"", b"")) self.assertNotEqual(result.returncode, 0) self.assertIn("contained attended login", json.loads(receipt.read_text())["error"]) bad = subprocess.run([sys.executable, str(SCRIPT), "--help"], capture_output=True, timeout=60) self.assertEqual((bad.stdout, bad.stderr, bad.returncode), (b"", b"", 2)) if __name__ == "__main__": unittest.main()