#!/usr/bin/env python3 """Tests for the operator backup and restore tool. Issue #39 asks for a production topology and a backup that a person can trust with the records Timmy is not allowed to lose. The corpus manifest is append-only and the tombstones in it are the only proof that a withdrawal happened, so a backup that quietly drops them, or a restore that rebuilds a training set from withdrawn contributions, is worse than no backup at all. The rules this suite locks in: - model weights and transient inference images are never archived, and every excluded path is named in the manifest with the reason it was left out; - every archived byte carries a SHA-256 that ``verify`` recomputes before a restore writes anything, so a tampered archive fails closed; - hostile members (absolute paths, traversal, links, unlisted extras) are refused instead of extracted; - a restore round-trips the append-only manifest byte-for-byte and reports the same record counts, including the tombstones. Every fixture is synthetic: the "images" are opaque bytes, no real medical image is read or written. """ from __future__ import annotations import hashlib import io import json import os import shutil import subprocess import sys import tarfile import tempfile import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[1] BACKUP = ROOT / "scripts" / "operator_backup.py" IMAGE_BYTES = b"\xff\xd8\xff\xe0synthetic-jpeg-payload-not-a-real-photo" WEIGHT_BYTES = b"GGUF-synthetic-weight-shard-not-a-real-model" TRANSIENT_BYTES = b"transient-inference-upload-must-not-be-archived" SECRET_BYTES = b"TIMMY_DATASET_SALT=demo\nTIMMY_VISION_TOKEN=do-not-archive-me\n" def digest(payload: bytes) -> str: return hashlib.sha256(payload).hexdigest() def run(args: list[str], env: dict[str, str] | None = None) -> subprocess.CompletedProcess: return subprocess.run( [sys.executable, str(BACKUP), *args], cwd=ROOT, capture_output=True, text=True, env={**os.environ, **(env or {})}, ) def records(*lines: dict) -> str: return "".join(json.dumps(line) + "\n" for line in lines) def sample(contribution_id: str, *, relative_path: str = "images/sample-1.jpg") -> dict: return { "recordType": "sample", "schemaVersion": 2, "contributionId": contribution_id, "relativePath": relative_path, "imageId": digest(IMAGE_BYTES), "lifecycle": {"state": "active", "updatedAt": "2026-08-01T10:00:00+00:00"}, } def tombstone(contribution_id: str) -> dict: return { "recordType": "tombstone", "schemaVersion": 2, "contributionId": contribution_id, "lifecycle": {"state": "revoked", "updatedAt": "2026-08-02T10:00:00+00:00"}, "reason": "participant withdrew consent", "deletedArtifacts": [{"contributionId": contribution_id, "sha256": digest(IMAGE_BYTES)}], } def build_source(root: Path, *, with_tombstone: bool = False) -> Path: """A synthetic server-side data root: the records that must survive, plus what must not.""" source = root / "srv-timmy" (source / "images").mkdir(parents=True) (source / "images" / "sample-1.jpg").write_bytes(IMAGE_BYTES) (source / "manifest.jsonl").write_text( records(sample("contribution-1"), *([tombstone("contribution-2")] if with_tombstone else [])), encoding="utf-8", ) # Must never be archived: weights, transient inference bytes, operator secrets, and logs. (source / "weights").mkdir() (source / "weights" / "model.gguf").write_bytes(WEIGHT_BYTES) (source / "transient").mkdir() (source / "transient" / "inference-upload-1.jpg").write_bytes(TRANSIENT_BYTES) (source / "secrets.env").write_bytes(SECRET_BYTES) (source / "operator.log").write_text("GET /api/vision 200\n", encoding="utf-8") return source def members(archive: Path) -> dict[str, bytes]: with tarfile.open(archive, "r:gz") as tar: out = {} for info in tar.getmembers(): if not info.isfile(): continue handle = tar.extractfile(info) assert handle is not None out[info.name] = handle.read() return out def craft_archive(path: Path, entries: list[tuple[str, bytes]], *, link: str | None = None) -> None: """Build an archive by hand so the hostile cases are hostile on purpose.""" with tarfile.open(path, "w:gz") as tar: for name, payload in entries: info = tarfile.TarInfo(name) info.size = len(payload) info.mode = 0o644 tar.addfile(info, io.BytesIO(payload)) if link: info = tarfile.TarInfo(link) info.type = tarfile.SYMTYPE info.linkname = "/etc/passwd" info.mode = 0o777 tar.addfile(info) class BackupContractTests(unittest.TestCase): def setUp(self) -> None: self.tmp = Path(tempfile.mkdtemp(prefix="timmy-backup-test-")) self.addCleanup(shutil.rmtree, self.tmp, True) self.archive = self.tmp / "backup.tar.gz" self.manifest = self.tmp / "backup.manifest.json" def backup(self, source: Path, *extra: str) -> subprocess.CompletedProcess: return run([ "backup", "--source", str(source), "--archive", str(self.archive), "--manifest", str(self.manifest), "--label", "nightly", *extra, ]) def test_backup_archives_records_and_excludes_weights_transient_images_secrets_and_logs(self) -> None: source = build_source(self.tmp, with_tombstone=True) result = self.backup(source) self.assertEqual(result.returncode, 0, result.stderr) archived = members(self.archive) self.assertEqual(sorted(archived), ["payload/images/sample-1.jpg", "payload/manifest.jsonl"]) self.assertEqual(archived["payload/images/sample-1.jpg"], IMAGE_BYTES) self.assertNotIn(WEIGHT_BYTES, b"".join(archived.values())) self.assertNotIn(TRANSIENT_BYTES, b"".join(archived.values())) self.assertNotIn(SECRET_BYTES, b"".join(archived.values())) manifest = json.loads(self.manifest.read_text(encoding="utf-8")) excluded = {entry["path"]: entry["reason"] for entry in manifest["excluded"]} self.assertEqual(excluded["weights/model.gguf"], "model-weights") self.assertEqual(excluded["transient/inference-upload-1.jpg"], "transient-inference") self.assertEqual(excluded["secrets.env"], "secret") self.assertEqual(excluded["operator.log"], "log") # Every archived member is listed with the digest of the bytes that were stored. listed = {entry["path"]: entry["sha256"] for entry in manifest["files"]} self.assertEqual(listed, { "images/sample-1.jpg": digest(IMAGE_BYTES), "manifest.jsonl": digest((source / "manifest.jsonl").read_bytes()), }) def test_backup_reports_record_counts_including_tombstones(self) -> None: source = build_source(self.tmp, with_tombstone=True) result = self.backup(source) self.assertEqual(result.returncode, 0, result.stderr) receipt = json.loads(result.stdout) self.assertEqual(receipt["records"], {"samples": 1, "tombstones": 1, "activeSamples": 1}) manifest = json.loads(self.manifest.read_text(encoding="utf-8")) self.assertEqual(manifest["records"], {"samples": 1, "tombstones": 1, "activeSamples": 1}) self.assertEqual(manifest["label"], "nightly") def test_backup_refuses_to_follow_a_symlink_out_of_the_data_root(self) -> None: source = build_source(self.tmp) (source / "images" / "escape.jpg").symlink_to("/etc/hostname") result = self.backup(source) self.assertEqual(result.returncode, 0, result.stderr) self.assertNotIn("payload/images/escape.jpg", members(self.archive)) manifest = json.loads(self.manifest.read_text(encoding="utf-8")) excluded = {entry["path"]: entry["reason"] for entry in manifest["excluded"]} self.assertEqual(excluded["images/escape.jpg"], "symlink") def test_backup_refuses_missing_source_and_existing_archive(self) -> None: missing = run(["backup", "--source", str(self.tmp / "nope"), "--archive", str(self.archive)]) self.assertNotEqual(missing.returncode, 0) self.assertIn("is not a directory", missing.stderr) source = build_source(self.tmp) self.assertEqual(self.backup(source).returncode, 0) again = self.backup(source) self.assertNotEqual(again.returncode, 0) self.assertIn("already exists", again.stderr) def test_verify_and_restore_round_trip_the_records(self) -> None: source = build_source(self.tmp, with_tombstone=True) self.assertEqual(self.backup(source).returncode, 0) verified = run(["verify", "--archive", str(self.archive), "--manifest", str(self.manifest)]) self.assertEqual(verified.returncode, 0, verified.stderr) target = self.tmp / "restored" restored = run([ "restore", "--archive", str(self.archive), "--manifest", str(self.manifest), "--target", str(target), ]) self.assertEqual(restored.returncode, 0, restored.stderr) receipt = json.loads(restored.stdout) self.assertEqual(receipt["records"], {"samples": 1, "tombstones": 1, "activeSamples": 1}) self.assertEqual((target / "manifest.jsonl").read_bytes(), (source / "manifest.jsonl").read_bytes()) self.assertEqual((target / "images" / "sample-1.jpg").read_bytes(), IMAGE_BYTES) # The manifest is append-only history: the tombstone line survives the round trip verbatim. lines = [json.loads(line) for line in (target / "manifest.jsonl").read_text().splitlines() if line.strip()] self.assertEqual([line["recordType"] for line in lines], ["sample", "tombstone"]) def test_verify_refuses_tampered_member_before_a_restore_writes_anything(self) -> None: source = build_source(self.tmp) self.assertEqual(self.backup(source).returncode, 0) manifest = json.loads(self.manifest.read_text(encoding="utf-8")) tampered = b"\xff\xd8\xff\xe0synthetic-jpeg-payload-not-a-real-photO" craft_archive(self.archive, [ ("payload/images/sample-1.jpg", tampered), ("payload/manifest.jsonl", (source / "manifest.jsonl").read_bytes()), ]) verified = run(["verify", "--archive", str(self.archive), "--manifest", str(self.manifest)]) self.assertNotEqual(verified.returncode, 0) self.assertIn("does not match", verified.stderr) self.assertEqual(manifest["files"][0]["sha256"], digest(IMAGE_BYTES)) target = self.tmp / "restored-tampered" restored = run([ "restore", "--archive", str(self.archive), "--manifest", str(self.manifest), "--target", str(target), ]) self.assertNotEqual(restored.returncode, 0) self.assertFalse(target.exists()) def test_verify_refuses_a_member_the_manifest_never_listed(self) -> None: source = build_source(self.tmp) self.assertEqual(self.backup(source).returncode, 0) craft_archive(self.archive, [ ("payload/images/sample-1.jpg", IMAGE_BYTES), ("payload/manifest.jsonl", (source / "manifest.jsonl").read_bytes()), ("payload/extra.bin", b"unlisted"), ]) verified = run(["verify", "--archive", str(self.archive), "--manifest", str(self.manifest)]) self.assertNotEqual(verified.returncode, 0) self.assertIn("not listed in the manifest", verified.stderr) def test_restore_refuses_traversal_and_absolute_members(self) -> None: for name, needle in (("payload/../../etc/passwd", "outside"), ("/etc/passwd", "absolute")): with self.subTest(member=name): archive = self.tmp / "hostile.tar.gz" manifest_path = self.tmp / "hostile.manifest.json" payload = b"root:x:0:0" craft_archive(archive, [(name, payload)]) manifest_path.write_text(json.dumps({ "schemaVersion": 1, "files": [{"path": "images/sample-1.jpg", "sha256": digest(payload), "bytes": len(payload)}], "records": {"samples": 0, "tombstones": 0, "activeSamples": 0}, }), encoding="utf-8") target = self.tmp / f"hostile-out-{needle}" restored = run([ "restore", "--archive", str(archive), "--manifest", str(manifest_path), "--target", str(target), ]) self.assertNotEqual(restored.returncode, 0) self.assertIn(needle, restored.stderr.lower()) self.assertFalse(target.exists()) def test_restore_refuses_link_members(self) -> None: archive = self.tmp / "link.tar.gz" manifest_path = self.tmp / "link.manifest.json" craft_archive(archive, [("payload/manifest.jsonl", b"{}\n")], link="payload/images/sample-1.jpg") manifest_path.write_text(json.dumps({ "schemaVersion": 1, "files": [{"path": "manifest.jsonl", "sha256": digest(b"{}\n"), "bytes": 3}], "records": {"samples": 0, "tombstones": 0, "activeSamples": 0}, }), encoding="utf-8") target = self.tmp / "link-out" restored = run([ "restore", "--archive", str(archive), "--manifest", str(manifest_path), "--target", str(target), ]) self.assertNotEqual(restored.returncode, 0) self.assertIn("link", restored.stderr) self.assertFalse(target.exists()) def test_restore_refuses_a_non_empty_target(self) -> None: source = build_source(self.tmp) self.assertEqual(self.backup(source).returncode, 0) target = self.tmp / "occupied" target.mkdir() (target / "keep.txt").write_text("operator file that must not be overwritten", encoding="utf-8") restored = run([ "restore", "--archive", str(self.archive), "--manifest", str(self.manifest), "--target", str(target), ]) self.assertNotEqual(restored.returncode, 0) self.assertIn("not empty", restored.stderr) self.assertEqual((target / "keep.txt").read_text(encoding="utf-8"), "operator file that must not be overwritten") def test_backup_is_reproducible_for_the_same_records(self) -> None: source = build_source(self.tmp, with_tombstone=True) self.assertEqual(self.backup(source).returncode, 0) first = json.loads(self.manifest.read_text(encoding="utf-8")) first_digest = hashlib.sha256(self.archive.read_bytes()).hexdigest() self.archive.unlink() self.manifest.unlink() self.assertEqual(self.backup(source).returncode, 0) second = json.loads(self.manifest.read_text(encoding="utf-8")) self.assertEqual(first["files"], second["files"]) self.assertEqual(first["excluded"], second["excluded"]) self.assertEqual(first_digest, hashlib.sha256(self.archive.read_bytes()).hexdigest()) self.assertNotEqual(first["createdAt"], "") if __name__ == "__main__": unittest.main(verbosity=2)