from __future__ import annotations import copy import json import shutil import subprocess import sys import tarfile import tempfile import unittest from datetime import datetime, timedelta, timezone from pathlib import Path from unittest import mock import yaml ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "scripts")) from s1_backup import ( # noqa: E402 BackupError, apply_prune, create_bundle, inspect_status, plan_prune, ) from s1_backup_contract import BackupContractError, load_spec, validate_spec # noqa: E402 from s1_restore import RestoreError, _safe_member, extract_bundle, validate_bundle # noqa: E402 @unittest.skipUnless(shutil.which("age") and shutil.which("age-keygen"), "age tools required") class S1BackupRecoveryTests(unittest.TestCase): def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) self.fixture = self.root / "fixture" (self.fixture / "etc/ssh/sshd_config.d").mkdir(parents=True) (self.fixture / "etc/ssh/sshd_config").write_text("PasswordAuthentication no\n") (self.fixture / "etc/ssh/sshd_config.d/10-hardening.conf").write_text( "PermitRootLogin no\n" ) (self.fixture / "etc/hosts").write_text("127.0.0.1 localhost\n") (self.fixture / "etc/hostname").write_text("fixture-host\n") self.packages = self.root / "packages.txt" self.packages.write_text("curl\tinstall\ngit\tinstall\n") self.identity = self.root / "identity.age" subprocess.run( ["age-keygen", "-o", str(self.identity)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True, ) self.recipient = subprocess.check_output( ["age-keygen", "-y", str(self.identity)], text=True ).strip() policy = { "creation_rules": [{"key_groups": [{"age": [self.recipient]}]}] } (self.root / ".sops.yaml").write_text(yaml.safe_dump(policy)) spec = yaml.safe_load((ROOT / "spec/s1-backup.yaml").read_text()) spec["recipients"] = [self.recipient] self.spec_dir = self.root / "spec" self.spec_dir.mkdir() self.spec = self.spec_dir / "s1-backup.yaml" self.spec.write_text(yaml.safe_dump(spec, sort_keys=False)) self.output = self.root / "bundles" def tearDown(self) -> None: self.temp.cleanup() def _create(self, **overrides: object) -> Path: arguments = { "spec_path": self.spec, "source_root": self.fixture, "output_dir": self.output, "packages_file": self.packages, "age_binary": shutil.which("age"), "source_revision": "a" * 40, "timestamp": "20260823T120000Z", "hostname": "fixture-host", } arguments.update(overrides) return create_bundle(**arguments) def test_repository_contract_validates_without_private_key(self) -> None: declaration = load_spec(ROOT / "spec/s1-backup.yaml") self.assertEqual(10, len(declaration["artifacts"]["os-config"]["paths"])) def test_relative_and_excluded_paths_fail(self) -> None: payload = yaml.safe_load(self.spec.read_text()) bad = copy.deepcopy(payload) bad["artifacts"]["os-config"]["paths"][0]["path"] = "etc/ssh/sshd_config" with self.assertRaisesRegex(BackupContractError, "absolute"): validate_spec(bad, policy_path=self.root / ".sops.yaml") bad = copy.deepcopy(payload) bad["artifacts"]["os-config"]["paths"][0]["path"] = "/root/id_ed25519" with self.assertRaisesRegex(BackupContractError, "secret|excluded"): validate_spec(bad, policy_path=self.root / ".sops.yaml") def test_atomic_encrypted_bundle_round_trip(self) -> None: bundle = self._create() self.assertEqual( {"manifest.json", "os-config.tar.gz.age", "packages.txt.age", "receipt.json"}, {path.name for path in bundle.iterdir()}, ) validated = validate_bundle(bundle, self.spec) self.assertEqual("pass", validated["receipt"]["status"]) restore = self.root / "restore" extract_bundle( bundle=bundle, spec_path=self.spec, destination=restore, identity=self.identity, age_binary=shutil.which("age"), ) self.assertEqual("fixture-host\n", (restore / "etc/hostname").read_text()) self.assertIn("curl", (restore / "packages.txt").read_text()) self.assertEqual("pass", json.loads((restore / "restore-receipt.json").read_text())["status"]) def test_encryption_failure_publishes_nothing(self) -> None: with self.assertRaisesRegex(BackupError, "no bundle was published"): self._create(age_binary="/bin/false") self.assertEqual([], list(self.output.iterdir())) def test_missing_required_input_publishes_nothing(self) -> None: (self.fixture / "etc/hostname").unlink() with self.assertRaisesRegex(BackupError, "required backup inputs"): self._create() self.assertEqual([], list(self.output.iterdir())) def test_runtime_secret_shaped_member_publishes_nothing(self) -> None: (self.fixture / "etc/ssh/sshd_config.d/operator-token").write_text("do-not-copy\n") with self.assertRaisesRegex(BackupError, "secret/private-key shaped"): self._create() self.assertEqual([], list(self.output.iterdir())) def test_critical_mode_drift_publishes_nothing(self) -> None: (self.fixture / "etc/ssh/sshd_config").chmod(0o600) with self.assertRaisesRegex(BackupError, "critical mode mismatch"): self._create() self.assertEqual([], list(self.output.iterdir())) def test_root_check_precedes_destination_creation(self) -> None: destination = self.root / "must-not-exist" with mock.patch("s1_backup.os.geteuid", return_value=1000): with self.assertRaisesRegex(BackupError, "requires root"): create_bundle( spec_path=self.spec, source_root=Path("/"), output_dir=destination, packages_file=None, age_binary=shutil.which("age"), source_revision="a" * 40, timestamp="20260823T120000Z", hostname="fixture-host", ) self.assertFalse(destination.exists()) def test_tampered_artifact_fails_before_decryption(self) -> None: bundle = self._create() with (bundle / "packages.txt.age").open("ab") as handle: handle.write(b"tamper") with self.assertRaisesRegex(RestoreError, "digest mismatch"): validate_bundle(bundle, self.spec) def test_archive_traversal_and_escaping_link_fail(self) -> None: member = tarfile.TarInfo("../../etc/shadow") with self.assertRaisesRegex(RestoreError, "unsafe archive member"): _safe_member(member) link = tarfile.TarInfo("etc/escape") link.type = tarfile.SYMTYPE link.linkname = "../../root/key" with self.assertRaisesRegex(RestoreError, "escapes"): _safe_member(link) def test_restore_destination_cannot_be_repository(self) -> None: bundle = self._create() with self.assertRaisesRegex(RestoreError, "repository"): extract_bundle( bundle=bundle, spec_path=self.spec, destination=ROOT / "restore-test", identity=self.identity, age_binary=shutil.which("age"), ) def test_status_fails_when_receipt_is_stale(self) -> None: bundle = self._create() created = datetime.fromisoformat( json.loads((bundle / "receipt.json").read_text())["created_at"].replace("Z", "+00:00") ) with mock.patch("s1_backup._utc_now", return_value=created + timedelta(hours=27)): self.assertEqual("stale", inspect_status(self.spec, self.output)["status"]) def test_prune_requires_current_exact_approval(self) -> None: declaration = yaml.safe_load(self.spec.read_text()) declaration["retention"]["keep_complete_bundles"] = 2 self.spec.write_text(yaml.safe_dump(declaration, sort_keys=False)) for index in range(3): self._create(timestamp=f"2026082{index + 1}T120000Z") plan = plan_prune(self.spec, self.output) self.assertEqual(["s1-backup-20260821T120000Z"], plan["candidates"]) with self.assertRaisesRegex(BackupError, "exact candidate set"): apply_prune(self.spec, self.output, "PRUNE-S1-BACKUPS-WRONG") self.assertTrue((self.output / plan["candidates"][0]).exists()) result = apply_prune(self.spec, self.output, plan["approval"]) self.assertEqual("pruned", result["status"]) self.assertFalse((self.output / plan["candidates"][0]).exists()) def test_status_rejects_tampered_passing_receipt(self) -> None: bundle = self._create() receipt_path = bundle / "receipt.json" receipt = json.loads(receipt_path.read_text()) receipt["created_at"] = "2020-01-01T00:00:00Z" receipt_path.write_text(json.dumps(receipt)) with self.assertRaisesRegex(BackupError, "invalid"): inspect_status(self.spec, self.output) def test_timer_calendar_and_unit_are_source_renderable(self) -> None: timer = ROOT / "ansible/roles/s1_backup/templates/railiance-backup-s1.timer.j2" service = ROOT / "ansible/roles/s1_backup/templates/railiance-backup-s1.service.j2" timer_text = timer.read_text() service_text = service.read_text() self.assertIn("OnCalendar=*-*-* 02:15:00 UTC", timer_text) self.assertIn("Persistent=true", timer_text) self.assertIn("ReadWritePaths=/opt/backup/railiance/infra", service_text) if shutil.which("systemd-analyze"): result = subprocess.run( ["systemd-analyze", "calendar", "*-*-* 02:15:00 UTC"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) self.assertEqual(0, result.returncode, result.stderr) playbook = yaml.safe_load((ROOT / "ansible/playbooks/s1-backup.yaml").read_text()) self.assertEqual("s1_backup", playbook[0]["roles"][0]["role"]) self.assertIn("DEPLOY-RAIL-HO-WP-0012-S1-BACKUP-TIMER", str(playbook)) bootstrap = (ROOT / "ansible/playbooks/bootstrap.yaml").read_text() self.assertNotIn("s1_backup", bootstrap) if __name__ == "__main__": unittest.main()