nixos/scripts/deploy-check.py

"""Fault-injection checks: python3 scripts/deploy-check.py.

All system/profile/bootloader operations are replaced with filesystem fixtures.
No Nix build, systemd unit, network request or production activation is performed.
"""

import contextlib
import importlib.util
import io
import json
import os
import signal
import subprocess
import sys
from pathlib import Path
import tempfile
import time
import unittest
from unittest.mock import patch, MagicMock

spec = importlib.util.spec_from_file_location(
    "deploy", Path(__file__).with_name("deploy.py")
)
m = importlib.util.module_from_spec(spec)
spec.loader.exec_module(m)


class Fixture(m.Deployment):
    def __init__(self, root):
        self.calls = []
        self.fail = set()
        self.health_calls = 0
        self.root = root
        self.candidate = root / "candidate"
        self.old_running = root / "running"
        self.old_boot = root / "boot"
        recovery = self.candidate / "etc/systemd/system/fred-deploy-recover.service"
        recovery.parent.mkdir(parents=True, exist_ok=True)
        recovery.write_text("fixture")
        for system in (self.candidate, self.old_running, self.old_boot):
            (system / "bin").mkdir(parents=True, exist_ok=True)
            (system / "bin/switch-to-configuration").write_text("fixture")
        runner = root / "tools/bin/worker"
        runner.parent.mkdir(parents=True, exist_ok=True)
        runner.write_text("fixture")
        super().__init__(
            runner, root / "state", root / "roots", root / "lock", self.fake
        )
        self.current = root / "current"
        self.profile = root / "profile"
        if not self.current.is_symlink():
            self.current.symlink_to(self.old_running)
        if not self.profile.is_symlink():
            self.profile.symlink_to(self.old_boot)
        self.boot_default = self.old_boot

    def fake(self, args, timeout=120):
        args = list(map(str, args))
        self.calls.append(args)
        step = args[0]
        if args[:2] == ["nix", "build"]:
            step = "build"
            self.pin("candidate", self.candidate)
        elif args[0] == "systemd-run":
            step = "arm"
        elif args[0] == "nix-env":
            target = Path(args[-1])
            self.profile.unlink()
            self.profile.symlink_to(target)
            step = "commit-profile" if target == self.candidate else "restore-profile"
        elif args[0].endswith("switch-to-configuration"):
            target = Path(args[0]).parents[1]
            if args[1] == "test":
                self.current.unlink()
                self.current.symlink_to(target)
            else:
                self.boot_default = target
            step = ("candidate-" if target == self.candidate else "restore-") + args[1]
        elif args[:2] == ["systemctl", "stop"]:
            step = "disarm" if args[-1].endswith(".timer") else "stop-worker"
        elif args[:2] == ["systemctl", "show"]:
            return "not-found"
        if step in self.fail:
            raise RuntimeError("Injected failure after side effect: " + step)
        return ""

    def healthy(self):
        self.health_calls += 1
        if self.current.resolve() == self.candidate:
            if "health" in self.fail or (
                "final-health" in self.fail and self.health_calls == 2
            ):
                raise RuntimeError("Injected unhealthy router")
        elif "restore-health" in self.fail:
            raise RuntimeError("Restored configuration is unhealthy")

    def wait_healthy(self, duration=90):
        self.healthy()

    def deploy(self, ident="a" * 32, flake="fixture", reboot=False):
        super().deploy(ident, flake, reboot)


class Transactions(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix="fred-deploy-check-")
        self.addCleanup(self.temp.cleanup)
        self.f = Fixture(Path(self.temp.name))
        self.output = contextlib.redirect_stdout(io.StringIO())
        self.output.__enter__()
        self.addCleanup(self.output.__exit__, None, None, None)

    def assert_restored(self):
        self.assertEqual(self.f.current.resolve(), self.f.old_running)
        self.assertEqual(self.f.profile.resolve(), self.f.old_boot)
        self.assertEqual(self.f.boot_default, self.f.old_boot)

    def pending(self, phase="activating", boot_id=None):
        self.f.pin("old-running", self.f.old_running)
        self.f.pin("old-boot", self.f.old_boot)
        record = {
            "id": "a" * 32,
            "unit": "fred-deploy-" + "a" * 32 + ".service",
            "old_running": str(self.f.old_running),
            "old_boot": str(self.f.old_boot),
            "candidate": str(self.f.candidate),
            "boot_id": boot_id or self.f.boot_id(),
        }
        self.f.save(record, phase)
        self.f.current.unlink()
        self.f.current.symlink_to(self.f.candidate)
        return record

    def test_success_orders_test_health_and_boot_commit(self):
        self.f.deploy()
        self.assertEqual(self.f.read()["phase"], "committed")
        self.assertEqual(self.f.profile.resolve(), self.f.candidate)
        commands = self.f.calls
        arm = next(i for i, c in enumerate(commands) if c[0] == "systemd-run")
        test = commands.index(
            [str(self.f.candidate / "bin/switch-to-configuration"), "test"]
        )
        boot = commands.index(
            [str(self.f.candidate / "bin/switch-to-configuration"), "boot"]
        )
        self.assertLess(arm, test)
        self.assertLess(test, boot)
        self.assertEqual(self.f.health_calls, 2)
        self.assertFalse(list(self.f.roots.iterdir()))

    def test_build_failure_never_activates_or_arms(self):
        self.f.fail.add("build")
        with self.assertRaises(RuntimeError):
            self.f.deploy()
        self.assertEqual(self.f.read()["phase"], "build-failed")
        self.assert_restored()
        self.assertFalse(
            any(
                c[0] == "systemd-run" or c[0].endswith("switch-to-configuration")
                for c in self.f.calls
            )
        )

    def test_candidate_must_retain_boot_recovery(self):
        (self.f.candidate / "etc/systemd/system/fred-deploy-recover.service").unlink()
        with self.assertRaisesRegex(RuntimeError, "missing the boot recovery"):
            self.f.deploy()
        self.assert_restored()
        self.assertEqual(self.f.read()["phase"], "build-failed")

    def test_activation_and_commit_failure_paths_restore_both_systems(self):
        for failure in [
            "arm",
            "candidate-test",
            "health",
            "commit-profile",
            "candidate-boot",
            "final-health",
        ]:
            with self.subTest(failure=failure), tempfile.TemporaryDirectory() as temp:
                f = Fixture(Path(temp))
                f.fail.add(failure)
                with self.assertRaises(RuntimeError):
                    f.deploy()
                self.assertEqual(f.read()["phase"], "rolled-back")
                self.assertEqual(f.current.resolve(), f.old_running)
                self.assertEqual(f.profile.resolve(), f.old_boot)
                self.assertEqual(f.boot_default, f.old_boot)

    def test_failed_boot_restore_still_restores_runtime_and_retains_roots(self):
        self.f.fail.update({"candidate-test", "restore-boot"})
        with self.assertRaisesRegex(RuntimeError, "needs attention"):
            self.f.deploy()
        self.assertEqual(self.f.current.resolve(), self.f.old_running)
        self.assertEqual(self.f.read()["phase"], "rollback-failed")
        self.assertTrue((self.f.roots / "old-boot").exists())
        with self.assertRaisesRegex(RuntimeError, "needs recovery"):
            self.f.deploy("b" * 32)
        self.f.fail.clear()
        self.f.recover()
        self.assertEqual(self.f.read()["phase"], "rolled-back")

    def test_unhealthy_restored_router_is_not_reported_as_success(self):
        self.f.fail.update({"candidate-test", "restore-health"})
        with self.assertRaisesRegex(RuntimeError, "needs attention"):
            self.f.deploy()
        self.assertEqual(self.f.read()["phase"], "rollback-failed")

    def test_late_timer_cannot_undo_commit(self):
        self.f.fail.add("disarm")
        self.f.deploy()
        self.assertEqual(self.f.read()["phase"], "committed")
        before = len(self.f.calls)
        self.f.recover("a" * 32)
        self.assertEqual(len(self.f.calls), before)

    def test_stale_timer_cannot_touch_new_transaction(self):
        self.pending()
        self.f.recover("b" * 32)
        self.assertFalse(self.f.calls)
        self.assertEqual(self.f.read()["phase"], "activating")

    def test_boot_service_does_not_cancel_its_own_installation(self):
        self.pending()
        self.f.recover(boot=True)
        self.assertFalse(self.f.calls)
        self.assertEqual(self.f.read()["phase"], "activating")

    def test_next_boot_recovers_even_if_worker_unit_was_collected(self):
        self.pending(boot_id="previous-boot")
        self.f.fail.add("stop-worker")
        self.f.recover(boot=True)
        self.assert_restored()
        self.assertEqual(self.f.read()["phase"], "rolled-back")

    def test_interrupted_build_can_be_retried_without_activation(self):
        self.pending(phase="building")
        # A building record never altered the running system.
        self.f.current.unlink()
        self.f.current.symlink_to(self.f.old_running)
        self.f.deploy("b" * 32)
        self.assertEqual(self.f.read()["phase"], "committed")

    def test_boot_time_deployment_recovers_previous_boot_first(self):
        self.pending(boot_id="previous-boot")
        self.f.deploy("b" * 32)
        restore = self.f.calls.index(
            [str(self.f.old_running / "bin/switch-to-configuration"), "test"]
        )
        build = next(
            i for i, args in enumerate(self.f.calls) if args[:2] == ["nix", "build"]
        )
        self.assertLess(restore, build)
        self.assertEqual(self.f.read()["phase"], "committed")

    def test_failed_state_write_does_not_prevent_restoring_system(self):
        record = self.pending()
        save = self.f.save

        def fail_start(record, phase, **fields):
            if phase == "rolling-back":
                raise OSError("disk full")
            return save(record, phase, **fields)

        self.f.save = fail_start
        self.f.rollback_locked(record)
        self.assert_restored()

    def test_recovery_rereads_state_after_stopping_worker(self):
        record = self.pending()
        command = self.f.command

        def race(args, **kw):
            result = command(args, **kw)
            if args[:2] == ["systemctl", "stop"]:
                self.f.save({**record, "id": "b" * 32}, "armed")
            return result

        self.f.command = race
        self.f.recover("a" * 32)
        self.assertEqual(self.f.read()["id"], "b" * 32)
        self.assertFalse(
            any(c[0].endswith("switch-to-configuration") for c in self.f.calls)
        )

    def test_transaction_lock_excludes_another_process(self):
        marker = self.f.root / "entered"
        code = "import fcntl,pathlib,sys; f=open(sys.argv[1],'a'); fcntl.flock(f,fcntl.LOCK_EX); pathlib.Path(sys.argv[2]).touch()"
        with self.f.exclusive():
            child = subprocess.Popen(
                [sys.executable, "-c", code, str(self.f.lock), str(marker)]
            )
            time.sleep(0.2)
            self.assertFalse(marker.exists())
        child.wait(timeout=3)
        self.assertEqual(child.returncode, 0)
        self.assertTrue(marker.exists())

    def test_independent_recovery_stops_hung_worker_before_taking_lock(self):
        self.pending()
        marker = self.f.root / "holding-lock"
        code = "import fcntl,pathlib,sys,time; f=open(sys.argv[1],'a'); fcntl.flock(f,fcntl.LOCK_EX); pathlib.Path(sys.argv[2]).touch(); time.sleep(30)"
        worker = subprocess.Popen(
            [sys.executable, "-c", code, str(self.f.lock), str(marker)]
        )
        try:
            deadline = time.monotonic() + 3
            while not marker.exists() and time.monotonic() < deadline:
                time.sleep(0.01)
            self.assertTrue(marker.exists())
            recovery = subprocess.run(
                [
                    sys.executable,
                    "-B",
                    __file__,
                    "recover-fixture",
                    str(self.f.root),
                    str(worker.pid),
                ],
                timeout=5,
                capture_output=True,
                text=True,
            )
            self.assertEqual(recovery.returncode, 0, recovery.stdout + recovery.stderr)
            worker.wait(timeout=2)
            self.assertEqual(worker.returncode, -signal.SIGKILL)
            self.assert_restored()
            self.assertEqual(self.f.read()["phase"], "rolled-back")
        finally:
            if worker.poll() is None:
                worker.kill()
                worker.wait()

    def test_command_timeout_kills_descendants(self):
        marker = self.f.root / "unexpected-child-write"
        child = (
            "import pathlib,time,sys; time.sleep(.5); pathlib.Path(sys.argv[1]).touch()"
        )
        parent = "import subprocess,sys,time; subprocess.Popen([sys.executable,'-c',sys.argv[1],sys.argv[2]]); time.sleep(30)"
        with self.assertRaisesRegex(RuntimeError, "timed out"):
            m.run([sys.executable, "-c", parent, child, str(marker)], timeout=0.1)
        time.sleep(0.6)
        self.assertFalse(marker.exists())

    def test_reboot_is_requested_only_after_commit(self):
        self.f.deploy(reboot=True)
        self.assertEqual(self.f.calls[-1], ["systemctl", "reboot", "--no-block"])
        self.assertEqual(self.f.read()["phase"], "committed")

    def test_router_health_rejects_missing_essential_functions(self):
        for failure in [
            None,
            "service",
            "address",
            "route",
            "firewall",
            "nat",
            "wan",
            "dns",
            "ssh",
            "proxy",
        ]:
            with self.subTest(failure=failure):

                def health_command(args, timeout=120):
                    if args[0] == "systemctl":
                        self.assertNotIn("jellyfin", args)
                        if failure == "service":
                            raise RuntimeError("inactive")
                        return ""
                    if args[:4] == ["ip", "-j", "-4", "address"]:
                        return json.dumps(
                            [
                                {
                                    "addr_info": []
                                    if failure == "address"
                                    else [{"local": "10.0.0.1", "prefixlen": 24}]
                                }
                            ]
                        )
                    if args[0] == "ip":
                        return json.dumps(
                            []
                            if failure == "route"
                            else [{"dst": "default", "dev": "eno1"}]
                        )
                    if args[0] == "nft":
                        if args[-1] == "filter":
                            return json.dumps(
                                {
                                    "nftables": [
                                        {
                                            "chain": {
                                                "hook": hook,
                                                "policy": "accept"
                                                if failure == "firewall"
                                                else "drop",
                                            }
                                        }
                                        for hook in ["input", "forward"]
                                    ]
                                }
                            )
                        return json.dumps(
                            {
                                "nftables": [
                                    {"chain": {"hook": hook}}
                                    for hook in ["prerouting", "postrouting"]
                                ]
                                + (
                                    []
                                    if failure == "nat"
                                    else [
                                        {
                                            "rule": {
                                                "expr": [
                                                    {"dnat": {}},
                                                    {"masquerade": None},
                                                ]
                                            }
                                        }
                                    ]
                                )
                            }
                        )
                    if args[0] == "ping":
                        if failure == "wan":
                            raise RuntimeError("unreachable")
                        return ""
                    if args[0] == "dig":
                        return "" if failure == "dns" else "93.184.215.14"
                    if args[0] == "curl":
                        return "503" if failure == "proxy" else "302"
                    raise AssertionError(args)

                self.f.command = health_command
                ssh = MagicMock()
                ssh.__enter__.return_value.recv.return_value = (
                    b"HTTP/1.1" if failure == "ssh" else b"SSH-2.0-fixture"
                )
                read_text = Path.read_text

                def read(path, *args, **kw):
                    if str(path) == "/proc/sys/net/ipv4/ip_forward":
                        return "1"
                    return read_text(path, *args, **kw)

                with (
                    patch.object(m.socket, "create_connection", return_value=ssh),
                    patch.object(Path, "read_text", read),
                ):
                    if failure:
                        with self.assertRaises(RuntimeError):
                            m.Deployment.healthy(self.f)
                    else:
                        m.Deployment.healthy(self.f)


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "recover-fixture":
        fixture = Fixture(Path(sys.argv[2]))
        worker_pid = int(sys.argv[3])
        fake = fixture.command

        def stop_fixture_worker(args, **kwargs):
            if args[:2] == ["systemctl", "stop"] and args[-1].endswith(".service"):
                os.kill(worker_pid, signal.SIGKILL)
            return fake(args, **kwargs)

        fixture.command = stop_fixture_worker
        fixture.recover("a" * 32)
    else:
        unittest.main()