diff options
Diffstat (limited to 'tests/post-rebuild-check')
| -rw-r--r-- | tests/post-rebuild-check/test_post_rebuild_check.py | 90 |
1 files changed, 86 insertions, 4 deletions
diff --git a/tests/post-rebuild-check/test_post_rebuild_check.py b/tests/post-rebuild-check/test_post_rebuild_check.py index 4894451..757039b 100644 --- a/tests/post-rebuild-check/test_post_rebuild_check.py +++ b/tests/post-rebuild-check/test_post_rebuild_check.py @@ -21,6 +21,8 @@ probe"): PRC_LOCAL_SCAN_ROOTS newline-separated roots to scan for *.example orphans PRC_PROJECT_ROOTS newline-separated project dirs for the tooling check PRC_SIGNAL_ACCOUNTS signal-cli listAccounts output ("" = no accounts); + PRC_NTP_SOURCES newline list of configured NTP server addresses + ("MISSING" = no NTP daemon active) the special value MISSING means the binary is absent Run from repo root: @@ -39,7 +41,8 @@ CHECK = os.path.join(REPO_ROOT, "scripts", "post-rebuild-check") def run_check(failed_units="", unit_states="", local_roots="", - project_roots="", signal_accounts="+15045551234"): + project_roots="", signal_accounts="+15045551234", + ntp_sources="162.159.200.1\npool.ntp.org"): """Run the script with every probe stubbed; defaults are all-clean. Roots are newline-separated. Empty means "the seam is set and names no @@ -52,11 +55,88 @@ def run_check(failed_units="", unit_states="", local_roots="", env["PRC_LOCAL_SCAN_ROOTS"] = local_roots env["PRC_PROJECT_ROOTS"] = project_roots env["PRC_SIGNAL_ACCOUNTS"] = signal_accounts + env["PRC_NTP_SOURCES"] = ntp_sources return subprocess.run( ["sh", CHECK], capture_output=True, text=True, timeout=30, env=env, ) +class NtpBootstrap(unittest.TestCase): + """Check 6 — the clock/DNS bootstrap deadlock. + + A wrong clock fails the DoT certificate and DNSSEC signature checks this + machine's DNS runs on, so nothing resolves; and an NTP daemon whose every + source is a hostname then cannot resolve the servers that would correct + the clock. One source addressed by IP is what makes the machine able to + recover on its own. + """ + + # --- Normal cases --------------------------------------------------- + + def test_an_ip_addressed_source_is_clean(self): + r = run_check(ntp_sources="162.159.200.1\npool.ntp.org") + self.assertIn("check 6/6: NTP bootstrap — ok", r.stdout) + self.assertEqual(r.returncode, 0, r.stdout) + + def test_all_hostname_sources_is_a_finding(self): + # The velox 2026-08-19 shape exactly: stock Arch chrony.conf, whose + # only source is a pool hostname. + r = run_check(ntp_sources="2.arch.pool.ntp.org") + self.assertIn("every NTP source is named by hostname", r.stdout) + self.assertEqual(r.returncode, 1) + + def test_an_ipv6_addressed_source_counts(self): + r = run_check(ntp_sources="2606:4700:f1::1") + self.assertIn("check 6/6: NTP bootstrap — ok", r.stdout) + + # --- Boundary cases ------------------------------------------------- + + def test_the_literal_may_sit_anywhere_in_the_list(self): + # Order must not matter; the property is "at least one", and the + # drop-in that carries it is read after the main config. + r = run_check(ntp_sources="a.pool.ntp.org\nb.pool.ntp.org\n162.159.200.1") + self.assertIn("check 6/6: NTP bootstrap — ok", r.stdout) + + def test_blank_lines_between_sources_are_ignored(self): + r = run_check(ntp_sources="\n\n162.159.200.1\n\n") + self.assertIn("check 6/6: NTP bootstrap — ok", r.stdout) + + def test_a_hostname_containing_digits_and_dots_is_not_an_address(self): + # The trap in any naive "looks like an IP" test: these resolve through + # DNS like any other name, so counting one as an address would hand a + # deadlocked machine a clean bill. + for host in ("0.arch.pool.ntp.org", "3.us.pool.ntp.org", "time1.google.com"): + with self.subTest(host=host): + r = run_check(ntp_sources=host) + self.assertIn("every NTP source is named by hostname", r.stdout) + + # --- Error cases ---------------------------------------------------- + + def test_no_ntp_daemon_is_a_finding(self): + r = run_check(ntp_sources="MISSING") + self.assertIn("no NTP implementation is active", r.stdout) + self.assertEqual(r.returncode, 1) + + def test_no_sources_configured_is_a_finding(self): + # Fails closed: an empty list proves nothing about the machine, and + # reporting ok would be a false pass on a box with no time sync at all. + r = run_check(ntp_sources="") + self.assertIn("no NTP sources are configured", r.stdout) + self.assertEqual(r.returncode, 1) + + def test_unset_seam_falls_through_to_the_real_probe(self): + # Same contract as every other seam: unset means "really look", so a + # caller who forgets the variable cannot silently skip the check. + env = dict(os.environ) + env.update({"PRC_FAILED_UNITS": "", "PRC_UNIT_STATES": "", + "PRC_LOCAL_SCAN_ROOTS": "", "PRC_PROJECT_ROOTS": "", + "PRC_SIGNAL_ACCOUNTS": "+15045551234"}) + env.pop("PRC_NTP_SOURCES", None) + r = subprocess.run(["sh", CHECK], capture_output=True, text=True, + timeout=30, env=env) + self.assertIn("check 6/6: NTP bootstrap", r.stdout) + + class AllClean(unittest.TestCase): # --- Normal cases --------------------------------------------------- @@ -482,6 +562,7 @@ class SignalAccount(unittest.TestCase): env = dict(os.environ) env.update({"PRC_FAILED_UNITS": "", "PRC_UNIT_STATES": "", "PRC_LOCAL_SCAN_ROOTS": "", "PRC_PROJECT_ROOTS": "", + "PRC_NTP_SOURCES": "162.159.200.1", "PRC_SIGNAL_ACCOUNTS": "+15045551234", "signal_missing": "1"}) r = subprocess.run(["sh", CHECK], capture_output=True, text=True, @@ -504,6 +585,7 @@ class ProbeFailure(unittest.TestCase): env = dict(os.environ) env.update({"PRC_FAILED_UNITS": "", "PRC_UNIT_STATES": "", "PRC_LOCAL_SCAN_ROOTS": "", "PRC_PROJECT_ROOTS": "", + "PRC_NTP_SOURCES": "162.159.200.1", "PRC_SIGNAL_ACCOUNTS": "+15045551234"}) for n in names: env.pop(n, None) @@ -624,7 +706,7 @@ class WedgedSystemctl(unittest.TestCase): self.assertEqual(r.returncode, 1) self.assertIn("could not query user units", r.stdout) # The run must reach the end rather than stopping at the first call. - self.assertIn("check 5/5", r.stdout) + self.assertIn("check 6/6", r.stdout) def test_a_hanging_systemctl_does_not_stall_the_whole_run(self): # The fake sleeps 8s against a 1s bound, so a bounded run lands near @@ -648,7 +730,7 @@ class Reporting(unittest.TestCase): unit_states="c.timer disabled") self.assertEqual(r.returncode, 1) summary = r.stdout.strip().splitlines()[-1] - self.assertEqual(summary, "3 finding(s) across 5 checks") + self.assertEqual(summary, "3 finding(s) across 6 checks") def test_the_summary_count_tracks_every_check(self): # One finding from each of the five, so a counter that drops or @@ -664,7 +746,7 @@ class Reporting(unittest.TestCase): local_roots=scan, project_roots=proj, signal_accounts="") summary = r.stdout.strip().splitlines()[-1] - self.assertEqual(summary, "5 finding(s) across 5 checks") + self.assertEqual(summary, "5 finding(s) across 6 checks") def test_help_exits_zero(self): r = subprocess.run(["sh", CHECK, "--help"], |