ap_scanner/parser.py (18329B)
1 """Parse 802.11 Beacon and Probe Response frames into AccessPoint objects."""
2
3 from datetime import datetime
4
5 from scapy.layers.dot11 import (
6 Dot11,
7 Dot11Beacon,
8 Dot11Elt,
9 Dot11ProbeReq,
10 Dot11ProbeResp,
11 RadioTap,
12 )
13
14 from ap_scanner.models import AccessPoint
15 from ap_scanner.oui import lookup_vendor
16 from ap_scanner.utils import CHANNELS_5G, channel_band, freq_to_channel
17
18
19 def parse_packet(pkt) -> AccessPoint | None:
20 """Parse a Scapy packet into an AccessPoint. Returns None if not a beacon/probe response."""
21 if not (pkt.haslayer(Dot11Beacon) or pkt.haslayer(Dot11ProbeResp)):
22 return None
23
24 try:
25 bssid = pkt[Dot11].addr3
26 if not bssid:
27 return None
28 bssid = bssid.upper()
29 except (IndexError, AttributeError):
30 return None
31
32 # Collect all Information Elements in a single pass
33 ies = _collect_ies(pkt)
34
35 # SSID
36 ssid = _extract_ssid(ies)
37
38 # Channel and frequency
39 channel, frequency = _extract_channel_and_freq(pkt, ies)
40 if channel == 0:
41 return None
42
43 # RSSI
44 rssi = _extract_rssi(pkt)
45
46 # Channel width and bonded channels
47 width, secondary_channels = _extract_channel_width(ies, channel)
48 channel_desc = _build_channel_desc(channel, secondary_channels)
49
50 # Security: encryption, cipher, auth + per-protocol detail
51 security = _extract_security(pkt, ies)
52
53 # Band (use frequency to disambiguate 6GHz from 2.4/5GHz)
54 band = channel_band(channel, frequency) + "GHz"
55
56 # WiFi standards (all supported, comma-separated)
57 wifi_standard = _detect_all_standards(ies, channel)
58
59 # Vendor from OUI
60 vendor = lookup_vendor(bssid)
61
62 # WPS
63 wps = _extract_wps(ies)
64
65 now = datetime.now()
66 return AccessPoint(
67 bssid=bssid,
68 ssid=ssid,
69 channel=channel,
70 rssi=rssi,
71 encryption=security["encryption"],
72 cipher=security["cipher"],
73 auth=security["auth"],
74 band=band,
75 wifi_standard=wifi_standard,
76 width=width,
77 channel_desc=channel_desc,
78 vendor=vendor,
79 wps=wps,
80 wep=security["wep"],
81 wpa=security["wpa"],
82 wpa2=security["wpa2"],
83 wpa3=security["wpa3"],
84 first_seen=now,
85 last_seen=now,
86 beacon_count=1,
87 frequency=frequency,
88 )
89
90
91 def parse_probe_request(pkt) -> tuple[str, str] | None:
92 """Parse a Probe Request frame to extract SSID and source MAC.
93
94 Clients looking for hidden networks send Probe Requests with the target SSID.
95 By capturing these, we can map SSIDs to hidden APs when the client is also
96 seen communicating with that AP.
97
98 Returns:
99 (ssid, client_mac) or None if not a probe request or SSID is empty/broadcast.
100 """
101 if not pkt.haslayer(Dot11ProbeReq):
102 return None
103
104 try:
105 client_mac = pkt[Dot11].addr2
106 if not client_mac:
107 return None
108 client_mac = client_mac.upper()
109 except (IndexError, AttributeError):
110 return None
111
112 # Extract SSID from IE
113 elt = pkt.getlayer(Dot11Elt)
114 while elt:
115 if elt.ID == 0 and elt.info:
116 ssid = elt.info.decode("utf-8", errors="replace").strip("\x00").strip()
117 if ssid: # Skip broadcast probes (empty SSID)
118 return ssid, client_mac
119 return None
120 elt = elt.payload.getlayer(Dot11Elt) if elt.payload else None
121
122 return None
123
124
125 # ---------------------------------------------------------------------------
126 # IE collection
127 # ---------------------------------------------------------------------------
128
129 def _collect_ies(pkt) -> dict[int, list[bytes]]:
130 """Walk all Dot11Elt layers and group by IE ID. Returns {id: [info_bytes, ...]}."""
131 ies: dict[int, list[bytes]] = {}
132 elt = pkt.getlayer(Dot11Elt)
133 while elt:
134 ie_id = elt.ID
135 info = elt.info if elt.info else b""
136 ies.setdefault(ie_id, []).append(info)
137 elt = elt.payload.getlayer(Dot11Elt) if elt.payload else None
138 return ies
139
140
141 # ---------------------------------------------------------------------------
142 # Field extraction
143 # ---------------------------------------------------------------------------
144
145 def _extract_ssid(ies: dict) -> str:
146 """Extract SSID from IE ID=0. Capped at 32 bytes per 802.11 spec."""
147 for info in ies.get(0, []):
148 try:
149 return info[:32].decode("utf-8", errors="replace")
150 except (UnicodeDecodeError, AttributeError):
151 return ""
152 return ""
153
154
155 def _extract_channel_and_freq(pkt, ies: dict) -> tuple[int, int]:
156 """Extract primary channel and frequency."""
157 channel = 0
158 frequency = 0
159
160 # DS Parameter Set (IE 3)
161 for info in ies.get(3, []):
162 if info and len(info) >= 1:
163 channel = info[0]
164 break
165
166 # RadioTap frequency
167 if pkt.haslayer(RadioTap):
168 rt = pkt[RadioTap]
169 if hasattr(rt, "ChannelFrequency") and rt.ChannelFrequency:
170 frequency = rt.ChannelFrequency
171 if channel == 0:
172 channel = freq_to_channel(frequency)
173
174 return channel, frequency
175
176
177 def _extract_rssi(pkt) -> int | None:
178 """Extract RSSI (dBm) from RadioTap header."""
179 if pkt.haslayer(RadioTap):
180 rt = pkt[RadioTap]
181 if hasattr(rt, "dBm_AntSignal") and rt.dBm_AntSignal is not None:
182 return rt.dBm_AntSignal
183 return None
184
185
186 # ---------------------------------------------------------------------------
187 # Channel width and bonded channels
188 # ---------------------------------------------------------------------------
189
190 # 5 GHz channel groupings for bonded channel calculation
191 _5G_40MHZ_PAIRS = [
192 (36, 40), (44, 48), (52, 56), (60, 64),
193 (100, 104), (108, 112), (116, 120), (124, 128),
194 (132, 136), (140, 144), (149, 153), (157, 161),
195 ]
196 _5G_80MHZ_GROUPS = [
197 (36, 40, 44, 48), (52, 56, 60, 64),
198 (100, 104, 108, 112), (116, 120, 124, 128),
199 (132, 136, 140, 144), (149, 153, 157, 161),
200 ]
201 _5G_160MHZ_GROUPS = [
202 (36, 40, 44, 48, 52, 56, 60, 64),
203 (100, 104, 108, 112, 116, 120, 124, 128),
204 (132, 136, 140, 144, 149, 153, 157, 161),
205 ]
206
207 # 2.4 GHz secondary channel offsets
208 def _2g_40mhz_channels(primary: int, above: bool) -> list[int]:
209 """Get secondary channel for 2.4GHz 40MHz."""
210 if above:
211 secondary = primary + 4
212 else:
213 secondary = primary - 4
214 if 1 <= secondary <= 14:
215 return [secondary]
216 return []
217
218
219 def _extract_channel_width(ies: dict, primary_channel: int) -> tuple[int, list[int]]:
220 """Extract channel width and secondary channel list from HT/VHT/HE Operation IEs.
221
222 Returns:
223 (width_mhz, [secondary_channels])
224 """
225 width = 20
226 secondary_channels: list[int] = []
227 is_5ghz = primary_channel in CHANNELS_5G
228
229 # HT Operation (IE 61): determines 20 vs 40 MHz
230 ht_secondary_offset = 0
231 for info in ies.get(61, []):
232 if len(info) >= 2:
233 # byte 0: primary channel
234 # byte 1, bits 0-1: secondary channel offset (0=none, 1=above, 3=below)
235 ht_secondary_offset = info[1] & 0x03
236 if ht_secondary_offset in (1, 3):
237 width = 40
238 above = (ht_secondary_offset == 1)
239 if is_5ghz:
240 # Find the 40MHz pair containing this channel
241 for pair in _5G_40MHZ_PAIRS:
242 if primary_channel in pair:
243 secondary_channels = [ch for ch in pair if ch != primary_channel]
244 break
245 else:
246 secondary_channels = _2g_40mhz_channels(primary_channel, above)
247 break
248
249 # VHT Operation (IE 192): determines 80/160 MHz
250 for info in ies.get(192, []):
251 if len(info) >= 1:
252 vht_width = info[0]
253 if vht_width == 1: # 80 MHz
254 width = 80
255 if is_5ghz:
256 for group in _5G_80MHZ_GROUPS:
257 if primary_channel in group:
258 secondary_channels = [ch for ch in group if ch != primary_channel]
259 break
260 elif vht_width in (2, 3): # 160 MHz or 80+80
261 width = 160
262 if is_5ghz:
263 for group in _5G_160MHZ_GROUPS:
264 if primary_channel in group:
265 secondary_channels = [ch for ch in group if ch != primary_channel]
266 break
267 break
268
269 # HE Operation (IE 255, ext ID=36): may indicate wider bandwidth
270 for info in ies.get(255, []):
271 if len(info) > 0 and info[0] == 36 and len(info) >= 7:
272 # HE Operation extension - check channel width
273 # byte 4 bits: channel width (bit 1 = 40MHz in 2.4, bit 2 = 40/80 in 5, bit 3 = 160 in 5)
274 he_params = info[1] if len(info) > 1 else 0
275 # This is complex; we rely on VHT/HT info primarily and only override if HE says wider
276 if len(info) >= 7:
277 he_ch_width = info[5] # Control channel width
278 if he_ch_width == 3 and width < 160:
279 width = 160
280 if is_5ghz:
281 for group in _5G_160MHZ_GROUPS:
282 if primary_channel in group:
283 secondary_channels = [ch for ch in group if ch != primary_channel]
284 break
285 break
286
287 return width, secondary_channels
288
289
290 def _build_channel_desc(primary: int, secondary_channels: list[int]) -> str:
291 """Build channel description string like '36+40+44+48'."""
292 all_channels = sorted([primary] + secondary_channels)
293 return "+".join(str(ch) for ch in all_channels)
294
295
296 # ---------------------------------------------------------------------------
297 # Security extraction
298 # ---------------------------------------------------------------------------
299
300 def _extract_security(pkt, ies: dict) -> dict[str, str]:
301 """Extract all security information.
302
303 Returns dict with keys: encryption, cipher, auth, wep, wpa, wpa2, wpa3
304 """
305 result = {
306 "encryption": "OPN",
307 "cipher": "--",
308 "auth": "--",
309 "wep": "",
310 "wpa": "",
311 "wpa2": "",
312 "wpa3": "",
313 }
314
315 # Check privacy bit (bit 4 = 0x0010 in 802.11 Capability Info)
316 # Direct bit check instead of sprintf() which is ~10x slower
317 has_privacy = False
318 if pkt.haslayer(Dot11Beacon):
319 has_privacy = bool(pkt[Dot11Beacon].cap & 0x0010)
320 elif pkt.haslayer(Dot11ProbeResp):
321 has_privacy = bool(pkt[Dot11ProbeResp].cap & 0x0010)
322
323 # Parse RSN IE (ID=48) -> WPA2/WPA3
324 rsn_info = None
325 for info in ies.get(48, []):
326 rsn_info = _parse_rsn_wpa_ie(info, is_rsn=True)
327 break
328
329 # Parse WPA IE (Vendor IE ID=221, OUI 00:50:F2, type 1)
330 wpa_info = None
331 for info in ies.get(221, []):
332 if info[:4] == b"\x00\x50\xf2\x01":
333 wpa_info = _parse_rsn_wpa_ie(info, is_rsn=False)
334 break
335
336 # Build per-protocol detail strings (AUTH-CIPHER format, matching wifi-inspector)
337 if rsn_info:
338 rsn_auth = rsn_info["auth"]
339 rsn_cipher = rsn_info["cipher"]
340 auth_cipher = f"{rsn_auth}-{rsn_cipher}" if rsn_cipher != "--" else rsn_auth
341
342 if "SAE" in rsn_auth:
343 result["wpa3"] = auth_cipher
344 if "PSK" in rsn_auth:
345 # Transition mode: WPA2 PSK + WPA3 SAE
346 result["wpa2"] = f"PSK-{rsn_cipher}"
347 result["wpa3"] = f"SAE-{rsn_cipher}"
348 result["encryption"] = "WPA2/WPA3"
349 else:
350 result["encryption"] = "WPA3"
351 else:
352 result["wpa2"] = auth_cipher
353 result["encryption"] = "WPA2"
354
355 result["cipher"] = rsn_cipher
356 result["auth"] = rsn_auth
357
358 if wpa_info:
359 wpa_auth = wpa_info["auth"]
360 wpa_cipher = wpa_info["cipher"]
361 auth_cipher = f"{wpa_auth}-{wpa_cipher}" if wpa_cipher != "--" else wpa_auth
362 result["wpa"] = auth_cipher
363
364 if not rsn_info:
365 result["encryption"] = "WPA"
366 result["cipher"] = wpa_cipher
367 result["auth"] = wpa_auth
368
369 if not rsn_info and not wpa_info:
370 if has_privacy:
371 result["encryption"] = "WEP"
372 result["wep"] = "WEP"
373 result["cipher"] = "WEP"
374 else:
375 result["encryption"] = "OPN"
376
377 return result
378
379
380 def _parse_rsn_wpa_ie(data: bytes, is_rsn: bool) -> dict[str, str]:
381 """Parse RSN (WPA2/WPA3) or WPA Information Element."""
382 result = {"cipher": "--", "auth": "--"}
383 try:
384 offset = 2 if is_rsn else 6 # RSN: skip version; WPA: skip OUI+type+version
385
386 # Group cipher suite (4 bytes)
387 offset += 4
388
389 # Pairwise cipher suites
390 if offset + 2 > len(data):
391 return result
392 pw_count = int.from_bytes(data[offset:offset + 2], "little")
393 offset += 2
394
395 ciphers = set()
396 for _ in range(pw_count):
397 if offset + 4 > len(data):
398 break
399 ciphers.add(_cipher_suite_name(data[offset + 3]))
400 offset += 4
401
402 result["cipher"] = "/".join(sorted(ciphers)) if ciphers else "--"
403
404 # AKM suites
405 if offset + 2 > len(data):
406 return result
407 akm_count = int.from_bytes(data[offset:offset + 2], "little")
408 offset += 2
409
410 auths = set()
411 for _ in range(akm_count):
412 if offset + 4 > len(data):
413 break
414 auths.add(_akm_suite_name(data[offset + 3]))
415 offset += 4
416
417 result["auth"] = "/".join(sorted(auths)) if auths else "--"
418
419 except (IndexError, ValueError):
420 pass
421
422 return result
423
424
425 def _cipher_suite_name(cipher_type: int) -> str:
426 return {
427 0: "GROUP", 1: "WEP40", 2: "TKIP", 3: "RESERVED",
428 4: "CCMP", 5: "WEP104", 6: "BIP-CMAC",
429 8: "GCMP", 9: "GCMP-256", 10: "CCMP-256",
430 }.get(cipher_type, f"CIPHER({cipher_type})")
431
432
433 def _akm_suite_name(akm_type: int) -> str:
434 return {
435 1: "MGT", 2: "PSK", 3: "FT-MGT", 4: "FT-PSK",
436 6: "MGT-SHA256", 8: "SAE", 9: "FT-SAE",
437 12: "OWE", 18: "OWE",
438 }.get(akm_type, f"AKM({akm_type})")
439
440
441 # ---------------------------------------------------------------------------
442 # WiFi standard detection
443 # ---------------------------------------------------------------------------
444
445 def _detect_all_standards(ies: dict, channel: int) -> str:
446 """Detect all supported WiFi standards from IEs.
447
448 Returns comma-separated string like "b, g, n" or "a, n, ac, ax".
449 """
450 standards: list[str] = []
451 is_5ghz = channel in CHANNELS_5G
452
453 # Base standard from band
454 if is_5ghz:
455 standards.append("a")
456 else:
457 # Check supported rates to determine b and/or g
458 supported_rates = _collect_rates(ies)
459 ofdm_rates = {6, 9, 12, 18, 24, 36, 48, 54}
460 b_rates = {1, 2, 5.5, 11}
461
462 has_b = any(r in b_rates for r in supported_rates)
463 has_g = any(r in ofdm_rates for r in supported_rates)
464
465 if has_b:
466 standards.append("b")
467 if has_g:
468 standards.append("g")
469 if not has_b and not has_g and supported_rates:
470 standards.append("b") # fallback
471
472 # HT Capabilities (IE 45) -> 802.11n
473 if 45 in ies:
474 standards.append("n")
475
476 # VHT Capabilities (IE 191) -> 802.11ac
477 if 191 in ies:
478 standards.append("ac")
479
480 # HE Capabilities (IE 255, ext ID 35) -> 802.11ax
481 for info in ies.get(255, []):
482 if info and len(info) > 0 and info[0] == 35:
483 standards.append("ax")
484 break
485
486 return ", ".join(standards)
487
488
489 def _collect_rates(ies: dict) -> set[float]:
490 """Collect supported rates from IE 1 (Supported Rates) and IE 50 (Extended Rates)."""
491 rates: set[float] = set()
492 for ie_id in (1, 50):
493 for info in ies.get(ie_id, []):
494 for byte in info:
495 rate = (byte & 0x7F) * 0.5
496 rates.add(rate)
497 return rates
498
499
500 # ---------------------------------------------------------------------------
501 # WPS detection
502 # ---------------------------------------------------------------------------
503
504 def _extract_wps(ies: dict) -> str:
505 """Extract WPS version from Vendor Specific IEs.
506
507 WPS IE: Vendor IE (221) with OUI 00:50:F2, type 4.
508 """
509 for info in ies.get(221, []):
510 if len(info) >= 4 and info[:4] == b"\x00\x50\xf2\x04":
511 # WPS IE found - parse WPS attributes to find version
512 return _parse_wps_version(info[4:])
513 return ""
514
515
516 def _parse_wps_version(data: bytes) -> str:
517 """Parse WPS attributes to extract version.
518
519 WPS TLV format: Type (2 bytes BE) | Length (2 bytes BE) | Value
520 Version attribute type = 0x104A
521 Version2 attribute type = 0x1032 (inside Vendor Extension 0x1049)
522 """
523 offset = 0
524 version = ""
525 while offset + 4 <= len(data):
526 attr_type = int.from_bytes(data[offset:offset + 2], "big")
527 attr_len = int.from_bytes(data[offset + 2:offset + 4], "big")
528 offset += 4
529
530 if offset + attr_len > len(data):
531 break
532
533 attr_data = data[offset:offset + attr_len]
534
535 if attr_type == 0x104A and attr_len >= 1:
536 # Version: single byte, major.minor (0x10 = 1.0, 0x20 = 2.0)
537 ver_byte = attr_data[0]
538 major = (ver_byte >> 4) & 0x0F
539 minor = ver_byte & 0x0F
540 version = f"{major}.{minor}"
541 elif attr_type == 0x1049:
542 # Vendor Extension - may contain Version2 (sub-element ID 0x00)
543 sub_offset = 3 # skip 3-byte WFA vendor ID
544 while sub_offset + 2 <= len(attr_data):
545 sub_id = attr_data[sub_offset]
546 sub_len = attr_data[sub_offset + 1]
547 sub_offset += 2
548 if sub_offset + sub_len > len(attr_data):
549 break # malformed sub-element
550 if sub_id == 0x00 and sub_len >= 1:
551 ver_byte = attr_data[sub_offset]
552 major = (ver_byte >> 4) & 0x0F
553 minor = ver_byte & 0x0F
554 version = f"{major}.{minor}"
555 sub_offset += sub_len
556
557 offset += attr_len
558
559 return version