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
332 lines
16 KiB
Python
332 lines
16 KiB
Python
"""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()
|