Merge pull request #1863 from mauricio-camayo/fix/wificanary-scan-timeout

WIFICANARY: bound the iw scan so a stuck one can't outlive the run

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,36 @@ 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 (at most half of
+    a short RUN_TIMEOUT) before core's limit so this script still gets to
+    log it."""
+    run_timeout = int(run_timeout)
+    margin = min(SCAN_MARGIN_S, run_timeout // 2)
+    scan_timeout = max(1, run_timeout - margin)
+    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 + min(5, margin // 2))
     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 (halved for a short RUN_TIMEOUT), 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,75 @@ def test_escalation_with_no_detections_does_not_query():
     MockDeviceInstance.return_value.getAllByMacs.assert_called_once_with([])
 
 
+# ---------------------------------------------------------------------------
+# scan()
+# ---------------------------------------------------------------------------
+
+def _completed(returncode=0, stdout='', stderr=''):
+    """Fake subprocess.run() result with the given exit code and output."""
+    return MagicMock(returncode=returncode, stdout=stdout, stderr=stderr)
+
+
+def test_scan_bounds_iw_inside_sudo():
+    """iw runs under sudo timeout -s KILL, RUN_TIMEOUT minus the margin."""
+    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
+
+
+@pytest.mark.parametrize('run_timeout, scan_timeout, guard', [(20, 10, 15), (10, 5, 7), (2, 1, 1)])
+def test_scan_timeout_stays_below_short_run_timeout(run_timeout, scan_timeout, guard):
+    """A short RUN_TIMEOUT halves the margin, so the scan and its guard still end before core's limit."""
+    with patch.object(wificanary.subprocess, 'run', return_value=_completed()) as run:
+        wificanary.scan('wlan0', run_timeout)
+
+    assert run.call_args.args[0][4] == str(scan_timeout)
+    assert run.call_args.kwargs['timeout'] == guard
+    assert guard <= run_timeout
+
+
+@pytest.mark.parametrize('returncode', [124, 137, -9])
+def test_scan_timeout_exit_returns_no_aps(returncode):
+    """timeout(1) expiry codes are logged as a timeout and yield no APs."""
+    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():
+    """subprocess's own TimeoutExpired yields no APs instead of raising."""
+    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():
+    """A non-timeout iw failure yields 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():
+    """A successful scan's output is parsed into AP dicts."""
+    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()
 # ---------------------------------------------------------------------------