feat: include recon identity in exports

This commit is contained in:
2026-08-21 08:31:47 -05:00
parent a67faecee9
commit 21e308386f
2 changed files with 88 additions and 8 deletions
@@ -2040,14 +2040,50 @@ def _csv_escape(v):
return v return v
def _identity_label(identity):
identity = identity or {}
manufacturer = identity.get('manufacturer') or 'Unknown'
model = identity.get('model')
if model:
return '%s %s' % (manufacturer, model)
if manufacturer == 'Unknown' and identity.get('oui'):
return 'Unknown (%s)' % identity['oui']
return manufacturer
def _association_label(association):
parts = [association.get('ssid') or '(hidden)']
if association.get('bssid'):
parts.append(association['bssid'])
if association.get('sources'):
parts.append('[' + ', '.join(association['sources']) + ']')
return ' '.join(parts)
def _aps_csv(data): def _aps_csv(data):
out = ['bssid,ssid,hidden,band,channel,freq,encryption,signal,vendor,first_seen,last_seen'] out = ['bssid,ssid,hidden,band,channel,freq,encryption,signal,vendor,first_seen,last_seen,Device Identity,Client Count,Confirmed SSIDs']
for a in (data or {}).get('aps') or []: for a in (data or {}).get('aps') or []:
confirmed = []
for client in a.get('clients') or []:
for association in client.get('associations') or []:
if association.get('bssid') == a.get('bssid'):
confirmed.append(association.get('ssid') or '(hidden)')
out.append(','.join(_csv_escape(x) for x in [ out.append(','.join(_csv_escape(x) for x in [
a.get('bssid'), a.get('ssid'), int(bool(a.get('hidden'))), a.get('bssid'), a.get('ssid'), int(bool(a.get('hidden'))),
a.get('band'), a.get('channel'), a.get('freq'), a.get('band'), a.get('channel'), a.get('freq'),
a.get('encryption'), a.get('signal'), a.get('vendor'), a.get('encryption'), a.get('signal'), a.get('vendor'),
_fmt_ts(a.get('first_seen')), _fmt_ts(a.get('last_seen'))])) _fmt_ts(a.get('first_seen')), _fmt_ts(a.get('last_seen')),
_identity_label(a.get('device_identity')), a.get('client_count', 0),
'; '.join(sorted(set(confirmed)))]))
clients = (data or {}).get('clients') or []
if clients:
out.append('client_mac,client_identity,confirmed_associations')
for client in clients:
associations = [_association_label(a)
for a in client.get('associations') or []]
out.append(','.join(_csv_escape(x) for x in [
client.get('mac'), _identity_label(client.get('vendor')),
'; '.join(associations)]))
out.append('unassociated,%d' % ((data or {}).get('unassociated') or 0)) out.append('unassociated,%d' % ((data or {}).get('unassociated') or 0))
return '\r\n'.join(out) + '\r\n' return '\r\n'.join(out) + '\r\n'
@@ -2249,14 +2285,30 @@ def _recon_html_download(scan_id, data, client_count, archive=None):
ap_rows.append([a.get('ssid') or '(hidden)', a.get('bssid'), ap_rows.append([a.get('ssid') or '(hidden)', a.get('bssid'),
a.get('band') or '--', a.get('band') or '--',
a.get('channel') if a.get('channel') is not None else '--', a.get('channel') if a.get('channel') is not None else '--',
_signal_html(a.get('signal')), _signal_html(a.get('signal')),
a.get('encryption') or '--', a.get('encryption') or '--',
a.get('vendor') or '--', _identity_label(a.get('device_identity')),
_fmt_ts(a.get('first_seen')), _fmt_ts(a.get('last_seen'))]) a.get('client_count', 0),
_fmt_ts(a.get('first_seen')), _fmt_ts(a.get('last_seen'))])
body_parts.append('<h2>Access Points</h2>') body_parts.append('<h2>Access Points</h2>')
body_parts.append(_html_table(['SSID', 'BSSID', 'Band', 'Ch', 'Signal', body_parts.append(_html_table(['SSID', 'BSSID', 'Band', 'Ch', 'Signal',
'Encryption', 'Vendor', 'First seen', 'Last seen'], 'Encryption', 'Device Identity', 'Clients',
ap_rows, raw_columns=(4,))) 'First seen', 'Last seen'],
ap_rows, raw_columns=(4,)))
confirmed_rows = []
for client in data.get('clients') or []:
for association in client.get('associations') or []:
confirmed_rows.append([
client.get('mac'), _identity_label(client.get('vendor')),
association.get('ssid') or '(hidden)', association.get('bssid') or '--',
', '.join(association.get('sources') or [])])
body_parts.append('<h2>Confirmed Clients</h2>')
if confirmed_rows:
body_parts.append(_html_table(
['Client MAC', 'Device Identity', 'SSID', 'BSSID', 'Evidence'],
confirmed_rows))
else:
body_parts.append('<p class="empty">No confirmed clients.</p>')
subtitle = ('Pager recon capture report (archived history)' subtitle = ('Pager recon capture report (archived history)'
if archive else 'Pager recon capture report') if archive else 'Pager recon capture report')
return Download(_html_doc('Scan #%d' % scan.get('id'), return Download(_html_doc('Scan #%d' % scan.get('id'),
+28
View File
@@ -1,3 +1,4 @@
import json
import os import os
import shutil import shutil
import sqlite3 import sqlite3
@@ -1109,6 +1110,23 @@ class ReconReportTest(unittest.TestCase):
self.assertIn('unassociated,1', text) self.assertIn('unassociated,1', text)
self.assertIn('C8:9E:43:64:80:80', 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): def test_html_download_contains_stats(self):
with mock.patch.object(server, '_gps_status_data', return_value={'lock': False}): with mock.patch.object(server, '_gps_status_data', return_value={'lock': False}):
status, payload = server.h_recon_scan_download_html(self._ctx(('1',))) status, payload = server.h_recon_scan_download_html(self._ctx(('1',)))
@@ -1137,6 +1155,16 @@ class ReconReportTest(unittest.TestCase):
# no GPS line without a fix # no GPS line without a fix
self.assertNotIn('GPS:', text) 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_report_includes_gps_when_locked(self): def test_html_report_includes_gps_when_locked(self):
with mock.patch.object(server, '_gps_status_data', with mock.patch.object(server, '_gps_status_data',
return_value={'lock': True, 'lat': 37.7, return_value={'lock': True, 'lat': 37.7,