mirror of
https://github.com/jokob-sk/NetAlertX.git
synced 2026-11-06 23:43:07 -05:00
WIFICANARY: keep the scan inside a short RUN_TIMEOUT
The 10s floor gave `iw` the whole deadline when RUN_TIMEOUT is 10s, so a stuck scan couldn't be logged before core killed the script. The margin is now min(15s, RUN_TIMEOUT // 2) and the subprocess guard stays within it, so both end before core's limit. Also adds docstrings to the new scan() tests. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
8cee6b51bf
commit
ad73316488
2 files changed
+24
-10
No files matched your search
@@ -147,12 +147,16 @@ def scan(iface, run_timeout):
|
||||
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)
|
||||
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=scan_timeout + 5)
|
||||
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 {scan_timeout}s'])
|
||||
return []
|
||||
|
||||
@@ -38,9 +38,9 @@ Layout:
|
||||
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.
|
||||
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.
|
||||
"""
|
||||
@@ -705,10 +705,12 @@ def test_escalation_with_no_detections_does_not_query():
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
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)
|
||||
|
||||
@@ -718,15 +720,20 @@ def test_scan_bounds_iw_inside_sudo():
|
||||
assert run.call_args.kwargs['timeout'] == 50
|
||||
|
||||
|
||||
def test_scan_timeout_has_a_floor():
|
||||
@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', 12)
|
||||
wificanary.scan('wlan0', run_timeout)
|
||||
|
||||
assert run.call_args.args[0][4] == '10'
|
||||
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) == []
|
||||
@@ -735,18 +742,21 @@ def test_scan_timeout_exit_returns_no_aps(returncode):
|
||||
|
||||
|
||||
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'
|
||||
|
||||
Reference in new issue
Block a user