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() # ---------------------------------------------------------------------------