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 <noreply@anthropic.com>

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 <iface> 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 []

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