From 8cee6b51bf656b0f46aa5e4aafad98f9508eae37 Mon Sep 17 00:00:00 2001 From: Mauricio Camayo Date: Mon, 26 Oct 2026 17:49:46 -0500 Subject: [PATCH] WIFICANARY: bound the iw scan so a stuck one can't outlive the run Core kills the plugin script when RUN_TIMEOUT is reached, but not its children. A scan stuck on a busy or wedged radio (seen on a Pi Zero W right after boot) left an orphaned `sudo iw dev wlan0 scan` spinning at ~70% CPU for as long as nobody killed it, slowing the whole host down. The scan used the same timeout as core (RUN_TIMEOUT), so core always won and the plugin's own TimeoutExpired handling never ran. Now `iw` runs under `timeout -s KILL` inside sudo, RUN_TIMEOUT - 15s (min 10s), so it is bounded even if the script itself gets killed, and the plugin logs the timeout before core steps in. Works with the Docker image's sudo shim (busybox timeout) and with real sudo (coreutils timeout). Co-Authored-By: Claude Opus 5.5 --- server/plugins/wificanary/script.py | 28 ++++++++++--- test/plugins/test_wificanary.py | 63 +++++++++++++++++++++++++++++ 2 files changed, 85 insertions(+), 6 deletions(-) diff --git a/server/plugins/wificanary/script.py b/server/plugins/wificanary/script.py index a24587472..35e9e31e6 100644 --- a/server/plugins/wificanary/script.py +++ b/server/plugins/wificanary/script.py @@ -50,6 +50,9 @@ PINEAPPLE_OUI_MID = ('13', '37') # BSSID octets [1:3] == 13:37 # Weakest-to-strongest, used to detect a downgrade. SECURITY_RANK = {'open': 0, 'wep': 1, 'wpa': 2, 'wpa2': 3, 'wpa3': 4} +# Seconds the scan leaves before core's RUN_TIMEOUT kills this script. +SCAN_MARGIN_S = 15 + def main(): """Scan once, compare against the configured trusted-AP baseline, and @@ -137,19 +140,32 @@ def parse_security_set(raw_value): return security_set or {'wpa2'} -def scan(iface, timeout): +def scan(iface, run_timeout): """Run `iw dev scan` and parse the output into a list of AP dicts - (bssid/ssid/security/signal/oui).""" - cmd = ['sudo', 'iw', 'dev', iface, 'scan'] + (bssid/ssid/security/signal/oui). + + Core kills this script once RUN_TIMEOUT is reached, but not its children: + a scan stuck on a busy or wedged radio used to leave an orphaned + `sudo iw` spinning at full CPU long after the run was gone. `timeout` + inside sudo bounds `iw` itself, finishing SCAN_MARGIN_S before core's + limit so this script still gets to log it.""" + scan_timeout = max(10, int(run_timeout) - SCAN_MARGIN_S) + cmd = ['sudo', 'timeout', '-s', 'KILL', str(scan_timeout), 'iw', 'dev', iface, 'scan'] try: - result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) + result = subprocess.run(cmd, capture_output=True, text=True, timeout=scan_timeout + 5) except subprocess.TimeoutExpired: - mylog('none', [f'[{pluginName}] scan on {iface} timed out after {timeout}s']) + mylog('none', [f'[{pluginName}] scan on {iface} timed out after {scan_timeout}s']) return [] except FileNotFoundError: - mylog('none', [f'[{pluginName}] `iw` not found - is it installed on this host?']) + mylog('none', [f'[{pluginName}] `sudo` not found - is it installed on this host?']) return [] + # timeout(1) exits 124 on expiry, or 128+9 when the signal is KILL; some + # coreutils builds re-raise KILL on themselves instead, so sudo dies of it + # too and subprocess reports -9 + if result.returncode in (124, 137, -9): + mylog('none', [f'[{pluginName}] scan on {iface} timed out after {scan_timeout}s']) + return [] if result.returncode != 0: mylog('none', [f'[{pluginName}] scan on {iface} failed: {result.stderr.strip()}']) return [] diff --git a/test/plugins/test_wificanary.py b/test/plugins/test_wificanary.py index 2f26218a3..c2eaef0b6 100644 --- a/test/plugins/test_wificanary.py +++ b/test/plugins/test_wificanary.py @@ -37,6 +37,10 @@ Layout: the only existing record is one WIFICANARY itself created on a prior run (self-escalation guard), and the motor/reason rewrite when it's a real pre-existing device from another source plugin. + - scan(): `iw` runs under `sudo timeout -s KILL`, RUN_TIMEOUT minus a + margin (10 s floor), so a stuck scan can't outlive a killed run; a + timeout (exit 124/137/-9, or the subprocess's own) and a failed scan both + yield no APs instead of raising. - main(): integration test with scan() mocked - covers the no-IFACE early-return and a run that finds one anomaly end to end. """ @@ -696,6 +700,65 @@ def test_escalation_with_no_detections_does_not_query(): MockDeviceInstance.return_value.getAllByMacs.assert_called_once_with([]) +# --------------------------------------------------------------------------- +# scan() +# --------------------------------------------------------------------------- + +def _completed(returncode=0, stdout='', stderr=''): + return MagicMock(returncode=returncode, stdout=stdout, stderr=stderr) + + +def test_scan_bounds_iw_inside_sudo(): + with patch.object(wificanary.subprocess, 'run', return_value=_completed()) as run: + wificanary.scan('wlan0', 60) + + cmd = run.call_args.args[0] + assert cmd == ['sudo', 'timeout', '-s', 'KILL', '45', 'iw', 'dev', 'wlan0', 'scan'] + # The subprocess's own limit stays above timeout(1)'s, so timeout(1) fires first + assert run.call_args.kwargs['timeout'] == 50 + + +def test_scan_timeout_has_a_floor(): + with patch.object(wificanary.subprocess, 'run', return_value=_completed()) as run: + wificanary.scan('wlan0', 12) + + assert run.call_args.args[0][4] == '10' + + +@pytest.mark.parametrize('returncode', [124, 137, -9]) +def test_scan_timeout_exit_returns_no_aps(returncode): + with patch.object(wificanary.subprocess, 'run', return_value=_completed(returncode)): + with patch.object(wificanary, 'mylog') as log: + assert wificanary.scan('wlan0', 60) == [] + + assert 'timed out after 45s' in str(log.call_args) + + +def test_scan_subprocess_timeout_returns_no_aps(): + expired = wificanary.subprocess.TimeoutExpired(cmd='iw', timeout=50) + with patch.object(wificanary.subprocess, 'run', side_effect=expired): + assert wificanary.scan('wlan0', 60) == [] + + +def test_scan_failure_returns_no_aps(): + failed = _completed(1, stderr='command failed: Device or resource busy (-16)') + with patch.object(wificanary.subprocess, 'run', return_value=failed): + assert wificanary.scan('wlan0', 60) == [] + + +def test_scan_success_is_parsed(): + output = ( + 'BSS aa:bb:cc:11:22:33(on wlan0)\n' + '\tsignal: -48.00 dBm\n' + '\tSSID: HomeWiFi\n' + '\tRSN:\t * Version: 1\n' + ) + with patch.object(wificanary.subprocess, 'run', return_value=_completed(0, output)): + aps = wificanary.scan('wlan0', 60) + + assert [(ap['bssid'], ap['ssid'], ap['security']) for ap in aps] == [('aa:bb:cc:11:22:33', 'HomeWiFi', 'wpa2')] + + # --------------------------------------------------------------------------- # main() # ---------------------------------------------------------------------------