bzuccaro/Mark-VIII
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: associate recon clients with confirmed networks

Browse Source
This commit is contained in:
bzuccaro
2026-08-21 08:26:58 -05:00
parent 2c9fa8d137
commit a67faecee9
3 changed files with 149 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_recon.py
+59 -11
View File
@@ -39,8 +39,22 @@ def make_db():
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)")
"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()
@@ -101,16 +115,16 @@ class ReconDataTest(unittest.TestCase):
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.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']), 2)
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')
@@ -125,16 +139,50 @@ class ReconDataTest(unittest.TestCase):
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'])
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']), 1)
hs = data['handshakes'][0]
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))
@@ -525,7 +573,7 @@ class ReconExtrasTest(unittest.TestCase):
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.assertEqual(data['last_activity'], 1786466535)
self.assertTrue(data['active'])
server.time.time = lambda: 1786466532 + 1000
status, data = server.h_recon_status(type('C', (), {'args': ()})())
@@ -580,7 +628,7 @@ class ReconExtrasTest(unittest.TestCase):
self.assertEqual(status, 200)
kinds = [e['type'] for e in data['events']]
self.assertIn('auth attempt', kinds)
self.assertEqual(data['events'][0]['time'], 1786466601)
self.assertEqual(data['events'][0]['time'], 1786466602)
class ReconExamineTest(unittest.TestCase):
@@ -988,7 +1036,7 @@ class ReconEnrichmentTest(unittest.TestCase):
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.assertEqual(len(data['aps']), 3)
self.assertLessEqual(len(data['clients']), 1)
self.assertEqual(data['scan']['id'], 1)