diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index 18f277d1..df3f0b4d 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -274,6 +274,7 @@ jobs: run: | python3 -B test/test_official_web_console.py -v python3 -B test/test_hil_linux_web_serial.py -v + python3 -B test/test_hil_slowfi_guard.py -v - name: Verify official setup GUI initial time command deadline run: python3 -B test/test_official_config_gui.py -v diff --git a/test/README.md b/test/README.md index a310169c..67063759 100644 --- a/test/README.md +++ b/test/README.md @@ -92,6 +92,33 @@ python3 -B test/test_hil_linux_web_serial.py -v See [the scoped G2 comparison and Chromium source evidence](../docs/research/linux_chromium_web_serial_tty.md). +### Timed WiFi return guard + +The host-only SlowFi guard tests use fake NetworkManager responses and local +locks. They cover recurring recovery after more than five failures, bounded +attempts, exact original UUID plus connected-state proof, overlapping attempts, +cleanup refusal while the original network is unproven, owned-profile collisions, +and timer-stop side effects with uncertain results. They do not test a real +systemd timer, WiFi association, a Pi, or a radio: + +```sh +python3 -B test/test_hil_slowfi_guard.py -v +``` + +The helper's CLI defaults to a plan and makes no NetworkManager/systemd calls: + +```sh +python3 -B -m tools.hil.slowfi_guard --config /path/to/reviewed-config.json +``` + +A future hardware test must separately bind and stage the reviewed host config, +use a fresh owned unit identity, and prove the actual first timer deadline before +changing the Pi's network. The rendered guard first attempts the original UUID +after 660 seconds, then retries 30 seconds after each bounded attempt until the +original UUID is connected. It never edits the original profile. Live recovery +and cleanup require explicit authorization, root-owned staged files and host +identity checks; cleanup also refuses unless the original network is proven. + The bootloader-version regression compiles the production reader with a C++17 `g++` (or `CXX`) compiler. Its offline cases include the MeshTower V2 SD 2.4.6 missing-UF2-text failure, OTAFIX metadata versus cached base-version precedence, diff --git a/test/test_hil_slowfi_guard.py b/test/test_hil_slowfi_guard.py new file mode 100644 index 00000000..821c0e4e --- /dev/null +++ b/test/test_hil_slowfi_guard.py @@ -0,0 +1,366 @@ +"""Pure fake-NM tests. No NetworkManager/systemd/SSH/hardware calls.""" +import copy +import contextlib +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +import importlib.util + +ROOT = Path(__file__).resolve().parents[1] +SPEC = importlib.util.spec_from_file_location( + "hil_slowfi_guard", ROOT / "tools/hil/slowfi_guard.py") +guard = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(guard) + +ORIGINAL = "11111111-1111-4111-8111-111111111111" +OWNED = "22222222-2222-4222-8222-222222222222" +OTHER = "33333333-3333-4333-8333-333333333333" +STEM = "meshcore-slowfi-return-synthetic-20261007" +CONFIG = { + "schema": 1, "original_uuid": ORIGINAL, + "expected_hostname": "synthetic-no-hardware-host", + "expected_machine_id_sha256": "1" * 64, "interface": "wlan0", + "unit_stem": STEM, "guard_dir": "/run/" + STEM, + "python_path": "/usr/bin/python3", + "helper_path": "/run/" + STEM + "/slowfi_guard.py", + "config_path": "/run/" + STEM + "/config.json", + "live_execution_authorized": False, + "owned_profiles": [{"uuid": OWNED, "id": "meshcore-owned-ap-synthetic-20261007"}], +} + + +class Clock: + def __init__(self): + self.now = 1000.0 + + def __call__(self): + return self.now + + +class FakeNM: + def __init__(self, clock=None, failures=0): + self.clock = clock or Clock() + self.failures = failures + self.attempts = 0 + self.current = {"uuid": OWNED, "state": 100} + self.timer_active = True + self.stop_ok = True + self.stop_side_effect = False + self.timer_state_unknown = False + self.state_sequence = [] + self.calls = [] + self.activation_return = None + self.activation_state = None + self.names = {OWNED: CONFIG["owned_profiles"][0]["id"], ORIGINAL: "original-profile"} + self.timeouts = False + + def state(self, interface, timeout): + self.calls.append(("state", interface, timeout)) + self.clock.now += timeout if self.timeouts else 0.1 + if self.timeouts: + return None + if self.state_sequence: + return copy.deepcopy(self.state_sequence.pop(0)) + return copy.deepcopy(self.current) + + def activate(self, original_uuid, timeout): + self.calls.append(("activate", original_uuid, timeout)) + self.attempts += 1 + self.clock.now += timeout if self.timeouts else 0.1 + if self.timeouts: + return False + if self.activation_state is not None: + self.current = copy.deepcopy(self.activation_state) + elif self.attempts > self.failures: + self.current = {"uuid": original_uuid, "state": 100} + return self.activation_return if self.activation_return is not None else self.attempts > self.failures + + def stop_timer(self, timer, timeout): + self.calls.append(("stop_timer", timer, timeout)) + self.clock.now += 0.1 + if self.stop_ok or self.stop_side_effect: + self.timer_active = False + return self.stop_ok + + def timer_state(self, timer, timeout): + self.calls.append(("timer_state", timer, timeout)) + self.clock.now += 0.1 + return None if self.timer_state_unknown else "active" if self.timer_active else "inactive" + + def stop_owned_units(self, timer, service, timeout): + self.calls.append(("stop_owned_units", timer, service, timeout)) + self.timer_active = False + return True + + def profile_name(self, value, timeout): + self.calls.append(("profile_name", value, timeout)) + return self.names.get(value) + + def delete_profile(self, value, timeout): + self.calls.append(("delete_profile", value, timeout)) + del self.names[value] + return True + + +class GuardTests(unittest.TestCase): + def setUp(self): + self.config = copy.deepcopy(CONFIG) + + def run_attempt(self, backend): + return guard.attempt(self.config, backend, clock=backend.clock) + + def run_cleanup(self, backend, acquired=True): + return guard.explicit_owned_cleanup(self.config, backend, clock=backend.clock, + lock_context=lambda _: contextlib.nullcontext(acquired)) + + def test_more_than_five_transient_failures_then_connected_stops_timer(self): + backend = FakeNM(failures=8) + receipts = [] + for number in range(9): + receipts.append(self.run_attempt(backend)) + if number < 8: + self.assertTrue(backend.timer_active) + self.assertTrue(receipts[-1]["recurring_retry_retained"]) + backend.clock.now += guard.RETRY_DELAY_SECONDS + self.assertEqual(backend.attempts, 9) + self.assertTrue(receipts[-1]["original_uuid_connected"]) + self.assertTrue(receipts[-1]["timer_stopped"]) + self.assertFalse(backend.timer_active) + + def test_persistent_failures_leave_unlimited_future_attempts(self): + backend = FakeNM(failures=10**9) + for _ in range(100): + result = self.run_attempt(backend) + self.assertFalse(result["original_uuid_connected"]) + self.assertTrue(result["recurring_retry_retained"]) + self.assertTrue(backend.timer_active) + backend.clock.now += guard.RETRY_DELAY_SECONDS + self.assertEqual(backend.attempts, 100) + self.assertFalse(any(row[0] == "stop_timer" for row in backend.calls)) + + def test_return_zero_is_not_association_proof(self): + backend = FakeNM(failures=10) + backend.activation_return = True + result = self.run_attempt(backend) + self.assertTrue(result["nmcli_activation_returned_success"]) + self.assertFalse(result["original_uuid_connected"]) + self.assertTrue(backend.timer_active) + + def test_wrong_uuid_or_nonconnected_state_never_stops_timer(self): + for state in ({"uuid": OTHER, "state": 100}, {"uuid": ORIGINAL, "state": 70}, {"uuid": ORIGINAL, "state": 30}): + with self.subTest(state=state): + backend = FakeNM() + backend.activation_state = state + result = self.run_attempt(backend) + self.assertFalse(result["original_uuid_connected"]) + self.assertTrue(backend.timer_active) + + def test_already_original_connected_has_no_nm_activation(self): + backend = FakeNM() + backend.current = {"uuid": ORIGINAL, "state": 100} + result = self.run_attempt(backend) + self.assertTrue(result["timer_stopped"]) + self.assertEqual(backend.attempts, 0) + + def test_stop_timer_failure_keeps_recurring_retry(self): + backend = FakeNM() + backend.stop_ok = False + result = self.run_attempt(backend) + self.assertTrue(result["original_uuid_connected"]) + self.assertFalse(result["timer_stopped"]) + self.assertTrue(result["recurring_retry_retained"]) + + def test_stop_timer_side_effect_then_failure_does_not_claim_retry_retained(self): + backend = FakeNM() + backend.stop_ok = False + backend.stop_side_effect = True + result = self.run_attempt(backend) + self.assertFalse(result["timer_stop_returned_success"]) + self.assertTrue(result["timer_stopped"]) + self.assertFalse(result["recurring_retry_retained"]) + self.assertEqual(result["timer_state_after_stop"], "inactive") + + def test_stop_timer_state_unknown_does_not_claim_retry_retained(self): + backend = FakeNM() + backend.timer_state_unknown = True + result = self.run_attempt(backend) + self.assertTrue(result["timer_stop_returned_success"]) + self.assertIsNone(result["recurring_retry_retained"]) + self.assertFalse(result["timer_stopped"]) + + def test_stop_exception_does_not_claim_retry_retained(self): + backend = FakeNM() + with patch.object(backend, "stop_timer", side_effect=TimeoutError("secret")): + result = self.run_attempt(backend) + self.assertIsNone(result["recurring_retry_retained"]) + self.assertFalse(result["timer_stopped"]) + + def test_all_subprocess_timeouts_stay_under_35_seconds(self): + backend = FakeNM() + backend.timeouts = True + result = self.run_attempt(backend) + self.assertLessEqual(result["elapsed_seconds"], guard.ATTEMPT_BUDGET_SECONDS) + self.assertTrue(result["recurring_retry_retained"]) + self.assertEqual([x[-1] for x in backend.calls], [3, 24, 3]) + + def test_nonblocking_lock_skips_overlap_then_allows_next_invocation(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "guard.lock" + with guard.owned_lock(path) as first: + self.assertTrue(first) + with guard.owned_lock(path) as second: + self.assertFalse(second) + with guard.owned_lock(path) as third: + self.assertTrue(third) + + def test_lock_refuses_symlink(self): + with tempfile.TemporaryDirectory() as directory: + actual = Path(directory) / "actual" + actual.touch() + alias = Path(directory) / "alias" + alias.symlink_to(actual) + with self.assertRaises(OSError): + with guard.owned_lock(alias): + pass + + def test_owned_cleanup_uses_only_captured_profile_and_units(self): + backend = FakeNM() + backend.current = {"uuid": ORIGINAL, "state": 100} + result = self.run_cleanup(backend) + self.assertTrue(result["ok"]) + self.assertTrue(result["original_profile_untouched"]) + self.assertIn(ORIGINAL, backend.names) + self.assertNotIn(OWNED, backend.names) + self.assertEqual([x[1] for x in backend.calls if x[0] == "delete_profile"], [OWNED]) + stopped = [row for row in backend.calls if row[0] == "stop_owned_units"] + self.assertEqual(stopped[0][:3], ("stop_owned_units", STEM + ".timer", STEM + ".service")) + self.assertEqual([row[0] for row in backend.calls[:4]], ["state", "state", "profile_name", "state"]) + + def test_owned_cleanup_refuses_changed_profile_name(self): + backend = FakeNM() + backend.current = {"uuid": ORIGINAL, "state": 100} + backend.names[OWNED] = "unowned-existing-profile" + result = self.run_cleanup(backend) + self.assertTrue(result["cleanup_refused"]) + self.assertTrue(result["no_mutation"]) + self.assertTrue(backend.timer_active) + self.assertEqual(result["reason"], "owned_profile_identity_changed") + self.assertFalse(any(x[0] == "delete_profile" for x in backend.calls)) + + def test_cleanup_wrong_uuid_connecting_or_inconclusive_keeps_guard_running(self): + for state in ({"uuid": OTHER, "state": 100}, {"uuid": ORIGINAL, "state": 70}, None): + with self.subTest(state=state): + backend = FakeNM() + backend.current = state + result = self.run_cleanup(backend) + self.assertTrue(result["cleanup_refused"]) + self.assertTrue(result["no_mutation"]) + self.assertTrue(backend.timer_active) + self.assertEqual([row[0] for row in backend.calls], ["state"]) + + def test_cleanup_association_lost_under_lock_refuses_without_mutation(self): + for state in ({"uuid": OWNED, "state": 100}, {"uuid": ORIGINAL, "state": 70}, None): + with self.subTest(state=state): + backend = FakeNM() + backend.state_sequence = [{"uuid": ORIGINAL, "state": 100}, state] + result = self.run_cleanup(backend) + self.assertTrue(result["cleanup_refused"]) + self.assertTrue(result["no_mutation"]) + self.assertTrue(backend.timer_active) + self.assertEqual([row[0] for row in backend.calls], ["state", "state"]) + + def test_later_ownership_collision_prevents_all_cleanup_mutations(self): + self.config["owned_profiles"].append({"uuid": OTHER, "id": "meshcore-owned-second-synthetic-20261007"}) + backend = FakeNM() + backend.current = {"uuid": ORIGINAL, "state": 100} + backend.names[OTHER] = "unowned-existing-profile" + result = self.run_cleanup(backend) + self.assertTrue(result["cleanup_refused"]) + self.assertTrue(result["no_mutation"]) + self.assertTrue(backend.timer_active) + self.assertIn(OWNED, backend.names) + self.assertFalse(any(row[0] in {"stop_owned_units", "delete_profile"} for row in backend.calls)) + + def test_cleanup_lost_before_stop_refuses_without_mutation(self): + backend = FakeNM() + backend.state_sequence = [{"uuid": ORIGINAL, "state": 100}, {"uuid": ORIGINAL, "state": 100}, {"uuid": ORIGINAL, "state": 70}] + result = self.run_cleanup(backend) + self.assertTrue(result["cleanup_refused"]) + self.assertTrue(result["no_mutation"]) + self.assertTrue(backend.timer_active) + self.assertFalse(any(row[0] in {"stop_owned_units", "delete_profile"} for row in backend.calls)) + + def test_cleanup_busy_lock_does_not_stop_any_unit_or_delete_profile(self): + backend = FakeNM() + backend.current = {"uuid": ORIGINAL, "state": 100} + result = self.run_cleanup(backend, acquired=False) + self.assertTrue(result["cleanup_refused"]) + self.assertTrue(result["no_mutation"]) + self.assertTrue(backend.timer_active) + self.assertEqual([row[0] for row in backend.calls], ["state"]) + + def test_cleanup_state_error_does_not_stop_guard(self): + backend = FakeNM() + with patch.object(backend, "state", side_effect=RuntimeError("secret")): + result = self.run_cleanup(backend) + self.assertTrue(result["cleanup_refused"]) + self.assertTrue(result["no_mutation"]) + self.assertTrue(backend.timer_active) + self.assertNotIn("secret", str(result)) + + def test_original_profile_cannot_be_declared_owned(self): + self.config["owned_profiles"][0]["uuid"] = ORIGINAL + with self.assertRaisesRegex(ValueError, "original_profile"): + guard.validate(self.config) + + def test_invalid_unit_or_path_rejected(self): + for key, value in (("unit_stem", "NetworkManager"), ("helper_path", "/tmp/unowned.py")): + with self.subTest(key=key): + config = copy.deepcopy(CONFIG) + config[key] = value + with self.assertRaises(ValueError): + guard.validate(config) + + def test_actual_backend_command_builder_never_edits_original_profile(self): + backend = guard.LiveBackend.__new__(guard.LiveBackend) + backend.interface = "wlan0" + with patch.object(backend, "call", return_value=(0, "SECRET-NOT-LOGGED")) as command: + self.assertTrue(backend.activate(ORIGINAL, 25)) + command.assert_called_once_with([guard.NMCLI, "--wait", "25", "connection", "up", "uuid", ORIGINAL, "ifname", "wlan0"], 25) + with patch.object(backend, "call", return_value=(0, "")) as command: + self.assertTrue(backend.stop_timer(STEM + ".timer", 1)) + command.assert_called_once_with([guard.SYSTEMCTL, "stop", STEM + ".timer"], 1) + + def test_backend_exception_is_safe_and_does_not_disarm_timer(self): + backend = FakeNM() + with patch.object(backend, "state", side_effect=RuntimeError("SECRET-not-for-receipt")): + result = self.run_attempt(backend) + self.assertEqual(result["errors"], ["RuntimeError"]) + self.assertTrue(result["recurring_retry_retained"]) + self.assertNotIn("SECRET", str(result)) + self.assertTrue(backend.timer_active) + + def test_parse_exact_uuid_and_connected_state(self): + self.assertEqual(guard.parse_state(ORIGINAL + "\n100 (connected)\n"), {"uuid": ORIGINAL, "state": 100}) + self.assertIsNone(guard.parse_state("--\n30 (disconnected)\n")) + self.assertIsNone(guard.parse_state(ORIGINAL + "\n100\nextra")) + + def test_plan_does_not_construct_live_backend_or_run_command(self): + with patch.object(guard.subprocess, "run", side_effect=AssertionError("unexpected command")), patch.object(guard, "LiveBackend", side_effect=AssertionError("unexpected live backend")): + result = guard.plan(self.config) + self.assertTrue(result["prepared_only"]) + files = result["unit_files"] + timer, service = files[STEM + ".timer"], files[STEM + ".service"] + self.assertIn("OnActiveSec=660s", timer) + self.assertIn("OnUnitInactiveSec=30s", timer) + self.assertIn("AccuracySec=1s", timer) + self.assertIn("TimeoutStartSec=35s", service) + self.assertIn("RemainAfterExit=no", service) + self.assertNotIn("Restart=", service) + self.assertNotIn("connection modify", timer + service) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/tools/hil/slowfi_guard.py b/tools/hil/slowfi_guard.py new file mode 100644 index 00000000..a34f5e9b --- /dev/null +++ b/tools/hil/slowfi_guard.py @@ -0,0 +1,326 @@ +#!/usr/bin/env python3 +"""Future host recovery helper. Default is a read-only plan; not staged/live-tested.""" +from __future__ import annotations + +import argparse +import contextlib +import fcntl +import hashlib +import json +import os +from pathlib import Path +import re +import socket +import subprocess +import time +import uuid + +FIRST_DELAY_SECONDS = 660 +RETRY_DELAY_SECONDS = 30 +# The worker has32s for subprocesses; the service enforces35s including startup. +ATTEMPT_BUDGET_SECONDS = 32 +CLEANUP_BUDGET_SECONDS = 75 +NMCLI = "/usr/bin/nmcli" +SYSTEMCTL = "/usr/bin/systemctl" + + +def require(condition, code): + if not condition: + raise ValueError(code) + + +def valid_uuid(value): + try: + return str(uuid.UUID(value)) == value + except (ValueError, TypeError, AttributeError): + return False + + +def validate(config): + require(config.get("schema") == 1, "schema_mismatch") + require(valid_uuid(config.get("original_uuid")), "original_uuid_unbound") + require(bool(re.fullmatch(r"[a-zA-Z0-9_.-]+", config.get("expected_hostname", ""))), "hostname_unbound") + require(bool(re.fullmatch(r"[a-f0-9]{64}", config.get("expected_machine_id_sha256", ""))), "machine_id_unbound") + require(bool(re.fullmatch(r"[a-zA-Z0-9_.-]{1,15}", config.get("interface", ""))), "interface_invalid") + stem = config.get("unit_stem", "") + require(bool(re.fullmatch(r"meshcore-slowfi-return-[a-z0-9-]{8,80}", stem)), "owned_unit_invalid") + root = config.get("guard_dir", "") + require(root == "/run/" + stem, "owned_guard_directory_invalid") + for key, name in (("helper_path", "slowfi_guard.py"), ("config_path", "config.json")): + require(config.get(key) == root + "/" + name, "owned_path_invalid") + require(config.get("python_path") == "/usr/bin/python3", "python_path_invalid") + require(type(config.get("live_execution_authorized")) is bool, "live_authorization_invalid") + owned = config.get("owned_profiles", []) + require(isinstance(owned, list) and len(owned) <= 4, "owned_profiles_invalid") + seen = set() + for item in owned: + require(isinstance(item, dict) and set(item) == {"uuid", "id"}, "owned_profile_invalid") + require(valid_uuid(item["uuid"]) and item["uuid"] != config["original_uuid"], "original_profile_must_not_be_owned") + require(item["uuid"] not in seen, "duplicate_owned_profile") + require(bool(re.fullmatch(r"meshcore-[a-z0-9-]{8,100}", item["id"])), "owned_profile_name_invalid") + seen.add(item["uuid"]) + return config + + +def render_units(config): + validate(config) + stem = config["unit_stem"] + timer = f"""[Unit] +Description=Return to original WiFi until connected ({stem}) + +[Timer] +OnActiveSec={FIRST_DELAY_SECONDS}s +OnUnitInactiveSec={RETRY_DELAY_SECONDS}s +AccuracySec=1s +RandomizedDelaySec=0 +Unit={stem}.service +RemainAfterElapse=yes +""" + service = f"""[Unit] +Description=One bounded original WiFi recovery attempt ({stem}) +StartLimitIntervalSec=0 + +[Service] +Type=oneshot +RemainAfterExit=no +User=root +ExecStart={config['python_path']} {config['helper_path']} attempt --config {config['config_path']} --allow-live-host-recovery +TimeoutStartSec=35s +TimeoutStopSec=1s +KillMode=control-group +SendSIGKILL=yes +UMask=0077 +""" + return {stem + ".timer": timer, stem + ".service": service} + + +def parse_state(output): + lines = output.strip().splitlines() + if len(lines) != 2 or not valid_uuid(lines[0].strip()): + return None + match = re.fullmatch(r"(\d+)(?:\s+\([^\r\n]*\))?", lines[1].strip()) + if not match: + return None + return {"uuid": lines[0].strip(), "state": int(match[1])} + + +class LiveBackend: + """No instantiation from plan/render/tests; commands never return raw output.""" + def __init__(self, config): + require(config["live_execution_authorized"], "live_execution_not_authorized") + require(os.geteuid() == 0, "live_requires_root") + require(socket.gethostname() == config["expected_hostname"], "wrong_host") + machine_id = Path("/etc/machine-id").read_text().strip().encode() + require(hashlib.sha256(machine_id).hexdigest() == config["expected_machine_id_sha256"], "wrong_machine_id") + root = Path(config["guard_dir"]) + stat = root.stat() + require(root.is_dir() and not root.is_symlink() and stat.st_uid == 0 and not stat.st_mode & 0o022, "guard_directory_not_owned") + for key in ("config_path", "helper_path"): + item = Path(config[key]) + stat = item.stat() + require(item.is_file() and not item.is_symlink() and stat.st_uid == 0 and not stat.st_mode & 0o022, "live_file_not_owned") + self.interface = config["interface"] + + @staticmethod + def call(argv, timeout): + try: + value = subprocess.run(argv, capture_output=True, text=True, timeout=timeout, check=False, stdin=subprocess.DEVNULL) + return value.returncode, value.stdout + except subprocess.TimeoutExpired: + return None, "" + + def state(self, interface, timeout): + rc, output = self.call([NMCLI, "--terse", "--escape", "no", "--get-values", "GENERAL.CON-UUID,GENERAL.STATE", "device", "show", interface], timeout) + return parse_state(output) if rc == 0 else None + + def activate(self, original_uuid, timeout): + rc, _ = self.call([NMCLI, "--wait", "25", "connection", "up", "uuid", original_uuid, "ifname", self.interface], timeout) + return rc == 0 + + def stop_timer(self, unit, timeout): + rc, _ = self.call([SYSTEMCTL, "stop", unit], timeout) + return rc == 0 + + def timer_state(self, unit, timeout): + rc, output = self.call([SYSTEMCTL, "show", "--property=ActiveState", "--value", unit], timeout) + state = output.strip() + return state if rc == 0 and state in {"active", "inactive", "failed", "activating", "deactivating"} else None + + def profile_name(self, profile_uuid, timeout): + rc, output = self.call([NMCLI, "--get-values", "connection.id", "connection", "show", "uuid", profile_uuid], timeout) + return output.strip() if rc == 0 else None + + def delete_profile(self, profile_uuid, timeout): + rc, _ = self.call([NMCLI, "--wait", "5", "connection", "delete", "uuid", profile_uuid], timeout) + return rc == 0 + + def stop_owned_units(self, timer, service, timeout): + rc, _ = self.call([SYSTEMCTL, "stop", timer, service], timeout) + return rc == 0 + + +@contextlib.contextmanager +def owned_lock(path, expected_uid=None): + fd = os.open(path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600) + try: + stat = os.fstat(fd) + require(stat.st_nlink == 1 and (expected_uid is None or stat.st_uid == expected_uid), "lock_not_owned") + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + yield False + return + try: + yield True + finally: + fcntl.flock(fd, fcntl.LOCK_UN) + finally: + os.close(fd) + + +def attempt(config, backend, *, clock=time.monotonic): + """Single bounded invocation; no sleep, global retry cap, or profile edit.""" + validate(config) + begun = clock() + deadline = begun + ATTEMPT_BUDGET_SECONDS + report = {"schema": 1, "original_uuid_connected": False, "activation_attempted": False, + "timer_stopped": False, "recurring_retry_retained": True, + "attempt_budget_seconds": ATTEMPT_BUDGET_SECONDS, "errors": []} + + def bounded(limit): + left = deadline - clock() + if left <= 0: + raise TimeoutError("attempt_deadline") + return min(limit, left) + + def connected(state): + return bool(state and state["uuid"] == config["original_uuid"] and state["state"] == 100) + + try: + state = backend.state(config["interface"], bounded(3)) + if not connected(state): + report["activation_attempted"] = True + # Reserve3s for association and1s each for stop and state proof. + allowance = min(24, deadline - clock() - 5) + if allowance <= 0: + raise TimeoutError("no_activation_budget") + report["nmcli_activation_returned_success"] = backend.activate(config["original_uuid"], allowance) + state = backend.state(config["interface"], bounded(3)) + report["original_uuid_connected"] = connected(state) + if report["original_uuid_connected"]: + timer = config["unit_stem"] + ".timer" + report["recurring_retry_retained"] = None + report["timer_state_after_stop"] = None + report["timer_stop_returned_success"] = backend.stop_timer(timer, bounded(1)) + # Stop failure/timeout can occur after the side effect. Read the + # exact unit rather than infer timer activity from the return code. + report["timer_state_after_stop"] = backend.timer_state(timer, bounded(1)) + report["timer_stopped"] = report["timer_state_after_stop"] == "inactive" + report["recurring_retry_retained"] = ( + True if report["timer_state_after_stop"] == "active" else + False if report["timer_state_after_stop"] in {"inactive", "failed"} else None) + except Exception as error: + # Never store command output, SSIDs, credentials, or exception messages. + report["errors"].append(type(error).__name__) + report["elapsed_seconds"] = round(clock() - begun, 6) + return report + + +def explicit_owned_cleanup(config, backend, *, clock=time.monotonic, lock_context=None): + """Refuse cleanup while original WiFi is unproven; one bounded lock owner.""" + validate(config) + deadline = clock() + CLEANUP_BUDGET_SECONDS + stem = config["unit_stem"] + report = {"ok": False, "cleanup_refused": False, "no_mutation": True, + "original_profile_untouched": True, "owned_profiles": [], + "cleanup_budget_seconds": CLEANUP_BUDGET_SECONDS} + + def bounded(limit): + left = deadline - clock() + if left <= 0: + raise TimeoutError("cleanup_deadline") + return min(left, limit) + + def original_connected(): + state = backend.state(config["interface"], bounded(3)) + return bool(state and state["uuid"] == config["original_uuid"] and state["state"] == 100) + + def refuse(code): + report.update({"cleanup_refused": True, "reason": code}) + return report + + try: + if not original_connected(): + return refuse("original_wifi_not_connected_before_lock") + context = lock_context or (lambda path: owned_lock(path, expected_uid=0)) + with context(config["guard_dir"] + "/attempt.lock") as acquired: + if not acquired: + return refuse("cleanup_attempt_lock_busy") + if not original_connected(): + return refuse("original_wifi_not_connected_under_lock") + # Check all ownership before any mutation, avoiding partial cleanup + # for a known collision in a later owned-profile entry. + for item in config["owned_profiles"]: + if backend.profile_name(item["uuid"], bounded(3)) != item["id"]: + return refuse("owned_profile_identity_changed") + if not original_connected(): + return refuse("original_wifi_not_connected_before_unit_stop") + report["no_mutation"] = False + require(backend.stop_owned_units(stem + ".timer", stem + ".service", bounded(5)), "owned_units_stop_failed") + report["owned_units_stopped"] = True + for item in config["owned_profiles"]: + # Recheck capturedUUID/name immediately before deletion too. + require(backend.profile_name(item["uuid"], bounded(3)) == item["id"], "owned_profile_identity_changed") + require(backend.delete_profile(item["uuid"], bounded(7)), "owned_profile_delete_failed") + report["owned_profiles"].append({"owned_profile_deleted": True}) + report["ok"] = True + except Exception as error: + report["error_type"] = type(error).__name__ + if report["no_mutation"]: + report["cleanup_refused"] = True + return report + + +def plan(config): + validate(config) + return {"prepared_only": True, "not_staged_or_executed": True, + "first_delay_seconds": FIRST_DELAY_SECONDS, "retry_delay_seconds": RETRY_DELAY_SECONDS, + "accuracy_seconds": 1, "per_attempt_budget_seconds": ATTEMPT_BUDGET_SECONDS, + "persistent_failure_retains_recurring_timer": True, "requires_fresh_owned_unit_stem": True, + "stop_requires_original_uuid_and_connected_state": True, + "original_profile_settings_are_never_modified": True, + "cannot_retroactively_repair_offline_host": True, "unit_files": render_units(config)} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("mode", choices=("plan", "attempt", "cleanup"), nargs="?", default="plan") + parser.add_argument("--config", required=True) + parser.add_argument("--allow-live-host-recovery", action="store_true") + args = parser.parse_args() + config_path = Path(args.config) + config = validate(json.loads(config_path.read_text())) + if args.mode == "plan": + print(json.dumps(plan(config), indent=2)) + return + require(args.allow_live_host_recovery, "explicit_live_flag_required") + require(str(config_path) == config["config_path"], "live_config_path_mismatch") + require(str(Path(__file__)) == config["helper_path"], "live_helper_path_mismatch") + backend = LiveBackend(config) + if args.mode == "cleanup": + result = explicit_owned_cleanup(config, backend) + else: + with owned_lock(config["guard_dir"] + "/attempt.lock", expected_uid=0) as acquired: + if not acquired: + result = {"schema": 1, "overlapping_attempt_skipped": True, "recurring_retry_retained": True} + else: + result = attempt(config, backend) + destination = Path(config["guard_dir"]) / "attempt-last-safe.json" + temporary = destination.with_suffix(".json.tmp") + temporary.write_text(json.dumps(result, indent=2) + "\n") + os.replace(temporary, destination) + print(json.dumps(result)) + + +if __name__ == "__main__": + main()