import json import os import shutil import sqlite3 import sys import tempfile import unittest from unittest import mock 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) SCHEMA = ''' CREATE TABLE scan(id INTEGER PRIMARY KEY AUTOINCREMENT, uuid TEXT, time INT, name TEXT); CREATE TABLE wifi_device(hash INT PRIMARY KEY, scan INT, mac TEXT, time INT, signal INT, freq INT, packets INT); CREATE TABLE ssid(hash INT PRIMARY KEY, wifi_device INT, scan INT, type INT, bssid TEXT, ssid BLOB, hidden INT, time INT, signal INT, freq INT, channel INT, encryption INT); CREATE TABLE handshake(hash INT PRIMARY KEY, scan INT, stahash INT, aphash INT, time INT, beacon BLOB, hs1 BLOB, hs2 BLOB, hs3 BLOB, hs4 BLOB); CREATE TABLE hostap_basic(id INT PRIMARY KEY, scan INT, time INT, type TEXT, identity TEXT, password TEXT, verified INT NOT NULL DEFAULT 0); CREATE TABLE hostap_chalresp(id INT PRIMARY KEY, scan INT, time INT, type TEXT, username TEXT, challenge BLOB, response BLOB, verified INT NOT NULL DEFAULT 0); CREATE TABLE hostap_client(id INT PRIMARY KEY, scan INT, hash INT, mac TEXT, ssid BLOB, connected_time INT, disconnected_time INT); CREATE TABLE hostap_handshake(id INT PRIMARY KEY, scan INT, time INT, type INT, mic BLOB, apmac BLOB, stamac BLOB, ssid BLOB, nonce BLOB, eapol BLOB); ''' def make_db(): fd, db = tempfile.mkstemp(suffix='.db') os.close(fd) conn = sqlite3.connect(db) conn.executescript(SCHEMA) conn.execute("INSERT INTO scan (uuid, time, name) VALUES ('u1', 1786466531, 'pager')") conn.execute("INSERT INTO scan (uuid, time, name) VALUES ('u2', 1786466848, 'pager')") conn.execute("INSERT INTO wifi_device (hash, scan, mac, time, signal, freq, packets) VALUES (1, 1, 'AE77C0EB3141', 1786466531, -71, 2412, 5)") conn.execute("INSERT INTO wifi_device (hash, scan, mac, time, signal, freq, packets) VALUES (2, 1, 'C89E43648080', 1786466532, -76, 5745, 9)") conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) " "VALUES (10, 2, 1, 8, 'C89E43648080', X'416E646572736F6E2D35', 0, 1786466532, -76, 5745, 149, 0x400400108)") conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) " "VALUES (11, 2, 1, 8, '506F9A010000', X'', 1, 1786466532, -64, 5745, 149, 0)") conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) " "VALUES (12, 1, 1, 4, NULL, X'5A6E6574', NULL, 1786466531, -40, 2412, NULL, NULL)") conn.execute("INSERT INTO handshake (hash, scan, stahash, aphash, time) VALUES (20, 1, 1, 2, 1786466600)") conn.execute("INSERT INTO wifi_device (hash, scan, mac, time, signal, freq, packets) VALUES (101, 1, 'AA11BB22CC33', 1786466533, -61, 2412, 4)") conn.execute("INSERT INTO wifi_device (hash, scan, mac, time, signal, freq, packets) VALUES (102, 1, 'DDEEFFEEDD00', 1786466534, -58, 2412, 7)") conn.execute("INSERT INTO wifi_device (hash, scan, mac, time, signal, freq, packets) VALUES (103, 1, '021122334455', 1786466535, -80, 2412, 1)") conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) " "VALUES (101, 102, 1, 8, 'DDEEFFEEDD00', X'5365636F6E642D4E6574', 0, 1786466534, -58, 2412, 1, 0)") conn.execute("INSERT INTO handshake (hash, scan, stahash, aphash, time) VALUES (101, 1, 101, 102, 1786466602)") conn.execute("INSERT INTO hostap_client (id, scan, hash, mac, ssid, connected_time, disconnected_time) " "VALUES (101, 1, 20, 'AE77C0EB3141', X'416E646572736F6E2D35', 1786466601, 1786466605)") conn.execute("INSERT INTO hostap_client (id, scan, hash, mac, ssid, connected_time, disconnected_time) " "VALUES (102, 1, 22, 'AA11BB22CC33', X'486F73744150', 1786466603, NULL)") conn.execute("INSERT INTO hostap_client (id, scan, hash, mac, ssid, connected_time, disconnected_time) " "VALUES (103, 1, 23, 'AA11BB22CC33', X'486F73744150', 1786466604, 1786466606)") conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) " "VALUES (102, 101, 1, 5, NULL, X'50726F62654F6E6C7953534944', 0, 1786466604, -30, 2412, NULL, NULL)") conn.execute("INSERT INTO hostap_basic (scan, time, type, identity, password, verified) VALUES (1, 1786466601, 'WPA', 'bob', '', 0)") conn.commit() conn.close() return db class DecodersTest(unittest.TestCase): def test_fmt_mac_colon_form(self): self.assertEqual(server.fmt_mac('AE77C0EB3141'), 'AE:77:C0:EB:31:41') def test_fmt_mac_noop(self): self.assertEqual(server.fmt_mac('AE:77:C0:EB:31:41'), 'AE:77:C0:EB:31:41') self.assertEqual(server.fmt_mac(''), '--') self.assertEqual(server.fmt_mac(None), '--') def test_norm_mac_12hex(self): self.assertEqual(server._norm_mac('C89E43648080'), 'C8:9E:43:64:80:80') def test_decode_ssid_bytes_and_str(self): self.assertEqual(server.decode_ssid(b'Znet'), 'Znet') self.assertEqual(server.decode_ssid('Znet'), 'Znet') self.assertEqual(server.decode_ssid(None), '') self.assertEqual(server.decode_ssid(b'\xff\xfeZnet'), '\ufffd\ufffdZnet') def test_decode_ssid_cli_escapes(self): # Device sqlite3 CLI -json emits \xNN escapes for non-UTF8 BLOBs. self.assertEqual(server.decode_ssid('casaalicia\\x00.\\xde_;'), 'casaalicia\x00.\ufffd_;') def test_decode_encryption_cases(self): self.assertEqual(server.decode_encryption(0), 'Open') self.assertEqual(server.decode_encryption(None), 'Open') self.assertEqual(server.decode_encryption(2), 'WEP') self.assertEqual(server.decode_encryption(0x04), 'WPA') self.assertEqual(server.decode_encryption(0x08), 'WPA2') self.assertEqual(server.decode_encryption(0x04 | 0x08), 'WPA2 WPA') self.assertEqual(server.decode_encryption(0x400400108), 'WPA3 WPA2 PSK') self.assertEqual(server.decode_encryption(0x400400108 | (1 << 33)), 'WPA3 WPA2 Enterprise') self.assertEqual(server.decode_encryption(0x400400108 | (1 << 40)), 'WPA3 WPA2 SAE') self.assertEqual(server.decode_encryption(0x400400108 | (1 << 33) | (1 << 40)), 'WPA3 WPA2 Enterprise') self.assertEqual(server.decode_encryption(0x400400108 | (1 << 45)), 'WPA3 WPA2 OWE') self.assertEqual(server.decode_encryption(0x400400110), 'WPA3 PSK') self.assertEqual(server.decode_encryption(0x20050004C), 'WPA2 WPA Enterprise') class ReconDataTest(unittest.TestCase): def setUp(self): self.db = make_db() server.RECON_DB = self.db def tearDown(self): os.unlink(self.db) def test_scans_list_uses_time_and_counts(self): data = server.recon_scans_data() self.assertEqual(len(data['scans']), 2) newest = data['scans'][0] self.assertEqual(newest['id'], 2) self.assertEqual(newest['time'], 1786466848) self.assertEqual(newest['name'], 'pager') old = data['scans'][1] self.assertEqual(old['devices'], 5) self.assertEqual(old['aps'], 3) self.assertEqual(old['handshakes'], 2) self.assertNotIn('uuid', old) def test_scan_detail_decodes_aps(self): data = server.recon_scan_data(1) self.assertEqual(data['scan']['id'], 1) self.assertEqual(data['scan']['time'], 1786466531) self.assertEqual(len(data['aps']), 3) aps = {a['bssid']: a for a in data['aps']} a = aps['C8:9E:43:64:80:80'] self.assertEqual(a['ssid'], 'Anderson-5') self.assertEqual(a['channel'], 149) self.assertEqual(a['signal'], -76) self.assertEqual(a['encryption'], 'WPA3 WPA2 PSK') self.assertFalse(a['hidden']) hidden = aps['50:6F:9A:01:00:00'] self.assertTrue(hidden['hidden']) self.assertEqual(hidden['encryption'], 'Open') def test_scan_detail_clients_exclude_ap_macs(self): data = server.recon_scan_data(1) macs = [c['mac'] for c in data['clients']] self.assertEqual(macs, ['AE:77:C0:EB:31:41', 'AA:11:BB:22:CC:33', '02:11:22:33:44:55']) def test_scan_detail_handshakes_resolve_macs(self): data = server.recon_scan_data(1) self.assertEqual(len(data['handshakes']), 2) hs = next(h for h in data['handshakes'] if h['client'] == 'AE:77:C0:EB:31:41') self.assertEqual(hs['ap'], 'C8:9E:43:64:80:80') self.assertEqual(hs['client'], 'AE:77:C0:EB:31:41') self.assertEqual(hs['time'], 1786466600) def test_scan_detail_associations_are_confirmed_only(self): data = server.recon_scan_data(1) client = next(c for c in data['clients'] if c['mac'] == 'AE:77:C0:EB:31:41') handshake = next(a for a in client['associations'] if 'bssid' in a) self.assertEqual(handshake['sources'], ['handshake']) self.assertEqual(handshake['ssid'], 'Anderson-5') self.assertEqual(handshake['bssid'], 'C8:9E:43:64:80:80') self.assertEqual(client['vendor']['manufacturer'], 'Local/Randomized') self.assertEqual(next(a for a in data['aps'] if a['bssid'] == 'C8:9E:43:64:80:80')['client_count'], 1) self.assertEqual(next(a for a in data['aps'] if a['bssid'] == 'C8:9E:43:64:80:80')['clients'][0]['mac'], client['mac']) self.assertNotIn('ProbeOnlySSID', [a['ssid'] for a in client['associations']]) hostap_client = next(c for c in data['clients'] if c['mac'] == 'AA:11:BB:22:CC:33') ssid_only = next(a for a in hostap_client['associations'] if a['ssid'] == 'HostAP') self.assertEqual(ssid_only['sources'], ['hostap_client']) self.assertNotIn('bssid', ssid_only) self.assertEqual(ssid_only['connected_time'], 1786466604) self.assertEqual(ssid_only['disconnected_time'], 1786466606) self.assertEqual(sum(a['ssid'] == 'HostAP' for a in hostap_client['associations']), 1) second_ap = next(a for a in hostap_client['associations'] if 'bssid' in a) self.assertEqual(second_ap['ssid'], 'Second-Net') unassociated = next(c for c in data['clients'] if c['mac'] == '02:11:22:33:44:55') self.assertEqual(unassociated['associations'], []) def test_scan_detail_allows_missing_hostap_client_table(self): conn = sqlite3.connect(self.db) conn.execute('DROP TABLE hostap_client') conn.commit() conn.close() data = server.recon_scan_data(1) client = next(c for c in data['clients'] if c['mac'] == 'AE:77:C0:EB:31:41') self.assertEqual([a['sources'] for a in client['associations']], [['handshake']]) def test_scan_detail_missing_returns_none(self): self.assertIsNone(server.recon_scan_data(999)) class FakeSock: def __init__(self, resp=b''): self.resp = resp self.sent = b'' self.connected = None def settimeout(self, t): pass def connect(self, addr): self.connected = addr def sendall(self, b): self.sent += b def recv(self, n): chunk, self.resp = self.resp[:n], self.resp[n:] return chunk def close(self): pass class ReconHopperTest(unittest.TestCase): def test_preflight_probes_one_channel_per_radio(self): calls = [] with mock.patch.object( server, '_set_monitor_channel', side_effect=lambda interface, channel: calls.append((interface, channel)) or (True, '')): with mock.patch.object(server, '_monitor_down', return_value=False): self.assertEqual( server._recon_hopper_preflight(), (True, 'monitor channel control ready')) expected = [ (interface, channels[0]) for interface, channels in server.RECON_CHANNELS.items() ] self.assertEqual(calls, expected) self.assertEqual(server._recon_hop_state['ifaces'], ['wlan0mon', 'wlan1mon']) def test_preflight_skips_busy_radio_and_keeps_the_other(self): def set_channel(interface, channel): if interface == 'wlan0mon': return False, 'wlan0mon channel 1: command failed: Resource busy (-16)' return True, '' with mock.patch.object(server, '_set_monitor_channel', side_effect=set_channel): with mock.patch.object(server, '_monitor_down', return_value=False): with mock.patch.object(server, '_sta_uplink_enabled', return_value=False): with mock.patch.object(server, '_wlan0_pinned', return_value=True): ok, detail = server._recon_hopper_preflight() self.assertTrue(ok) self.assertIn('wlan0mon', server._recon_hop_state['skipped']) self.assertEqual(server._recon_hop_state['ifaces'], ['wlan1mon']) self.assertIn('2.4 GHz hopping skipped', detail) self.assertIn('Scanning 5 GHz only', detail) def test_preflight_fails_when_no_monitor_is_usable(self): with mock.patch.object( server, '_set_monitor_channel', return_value=(False, 'wlan0mon channel 1: No such device')): with mock.patch.object(server, '_monitor_down', return_value=True): ok, detail = server._recon_hopper_preflight() self.assertFalse(ok) self.assertIn('unavailable', detail.lower()) def test_busy_error_is_classified(self): self.assertEqual( server._iw_error_kind('wlan0mon channel 1: command failed: Resource busy (-16)'), 'busy') self.assertEqual(server._iw_error_kind('No such device'), 'missing') def test_set_channel_surfaces_iw_failure(self): with mock.patch.object( server, 'device_run', return_value=(240, '', 'Device or resource busy')): ok, detail = server._set_monitor_channel('wlan0mon', 6) self.assertFalse(ok) self.assertIn('wlan0mon channel 6', detail) self.assertIn('Device or resource busy', detail) def test_dummy_sta_not_borrowable_when_client_mode_on(self): with mock.patch.object(server, '_wifi_client_mode_enabled', return_value=True): with mock.patch.object(server, '_wlan0_pinned', return_value=False): with mock.patch.object(server, '_wlan0_mgmt_enabled', return_value=False): self.assertFalse(server._dummy_sta_borrowable()) def test_dummy_sta_borrowable_when_only_dummy_is_up(self): with mock.patch.object(server, '_wifi_client_mode_enabled', return_value=False): with mock.patch.object(server, '_wlan0_pinned', return_value=False): with mock.patch.object(server, '_wlan0_mgmt_enabled', return_value=False): with mock.patch.object(server, '_iface_associated', return_value=False): with mock.patch.object(server, '_sta_uplink_enabled', return_value=True): self.assertTrue(server._dummy_sta_borrowable()) def test_preflight_parks_dummy_sta_and_hops_24ghz(self): def set_channel(interface, channel): if interface == 'wlan0mon' and not server._recon_hop_state.get('borrowed_wlan0'): return False, 'wlan0mon channel 1: command failed: Resource busy (-16)' return True, '' def borrow(): server._recon_hop_state['borrowed_wlan0'] = True return True with mock.patch.object(server, '_set_monitor_channel', side_effect=set_channel): with mock.patch.object(server, '_monitor_down', return_value=False): with mock.patch.object(server, '_dummy_sta_borrowable', return_value=True): with mock.patch.object(server, '_borrow_dummy_sta', side_effect=borrow): ok, detail = server._recon_hopper_preflight() self.assertTrue(ok) self.assertEqual(detail, 'monitor channel control ready') self.assertEqual(server._recon_hop_state['ifaces'], ['wlan0mon', 'wlan1mon']) self.assertTrue(server._recon_hop_state['borrowed_wlan0']) self.assertEqual(server._recon_hop_state['skipped'], {}) def test_preflight_does_not_park_when_ap_holds_phy0(self): def set_channel(interface, channel): if interface == 'wlan0mon': return False, 'wlan0mon channel 1: command failed: Resource busy (-16)' return True, '' with mock.patch.object(server, '_set_monitor_channel', side_effect=set_channel): with mock.patch.object(server, '_monitor_down', return_value=False): with mock.patch.object(server, '_dummy_sta_borrowable', return_value=False): with mock.patch.object(server, '_borrow_dummy_sta') as borrow: with mock.patch.object(server, '_sta_uplink_enabled', return_value=False): with mock.patch.object(server, '_wlan0_pinned', return_value=True): ok, detail = server._recon_hopper_preflight() self.assertTrue(ok) borrow.assert_not_called() self.assertEqual(server._recon_hop_state['ifaces'], ['wlan1mon']) self.assertIn('Open AP / Evil WPA', detail) def test_reset_restores_parked_dummy_sta(self): server._recon_hop_state['borrowed_wlan0'] = True with mock.patch.object( server, 'device_run', return_value=(0, '', '')) as run: server._reset_recon_hop_state() run.assert_any_call(['ip', 'link', 'set', 'wlan0', 'up'], timeout=10) self.assertFalse(server._recon_hop_state['borrowed_wlan0']) class DaemonSockTest(unittest.TestCase): def setUp(self): # h_recon_start now reads shared scan state; keep these isolated. server._recon_scan_state = {'active': False, 'started': 0, 'duration': 0} preflight = mock.patch.object( server, '_recon_hopper_preflight', return_value=(True, 'ready')) start = mock.patch.object(server, '_start_recon_hopper') preflight.start() start.start() self.addCleanup(preflight.stop) self.addCleanup(start.stop) def test_socket_call_posts_json_to_sock(self): server.DAEMON_SOCK = '/tmp/api.sock' fake = FakeSock(b'HTTP/1.1 200 OK\r\nContent-Length: 17\r\n\r\n{"success":true}') with mock.patch.object(server.socket, 'socket', lambda *a, **k: fake): status, data = server.daemon_sock_call('POST', '/api/pineap/recon/new', body={'x': 1}) self.assertEqual(status, 200) self.assertEqual(data, {'success': True}) self.assertEqual(fake.connected, '/tmp/api.sock') self.assertTrue(fake.sent.startswith(b'POST /api/pineap/recon/new HTTP/1.1')) self.assertTrue(b'Content-Length: 8' in fake.sent) self.assertTrue(fake.sent.endswith(b'{"x": 1}')) def test_socket_call_connect_error(self): def boom(*a, **k): raise OSError('nope') with mock.patch.object(server.socket, 'socket', boom): status, data = server.daemon_sock_call('POST', '/api/pineap/recon/new') self.assertEqual(status, 0) self.assertIsNone(data) def test_start_stop_handlers_call_socket(self): calls = [] server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p, body)) or (200, {'success': True}) server.h_recon_start(type('C', (), {'args': ()})()) self.assertEqual(calls, [ ('POST', '/api/pineap/recon/new', {'scan_time': 30})]) def test_start_forwards_scan_time(self): calls = [] server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p, body)) or (200, {'success': True}) ctx = type('C', (), {'args': (), 'body': {'scan_time': 60}})() status, data = server.h_recon_start(ctx) self.assertEqual(status, 200) self.assertEqual(calls[0], ('POST', '/api/pineap/recon/new', {'scan_time': 60})) server._start_recon_hopper.assert_called_once_with(60) def test_start_defaults_empty_body(self): calls = [] server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p, body)) or (200, {'success': True}) server.h_recon_start(type('C', (), {'args': ()})()) self.assertEqual(calls[0], ('POST', '/api/pineap/recon/new', {'scan_time': 30})) def test_start_rejects_invalid_scan_time(self): calls = [] server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p, body)) ctx = type('C', (), {'args': (), 'body': {'scan_time': 'forever'}})() status, data = server.h_recon_start(ctx) self.assertEqual(status, 400) self.assertIn('scan_time', data['error']) self.assertEqual(calls, []) def test_start_reports_native_failure(self): server._recon_scan_state = {'active': False, 'started': 0, 'duration': 0} server.daemon_sock_call = lambda m, p, body=None: (500, {'error': 'no radio'}) status, data = server.h_recon_start( type('C', (), {'args': (), 'body': {'scan_time': 30}})()) self.assertEqual(status, 502) self.assertEqual(data['error'], 'native recon scan failed') self.assertEqual(data['detail'], 'recon/new: no radio') self.assertEqual(data['daemon'], {'error': 'no radio'}) self.assertFalse(server._recon_scan_state['active']) def test_start_reports_hopper_preflight_failure(self): calls = [] server._recon_hopper_preflight.return_value = ( False, 'Recon radios are unavailable. wlan0mon is missing.') server.daemon_sock_call = lambda *args, **kwargs: calls.append(args) status, data = server.h_recon_start( type('C', (), {'args': (), 'body': {'scan_time': 30}})()) self.assertEqual(status, 503) self.assertEqual(data['error'], 'Could not prepare recon radios') self.assertIn('unavailable', data['detail']) self.assertEqual(calls, []) def test_start_returns_warning_when_a_radio_is_skipped(self): calls = [] def fake_preflight(): server._recon_hop_state.update({ 'warning': '2.4 GHz hopping skipped: Open AP is holding phy0. Scanning 5 GHz only.', 'ifaces': ['wlan1mon'], 'skipped': {'wlan0mon': '2.4 GHz hopping skipped: Open AP is holding phy0.'}, 'hint': 'Stop the 2.4 GHz AP to hop 2.4 GHz.', }) return True, server._recon_hop_state['warning'] server._recon_hopper_preflight.side_effect = fake_preflight server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p, body)) or (200, {'success': True}) status, data = server.h_recon_start( type('C', (), {'args': (), 'body': {'scan_time': 30}})()) self.assertEqual(status, 200) self.assertTrue(data.get('ok')) self.assertIn('2.4 GHz hopping skipped', data.get('warning')) self.assertEqual(data.get('hopping'), ['wlan1mon']) self.assertEqual(calls[0][1], '/api/pineap/recon/new') server._start_recon_hopper.assert_called_once_with(30) class ReconScanStateTest(unittest.TestCase): """The webui mirrors the duration of the Pager's native timed scan.""" def setUp(self): self.db = make_db() server.RECON_DB = self.db server._recon_scan_state = {'active': False, 'started': 0, 'duration': 0} preflight = mock.patch.object( server, '_recon_hopper_preflight', return_value=(True, 'ready')) start = mock.patch.object(server, '_start_recon_hopper') preflight.start() start.start() self.addCleanup(preflight.stop) self.addCleanup(start.stop) def tearDown(self): os.unlink(self.db) def _start(self, scan_time=None, fail=False): body = {} if scan_time is not None: body['scan_time'] = scan_time server.daemon_sock_call = (lambda m, p, body=None: (502, {})) if fail \ else (lambda m, p, body=None: (200, {'success': True})) return server.h_recon_start(type('C', (), {'args': (), 'body': body})()) def _stop(self, fail=False): server.daemon_sock_call = (lambda m, p, body=None: (502, {})) if fail \ else (lambda m, p, body=None: (200, {'success': True})) return server.h_recon_stop(type('C', (), {'args': ()})()) def _status(self): server.daemon_sock_call = lambda m, p, body=None: (200, {'success': True}) return server.h_recon_status(type('C', (), {'args': ()})()) def test_timed_start_marks_scanning_with_remaining(self): server.time.time = lambda: 1000.0 status, data = self._start(scan_time=30) self.assertEqual(status, 200) status, data = self._status() self.assertTrue(data['scanning']) self.assertEqual(data['scan_remaining'], 30) def test_timed_scan_expires_when_duration_elapses(self): base = [1000.0] server.time.time = lambda: base[0] self._start(scan_time=30) base[0] = 1031.0 status, data = self._status() self.assertFalse(data['scanning']) self.assertEqual(data['scan_remaining'], 0) def test_zero_duration_is_rejected(self): server.time.time = lambda: 1000.0 status, data = self._start(scan_time=0) self.assertEqual(status, 400) self.assertFalse(server._recon_scan_state['active']) def test_default_start_uses_thirty_seconds(self): server.time.time = lambda: 1000.0 self._start() status, data = self._status() self.assertTrue(data['scanning']) self.assertEqual(data['scan_remaining'], 30) def test_stop_rejects_active_native_scan(self): server.time.time = lambda: 1000.0 self._start(scan_time=30) status, data = self._stop() self.assertEqual(status, 409) self.assertIn('finish automatically', data['error']) self.assertEqual(data['scan_remaining'], 30) status, data = self._status() self.assertTrue(data['scanning']) def test_stop_is_idempotent_when_inactive(self): status, data = self._stop() self.assertEqual(status, 200) self.assertEqual(data, {'ok': True}) def test_start_failure_does_not_mark_scanning(self): server.time.time = lambda: 1000.0 status, data = self._start(scan_time=30, fail=True) self.assertEqual(status, 502) status, data = self._status() self.assertFalse(data['scanning']) def test_start_while_scanning_returns_409_without_restart(self): server.time.time = lambda: 1000.0 self._start(scan_time=30) calls = [] server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p)) or (200, {'success': True}) status, data = server.h_recon_start( type('C', (), {'args': (), 'body': {'scan_time': 30}})()) self.assertEqual(status, 409) self.assertEqual(data['scan_remaining'], 30) self.assertEqual(calls, []) status, data = self._status() self.assertTrue(data['scanning']) def test_watchdog_stops_expired_timed_scan(self): server.time.time = lambda: 1000.0 self._start(scan_time=10) server.time.time = lambda: 1012.0 calls = [] server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p)) or (200, {'success': True}) server._recon_watchdog_tick() self.assertEqual(calls, []) self.assertFalse(server._recon_scan_state['active']) def test_watchdog_leaves_active_scan_alone(self): server.time.time = lambda: 1000.0 self._start(scan_time=30) server.time.time = lambda: 1010.0 calls = [] server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p)) or (200, {'success': True}) server._recon_watchdog_tick() self.assertEqual(calls, []) self.assertTrue(server._recon_scan_state['active']) class ReconExtrasTest(unittest.TestCase): def setUp(self): self.db = make_db() server.RECON_DB = self.db def tearDown(self): os.unlink(self.db) def test_status_reports_last_scan_and_active(self): server.time.time = lambda: 1786466532 + 100 status, data = server.h_recon_status(type('C', (), {'args': ()})()) self.assertEqual(status, 200) self.assertEqual(data['last_scan'], 1786466848) self.assertEqual(data['last_activity'], 1786466535) self.assertTrue(data['active']) server.time.time = lambda: 1786466532 + 1000 status, data = server.h_recon_status(type('C', (), {'args': ()})()) self.assertFalse(data['active']) def test_status_includes_hopper_and_history_flags(self): with mock.patch.object(server, '_hopper_online', return_value=False), \ mock.patch.object(server, '_recon_history_reset', return_value=True): status, data = server.h_recon_status(type('C', (), {'args': ()})()) self.assertEqual(status, 200) self.assertFalse(data['hopper_online']) self.assertIn('hopper_error', data) self.assertTrue(data['history_reset']) def test_hopper_online_cached(self): server._hopper_cache.update({'updated': 0, 'online': None}) with mock.patch.object(server, 'wifi_ifaces', return_value=['wlan0mon', 'wlan1mon']): self.assertTrue(server._hopper_online()) # Second call within the cache window must not re-run iwinfo. with mock.patch.object(server, 'wifi_ifaces', side_effect=AssertionError('cached, must not call iwinfo')): self.assertTrue(server._hopper_online()) server._hopper_cache['updated'] = 0 # let other tests start fresh def test_delete_cascades(self): server.recon_delete_scan(1) rows = server._db_rows(self.db, 'SELECT count(*) AS c FROM wifi_device') self.assertEqual(rows[0]['c'], 0) rows = server._db_rows(self.db, 'SELECT count(*) AS c FROM ssid') self.assertEqual(rows[0]['c'], 0) rows = server._db_rows(self.db, 'SELECT count(*) AS c FROM handshake') self.assertEqual(rows[0]['c'], 0) rows = server._db_rows(self.db, 'SELECT count(*) AS c FROM scan') self.assertEqual(rows[0]['c'], 1) def test_delete_handler_404_for_missing(self): status, data = server.h_recon_delete(type('C', (), {'args': ('999',)})()) self.assertEqual(status, 404) def test_delete_all_clears_every_scan(self): status, data = server.h_recon_delete_all(type('C', (), {'args': ()})()) self.assertEqual(status, 200) self.assertEqual(data['deleted'], 2) for table in ('scan', 'ssid', 'wifi_device', 'handshake', 'hostap_basic', 'hostap_chalresp'): rows = server._db_rows(self.db, 'SELECT count(*) AS c FROM %s' % table) self.assertEqual(rows[0]['c'], 0, table) def test_events_lists_db_rows(self): status, data = server.h_recon_events(type('C', (), {'args': ()})()) self.assertEqual(status, 200) kinds = [e['type'] for e in data['events']] self.assertIn('auth attempt', kinds) self.assertEqual(data['events'][0]['time'], 1786466602) class ReconExamineTest(unittest.TestCase): def test_examine_bssid_calls_hak5(self): calls = [] server.hak5 = lambda *args, **kw: calls.append(args) or '' ctx = type('C', (), {'args': (), 'body': {'bssid': 'AA:BB:CC:DD:EE:FF'}})() status, data = server.h_recon_examine(ctx) self.assertEqual(status, 200) self.assertEqual(calls[0], ('PINEAPPLE_EXAMINE_BSSID', 'AA:BB:CC:DD:EE:FF', '30')) def test_examine_channel_calls_hak5(self): calls = [] server.hak5 = lambda *args, **kw: calls.append(args) or '' ctx = type('C', (), {'args': (), 'body': {'channel': 6}})() status, data = server.h_recon_examine(ctx) self.assertEqual(status, 200) self.assertEqual(calls[0], ('PINEAPPLE_EXAMINE_CHANNEL', '6', '30')) def test_examine_channel_5ghz_sends_duration(self): calls = [] server.hak5 = lambda *args, **kw: calls.append(args) or '' ctx = type('C', (), {'args': (), 'body': {'channel': 140, 'seconds': 15}})() status, data = server.h_recon_examine(ctx) self.assertEqual(status, 200) self.assertEqual(calls[0], ('PINEAPPLE_EXAMINE_CHANNEL', '140', '15')) self.assertEqual(data.get('seconds'), 15) def test_examine_compact_bssid_is_colonized(self): calls = [] server.hak5 = lambda *args, **kw: calls.append(args) or '' ctx = type('C', (), {'args': (), 'body': {'bssid': 'aabbccddeeff'}})() status, data = server.h_recon_examine(ctx) self.assertEqual(status, 200) self.assertEqual(calls[0], ('PINEAPPLE_EXAMINE_BSSID', 'AA:BB:CC:DD:EE:FF', '30')) def test_examine_requires_target(self): server.hak5 = lambda *args, **kw: '' ctx = type('C', (), {'args': (), 'body': {}})() status, data = server.h_recon_examine(ctx) self.assertEqual(status, 400) class HandshakeFileTest(unittest.TestCase): def test_file_download_decodes_name(self): import shutil d = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, d) server.LOOT_HS_DIR = d name = '00 11_scan.cap' with open(os.path.join(d, name), 'wb') as f: f.write(b'PCAPDATA') status, payload = server.h_handshake_file(type('C', (), {'args': ('00%2011_scan.cap',)})) self.assertEqual(status, 200) self.assertEqual(payload.data, b'PCAPDATA') self.assertEqual(payload.filename, name) @unittest.skipIf(os.name == 'nt', 'colons are not valid in filenames on Windows') def test_file_download_decodes_mac_colon_name(self): import shutil d = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, d) server.LOOT_HS_DIR = d name = '00:11:22:33:44:55_scan.cap' with open(os.path.join(d, name), 'wb') as f: f.write(b'PCAPDATA') encoded = '00%3A11%3A22%3A33%3A44%3A55_scan.cap' status, payload = server.h_handshake_file(type('C', (), {'args': (encoded,)})) self.assertEqual(status, 200) self.assertEqual(payload.data, b'PCAPDATA') self.assertEqual(payload.filename, name) def test_file_download_404_missing(self): import shutil d = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, d) server.LOOT_HS_DIR = d status, payload = server.h_handshake_file(type('C', (), {'args': ('nope.cap',)})) self.assertEqual(status, 404) class CliFallbackTest(unittest.TestCase): def setUp(self): self.db = make_db() server.RECON_DB = self.db self._orig_sqlite3 = server.sqlite3 self._orig_device_run = server.device_run server.sqlite3 = None server.device_run = lambda args, timeout=20: (0, self._run(args), '') def _run(self, args): # emulate `sqlite3 -json ` over an in-memory copy import sqlite3 as s3 import subprocess conn = s3.connect(self.db) try: cur = conn.execute(args[-1]) conn.commit() if cur.description: import json rows = cur.fetchall() cols = [d[0] for d in cur.description] objs = [] for row in rows: obj = {} for c, v in zip(cols, row): if isinstance(v, bytes): try: v = v.decode('utf-8') except Exception: v = ''.join('\\x%02x' % b for b in v) obj[c] = v objs.append(obj) return json.dumps(objs) return '' finally: conn.close() def tearDown(self): server.sqlite3 = self._orig_sqlite3 server.device_run = self._orig_device_run os.unlink(self.db) def test_delete_cascade_via_cli(self): server.recon_delete_scan(1) rows = server._db_rows(self.db, 'SELECT count(*) AS c FROM scan') self.assertEqual(rows[0]['c'], 1) def test_cli_ssid_escapes_decoded(self): # emulate the CLI emitting \xNN escapes for a non-UTF8 blob server.device_run = lambda args, timeout=20: (0, '[{"ssid": "casaalicia\\\\x00.\\\\xde_"}]', '') server.RECON_DB = '/nonexistent.db' self.assertEqual(server.decode_ssid('casaalicia\\x00.\\xde_'), 'casaalicia\x00.\ufffd_') def test_completed_recon_lock_uses_immutable_read(self): calls = [] server._recon_scan_state = {'active': False, 'started': 0, 'duration': 0} def locked_then_read(args, timeout=20): calls.append(args) if len(calls) == 1: return 5, '', 'Error: database is locked' return 0, '[{"id": 2}]', '' server.device_run = locked_then_read rows = server._db_rows(self.db, 'SELECT MAX(id) AS id FROM scan') self.assertEqual(rows, [{'id': 2}]) self.assertEqual(calls[1][-2], 'file:%s?immutable=1' % self.db) def make_hs_db(): db = make_db() conn = sqlite3.connect(db) conn.execute( "INSERT INTO handshake (hash, scan, stahash, aphash, time, beacon, hs1, hs2, hs3, hs4) " "VALUES (21, 1, 1, 2, 1786466650, X'424541434F4E', X'01', X'02', X'03', X'04')") conn.commit() conn.close() return db class ParseHsFilenameTest(unittest.TestCase): def test_parse_full_pcap(self): p = server.parse_hs_filename('1786466650_C8:9E:43:64:80:80_AE:77:C0:EB:31:41_handshake.pcap') self.assertEqual(p['ts'], 1786466650) self.assertEqual(p['ap'], 'C8:9E:43:64:80:80') self.assertEqual(p['client'], 'AE:77:C0:EB:31:41') self.assertEqual(p['kind'], 'full') self.assertEqual(p['ext'], 'pcap') def test_parse_partial_and_incomplete(self): p = server.parse_hs_filename('1_C8-9E-43-64-80-80_AE-77-C0-EB-31-41_handshake_partial.22000') self.assertEqual(p['kind'], 'partial') self.assertEqual(p['ext'], '22000') p = server.parse_hs_filename('1_C8:9E:43:64:80:80_AE:77:C0:EB:31:41_handshake_incomplete.pcap') self.assertEqual(p['kind'], 'incomplete') self.assertEqual(p['ext'], 'pcap') def test_parse_dash_macs_and_no_ts(self): p = server.parse_hs_filename('C8-9E-43-64-80-80_AE-77-C0-EB-31-41_handshake.pcap') self.assertIsNone(p['ts']) self.assertEqual(p['ap'], 'C8:9E:43:64:80:80') self.assertEqual(p['client'], 'AE:77:C0:EB:31:41') def test_parse_unrecognized(self): self.assertIsNone(server.parse_hs_filename('random.cap')) self.assertIsNone(server.parse_hs_filename('notes.txt')) self.assertIsNone(server.parse_hs_filename('')) self.assertIsNone(server.parse_hs_filename('123_mac1_mac2_handshake')) class HandshakesDataTest(unittest.TestCase): def setUp(self): self.db = make_hs_db() self.dir = tempfile.mkdtemp() self.addCleanup(server.__dict__.update, {'RECON_DB': server.RECON_DB, 'LOOT_HS_DIR': server.LOOT_HS_DIR}) server.RECON_DB = self.db server.LOOT_HS_DIR = self.dir def tearDown(self): shutil.rmtree(self.dir) os.unlink(self.db) def _write(self, name, ts): path = os.path.join(self.dir, name) open(path, 'w').close() os.utime(path, (ts, ts)) def test_empty_dir_skips_db(self): with mock.patch.object(server, '_db_rows', side_effect=AssertionError('db should not be touched')): data = server.handshakes_data() self.assertEqual(data, {'files': [], 'handshakes': []}) def test_correlation_composes_full_record(self): self._write('1786466650_C8-9E-43-64-80-80_AE-77-C0-EB-31-41_handshake.pcap', 1786466650) data = server.handshakes_data() self.assertEqual(len(data['files']), 1) hs = data['handshakes'][0] self.assertEqual(hs['mac'], 'C8:9E:43:64:80:80') self.assertEqual(hs['client'], 'AE:77:C0:EB:31:41') self.assertEqual(hs['source'], 'Recon') self.assertEqual(hs['type'], 'full') self.assertEqual(hs['extension'], 'pcap') self.assertEqual(hs['timestamp'], 1786466650) self.assertTrue(hs['in_db']) self.assertEqual(hs['part_mask'], 15) self.assertTrue(hs['beacon']) self.assertEqual(hs['name'], '1786466650_C8-9E-43-64-80-80_AE-77-C0-EB-31-41_handshake.pcap') self.assertTrue(hs['file_exists']) self.assertEqual(hs['location'], os.path.join(server.LOOT_HS_DIR, hs['name'])) def test_file_not_in_db_has_question_mark_fields(self): self._write('1786467000_AA-BB-CC-DD-EE-FF_00-11-22-33-44-55_handshake.pcap', 1786467000) hs = server.handshakes_data()['handshakes'][0] self.assertFalse(hs['in_db']) self.assertEqual(hs['part_mask'], 0) self.assertFalse(hs['beacon']) self.assertEqual(hs['timestamp'], 1786467000) def test_unparseable_file_still_listed_with_placeholders(self): self._write('random.cap', 1786467005) hs = server.handshakes_data()['handshakes'][0] self.assertEqual(hs['mac'], '--') self.assertEqual(hs['client'], '--') self.assertFalse(hs['in_db']) self.assertEqual(hs['extension'], 'cap') class HandshakeRoutesTest(unittest.TestCase): def setUp(self): self.dir = tempfile.mkdtemp() self.addCleanup(server.__dict__.update, {'LOOT_HS_DIR': server.LOOT_HS_DIR}) server.LOOT_HS_DIR = self.dir def tearDown(self): shutil.rmtree(self.dir) def _write(self, name, data=b'data'): with open(os.path.join(self.dir, name), 'wb') as f: f.write(data) def test_location_returns_loot_dir(self): status, data = server.h_handshakes_location(type('C', (), {'args': ()})()) self.assertEqual(status, 200) self.assertEqual(data['location'], self.dir) def test_location_route_precedes_file_download(self): h, args = server.ROUTER.dispatch('GET', '/api/pineap/handshakes/location') self.assertIs(h, server.h_handshakes_location) def test_delete_all_removes_files(self): self._write('1_C8-9E-43-64-80-80_AE-77-C0-EB-31-41_handshake.pcap') self._write('2_C8-9E-43-64-80-80_AE-77-C0-EB-31-41_handshake.22000') status, data = server.h_handshakes_delete_all(type('C', (), {'args': ()})()) self.assertEqual(status, 200) self.assertEqual(data['files'], []) self.assertEqual(data['handshakes'], []) self.assertEqual(os.listdir(self.dir), []) def test_delete_all_empty_dir_is_ok(self): status, data = server.h_handshakes_delete_all(type('C', (), {'args': ()})()) self.assertEqual(status, 200) self.assertEqual(data['files'], []) def test_delete_all_skips_dotfiles(self): self._write('.hidden') self._write('1_C8-9E-43-64-80-80_AE-77-C0-EB-31-41_handshake.pcap') status, data = server.h_handshakes_delete_all(type('C', (), {'args': ()})()) self.assertEqual(status, 200) self.assertEqual(data['files'], []) self.assertEqual(data['handshakes'], []) self.assertEqual(os.listdir(self.dir), ['.hidden']) class OuiVendorTest(unittest.TestCase): def test_oui_prefix_forms(self): self.assertEqual(server._oui_prefix('C8:9E:43:64:80:80'), 'C89E43') self.assertEqual(server._oui_prefix('C89E43648080'), 'C89E43') self.assertEqual(server._oui_prefix('c8:9e:43:64:80:80'), 'C89E43') self.assertIsNone(server._oui_prefix(None)) self.assertIsNone(server._oui_prefix('')) self.assertIsNone(server._oui_prefix('XX:YY:ZZ:00:00:00')) def test_oui_vendor_lookup(self): self.assertEqual(server.oui_vendor('B8:27:EB:00:00:00'), 'Raspberry Pi') self.assertEqual(server.oui_vendor('10:BF:48:00:00:00'), 'Apple') self.assertEqual(server.oui_vendor('14:CC:20:00:00:00'), 'TP-Link') self.assertEqual(server.oui_vendor('FC:63:3E:00:00:00'), 'Google') def test_oui_vendor_unknown_and_local(self): self.assertEqual(server.oui_vendor('C8:9E:43:64:80:80'), 'Unknown') self.assertEqual(server.oui_vendor('AE:77:C0:EB:31:41'), 'Local') self.assertEqual(server.oui_vendor(None), 'Unknown') self.assertEqual(server.oui_vendor('--'), 'Unknown') def test_oui_identity_prefers_nmap_then_macchanger(self): server._oui_identity_cache = None files = { '/nmap': 'C89E43 Apple Corporation\n', '/mac': 'C89E43 fallback\n', } with mock.patch.object(server, 'OUI_DATA_PATHS', ['/nmap', '/mac']), \ mock.patch('builtins.open', side_effect=lambda p, *a, **k: mock.mock_open(read_data=files[p]).return_value): value = server.oui_identity('C89E43648080') self.assertEqual(value['manufacturer'], 'Apple Corporation') self.assertEqual(value['source'], 'nmap') self.assertIsNone(value['model']) def test_oui_identity_uses_macchanger_when_nmap_is_missing(self): server._oui_identity_cache = None files = {'/mac': 'C8-9E-43 fallback\n'} def open_file(path, *args, **kwargs): if path not in files: raise OSError('missing') return mock.mock_open(read_data=files[path]).return_value with mock.patch.object(server, 'OUI_DATA_PATHS', ['/nmap', '/mac']), \ mock.patch('builtins.open', side_effect=open_file): value = server.oui_identity('C89E43648080') self.assertEqual(value['manufacturer'], 'fallback') self.assertEqual(value['source'], 'macchanger') def test_oui_identity_handles_missing_files_unknown_and_local(self): server._oui_identity_cache = None with mock.patch.object(server, 'OUI_DATA_PATHS', ['/missing']), \ mock.patch('builtins.open', side_effect=OSError('missing')): unknown = server.oui_identity('AC:BB:CC:00:00:01') self.assertEqual(unknown['manufacturer'], 'Unknown') self.assertEqual(unknown['source'], 'unknown') self.assertEqual(unknown['oui'], 'ACBBCC') self.assertIsNone(unknown['model']) local = server.oui_identity('02:11:22:33:44:55') self.assertEqual(local['manufacturer'], 'Local/Randomized') self.assertEqual(local['source'], 'local') self.assertEqual(local['oui'], '021122') def test_oui_identity_falls_back_to_builtin_vendors(self): server._oui_identity_cache = None with mock.patch.object(server, 'OUI_DATA_PATHS', []): value = server.oui_identity('B8:27:EB:00:00:00') self.assertEqual(value['manufacturer'], 'Raspberry Pi') self.assertEqual(value['source'], 'builtin') self.assertIsNone(value['model']) def test_band_of_frequencies(self): self.assertEqual(server.band_of(2412), '2.4') self.assertEqual(server.band_of(5200), '5') self.assertEqual(server.band_of(6180), '6') self.assertEqual(server.band_of(0), '--') self.assertEqual(server.band_of(None), '--') def test_curated_table_has_no_garbage_keys(self): for key in server.OUI_VENDORS: self.assertRegex(key, r'^[0-9A-F]{6}$') self.assertNotIn('349A...', server.OUI_VENDORS) class ReconEnrichmentTest(unittest.TestCase): def setUp(self): self.db = make_db() server.RECON_DB = self.db def tearDown(self): os.unlink(self.db) def test_scan_detail_enriches_aps(self): data = server.recon_scan_data(1) aps = {a['bssid']: a for a in data['aps']} a = aps['C8:9E:43:64:80:80'] self.assertEqual(a['band'], '5') self.assertEqual(a['vendor'], 'Unknown') self.assertEqual(a['first_seen'], 1786466532) self.assertEqual(a['last_seen'], 1786466532) hidden = aps['50:6F:9A:01:00:00'] self.assertEqual(hidden['band'], '5') self.assertEqual(hidden['vendor'], 'Unknown') def test_scan_detail_unassociated_count(self): data = server.recon_scan_data(1) self.assertEqual(data['unassociated'], 1) def test_scan_detail_bounded_mode_counts_unassociated(self): data = server.recon_scan_data(1, _limit=1) self.assertEqual(data['unassociated'], 1) self.assertEqual(len(data['aps']), 3) self.assertLessEqual(len(data['clients']), 1) self.assertEqual(data['scan']['id'], 1) def test_first_last_seen_span_multiple_rows(self): conn = sqlite3.connect(self.db) conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) " "VALUES (30, 2, 1, 8, 'C89E43648080', X'416E646572736F6E2D35', 0, 1786466540, -80, 5745, 149, 0x400400108)") conn.commit() conn.close() data = server.recon_scan_data(1) a = [a for a in data['aps'] if a['bssid'] == 'C8:9E:43:64:80:80'][0] self.assertEqual(a['first_seen'], 1786466532) self.assertEqual(a['last_seen'], 1786466540) def test_band_for_24ghz_ap(self): conn = sqlite3.connect(self.db) conn.execute("INSERT INTO wifi_device (hash, scan, mac, time, signal, freq, packets) VALUES (3, 1, 'FC633E000001', 1786466533, -60, 2412, 4)") conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) " "VALUES (31, 3, 1, 8, 'FC633E000001', X'4E6574776F726B', 0, 1786466533, -60, 2412, 6, 0x08)") conn.commit() conn.close() data = server.recon_scan_data(1) a = [a for a in data['aps'] if a['bssid'] == 'FC:63:3E:00:00:01'][0] self.assertEqual(a['band'], '2.4') self.assertEqual(a['vendor'], 'Google') def test_scan_detail_handler_is_bounded_and_retried(self): seen = {} orig = server.recon_scan_data def fake(scan_id, _timeout=20, _limit=None, db=None): seen['limit'] = _limit seen['db'] = db return orig(scan_id, _timeout=_timeout, _limit=_limit, db=db) with mock.patch.object(server, 'recon_scan_data', side_effect=fake), \ mock.patch.object(server.time, 'sleep'): status, data = server.h_recon_scan_detail(type('C', (), {'args': ('1',)})()) self.assertEqual(status, 200) self.assertEqual(seen['limit'], 300) # The live handler relies on the default, which resolves to RECON_DB. self.assertIsNone(seen['db']) self.assertEqual(data['scan']['id'], 1) def test_scan_detail_handler_503_on_lock(self): with mock.patch.object(server, 'recon_scan_data', side_effect=RuntimeError('sqlite read failed: locked')), \ mock.patch.object(server.time, 'sleep'): status, data = server.h_recon_scan_detail(type('C', (), {'args': ('1',)})()) self.assertEqual(status, 503) self.assertIn('temporarily unavailable', data['error']) class ReconReportTest(unittest.TestCase): def setUp(self): self.db = make_db() server.RECON_DB = self.db def tearDown(self): os.unlink(self.db) def _ctx(self, args=()): return type('C', (), {'args': args, 'body': {}})() def test_csv_download_contains_aps_and_unassociated(self): status, payload = server.h_recon_scan_download_csv(self._ctx(('1',))) self.assertEqual(status, 200) self.assertEqual(payload.ctype, 'text/csv') self.assertEqual(payload.filename, 'scan-1.csv') text = payload.data.decode('utf-8') self.assertIn('Anderson-5', text) self.assertIn('unassociated,1', text) self.assertIn('C8:9E:43:64:80:80', text) def test_json_download_preserves_enriched_recon_data(self): status, payload = server.h_recon_scan_download(self._ctx(('1',))) self.assertEqual(status, 200) data = json.loads(payload.data.decode('utf-8')) self.assertIn('device_identity', data['aps'][0]) self.assertIn('associations', data['clients'][0]) def test_csv_download_contains_identity_counts_and_associations(self): status, payload = server.h_recon_scan_download_csv(self._ctx(('1',))) self.assertEqual(status, 200) text = payload.data.decode('utf-8') self.assertIn('Device Identity', text) self.assertIn('Client Count', text) self.assertIn('Confirmed SSIDs', text) self.assertIn('Anderson-5', text) self.assertIn('Local/Randomized', text) def test_html_download_contains_stats(self): with mock.patch.object(server, '_gps_status_data', return_value={'lock': False}): status, payload = server.h_recon_scan_download_html(self._ctx(('1',))) self.assertEqual(status, 200) self.assertEqual(payload.ctype, 'text/html') self.assertEqual(payload.filename, 'scan-1.html') text = payload.data.decode('utf-8') self.assertIn('Scan #1', text) self.assertIn('Anderson-5', text) self.assertIn('WPA3 WPA2', text) # stat cards self.assertIn('stat-card', text) self.assertIn('Unassociated', text) # band + encryption breakdowns self.assertIn('Band Breakdown', text) self.assertIn('Encryption Breakdown', text) self.assertIn('WPA2', text) # channel occupancy table self.assertIn('Channel Occupancy', text) self.assertIn('5 GHz', text) self.assertIn('>149<', text) # color-coded signal cells self.assertIn('sig-weak', text) self.assertIn('sig-good', text) self.assertIn('-76 dBm', text) # no GPS line without a fix self.assertNotIn('GPS:', text) def test_html_report_includes_confirmed_clients(self): with mock.patch.object(server, '_gps_status_data', return_value={'lock': False}): status, payload = server.h_recon_scan_download_html(self._ctx(('1',))) self.assertEqual(status, 200) text = payload.data.decode('utf-8') self.assertIn('Confirmed Clients', text) self.assertIn('AE:77:C0:EB:31:41', text) self.assertIn('handshake', text) self.assertNotIn('ProbeOnlySSID', text) def test_html_download_orders_strongest_signal_first(self): with mock.patch.object(server, '_gps_status_data', return_value={'lock': False}): status, payload = server.h_recon_scan_download_html(self._ctx(('1',))) self.assertEqual(status, 200) text = payload.data.decode('utf-8') # Fixture: hidden AP at -64 dBm is stronger than Anderson-5 at -76 dBm. strong = '50:6F:9A:01:00:00' weak = 'C8:9E:43:64:80:80' self.assertLess(text.index(strong), text.index(weak)) def test_html_report_includes_gps_when_locked(self): with mock.patch.object(server, '_gps_status_data', return_value={'lock': True, 'lat': 37.7, 'lon': -122.4, 'satellites': 8}): status, payload = server.h_recon_scan_download_html(self._ctx(('1',))) text = payload.data.decode('utf-8') self.assertIn('GPS: 37.70000, -122.40000', text) self.assertIn('8 sats', text) def test_download_404_for_missing_scan(self): status, payload = server.h_recon_scan_download_csv(self._ctx(('999',))) self.assertEqual(status, 404) status, payload = server.h_recon_scan_download_html(self._ctx(('999',))) self.assertEqual(status, 404) def test_download_503_when_db_unavailable(self): with mock.patch.object(server, 'recon_scan_data', side_effect=RuntimeError('sqlite read failed: locked')), \ mock.patch.object(server.time, 'sleep'): status, payload = server.h_recon_scan_download_csv(self._ctx(('1',))) self.assertEqual(status, 503) status, payload = server.h_recon_scan_download_html(self._ctx(('1',))) self.assertEqual(status, 503) class ReconFrontendTest(unittest.TestCase): def test_recon_view_supports_identity_and_confirmed_associations(self): path = os.path.join(os.path.dirname(__file__), '..', 'payload', 'user', 'remote_access', 'pager-webui', 'www', 'js', 'views.js') with open(path, encoding='utf-8') as source_file: source = source_file.read() for expected in ( "identity: true", "clients: true", "associated_ssid", "device_identity", "associations", "Confirmed Clients", "No confirmed clients", "reconLoadCols", "Object.assign"): self.assertIn(expected, source) class ReconArchivesTest(unittest.TestCase): """Read-only history from pineapd-rotated databases (error-*-recon.db).""" def setUp(self): self.dir = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, self.dir) self.archive_name = 'error-2026-08-18-19:21:42Z-recon.db' tmp = make_db() self.addCleanup(os.unlink, tmp) server.RECON_DB = os.path.join(self.dir, 'recon.db') shutil.copy(tmp, server.RECON_DB) archive = os.path.join(self.dir, self.archive_name) shutil.copy(tmp, archive) conn = sqlite3.connect(archive) conn.execute("INSERT INTO scan (uuid, time, name) VALUES ('u3', 1786467000, 'pager')") conn.commit() conn.close() def tearDown(self): server._recon_archives_cache.update({'dir': None, 'updated': 0, 'data': None}) def _ctx(self, args=()): return type('C', (), {'args': args, 'body': {}})() def test_archives_list_finds_rotated_db(self): data = server.recon_archives_data() self.assertEqual(len(data['archives']), 1) a = data['archives'][0] self.assertEqual(a['id'], self.archive_name) self.assertEqual(a['min_id'], 1) self.assertEqual(a['max_id'], 3) self.assertEqual(a['scans_count'], 3) self.assertEqual([s['id'] for s in a['scans']], [3, 2, 1]) def test_archives_list_is_cached_per_directory(self): orig = server.recon_scans_data calls = [] def fake(limit=50, _timeout=20, db=None): calls.append(db) return orig(limit=limit, _timeout=_timeout, db=db) with mock.patch.object(server, 'recon_scans_data', side_effect=fake): server.recon_archives_data() server.recon_archives_data() # Second call within the cache window must not re-scan the archives. self.assertEqual(len(calls), 1) self.assertTrue(calls[0].endswith(self.archive_name)) def test_archive_path_rejects_traversal(self): self.assertIsNone(server._recon_archive_path('..%2F..%2Fetc%2Fpasswd')) self.assertIsNone(server._recon_archive_path('../etc/passwd')) self.assertIsNone(server._recon_archive_path('error-x-recon.db/../evil')) self.assertIsNone(server._recon_archive_path('random.db')) self.assertIsNone(server._recon_archive_path('')) self.assertIsNotNone(server._recon_archive_path(self.archive_name)) def test_archive_scan_detail_reads_archive_not_live(self): status, data = server.h_recon_archive_scan_detail( self._ctx((self.archive_name, '3'))) self.assertEqual(status, 200) self.assertEqual(data['scan']['id'], 3) # The live db has no scan 3; the archive must be the source. self.assertIsNone(server.recon_scan_data(3)) def test_archive_scan_detail_is_bounded(self): seen = {} orig = server.recon_scan_data def fake(scan_id, _timeout=20, _limit=None, db=None): seen.update({'limit': _limit, 'db': db}) return orig(scan_id, _timeout=_timeout, _limit=_limit, db=db) with mock.patch.object(server, 'recon_scan_data', side_effect=fake), \ mock.patch.object(server.time, 'sleep'): status, data = server.h_recon_archive_scan_detail( self._ctx((self.archive_name, '1'))) self.assertEqual(status, 200) self.assertEqual(seen['limit'], 300) self.assertTrue(seen['db'].endswith(self.archive_name)) def test_archive_scans_list(self): status, data = server.h_recon_archive_scans(self._ctx((self.archive_name,))) self.assertEqual(status, 200) self.assertEqual([s['id'] for s in data['scans']], [3, 2, 1]) def test_archive_handlers_404_for_missing(self): status, data = server.h_recon_archive_scans(self._ctx(('nope.db',))) self.assertEqual(status, 404) status, data = server.h_recon_archive_scan_detail(self._ctx(('nope.db', '1'))) self.assertEqual(status, 404) status, data = server.h_recon_archive_scan_detail( self._ctx((self.archive_name, '999'))) self.assertEqual(status, 404) def test_archive_downloads(self): status, payload = server.h_recon_archive_scan_download( self._ctx((self.archive_name, '1'))) self.assertEqual(status, 200) self.assertEqual(payload.filename, 'scan-1.json') self.assertIn(b'Anderson-5', payload.data) status, payload = server.h_recon_archive_scan_download_csv( self._ctx((self.archive_name, '1'))) self.assertEqual(status, 200) self.assertEqual(payload.ctype, 'text/csv') self.assertIn(b'C8:9E:43:64:80:80', payload.data) status, payload = server.h_recon_archive_scan_download_html( self._ctx((self.archive_name, '1'))) self.assertEqual(status, 200) self.assertEqual(payload.ctype, 'text/html') text = payload.data.decode('utf-8') self.assertIn('archived history', text) self.assertIn('Anderson-5', text) def test_archive_html_omits_current_gps(self): # An archive is historical; attaching a current fix would mislead. with mock.patch.object(server, '_gps_status_data', return_value={'lock': True, 'lat': 37.7, 'lon': -122.4}): status, payload = server.h_recon_archive_scan_download_html( self._ctx((self.archive_name, '1'))) self.assertEqual(status, 200) self.assertNotIn('GPS:', payload.data.decode('utf-8')) def test_archive_detail_503_on_lock(self): with mock.patch.object(server, 'recon_scan_data', side_effect=RuntimeError('sqlite read failed: locked')), \ mock.patch.object(server.time, 'sleep'): status, data = server.h_recon_archive_scan_detail( self._ctx((self.archive_name, '1'))) self.assertEqual(status, 503) self.assertIn('temporarily unavailable', data['error']) class GpsTest(unittest.TestCase): def setUp(self): server._gps_cache.update({'updated': 0, 'data': None}) @unittest.skipIf(os.name == 'nt', 'symlinks are not reliably available on Windows') def test_serial_candidates_detect_bypath_targets(self): d = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, d) self.addCleanup(setattr, server, 'SERIAL_DIR', server.SERIAL_DIR) server.SERIAL_DIR = d os.symlink('/dev/ttyACM0', os.path.join(d, '1.3_1-1.3:1.0')) os.symlink('/dev/ttyACM1', os.path.join(d, '1.3_1-1.3:1.2')) with open(os.path.join(d, 'not-a-serial'), 'w') as f: f.write('x') candidates = server._gps_serial_candidates() names = [name for name, _ in candidates] self.assertEqual(names, ['1.3_1-1.3:1.0', '1.3_1-1.3:1.2']) def test_gps_status_passthrough(self): with mock.patch.object(server, '_gps_status_data_nocache', return_value={'present': True, 'wigle': True}): status, data = server.h_recon_gps(type('C', (), {'args': ()})()) self.assertEqual(status, 200) self.assertTrue(data['present']) self.assertTrue(data['wigle']) def test_gps_status_skips_hak5cmd_when_gpsd_down(self): # GPS_GET can block for seconds when gpsd is down; it must not run. with mock.patch.object(server, '_uci_gps_get', return_value=None), \ mock.patch.object(server, '_gps_serial_candidates', return_value=[]), \ mock.patch.object(server, '_wigle_config', return_value={'logwigle': False}), \ mock.patch.object(server, '_gpsd_running', return_value=False), \ mock.patch.object(server, '_gps_from_gpspipe', return_value=None), \ mock.patch.object(server, '_gps_from_hak5cmd', side_effect=AssertionError('GPS_GET must not run when gpsd is down')) as hak5: data = server._gps_status_data_nocache() self.assertFalse(data['gpsd_running']) hak5.assert_not_called() def test_gps_status_falls_back_to_hak5cmd_when_gpsd_up(self): with mock.patch.object(server, '_uci_gps_get', return_value=None), \ mock.patch.object(server, '_gps_serial_candidates', return_value=[]), \ mock.patch.object(server, '_wigle_config', return_value={'logwigle': False}), \ mock.patch.object(server, '_gpsd_running', return_value=True), \ mock.patch.object(server, '_gps_from_gpspipe', return_value=None), \ mock.patch.object(server, '_gps_from_hak5cmd', return_value={'fix': 3, 'lat': 37.7, 'lon': -122.4, 'satellites': 8}): data = server._gps_status_data_nocache() self.assertEqual(data['lat'], 37.7) self.assertEqual(data['satellites'], 8) def test_configure_binds_preferred_device_and_locks(self): candidates = [('1.2_1-1.2:1.0', '/dev/1.2'), ('1.3_2-1.3:1.0', '/dev/1.3')] with mock.patch.object(server, '_gps_serial_candidates', return_value=candidates), \ mock.patch.object(server, '_uci_gps_get', return_value='1.3_2-1.3:1.0'), \ mock.patch.object(server, '_uci_gps_set') as uci_set, \ mock.patch.object(server, '_gpsd_restart'), \ mock.patch.object(server, 'time', mock.Mock(sleep=lambda s: None)), \ mock.patch.object(server, '_gps_from_gpspipe', return_value={'fix': 3, 'lat': 37.7, 'lon': -122.4, 'satellites': 8}), \ mock.patch.object(server, '_gps_status_data_nocache', return_value={'present': True}): status, data = server.h_recon_gps_configure(type('C', (), {'args': ()})()) self.assertEqual(status, 200) self.assertTrue(data['lock']) self.assertEqual(data['tried'], ['1.3_2-1.3:1.0']) uci_set.assert_called_once_with('1.3_2-1.3:1.0') def test_configure_no_candidates_errors(self): with mock.patch.object(server, '_gps_serial_candidates', return_value=[]): status, data = server.h_recon_gps_configure(type('C', (), {'args': ()})()) self.assertEqual(status, 200) self.assertIn('error', data) def test_configure_fallback_binds_first_with_note(self): candidates = [('1.2_1-1.2:1.0', '/dev/1.2'), ('1.3_2-1.3:1.0', '/dev/1.3')] with mock.patch.object(server, '_gps_serial_candidates', return_value=candidates), \ mock.patch.object(server, '_uci_gps_get', return_value=None), \ mock.patch.object(server, '_uci_gps_set'), \ mock.patch.object(server, '_gpsd_restart'), \ mock.patch.object(server, 'time', mock.Mock(sleep=lambda s: None)), \ mock.patch.object(server, '_gps_from_gpspipe', return_value=None), \ mock.patch.object(server, '_gps_status_data_nocache', return_value={'present': True}): status, data = server.h_recon_gps_configure(type('C', (), {'args': ()})()) self.assertEqual(status, 200) self.assertEqual(data['device'], '1.2_1-1.2:1.0') self.assertIn('waiting for a fix', data['note']) def test_configure_tries_at_most_three_candidates(self): candidates = [(str(i), '/dev/%d' % i) for i in range(5)] with mock.patch.object(server, '_gps_serial_candidates', return_value=candidates), \ mock.patch.object(server, '_uci_gps_get', return_value=None), \ mock.patch.object(server, '_uci_gps_set'), \ mock.patch.object(server, '_gpsd_restart'), \ mock.patch.object(server, 'time', mock.Mock(sleep=lambda s: None)), \ mock.patch.object(server, '_gps_from_gpspipe', return_value=None), \ mock.patch.object(server, '_gps_status_data_nocache', return_value={'present': True}): status, data = server.h_recon_gps_configure(type('C', (), {'args': ()})()) self.assertEqual(status, 200) self.assertEqual(data['tried'], ['0', '1', '2']) class WigleTest(unittest.TestCase): def setUp(self): self.dir = tempfile.mkdtemp() self._orig = server.WIGLE_DIR server.WIGLE_DIR = self.dir def tearDown(self): server.WIGLE_DIR = self._orig shutil.rmtree(self.dir) def _ctx(self, args=(), body=None): return type('C', (), {'args': args, 'body': body or {}})() def _write(self, name, content): raw = content.encode('utf-8') if isinstance(content, str) else content path = os.path.join(self.dir, name) fd = os.open(path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o644) try: os.write(fd, raw) finally: os.close(fd) def test_file_rows_count_excludes_header(self): payload = b'header\nr1\nr2\n' self._write('a.csv', payload) self._write('b.csv', b'onlyheader\n') status, data = server.h_recon_wigle_files(self._ctx()) self.assertEqual(status, 200) files = {f['name']: f for f in data['files']} self.assertEqual(files['a.csv']['rows'], 2) self.assertEqual(files['b.csv']['rows'], 0) self.assertEqual(files['a.csv']['size'], os.path.getsize(os.path.join(self.dir, 'a.csv'))) def test_file_rows_count_ignores_wigle_meta_and_header(self): meta = 'WigleWifi-1.6,appRelease=0.0.0,model=pineapplepager,release=0.0.0\n' header = 'MAC,SSID,AuthMode,FirstSeen,Channel,Frequency,RSSI,CurrentLatitude,CurrentLongitude\n' self._write('empty.csv', meta + header) self._write('full.csv', meta + header + 'AA:BB:CC:DD:EE:FF,test,0,,1,2412,-60,37.7,-122.4\n') status, data = server.h_recon_wigle_files(self._ctx()) files = {f['name']: f for f in data['files']} self.assertEqual(files['empty.csv']['rows'], 0) self.assertEqual(files['full.csv']['rows'], 1) def test_file_download(self): self._write('wigle-1.csv', 'lat,lon\n37.7,-122.4\n') status, payload = server.h_recon_wigle_file(self._ctx(('wigle-1.csv',))) self.assertEqual(status, 200) self.assertEqual(payload.filename, 'wigle-1.csv') self.assertIn(b'37.7', payload.data) def test_file_download_404_and_traversal(self): status, payload = server.h_recon_wigle_file(self._ctx(('missing.csv',))) self.assertEqual(status, 404) status, payload = server.h_recon_wigle_file(self._ctx(('..%2F..%2Fetc%2Fpasswd',))) self.assertEqual(status, 404) def test_toggle_enable_and_disable(self): with mock.patch.object(server, '_wigle_set', return_value=(200, {'ok': True})), \ mock.patch.object(server, 'hak5') as hak5, \ mock.patch.object(server, 'wigle_files_data', return_value={'files': [{'name': 'w.csv'}]}): status, data = server.h_recon_wigle(self._ctx(body={'enable': True})) self.assertEqual(status, 200) self.assertTrue(data['wigle']) self.assertEqual(data['filename'], 'w.csv') hak5.assert_called_once_with('WIGLE_START', timeout=10) status, data = server.h_recon_wigle(self._ctx(body={'enable': False})) self.assertEqual(status, 200) self.assertFalse(data['wigle']) hak5.assert_called_with('WIGLE_STOP', timeout=10) class ReconRoutesTest(unittest.TestCase): def test_new_routes_registered(self): expected = [ ('GET', '/api/recon/scans/1/download/csv', 'h_recon_scan_download_csv'), ('GET', '/api/recon/scans/1/download/html', 'h_recon_scan_download_html'), ('GET', '/api/recon/gps', 'h_recon_gps'), ('POST', '/api/recon/gps/configure', 'h_recon_gps_configure'), ('POST', '/api/recon/wigle', 'h_recon_wigle'), ('GET', '/api/recon/wigle/files', 'h_recon_wigle_files'), ('GET', '/api/recon/wigle/files/x.csv', 'h_recon_wigle_file'), ('GET', '/api/recon/archives', 'h_recon_archives'), ('GET', '/api/recon/archives/error-2026-08-18-19:21:42Z-recon.db/scans', 'h_recon_archive_scans'), ('GET', '/api/recon/archives/error-2026-08-18-19:21:42Z-recon.db/scans/1', 'h_recon_archive_scan_detail'), ('GET', '/api/recon/archives/error-2026-08-18-19:21:42Z-recon.db/scans/1/download/json', 'h_recon_archive_scan_download'), ('GET', '/api/recon/archives/error-2026-08-18-19:21:42Z-recon.db/scans/1/download/csv', 'h_recon_archive_scan_download_csv'), ('GET', '/api/recon/archives/error-2026-08-18-19:21:42Z-recon.db/scans/1/download/html', 'h_recon_archive_scan_download_html'), ] for method, path, handler in expected: h, args = server.ROUTER.dispatch(method, path) self.assertIsNotNone(h, '%s %s' % (method, path)) self.assertEqual(h.__name__, handler, '%s %s' % (method, path)) def test_survey_routes_are_gone(self): for method, path in [('GET', '/api/recon/survey/live'), ('POST', '/api/recon/survey/start'), ('GET', '/api/recon/surveys'), ('GET', '/api/recon/surveys/abc'), ('DELETE', '/api/recon/surveys/abc')]: h, args = server.ROUTER.dispatch(method, path) self.assertIsNone(h, '%s %s should not be registered' % (method, path)) def test_original_recon_routes_unchanged(self): for path in ['/api/recon/start', '/api/recon/status', '/api/recon/scans', '/api/recon/events']: method = 'GET' if path.endswith(('status', 'scans', 'events')) else 'POST' h, args = server.ROUTER.dispatch(method, path) self.assertIsNotNone(h, path)