Enhance SSID metrics extraction by adding client tracking and updating function signatures

This commit is contained in:
Yaro Kasear 2025-04-28 11:52:45 -05:00
parent dc9da2e2e3
commit 5ea493b3f8
4 changed files with 63 additions and 49 deletions

View file

@ -65,7 +65,8 @@ def analyze_pcap(pcapng_path, start_ts, end_ts, ap_bssid, ap_channel):
ssid_signals,
cisco_ssid_clients,
cisco_reported_clients,
ssid_packet_counts
ssid_packet_counts,
ssid_clients
) = extract_ssid_metrics(filtered_packets)
our_ssid = bssid_to_ssid.get(ap_bssid, None)

View file

@ -1,4 +1,3 @@
from collections import defaultdict
from statistics import mean
def merge_ssid_summaries(summary_lists):

View file

@ -2,7 +2,7 @@ from collections import defaultdict
from enrichment.utils import get_channel_from_freq
def get_clients_on_ap(capture, ap_bssid):
clients = defaultdict(list)
clients = defaultdict(int)
ap_bssid = ap_bssid.lower()
for packet in capture:

View file

@ -3,6 +3,7 @@ from collections import defaultdict
from enrichment.utils import get_channel_from_freq
def extract_ssid_metrics(packets):
ssid_clients = defaultdict(set)
bssid_to_ssid = {}
ssid_to_bssids = defaultdict(set)
ssid_hidden_status = {}
@ -24,63 +25,75 @@ def extract_ssid_metrics(packets):
continue
packet_freq = int(radio.channel.freq)
get_channel_from_freq(packet_freq) # validate channel, or skip
get_channel_from_freq(packet_freq) # Validate channel or skip
subtype = int(getattr(wlan, 'type_subtype', 0), 16)
if subtype not in (5, 8): # Beacon or Probe Response
continue
try:
mgt = packet.get_multiple_layers('wlan.mgt')[0]
tags = mgt._all_fields.get('wlan.tagged.all', {}).get('wlan.tag', [])
except Exception:
continue
# --- EXTRACT SSID from Management Frames ---
if subtype in (5, 8): # Beacon or Probe Response
try:
mgt = packet.get_multiple_layers('wlan.mgt')[0]
tags = mgt._all_fields.get('wlan.tagged.all', {}).get('wlan.tag', [])
except Exception:
continue
ssid = None
hidden_ssid = False
ssid = None
hidden_ssid = False
privacy_bit = mgt._all_fields.get('wlan_mgt.fixed.capabilities.privacy')
is_open = (str(privacy_bit) != '1')
privacy_bit = mgt._all_fields.get('wlan_mgt.fixed.capabilities.privacy')
is_open = (str(privacy_bit) != '1')
for tag in tags:
tag_number = tag.get('wlan.tag.number')
if tag_number == '0':
raw_ssid = tag.get('wlan.ssid', '')
if not raw_ssid:
hidden_ssid = True
ssid = '<hidden>'
else:
for tag in tags:
tag_number = tag.get('wlan.tag.number')
if tag_number == '0':
raw_ssid = tag.get('wlan.ssid', '')
if not raw_ssid:
hidden_ssid = True
ssid = '<hidden>'
else:
try:
ssid_bytes = bytes.fromhex(raw_ssid.replace(':', ''))
ssid = ssid_bytes.decode('utf-8', errors='replace')
except Exception:
ssid = None
if tag_number == '133':
try:
ssid_bytes = bytes.fromhex(raw_ssid.replace(':', ''))
ssid = ssid_bytes.decode('utf-8', errors='replace')
except Exception:
ssid = None
if tag_number == '133':
try:
num_clients = int(tag.get('wlan.cisco.ccx1.clients'))
if ssid:
cisco_ssid_clients[ssid].append(num_clients)
cisco_reported_clients.append(num_clients)
except (TypeError, ValueError):
pass
num_clients = int(tag.get('wlan.cisco.ccx1.clients'))
if ssid:
cisco_ssid_clients[ssid].append(num_clients)
cisco_reported_clients.append(num_clients)
except (TypeError, ValueError):
pass
if not ssid:
continue
if not ssid:
continue
ssid_hidden_status[ssid] = hidden_ssid
ssid_encryption_status.setdefault(ssid, is_open)
ssid_packet_counts[ssid] += 1
ssid_hidden_status[ssid] = hidden_ssid
ssid_encryption_status.setdefault(ssid, is_open)
ssid_packet_counts[ssid] += 1
bssid = getattr(wlan, 'bssid', '').lower()
if not bssid or bssid == 'ff:ff:ff:ff:ff:ff':
continue
bssid_to_ssid[bssid] = ssid
ssid_to_bssids[ssid].add(bssid)
signal = getattr(radio, 'dbm_antsignal', None)
if signal:
ssid_signals[ssid].append(int(signal))
# --- CAPTURE CLIENTS on any frames ---
# This runs even on non-beacon frames
bssid = getattr(wlan, 'bssid', '').lower()
if not bssid or bssid == 'ff:ff:ff:ff:ff:ff':
continue
sa = getattr(wlan, 'sa', '').lower()
da = getattr(wlan, 'da', '').lower()
bssid_to_ssid[bssid] = ssid
ssid_to_bssids[ssid].add(bssid)
signal = getattr(radio, 'dbm_antsignal', None)
if signal:
ssid_signals[ssid].append(int(signal))
if bssid in bssid_to_ssid:
ssid = bssid_to_ssid[bssid]
for mac in (sa, da):
if mac and mac != "ff:ff:ff:ff:ff:ff" and mac != bssid:
ssid_clients[ssid].add(mac)
except Exception:
continue
@ -93,5 +106,6 @@ def extract_ssid_metrics(packets):
ssid_signals,
cisco_ssid_clients,
cisco_reported_clients,
ssid_packet_counts
ssid_packet_counts,
ssid_clients
)