felhom-agent/configs/test_felhom_crash_guard.py

#!/usr/bin/python3
"""Tests for felhom-crash-guard (`11` §5.9). Temp dirs only; nothing real is touched. Red-proof seam: CRASHGUARD_UNDER_TEST."""
import importlib.machinery
import importlib.util
import json
import os
import pathlib
import tempfile
import unittest

HERE = pathlib.Path(__file__).resolve().parent
_loader = importlib.machinery.SourceFileLoader("crashguard", os.environ.get("CRASHGUARD_UNDER_TEST", str(HERE / "felhom-crash-guard")))
_spec = importlib.util.spec_from_loader("crashguard", _loader)
cg = importlib.util.module_from_spec(_spec)
_loader.exec_module(cg)

T0 = 1791115200.0  # 2026-10-04T12:00:00Z


class FakeEnv(cg.Env):
    def __init__(self, d):
        super().__init__(conf=os.path.join(d, "conf"), state_dir=os.path.join(d, "state"),
                         panic_path=os.path.join(d, "panic"), uptime_path=os.path.join(d, "uptime"),
                         boot_id_path=os.path.join(d, "bootid"))
        self.t = T0
        self.logs = []
        open(self.panic_path, "w").write("0\n")
        open(self.uptime_path, "w").write("20.00 10.00\n")

    def now(self):
        return self.t

    def log(self, line):
        self.logs.append(line)

    def panic(self):
        return int(open(self.panic_path).read())

    def state(self):
        return json.load(open(os.path.join(self.state_dir, "state.json")))


class Guard(unittest.TestCase):
    def setUp(self):
        self.d = tempfile.TemporaryDirectory()
        self.e = FakeEnv(self.d.name)

    def tearDown(self):
        self.d.cleanup()

    def crash_boot(self, minutes_later):
        self.e.t += minutes_later * 60
        cg.main(["x", "boot"], self.e)  # no clean-stop before it: an unclean stop

    def clean_reboot(self, minutes_later):
        cg.main(["x", "clean-stop"], self.e)
        self.e.t += minutes_later * 60
        cg.main(["x", "boot"], self.e)

    def test_first_boot_is_not_a_crash_and_arms(self):
        cg.main(["x", "boot"], self.e)
        s = self.e.state()
        self.assertFalse(s["last_boot_unclean"])
        self.assertEqual(self.e.panic(), 10)
        self.assertTrue(s["armed"])

    def test_clean_reboots_never_count(self):
        cg.main(["x", "boot"], self.e)
        for _ in range(5):
            self.clean_reboot(1)
        s = self.e.state()
        self.assertEqual(s["unclean_boots_in_window"], 0)
        self.assertEqual(self.e.panic(), 10)

    def test_third_crash_in_an_hour_leaves_the_box_off(self):
        # operator's words: "if it crashes 3 times within one hour, it stays off" — after crash 2 the guard trips,
        # so crash 3 (kernel.panic = 0) does not restart the box.
        cg.main(["x", "boot"], self.e)
        self.crash_boot(5)
        self.assertEqual(self.e.panic(), 10, "one crash: still restarts")
        self.crash_boot(5)
        s = self.e.state()
        self.assertTrue(s["tripped"], s)
        self.assertEqual(self.e.panic(), 0, "after the 2nd crash boot the 3rd crash must leave the box off")
        self.assertIn("2 unclean boots within 60 minutes", s["tripped_reason"])

    def test_crashes_spread_over_more_than_the_window_do_not_trip(self):
        cg.main(["x", "boot"], self.e)
        self.crash_boot(5)
        self.crash_boot(61)
        self.assertFalse(self.e.state()["tripped"])
        self.assertEqual(self.e.panic(), 10)

    def test_tripped_stays_tripped_across_boots(self):
        cg.main(["x", "boot"], self.e)
        self.crash_boot(5)
        self.crash_boot(5)
        self.clean_reboot(30)  # the operator switched it on; even a clean boot keeps the trip
        self.assertTrue(self.e.state()["tripped"])
        self.assertEqual(self.e.panic(), 0)

    def test_rearms_after_24h_of_normal_running(self):
        cg.main(["x", "boot"], self.e)
        self.crash_boot(5)
        self.crash_boot(5)
        self.e.t += 23 * 3600
        cg.main(["x", "check"], self.e)
        self.assertTrue(self.e.state()["tripped"], "not before 24 h")
        self.e.t += 3600
        cg.main(["x", "check"], self.e)
        s = self.e.state()
        self.assertFalse(s["tripped"])
        self.assertEqual(self.e.panic(), 10)
        self.assertIn("timer", s["rearmed_by"])

    def test_operator_rearm_starts_a_fresh_window(self):
        cg.main(["x", "boot"], self.e)
        self.crash_boot(5)
        self.crash_boot(5)
        self.e.t += 60
        cg.main(["x", "rearm"], self.e)
        s = self.e.state()
        self.assertFalse(s["tripped"])
        self.assertEqual(s["rearmed_by"], "operator")
        self.assertEqual(s["unclean_boots_24h"], 2, "the history stays")
        self.crash_boot(5)
        self.assertFalse(self.e.state()["tripped"], "one crash after a re-arm must not trip at once")

    def test_config_numbers_are_read(self):
        open(self.e.conf, "w").write("LIMIT=2\nPANIC_SECONDS=30\n")
        cg.main(["x", "boot"], self.e)
        self.assertEqual(self.e.panic(), 30)
        self.crash_boot(1)
        self.assertTrue(self.e.state()["tripped"], "LIMIT=2: the first crash boot trips")

    def test_state_is_world_readable_for_the_agent(self):
        cg.main(["x", "boot"], self.e)
        mode = os.stat(os.path.join(self.e.state_dir, "state.json")).st_mode & 0o777
        self.assertEqual(mode, 0o644)


class KernelStepCannotLeaveTheBoxOff(unittest.TestCase):
    """R-836 / `11` §5.11 Part B 4: a kernel step's planned reboot, one crash and one self-revert cannot add up to the
    box staying off. The wrapper starts a step only when the guard is armed with NO unclean boot in its window
    (felhom-os-apply Kernel.check_guard, R21); the planned reboot and the self-revert are orderly (`systemctl reboot` —
    the clean-stop marker); so the step adds at most ONE unclean boot, and the box stays off only after the LIMIT-th
    (3rd) within the hour. Red-proof: audits/kernel-lane-2026-10-07/A/redproof.txt."""

    def setUp(self):
        self.d = tempfile.TemporaryDirectory()
        self.e = FakeEnv(self.d.name)
        cg.main(["x", "boot"], self.e)
        s = self.e.state()
        self.assertTrue(s["armed"])
        self.assertEqual(s["unclean_boots_in_window"], 0, "the wrapper's precondition (R21)")

    def tearDown(self):
        self.d.cleanup()

    def planned(self, minutes):
        cg.main(["x", "clean-stop"], self.e)
        self.e.t += minutes * 60
        cg.main(["x", "boot"], self.e)

    def crash(self, minutes):
        self.e.t += minutes * 60
        cg.main(["x", "boot"], self.e)

    def test_planned_reboot_one_crash_one_self_revert(self):
        self.planned(2)      # the step's one-shot reboot (orderly)
        self.crash(3)        # the new kernel crashes after the guard ran; the box restarts (panic=10)
        self.planned(2)      # the self-revert (orderly)
        s = self.e.state()
        self.assertFalse(s["tripped"], s)
        self.assertTrue(s["armed"])
        self.assertEqual(self.e.panic(), 10, "the box still restarts after a crash")
        self.assertEqual(s["unclean_boots_in_window"], 1, "the step added exactly one unclean boot")

    def test_a_panic_before_userspace_is_not_even_counted(self):
        # the one-shot kernel panics before the guard's unit runs (measured: rdinit= and init= missing): the planned
        # reboot's clean-stop marker is still there when the old kernel boots, so this boot counts as clean.
        cg.main(["x", "clean-stop"], self.e)
        self.e.t += 120      # the panicking boot: no userspace, the guard never ran
        cg.main(["x", "boot"], self.e)
        s = self.e.state()
        self.assertEqual(s["unclean_boots_in_window"], 0, s)
        self.assertEqual(self.e.panic(), 10)

    def test_the_box_stays_off_only_after_two_more_crashes_than_the_step_makes(self):
        self.planned(2)
        self.crash(3)        # the step's one crash
        self.planned(2)      # the self-revert
        self.crash(5)        # an UNRELATED crash within the hour: the guard trips (the 3rd would leave it off)
        s = self.e.state()
        self.assertTrue(s["tripped"])
        self.assertEqual(s["unclean_boots_in_window"], 2, "two unclean boots: one from the step, one not")


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