1210 lines
55 KiB
Python
1210 lines
55 KiB
Python
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 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'], 2)
|
|
self.assertEqual(old['aps'], 2)
|
|
self.assertEqual(old['handshakes'], 1)
|
|
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']), 2)
|
|
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'])
|
|
|
|
def test_scan_detail_handshakes_resolve_macs(self):
|
|
data = server.recon_scan_data(1)
|
|
self.assertEqual(len(data['handshakes']), 1)
|
|
hs = data['handshakes'][0]
|
|
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_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 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}
|
|
|
|
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}))
|
|
|
|
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'], {'error': 'no radio'})
|
|
self.assertFalse(server._recon_scan_state['active'])
|
|
|
|
|
|
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}
|
|
|
|
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'], 1786466532)
|
|
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.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', 'wlan2mon']):
|
|
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'], 1786466601)
|
|
|
|
|
|
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'))
|
|
|
|
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'))
|
|
|
|
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 <db> <sql>` 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_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']), 2)
|
|
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_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_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 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):
|
|
with open(os.path.join(self.dir, name), 'w') as f:
|
|
f.write(content)
|
|
|
|
def test_file_rows_count_excludes_header(self):
|
|
self._write('a.csv', 'header\nr1\nr2\n')
|
|
self._write('b.csv', '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'], len('header\nr1\nr2\n'))
|
|
|
|
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)
|