diff --git a/payload/user/remote_access/pager-webui/server.py b/payload/user/remote_access/pager-webui/server.py index 32b6db8..a9e294e 100644 --- a/payload/user/remote_access/pager-webui/server.py +++ b/payload/user/remote_access/pager-webui/server.py @@ -3163,6 +3163,86 @@ def _pineap(*args, timeout=30): return rc, out, err +# -------------------------------------------------------------------------- +# Health monitor: keep pineapd alive and the monitor radios up. The stock +# SSID-pool broadcast segfaults pineapd on this firmware; when a crash-loop +# is detected the pool broadcast is disabled and pineapd restarted. +# -------------------------------------------------------------------------- + +HEALTH_POLL_SECONDS = 15 +HEALTH_FIX_COOLDOWN = 20.0 +HEALTH_STOP = threading.Event() +_health = { + 'sigsegv_last': None, + 'last_fix': 0.0, + 'fixes': 0, + 'last_action': None, + 'pineap_up': False, +} + + +def _sigsegv_count(): + rc, out, err = device_run(['logread'], timeout=15) + return out.count('SIGSEGV') + + +def _monitor_down(name): + rc, out, err = device_run(['iw', 'dev'], timeout=10) + return name not in out + + +def health_check(): + """One health pass. Returns the health dict. Fix actions are + rate-limited by HEALTH_FIX_COOLDOWN.""" + h = _health + rc, out, err = device_run([HAK5CMD, 'PING'], timeout=10) + h['pineap_up'] = rc == 0 and 'PONG' in (out or '') + if h['pineap_up']: + return dict(h) + now = time.time() + if now - h['last_fix'] < HEALTH_FIX_COOLDOWN: + return dict(h) + count = _sigsegv_count() + if h['sigsegv_last'] is not None and count > h['sigsegv_last']: + # Crash-loop signature: disable the SSID pool broadcast and restart. + device_run(['uci', 'set', 'pineapd.@ssidpool[0].disable=1']) + device_run(['uci', 'commit', 'pineapd']) + device_run(['/etc/init.d/pineapd', 'restart'], timeout=30) + h['last_action'] = 'pool-disabled + pineapd restart (SIGSEGV crash-loop)' + elif _monitor_down('wlan1mon'): + device_run(['ip', 'link', 'set', 'wlan1mon', 'up'], timeout=10) + h['last_action'] = 'wlan1mon brought up' + else: + device_run(['/etc/init.d/pineapd', 'restart'], timeout=30) + h['last_action'] = 'pineapd restart' + h['sigsegv_last'] = count + h['last_fix'] = now + h['fixes'] += 1 + return dict(h) + + +def h_health(ctx): + h = dict(_health) + h['sigsegv_count'] = h.pop('sigsegv_last') + 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') + return 200, h + + +def _health_loop(): + while not HEALTH_STOP.is_set(): + try: + health_check() + except Exception: + pass + HEALTH_STOP.wait(HEALTH_POLL_SECONDS) + + +def start_health_monitor(): + threading.Thread(target=_health_loop, daemon=True).start() + + def h_attacks_clients(ctx): """Clients (recon devices) plus the APs matching an SSID, for targeting.""" ssid = ((ctx.query or {}).get('ssid') or '').strip() @@ -4199,6 +4279,7 @@ ROUTER.add('POST', r'/api/attacks/capture', h_attacks_capture) ROUTER.add('GET', r'/api/attacks/export/hc22000', h_attacks_export_hc22000) ROUTER.add('POST', r'/api/attacks/deauth', h_attacks_deauth) ROUTER.add('GET', r'/api/attacks/clients', h_attacks_clients) +ROUTER.add('GET', r'/api/health', h_health) ROUTER.add('POST', r'/api/recon/start', h_recon_start) ROUTER.add('POST', r'/api/recon/stop', h_recon_stop) ROUTER.add('GET', r'/api/recon/status', h_recon_status) @@ -4324,6 +4405,7 @@ def _recon_watchdog_loop(): def serve(): threading.Thread(target=live_loop, daemon=True).start() threading.Thread(target=_recon_watchdog_loop, daemon=True).start() + start_health_monitor() sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind((HOST, PORT)) diff --git a/tests/test_health.py b/tests/test_health.py new file mode 100644 index 0000000..79eceac --- /dev/null +++ b/tests/test_health.py @@ -0,0 +1,91 @@ +import os +import sys +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'payload', 'user', 'remote_access', 'pager-webui')) +import server + + +def setUpModule(): + __import__('importlib').reload(server) + + +class HealthCheckTest(unittest.TestCase): + def setUp(self): + self.runs = [] + self.ping_ok = True + self.sigsegvs = 0 + self.iw_out = 'wlan0mon\nwlan1mon\n' + self.uci_state = {} + server._health.update({ + 'sigsegv_last': None, 'last_fix': 0.0, 'fixes': 0, + 'last_action': None, 'pineap_up': False}) + + def fake_run(args, timeout=20, input_data=None): + self.runs.append((list(args), timeout)) + a = list(args) + if a[0] == '/usr/bin/hak5cmd' and a[1] == 'PING': + if self.ping_ok: + return (0, 'Sending PING...\nGot PONG response\n', '') + return (1, '', 'could not connect to pineap: connection refused') + if a[0] == 'logread': + return (0, 'SIGSEGV\n' * self.sigsegs if hasattr(self, 'sigsegs') else '', '') + if a[0] == 'iw': + return (0, self.iw_out, '') + if a[:2] == ['uci', 'set']: + k, _, v = a[2].partition('=') + self.uci_state[k] = v + if a[:2] == ['uci', 'get']: + return (0, self.uci_state.get(a[2], '') + '\n', '') + return (0, '', '') + + server.device_run = fake_run + + def test_pineap_up_reports_no_action(self): + result = server.health_check() + self.assertTrue(result['pineap_up']) + self.assertIsNone(result['last_action']) + + def test_down_with_growing_sigsegv_disables_pool(self): + self.ping_ok = False + self.sigsegs = 5 + server._health['sigsegv_last'] = 3 + result = server.health_check() + self.assertIn('pool-disabled', result['last_action']) + self.assertEqual(self.uci_state['pineapd.@ssidpool[0].disable'], '1') + self.assertIn(['/etc/init.d/pineapd', 'restart'], [r[0] for r in self.runs]) + self.assertEqual(result['fixes'], 1) + + def test_down_without_crash_brings_monitor_up(self): + self.ping_ok = False + self.iw_out = 'wlan0mon\n' + result = server.health_check() + self.assertEqual(result['last_action'], 'wlan1mon brought up') + self.assertIn(['ip', 'link', 'set', 'wlan1mon', 'up'], [r[0] for r in self.runs]) + + def test_down_with_no_sigsegv_growth_restarts_pineapd(self): + self.ping_ok = False + server._health['sigsegv_last'] = 4 + result = server.health_check() + self.assertEqual(result['last_action'], 'pineapd restart') + self.assertIn(['/etc/init.d/pineapd', 'restart'], [r[0] for r in self.runs]) + + def test_fix_cooldown_prevents_thrash(self): + self.ping_ok = False + server._health['last_fix'] = server.time.time() - 30 + server.health_check() + server.health_check() + fixes = [r for r in self.runs if r[0][0] in ('/etc/init.d/pineapd', 'ip', 'uci')] + self.assertEqual(len(fixes), 1, 'cooldown must allow only one fix action') + + def test_health_endpoint_shape(self): + server._health['sigsegv_last'] = 7 + status, payload = server.h_health(type('C', (), {'query': {}})()) + self.assertEqual(status, 200) + self.assertEqual(payload['sigsegv_count'], 7) + self.assertIn('wlan1mon_up', payload) + self.assertIn('pool_disabled', payload) + + +if __name__ == '__main__': + unittest.main()