From 21e308386f03bb5a159d21a54b94ed97b007b0fa Mon Sep 17 00:00:00 2001 From: c4ch3c4d3 Date: Fri, 21 Aug 2026 08:31:47 -0500 Subject: [PATCH] feat: include recon identity in exports --- .../user/remote_access/pager-webui/server.py | 68 ++++++++++++++++--- tests/test_recon.py | 28 ++++++++ 2 files changed, 88 insertions(+), 8 deletions(-) diff --git a/payload/user/remote_access/pager-webui/server.py b/payload/user/remote_access/pager-webui/server.py index f9dcf8c..7d6c991 100644 --- a/payload/user/remote_access/pager-webui/server.py +++ b/payload/user/remote_access/pager-webui/server.py @@ -2040,14 +2040,50 @@ def _csv_escape(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): - 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 []: + 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 [ a.get('bssid'), a.get('ssid'), int(bool(a.get('hidden'))), a.get('band'), a.get('channel'), a.get('freq'), 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)) 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'), a.get('band') or '--', a.get('channel') if a.get('channel') is not None else '--', - _signal_html(a.get('signal')), - a.get('encryption') or '--', - a.get('vendor') or '--', - _fmt_ts(a.get('first_seen')), _fmt_ts(a.get('last_seen'))]) + _signal_html(a.get('signal')), + a.get('encryption') or '--', + _identity_label(a.get('device_identity')), + a.get('client_count', 0), + _fmt_ts(a.get('first_seen')), _fmt_ts(a.get('last_seen'))]) body_parts.append('

Access Points

') body_parts.append(_html_table(['SSID', 'BSSID', 'Band', 'Ch', 'Signal', - 'Encryption', 'Vendor', 'First seen', 'Last seen'], - ap_rows, raw_columns=(4,))) + 'Encryption', 'Device Identity', 'Clients', + '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('

Confirmed Clients

') + if confirmed_rows: + body_parts.append(_html_table( + ['Client MAC', 'Device Identity', 'SSID', 'BSSID', 'Evidence'], + confirmed_rows)) + else: + body_parts.append('

No confirmed clients.

') subtitle = ('Pager recon capture report (archived history)' if archive else 'Pager recon capture report') return Download(_html_doc('Scan #%d' % scan.get('id'), diff --git a/tests/test_recon.py b/tests/test_recon.py index 1b811a7..dd9c612 100644 --- a/tests/test_recon.py +++ b/tests/test_recon.py @@ -1,3 +1,4 @@ +import json import os import shutil import sqlite3 @@ -1109,6 +1110,23 @@ class ReconReportTest(unittest.TestCase): 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',))) @@ -1137,6 +1155,16 @@ class ReconReportTest(unittest.TestCase): # 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_report_includes_gps_when_locked(self): with mock.patch.object(server, '_gps_status_data', return_value={'lock': True, 'lat': 37.7,