railiance-infra/tests/test_s1_backup_recovery.py

from __future__ import annotations

import copy
import base64
import http.server
import json
import os
import shutil
import subprocess
import sys
import tarfile
import tempfile
import threading
import unittest
import uuid
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
from s1_offsite import (  # noqa: E402
    OffsiteError,
    contract_sha256,
    plan_upload,
    review_contract,
    upload,
    validate_prune_receipt,
)

EXPECTED_OFFSITE_MEMBERS = {
    "manifest.json",
    "receipt.json",
    "os-config.tar.gz.age",
    "packages.txt.age",
}


@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_isolated_restore_supports_python_311_tar_api(self) -> None:
        bundle = self._create()
        with mock.patch("s1_restore.sys.version_info", (3, 11, 0)):
            restored = extract_bundle(
                bundle=bundle,
                spec_path=self.spec,
                destination=self.root / "restore-python311",
                identity=self.identity,
                age_binary=shutil.which("age"),
            )
        self.assertEqual("fixture-host\n", (restored / "etc/hostname").read_text())

    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
        declaration["offsite"]["required_before_prune"] = False
        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_prune_blocks_without_offsite_evidence(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("blocked-missing-offsite-evidence", plan["status"])
        self.assertIsNone(plan["approval"])
        self.assertEqual(["s1-backup-20260821T120000Z"], plan["missing_offsite_evidence"])

    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)

    def _offsite_contract(self, base_url: str, *, accepted: bool) -> Path:
        contract = yaml.safe_load((ROOT / "spec/s1-offsite.yaml").read_text())
        contract["provider"]["base_url"] = base_url
        contract_path = self.spec_dir / "s1-offsite.yaml"
        contract_path.write_text(yaml.safe_dump(contract, sort_keys=False))
        acceptance = {
            "schema_version": "1.0",
            "contract_sha256": contract_sha256(contract_path) if accepted else "pending",
            "status": "accepted" if accepted else "pending",
            "owner_repo": "railiance-platform",
            "decision_id": str(uuid.uuid4()) if accepted else None,
            "accepted_at": "2026-08-23T12:00:00Z" if accepted else None,
        }
        (self.spec_dir / "s1-offsite-owner-acceptance.yaml").write_text(
            yaml.safe_dump(acceptance, sort_keys=False)
        )
        return contract_path

    def test_offsite_contract_is_pending_owner_review(self) -> None:
        review = review_contract(ROOT / "spec/s1-offsite.yaml")
        self.assertEqual("review-required", review["status"])
        self.assertTrue(review["approval"].startswith("APPROVE S1-OFFSITE-CONTRACT-"))

    def test_deterministic_single_object_offsite_upload_and_collision(self) -> None:
        token = "fixture-upload-token"
        objects: dict[str, bytes] = {}

        class Handler(http.server.BaseHTTPRequestHandler):
            def do_PUT(self) -> None:  # noqa: N802
                expected_auth = "Basic " + base64.b64encode(f"{token}:".encode()).decode()
                if self.headers.get("Authorization") != expected_auth:
                    self.send_response(401)
                    self.end_headers()
                    return
                if self.headers.get("If-None-Match") != "*":
                    self.send_response(428)
                    self.end_headers()
                    return
                if self.path in objects:
                    self.send_response(412)
                    self.end_headers()
                    return
                length = int(self.headers["Content-Length"])
                objects[self.path] = self.rfile.read(length)
                self.send_response(201)
                self.send_header("ETag", '"fixture-etag"')
                self.end_headers()

            def log_message(self, format: str, *args: object) -> None:
                pass

        server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        try:
            bundle = self._create()
            contract = self._offsite_contract(
                f"http://127.0.0.1:{server.server_port}/filesdrop", accepted=True
            )
            plan = plan_upload(
                bundle=bundle,
                backup_spec=self.spec,
                contract_path=contract,
                fixture_http=True,
            )
            self.assertTrue(plan["executable"])
            second_plan = plan_upload(
                bundle=bundle,
                backup_spec=self.spec,
                contract_path=contract,
                fixture_http=True,
            )
            self.assertEqual(plan["envelope_sha256"], second_plan["envelope_sha256"])
            receipt_dir = self.root / "offsite-receipts"
            with mock.patch.dict(os.environ, {"RAILIANCE_BACKUP_NC_TOKEN": token}):
                with self.assertRaisesRegex(OffsiteError, "exact envelope"):
                    upload(
                        bundle=bundle,
                        backup_spec=self.spec,
                        contract_path=contract,
                        approval="UPLOAD-S1-OFFSITE-WRONG",
                        receipt_dir=receipt_dir,
                        fixture_http=True,
                    )
                self.assertEqual({}, objects)
                receipt_path = upload(
                    bundle=bundle,
                    backup_spec=self.spec,
                    contract_path=contract,
                    approval=plan["approval"],
                    receipt_dir=receipt_dir,
                    fixture_http=True,
                )
                with self.assertRaisesRegex(OffsiteError, "collision"):
                    upload(
                        bundle=bundle,
                        backup_spec=self.spec,
                        contract_path=contract,
                        approval=plan["approval"],
                        receipt_dir=receipt_dir,
                        fixture_http=True,
                    )
            receipt = validate_prune_receipt(bundle, self.spec, receipt_path)
            self.assertEqual(plan["remote_object_identity"], receipt["remote_object_identity"])
            self.assertNotIn(token, receipt_path.read_text())
            self.assertEqual(1, len(objects))
            envelope_path = self.root / "observed-envelope.tar"
            envelope_path.write_bytes(next(iter(objects.values())))
            with tarfile.open(envelope_path, "r:") as archive:
                self.assertEqual(EXPECTED_OFFSITE_MEMBERS, {item.name for item in archive})
        finally:
            server.shutdown()
            server.server_close()
            thread.join(timeout=2)

    def test_offsite_upload_refuses_pending_owner_and_wrong_approval(self) -> None:
        bundle = self._create()
        contract = self._offsite_contract("http://127.0.0.1:9/filesdrop", accepted=False)
        plan = plan_upload(
            bundle=bundle,
            backup_spec=self.spec,
            contract_path=contract,
            fixture_http=True,
        )
        self.assertFalse(plan["executable"])
        self.assertIsNone(plan["approval"])
        with self.assertRaisesRegex(OffsiteError, "has not accepted"):
            upload(
                bundle=bundle,
                backup_spec=self.spec,
                contract_path=contract,
                approval="UPLOAD-S1-OFFSITE-WRONG",
                receipt_dir=self.root / "receipts",
                fixture_http=True,
            )

    def test_offsite_upload_never_follows_redirect(self) -> None:
        paths: list[str] = []

        class RedirectHandler(http.server.BaseHTTPRequestHandler):
            def do_PUT(self) -> None:  # noqa: N802
                paths.append(self.path)
                self.send_response(307)
                self.send_header("Location", "/credential-capture")
                self.end_headers()

            def log_message(self, format: str, *args: object) -> None:
                pass

        server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), RedirectHandler)
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        try:
            bundle = self._create()
            contract = self._offsite_contract(
                f"http://127.0.0.1:{server.server_port}/filesdrop", accepted=True
            )
            plan = plan_upload(
                bundle=bundle,
                backup_spec=self.spec,
                contract_path=contract,
                fixture_http=True,
            )
            with mock.patch.dict(os.environ, {"RAILIANCE_BACKUP_NC_TOKEN": "redirect-token"}):
                with self.assertRaisesRegex(OffsiteError, "HTTP 307"):
                    upload(
                        bundle=bundle,
                        backup_spec=self.spec,
                        contract_path=contract,
                        approval=plan["approval"],
                        receipt_dir=self.root / "redirect-receipts",
                        fixture_http=True,
                    )
            self.assertEqual(1, len(paths))
            self.assertNotEqual("/credential-capture", paths[0])
            self.assertFalse((self.root / "redirect-receipts").exists())
        finally:
            server.shutdown()
            server.server_close()
            thread.join(timeout=2)


if __name__ == "__main__":
    unittest.main()