diff --git a/README.md b/README.md index 12f36e1..7add2b6 100644 --- a/README.md +++ b/README.md @@ -51,6 +51,13 @@ Then on the Pager menu, run **Mark VIII**: - Re-run the payload while running to **Stop** the service. - `PAYLOAD_GET_CONFIG pager_webui auto_mode/run_mode` skip the prompt. +Every payload run (and every service startup) first runs an **environment +check** that prints on the payload screen / `/tmp/pagerwebui.log`: daemon +reachable, pineapd alive, sane-off PineAP UCI defaults applied (SSID pool +broadcast off, 6GHz hopper disabled, monitor bands), runtime pool sync, monitor +interfaces up, and recon DB readable. Anything fixable is fixed and re-verified; +startup aborts only if a core dependency fails. + Browse `http://172.16.52.1:8080/` and log in with the device password. ## Uninstall / recovery diff --git a/payload/user/remote_access/pager-webui/_hak5_manifest.json b/payload/user/remote_access/pager-webui/_hak5_manifest.json index 90578e8..4dfdc5a 100644 --- a/payload/user/remote_access/pager-webui/_hak5_manifest.json +++ b/payload/user/remote_access/pager-webui/_hak5_manifest.json @@ -8,7 +8,7 @@ "title": "Mark VIII", "author": "c4ch3c4d3", "description": "Mark VII-style web management UI for the WiFi Pineapple Pager", - "version": "1.2", + "version": "1.3", "category": "remote_access", "tags": ["remote-access", "web-interface", "device-management", "pineap"], "firmware": "Pineapple Pager 24.10.1" diff --git a/payload/user/remote_access/pager-webui/payload.sh b/payload/user/remote_access/pager-webui/payload.sh index a904def..277be54 100755 --- a/payload/user/remote_access/pager-webui/payload.sh +++ b/payload/user/remote_access/pager-webui/payload.sh @@ -2,7 +2,7 @@ # Title: Mark VIII # Description: Mark VII-style web management UI for the WiFi Pineapple Pager # Author: c4ch3c4d3 -# Version: 1.1 +# Version: 1.3 # Category: Remote-Access # Tags: remote-access, web-interface, device-management, pineap # Firmware: Pineapple Pager 24.10.1 @@ -28,7 +28,7 @@ get_pager_ip() { } LOG "cyan" "+---------------------------+" -LOG "cyan" "| Mark VIII v1.1 |" +LOG "cyan" "| Mark VIII v1.3 |" LOG "cyan" "+---------------------------+" if ! command -v python3 >/dev/null 2>&1; then @@ -36,6 +36,16 @@ if ! command -v python3 >/dev/null 2>&1; then exit 1 fi +run_env_check() { + LOG "cyan" "Running environment check..." + if ! python3 "$SCRIPT_DIR/server.py" --env-check; then + LOG "red" "Environment check FAILED. Fix the issues above and re-run the payload." + sleep 3 + exit 1 + fi + LOG "green" "Environment check passed." +} + if [ -f "$INIT_SCRIPT" ] && "$INIT_SCRIPT" running 2>/dev/null; then PAGER_IP=$(get_pager_ip) LOG "green" "Mark VIII service is running" @@ -67,6 +77,7 @@ else fi if user_confirmed "$resp"; then + run_env_check LOG "cyan" "Starting as background service..." [ ! -f "$SCRIPT_DIR/server.py" ] && { LOG "red" "server.py not found!"; exit 1; } cp "$SCRIPT_DIR/pagerwebui.init" "$INIT_SCRIPT" @@ -82,6 +93,7 @@ if user_confirmed "$resp"; then exit 0 fi +run_env_check LOG "cyan" "Starting foreground mode..." cleanup() { LOG "yellow" "Stopping Mark VIII..." diff --git a/payload/user/remote_access/pager-webui/server.py b/payload/user/remote_access/pager-webui/server.py index feac7a0..6f5ec22 100644 --- a/payload/user/remote_access/pager-webui/server.py +++ b/payload/user/remote_access/pager-webui/server.py @@ -1255,7 +1255,13 @@ def h_recon_start(ctx): # appends a scan without discarding history. status, data = daemon_sock_call('POST', '/api/pineap/recon/new', body=body) if status != 200 or not (data or {}).get('success'): - return 502, {'error': 'native recon scan failed', 'detail': data} + detail = data if isinstance(data, dict) else {} + reason = detail.get('error') or detail.get('detail') + if not reason and status == 0: + reason = 'daemon unreachable' + return 502, {'error': 'native recon scan failed', + 'detail': ('recon/new: %s' % reason) if reason else None, + 'daemon': detail or None} _recon_scan_state['active'] = True _recon_scan_state['started'] = time.time() _recon_scan_state['duration'] = scan_time @@ -1352,6 +1358,7 @@ def h_recon_status(ctx): 'active': last_activity is not None and int(time.time()) - last_activity < 300, 'scanning': scanning, 'scan_remaining': remaining, 'stale': stale, 'hopper_online': _hopper_online(), + 'wlan0_pinned': _wlan0_pinned(), 'history_reset': _recon_history_reset()} @@ -3502,6 +3509,41 @@ def _sigsegv_count(): return out.count('SIGSEGV') +PINEAPD_SAFE_UCI = { + 'pineapd.@ssidpool[0].disable': '1', + 'pineapd.wlan2mon.disable': '1', + 'pineapd.wlan2mon.hop': '0', + 'pineapd.wlan1mon.bands': '5', + 'pineapd.wlan0mon.bands': '2', + 'pineapd.wlan1mon.hop': '0', +} + + +def _apply_uci_wanted(wanted): + """Idempotently apply a wanted UCI key/value set. Returns changed keys.""" + actions = [] + for key, value in wanted.items(): + rc, out, err = device_run(['uci', 'get', key]) + if rc != 0 or out.strip() != value: + device_run(['uci', 'set', '%s=%s' % (key, value)]) + actions.append(key.split('.')[-1]) + return actions + + +def _stabilize_uci(): + """Sane-off UCI pass shared by the health monitor and the startup env + check. Clears the refilled pool list too. Commits when changed and returns + the action labels.""" + actions = _apply_uci_wanted(PINEAPD_SAFE_UCI) + rc, out, err = device_run(['uci', 'get', 'pineapd.@ssidpool[0].ssid']) + if rc == 0 and out.strip(): + device_run(['uci', 'delete', 'pineapd.@ssidpool[0].ssid']) + actions.append('pool-list cleared') + if actions: + device_run(['uci', 'commit', 'pineapd']) + return actions + + def _stabilize_pineapd(): """Idempotent crash-source pass. Field-verified SIGSEGV/terminate sources on this firmware: @@ -3513,26 +3555,8 @@ def _stabilize_pineapd(): 4. a large refilled pool (collect refills it; crashes observed even with broadcast disabled) The pool list itself is cleared. Returns what changed.""" - actions = [] - wanted = { - 'pineapd.@ssidpool[0].disable': '1', - 'pineapd.wlan2mon.disable': '1', - 'pineapd.wlan2mon.hop': '0', - 'pineapd.wlan1mon.bands': '5', - 'pineapd.wlan0mon.bands': '2', - 'pineapd.wlan1mon.hop': '0', - } - for key, value in wanted.items(): - rc, out, err = device_run(['uci', 'get', key]) - if rc != 0 or out.strip() != value: - device_run(['uci', 'set', '%s=%s' % (key, value)]) - actions.append(key.split('.')[-1]) - rc, out, err = device_run(['uci', 'get', 'pineapd.@ssidpool[0].ssid']) - if rc == 0 and out.strip(): - device_run(['uci', 'delete', 'pineapd.@ssidpool[0].ssid']) - actions.append('pool-list cleared') + actions = _stabilize_uci() if actions: - device_run(['uci', 'commit', 'pineapd']) return 'stabilized: ' + ', '.join(actions) return 'pineapd restart' @@ -3589,10 +3613,18 @@ def health_check(): return dict(h) -def _bring_monitors_up(h): +def _raise_monitors(): + """Bring any down monitor interfaces up. Returns the interfaces raised.""" + raised = [] for name in ('wlan1mon', 'wlan0mon'): if _monitor_down(name): device_run(['ip', 'link', 'set', name, 'up'], timeout=10) + raised.append(name) + return raised + + +def _bring_monitors_up(h): + _raise_monitors() h['last_action'] = 'monitor interfaces brought up' h['monitor_fixes'] = h.get('monitor_fixes', 0) + 1 @@ -3603,6 +3635,15 @@ def h_health(ctx): h['pool_disabled'] = _uci_section('pineapd.@ssidpool[0]').get('disable') == '1' h['wlan1mon_up'] = not _monitor_down('wlan1mon') h['wlan0mon_up'] = not _monitor_down('wlan0mon') + h['pool_runtime'] = ENV_CHECK_STATE.get('pool_runtime') + if ENV_CHECK_STATE.get('report'): + h['env'] = { + 'overall': ENV_CHECK_STATE['overall'], + 'updated': ENV_CHECK_STATE['updated'], + 'counts': {k: _env_count(ENV_CHECK_STATE['report'], k) + for k in ('pass', 'fixed', 'warn', 'fail')}, + 'steps': ENV_CHECK_STATE['report'], + } return 200, h @@ -3619,6 +3660,126 @@ def start_health_monitor(): threading.Thread(target=_health_loop, daemon=True).start() +# -------------------------------------------------------------------------- +# Startup environment check: run once at service startup (and via +# ``server.py --env-check`` on the payload screen) to make the device match +# the UI before the user interacts with it. +# -------------------------------------------------------------------------- + +ENV_CHECK_STATE = {'report': None, 'overall': None, 'updated': 0, 'pool_runtime': None} + + +def _env_step(report, ok, detail, action=None): + step = {'ok': ok, 'detail': detail} + if action: + step['action'] = action + report.append(step) + + +def _env_overall(report): + if any(r['ok'] == 'fail' for r in report): + return 'fail' + if any(r['ok'] == 'fixed' for r in report): + return 'fixed' + if any(r['ok'] == 'warn' for r in report): + return 'warn' + return 'pass' + + +def _env_count(report, kind): + return sum(1 for r in report if r['ok'] == kind) + + +def _pineapd_alive(): + rc, out, err = device_run(['pidof', 'pineapd'], timeout=10) + return rc == 0 and bool((out or '').strip()) + + +def _sync_pool_runtime(): + """Force pineapd's runtime SSID-pool broadcast off so it matches the + sane-off UCI default the UI derives from. Returns (state, detail).""" + rc, out, err = _pineap('SSIDPOOL', 'DISABLE', timeout=15) + if rc != 0: + detail = (err or out or '').strip() or 'no output' + return 'unknown', 'SSIDPOOL DISABLE failed: %s' % detail[-200:] + return 'disabled', 'SSIDPOOL DISABLE sent' + + +def _wlan0_pinned(): + """True when a radio0 AP (OpenAP/Evil WPA) is enabled. A phy's channel is + held by its AP interface, so an enabled radio0 AP pins wlan0mon to one + 2.4GHz channel and 2.4GHz recon results are under-sampled.""" + for name in ('wlan0open', 'wlan0wpa'): + cfg = _uci_wifi_iface(name) or {} + if cfg and cfg.get('disabled') != '1': + return True + return False + + +def env_check(): + """Full environment pass, run at service startup and via + ``server.py --env-check``. Auto-fixes everything fixable, re-verifies, and + returns a list of step reports (ok: pass|fixed|warn|fail). Only core + dependencies (daemon, pineapd, recon DB) can fail.""" + report = [] + + status, data = daemon_sock_call('GET', '/api/pineap/get_config') + if status != 200: + _env_step(report, 'fail', 'daemon unreachable (status %s)' % status) + else: + _env_step(report, 'pass', 'daemon reachable') + + if _pineapd_alive(): + _env_step(report, 'pass', 'pineapd running') + else: + actions = _stabilize_uci() + device_run(['/etc/init.d/pineapd', 'restart'], timeout=30) + if _pineapd_alive(): + _env_step(report, 'fixed', 'pineapd was down; stabilized and restarted', + ', '.join(actions) or 'restart') + else: + _env_step(report, 'fail', 'pineapd did not come back after restart', + ', '.join(actions) or 'restart') + + actions = _stabilize_uci() + if actions: + _env_step(report, 'fixed', 'pineapd sane-off UCI defaults applied', + ', '.join(actions)) + else: + _env_step(report, 'pass', 'pineapd sane-off UCI defaults already set') + + pool_runtime, pool_detail = _sync_pool_runtime() + ENV_CHECK_STATE['pool_runtime'] = pool_runtime + if pool_runtime == 'disabled': + _env_step(report, 'pass', 'SSID-pool broadcast off (runtime)') + else: + _env_step(report, 'warn', pool_detail) + + raised = _raise_monitors() + if raised: + _env_step(report, 'fixed', 'monitor interfaces brought up', ', '.join(raised)) + else: + _env_step(report, 'pass', 'monitors up (wlan0mon, wlan1mon)') + + try: + rows = _db_rows(RECON_DB, 'SELECT count(*) AS c FROM scan', timeout=15) + count = rows[0]['c'] if rows else 0 + _env_step(report, 'pass', 'recon DB readable (%d scans)' % count) + except RuntimeError as exc: + _env_step(report, 'fail', 'recon DB unreadable: %s' % exc) + + if _wlan0_pinned(): + _env_step(report, 'warn', '2.4GHz under-sampled: a radio0 AP is up and pins ' + 'wlan0mon to one channel during scans') + else: + _env_step(report, 'pass', 'no radio0 AP pins wlan0mon') + + ENV_CHECK_STATE['report'] = report + ENV_CHECK_STATE['overall'] = _env_overall(report) + ENV_CHECK_STATE['updated'] = time.time() + return report + + def h_attacks_clients(ctx): """Clients (recon devices) plus the APs matching an SSID, for targeting.""" ssid = ((ctx.query or {}).get('ssid') or '').strip() @@ -5164,6 +5325,10 @@ def _recon_watchdog_loop(): def serve(): + try: + env_check() + except Exception: + pass threading.Thread(target=live_loop, daemon=True).start() threading.Thread(target=_recon_watchdog_loop, daemon=True).start() start_health_monitor() @@ -5177,5 +5342,23 @@ def serve(): threading.Thread(target=_handle_conn, args=(conn, addr), daemon=True).start() +def env_check_cli(): + """Run the environment check and print a verbose report to stdout. + Returns the process exit code (0 = pass/warn, 1 = core failure).""" + report = env_check() + for step in report: + line = '[%s] %s' % (step['ok'].upper(), step['detail']) + if step.get('action'): + line += ' (%s)' % step['action'] + print(line) + counts = {k: _env_count(report, k) for k in ('pass', 'fixed', 'warn', 'fail')} + print('ENVIRONMENT CHECK: %s (%d pass, %d fixed, %d warn, %d fail)' % ( + ENV_CHECK_STATE['overall'].upper(), counts['pass'], counts['fixed'], + counts['warn'], counts['fail'])) + return 0 if ENV_CHECK_STATE['overall'] != 'fail' else 1 + + if __name__ == '__main__': + if '--env-check' in sys.argv: + sys.exit(env_check_cli()) serve() diff --git a/payload/user/remote_access/pager-webui/www/js/app.js b/payload/user/remote_access/pager-webui/www/js/app.js index 56e5089..a358b8c 100644 --- a/payload/user/remote_access/pager-webui/www/js/app.js +++ b/payload/user/remote_access/pager-webui/www/js/app.js @@ -516,6 +516,9 @@ const Live = (() => { if (h.pineap_up === false) { el.textContent = 'PINEAPD DOWN'; el.className = 'health-chip bad'; + } else if (h.env && h.env.overall === 'fail') { + el.textContent = 'ENV CHECK FAIL'; + el.className = 'health-chip bad'; } else if (h.pool_disabled) { el.textContent = 'POOL OFF'; el.className = 'health-chip warn'; diff --git a/payload/user/remote_access/pager-webui/www/js/views.js b/payload/user/remote_access/pager-webui/www/js/views.js index 46eadfd..00d7422 100644 --- a/payload/user/remote_access/pager-webui/www/js/views.js +++ b/payload/user/remote_access/pager-webui/www/js/views.js @@ -177,7 +177,8 @@ views.dashboard = (root) => { liveCards.health.textContent = (h2.pineap_up ? 'pineapd up' : 'pineapd DOWN') + ' · wlan0mon ' + (h2.wlan0mon_up ? 'up' : 'down') + ' · wlan1mon ' + (h2.wlan1mon_up ? 'up' : 'down') + - (h2.pool_disabled ? ' · pool off' : ''); + (h2.pool_disabled ? ' · pool off' : '') + + ((h2.env || {}).overall ? ' · env ' + h2.env.overall : ''); liveCards.health.style.color = h2.pineap_up ? '' : '#b71c1c'; }).catch(() => {}); PagerAPI.get('/api/recon/status').then((r) => { @@ -1460,7 +1461,8 @@ views.recon = (root) => { apBand: 'all', apEnc: 'all', gps: null, wigle: null, compare: [], history: {}, mapBand: null, archive: null, archives: [], scanRemaining: null, - hopperOnline: null, historyReset: false, scanErr: null }; + hopperOnline: null, historyReset: false, scanErr: null, + wlan0Pinned: false }; const cols = reconLoadCols(); // ---- title cards (stat cards with optional mini charts) ---- @@ -1658,9 +1660,10 @@ views.recon = (root) => { } if (state.hopperOnline === false) bits.push('Hopper radio offline — fewer networks seen'); if (state.historyReset) bits.push('History reset — previous scans archived (see Previous Scans)'); + if (state.wlan0Pinned) bits.push('2.4GHz under-sampled — OpenAP/Evil WPA holds wlan0mon'); if (state.scanErr) bits.push(state.scanErr); scanStatus.textContent = bits.join(' · '); - scanStatus.classList.toggle('warn', state.hopperOnline === false || state.historyReset || !!state.scanErr); + scanStatus.classList.toggle('warn', state.hopperOnline === false || state.historyReset || state.wlan0Pinned || !!state.scanErr); } scanToggle.addEventListener('change', () => { if (pendingScan) { scanToggle.checked = !scanToggle.checked; return; } @@ -2522,6 +2525,7 @@ views.recon = (root) => { state.scanRemaining = r.data.scan_remaining != null ? r.data.scan_remaining : null; state.hopperOnline = r.data.hopper_online; state.historyReset = !!r.data.history_reset; + state.wlan0Pinned = !!r.data.wlan0_pinned; if (!pendingScan) scanToggle.checked = scanning; renderScanBar(); if (wasScanning !== scanning) restartPoll(); diff --git a/tests/test_env_check.py b/tests/test_env_check.py new file mode 100644 index 0000000..9ac8b97 --- /dev/null +++ b/tests/test_env_check.py @@ -0,0 +1,230 @@ +import importlib +import io +import os +import sqlite3 +import sys +import tempfile +import unittest +from contextlib import redirect_stdout + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'payload', 'user', 'remote_access', 'pager-webui')) +import server + + +def setUpModule(): + importlib.reload(server) + + +class EnvCheckTest(unittest.TestCase): + def setUp(self): + fd, self.db = tempfile.mkstemp(suffix='.db') + os.close(fd) + conn = sqlite3.connect(self.db) + conn.execute('CREATE TABLE scan(id INTEGER PRIMARY KEY, time INT, name TEXT)') + conn.execute("INSERT INTO scan (time, name) VALUES (1786466531, 'pager')") + conn.commit() + conn.close() + server.RECON_DB = self.db + self.runs = [] + self.ping_ok = True + self.daemon_ok = True + self.iface_up = {'wlan0mon': True, 'wlan1mon': True} + self.uci_state = {} + server.ENV_CHECK_STATE.update({'report': None, 'overall': None, 'updated': 0, + 'pool_runtime': None}) + self.old_iface_up = server._iface_up + server._iface_up = lambda name: self.iface_up.get(name, True) + self.old_daemon = server.daemon_sock_call + self.old_run = server.device_run + server.daemon_sock_call = self.fake_daemon + server.device_run = self.fake_run + + def tearDown(self): + server._iface_up = self.old_iface_up + server.daemon_sock_call = self.old_daemon + server.device_run = self.old_run + try: + os.unlink(self.db) + except OSError: + pass + + def fake_daemon(self, method, path, body=None, timeout=10): + if self.daemon_ok: + return 200, {'autossidpool': False} + return 0, None + + def fake_run(self, args, timeout=20, input_data=None): + self.runs.append(list(args)) + a = list(args) + if a[0] == 'pidof' and a[1] == 'pineapd': + return (0, '12345\n', '') if self.ping_ok else (1, '', '') + if a[0] == 'uci': + if a[1] == 'set': + k, _, v = a[2].partition('=') + self.uci_state[k] = v + return (0, '', '') + if a[1] == 'delete': + for k in list(self.uci_state): + if k == a[2] or k.startswith(a[2] + '.'): + del self.uci_state[k] + return (0, '', '') + if a[1] == 'get': + return (0, self.uci_state.get(a[2], '') + '\n', '') + if a[1] == 'commit': + return (0, '', '') + if a[1] == 'show': + sec = a[2] + return (0, ''.join("%s=%s\n" % (k, v) for k, v in self.uci_state.items() + if k.startswith(sec + '.')), '') + if a[0] == '_pineap': + return (0, '', '') + if a[0] in ('ip', '/etc/init.d/pineapd'): + return (0, '', '') + return (0, '', '') + + def safe_set(self): + for key, value in server.PINEAPD_SAFE_UCI.items(): + self.uci_state[key] = value + + def steps(self, report, needle): + return [r for r in report if needle in r['detail']] + + def test_pass_when_state_sane(self): + self.safe_set() + report = server.env_check() + self.assertEqual(server.ENV_CHECK_STATE['overall'], 'pass') + self.assertEqual([r['ok'] for r in report], + ['pass'] * len(report)) + self.assertEqual(server.ENV_CHECK_STATE['pool_runtime'], 'disabled') + self.assertIn(['_pineap', 'SSIDPOOL', 'DISABLE'], self.runs) + + def test_applies_sane_defaults_when_missing(self): + report = server.env_check() + self.assertEqual(self.steps(report, 'sane-off UCI defaults applied')[0]['ok'], 'fixed') + for key, value in server.PINEAPD_SAFE_UCI.items(): + self.assertEqual(self.uci_state[key], value) + + def test_uci_pass_when_already_set(self): + self.safe_set() + report = server.env_check() + self.assertEqual(self.steps(report, 'sane-off UCI defaults already set')[0]['ok'], 'pass') + + def test_clears_refilled_pool_list(self): + self.uci_state['pineapd.@ssidpool[0].ssid'] = 'QmVlcg==' + report = server.env_check() + actions = ' | '.join((r.get('action') or '') for r in report) + self.assertIn('pool-list cleared', actions) + self.assertNotIn('pineapd.@ssidpool[0].ssid', self.uci_state) + + def test_restarts_pineapd_when_down(self): + self.safe_set() + self.ping_ok = False + self.pidof_calls = 0 + real_ping = self.fake_run + + def ping_then_up(args, timeout=20, input_data=None): + if args[0] == 'pidof' and args[1] == 'pineapd': + self.pidof_calls += 1 + if self.pidof_calls > 1: + return (0, '12345\n', '') + return real_ping(args, timeout=timeout, input_data=input_data) + + server.device_run = ping_then_up + report = server.env_check() + self.assertEqual(self.steps(report, 'pineapd was down')[0]['ok'], 'fixed') + self.assertIn(['/etc/init.d/pineapd', 'restart'], self.runs) + + def test_fail_when_pineapd_stays_down(self): + self.safe_set() + self.ping_ok = False + report = server.env_check() + self.assertEqual(self.steps(report, 'did not come back')[0]['ok'], 'fail') + self.assertEqual(server.ENV_CHECK_STATE['overall'], 'fail') + + def test_fail_when_daemon_unreachable(self): + self.daemon_ok = False + report = server.env_check() + self.assertEqual(self.steps(report, 'daemon unreachable')[0]['ok'], 'fail') + self.assertEqual(server.ENV_CHECK_STATE['overall'], 'fail') + + def test_raises_down_monitors(self): + self.safe_set() + self.iface_up = {'wlan0mon': False, 'wlan1mon': True} + report = server.env_check() + self.assertEqual(self.steps(report, 'monitor interfaces brought up')[0]['ok'], 'fixed') + self.assertIn(['ip', 'link', 'set', 'wlan0mon', 'up'], self.runs) + + def test_monitors_up_pass(self): + self.safe_set() + report = server.env_check() + self.assertEqual(self.steps(report, 'monitors up')[0]['ok'], 'pass') + + def test_recon_db_unreadable_fails(self): + self.safe_set() + os.unlink(self.db) + report = server.env_check() + self.assertEqual(self.steps(report, 'recon DB unreadable')[0]['ok'], 'fail') + self.assertEqual(server.ENV_CHECK_STATE['overall'], 'fail') + + def test_recon_db_readable_reports_count(self): + self.safe_set() + report = server.env_check() + step = self.steps(report, 'recon DB readable')[0] + self.assertEqual(step['ok'], 'pass') + self.assertIn('(1 scans)', step['detail']) + + def test_wlan0_pinned_warns(self): + self.safe_set() + self.uci_state['wireless.wlan0wpa.disabled'] = '0' + report = server.env_check() + self.assertEqual(self.steps(report, '2.4GHz under-sampled')[0]['ok'], 'warn') + + def test_wlan0_not_pinned_when_absent(self): + self.safe_set() + report = server.env_check() + self.assertEqual(self.steps(report, 'no radio0 AP pins wlan0mon')[0]['ok'], 'pass') + + def test_wlan0_not_pinned_when_disabled(self): + self.safe_set() + self.uci_state['wireless.wlan0open.disabled'] = '1' + self.uci_state['wireless.wlan0wpa.disabled'] = '1' + report = server.env_check() + self.assertEqual(self.steps(report, 'no radio0 AP pins wlan0mon')[0]['ok'], 'pass') + + def test_cli_exits_zero_on_pass(self): + self.safe_set() + buf = io.StringIO() + with redirect_stdout(buf): + code = server.env_check_cli() + self.assertEqual(code, 0) + self.assertIn('[PASS]', buf.getvalue()) + self.assertIn('ENVIRONMENT CHECK: PASS', buf.getvalue()) + + def test_cli_exits_one_on_fail(self): + self.safe_set() + self.daemon_ok = False + buf = io.StringIO() + with redirect_stdout(buf): + code = server.env_check_cli() + self.assertEqual(code, 1) + self.assertIn('[FAIL]', buf.getvalue()) + + def test_health_exposes_env_and_pool_runtime(self): + self.safe_set() + server.env_check() + status, payload = server.h_health(type('C', (), {'query': {}})()) + self.assertEqual(status, 200) + self.assertEqual(payload['pool_runtime'], 'disabled') + self.assertEqual(payload['env']['overall'], 'pass') + self.assertEqual(payload['env']['counts']['pass'], len(payload['env']['steps'])) + + def test_recon_status_exposes_wlan0_pinned(self): + self.safe_set() + self.uci_state['wireless.wlan0open.disabled'] = '0' + status, payload = server.h_recon_status(type('C', (), {'query': {}})()) + self.assertEqual(status, 200) + self.assertTrue(payload['wlan0_pinned']) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_recon.py b/tests/test_recon.py index c7c7a40..cdcad9c 100644 --- a/tests/test_recon.py +++ b/tests/test_recon.py @@ -224,7 +224,8 @@ class DaemonSockTest(unittest.TestCase): type('C', (), {'args': (), 'body': {'scan_time': 30}})()) self.assertEqual(status, 502) self.assertEqual(data['error'], 'native recon scan failed') - self.assertEqual(data['detail'], {'error': 'no radio'}) + self.assertEqual(data['detail'], 'recon/new: no radio') + self.assertEqual(data['daemon'], {'error': 'no radio'}) self.assertFalse(server._recon_scan_state['active'])