ap_scanner/scanner.py (15874B)
1 """Scanner engine: packet capture, channel hopping, AP management."""
2
3 import logging
4 import subprocess
5 import threading
6 import time
7 from datetime import datetime
8
9 from scapy.all import sendp, sniff
10 from scapy.layers.dot11 import Dot11, Dot11Elt, Dot11ProbeReq, RadioTap
11
12 from ap_scanner.models import AccessPoint
13 from ap_scanner.parser import parse_packet, parse_probe_request
14
15 logger = logging.getLogger("ap_scanner.scanner")
16
17 # Memory bounds for tracking maps
18 MAX_CLIENTS = 10000
19 MAX_PROBE_SSIDS = 5000
20 _CLEANUP_INTERVAL = 60 # seconds
21
22
23 class ChannelHopper:
24 """Hops through WiFi channels in a background thread.
25
26 Uses a weighted schedule so primary 2.4 GHz channels (1, 6, 11) are
27 visited more frequently, improving discovery of the most common APs.
28 """
29
30 # Primary 2.4 GHz channels — visited twice per cycle for better coverage
31 _PRIMARY_24 = {1, 6, 11}
32
33 def __init__(
34 self,
35 interface: str,
36 channels: list[int],
37 dwell_time: float = 0.5,
38 ):
39 self.interface = interface
40 self.channels = channels
41 self.dwell_time = dwell_time
42 self._stop_event = threading.Event()
43 self._thread: threading.Thread | None = None
44 self._current_channel: int = 0
45 self._lock = threading.Lock()
46 self._fail_count: dict[int, int] = {} # ch -> consecutive failure count
47 self._failed_channels: set[int] = set() # permanently skipped after 3 failures
48 self._probe_src_mac = self._get_mac()
49 self._schedule = self._build_schedule()
50
51 def _get_mac(self) -> str:
52 """Read the interface MAC address for Probe Request source."""
53 try:
54 mac_path = f"/sys/class/net/{self.interface}/address"
55 with open(mac_path) as f:
56 return f.read().strip()
57 except (OSError, ValueError):
58 return "00:00:00:00:00:00"
59
60 def _build_schedule(self) -> list[int]:
61 """Build a weighted channel schedule.
62
63 Primary 2.4 GHz channels (1, 6, 11) are interleaved between other
64 channels so they get visited roughly twice as often.
65 """
66 primary = [ch for ch in self.channels if ch in self._PRIMARY_24]
67 others = [ch for ch in self.channels if ch not in self._PRIMARY_24]
68
69 if not primary:
70 return list(self.channels)
71
72 # Interleave: after every N other channels, insert a primary channel
73 # With ~36 other channels and 3 primary, insert one primary every 12 channels
74 schedule: list[int] = []
75 step = max(len(others) // len(primary), 3) if others else 1
76 pri_idx = 0
77
78 for i, ch in enumerate(others):
79 schedule.append(ch)
80 if (i + 1) % step == 0 and pri_idx < len(primary):
81 schedule.append(primary[pri_idx])
82 pri_idx += 1
83
84 # Append any remaining primary channels
85 while pri_idx < len(primary):
86 schedule.append(primary[pri_idx])
87 pri_idx += 1
88
89 # Prepend all primary channels at the start for quick first results
90 return primary + schedule
91
92 @property
93 def current_channel(self) -> int:
94 with self._lock:
95 return self._current_channel
96
97 def start(self) -> None:
98 self._stop_event.clear()
99 self._fail_count.clear()
100 self._failed_channels.clear()
101 self._thread = threading.Thread(target=self._hop_loop, daemon=True)
102 self._thread.start()
103
104 def stop(self) -> None:
105 self._stop_event.set()
106 if self._thread and self._thread.is_alive():
107 self._thread.join(timeout=3)
108 if self._thread.is_alive():
109 logger.warning("Channel hopper thread did not stop within timeout")
110
111 def _send_probe_request(self) -> None:
112 """Inject a broadcast Probe Request to trigger Probe Responses from APs.
113
114 This is the key difference between passive and active scanning.
115 APs respond to Probe Requests even when the next beacon is far away.
116 """
117 try:
118 probe = (
119 RadioTap()
120 / Dot11(
121 type=0,
122 subtype=4,
123 addr1="ff:ff:ff:ff:ff:ff",
124 addr2=self._probe_src_mac,
125 addr3="ff:ff:ff:ff:ff:ff",
126 )
127 / Dot11ProbeReq()
128 / Dot11Elt(ID=0, info=b"") # broadcast SSID
129 / Dot11Elt(ID=1, info=b"\x02\x04\x0b\x16\x0c\x12\x18\x24") # rates
130 )
131 sendp(probe, iface=self.interface, verbose=False, count=1)
132 except Exception:
133 pass # best-effort; injection may not be supported
134
135 def _hop_loop(self) -> None:
136 _BLACKLIST_THRESHOLD = 3 # failures before permanent skip
137
138 while not self._stop_event.is_set():
139 for ch in self._schedule:
140 if self._stop_event.is_set():
141 break
142 # Skip channels that have consistently failed
143 if ch in self._failed_channels:
144 continue
145 try:
146 result = subprocess.run(
147 ["iw", "dev", self.interface, "set", "channel", str(ch)],
148 capture_output=True,
149 timeout=2,
150 )
151 if result.returncode != 0:
152 count = self._fail_count.get(ch, 0) + 1
153 self._fail_count[ch] = count
154 if count >= _BLACKLIST_THRESHOLD:
155 self._failed_channels.add(ch)
156 logger.debug("Channel %d failed %d times — skipping", ch, count)
157 continue
158 # Success — reset failure counter
159 self._fail_count.pop(ch, None)
160 with self._lock:
161 self._current_channel = ch
162 except (subprocess.TimeoutExpired, FileNotFoundError):
163 count = self._fail_count.get(ch, 0) + 1
164 self._fail_count[ch] = count
165 if count >= _BLACKLIST_THRESHOLD:
166 self._failed_channels.add(ch)
167 continue
168
169 # Send Probe Request to trigger AP responses on this channel
170 self._send_probe_request()
171
172 self._stop_event.wait(self.dwell_time)
173
174
175 class ScannerEngine:
176 """Orchestrates packet capture and channel hopping to discover APs."""
177
178 # Check time-based cleanup every N packets to avoid syscall on every packet
179 _CLEANUP_CHECK_INTERVAL = 5000
180
181 def __init__(
182 self,
183 interface: str,
184 channels: list[int],
185 dwell_time: float = 0.5,
186 ):
187 self.interface = interface
188 self.channels = channels
189
190 self._aps: dict[str, AccessPoint] = {}
191 # client_mac -> set of BSSIDs they communicate with
192 self._client_ap_map: dict[str, set[str]] = {}
193 # SSID -> set of client MACs that probed for it
194 self._probe_ssids: dict[str, set[str]] = {}
195 self._lock = threading.Lock()
196 self._stop_event = threading.Event()
197 self._hopper = ChannelHopper(interface, channels, dwell_time=dwell_time)
198 self._capture_thread: threading.Thread | None = None
199 self._last_cleanup = time.monotonic()
200 self._pkt_count = 0
201
202 @property
203 def access_points(self) -> dict[str, AccessPoint]:
204 """Return a copy of the current AP dictionary."""
205 with self._lock:
206 return dict(self._aps)
207
208 @property
209 def current_channel(self) -> int:
210 return self._hopper.current_channel
211
212 def start(self) -> None:
213 """Start scanning: channel hopper + packet capture."""
214 self._stop_event.clear()
215 self._hopper.start()
216 self._capture_thread = threading.Thread(target=self._capture_loop, daemon=True)
217 self._capture_thread.start()
218
219 def stop(self) -> None:
220 """Stop scanning and clean up threads."""
221 self._stop_event.set()
222 # Suppress Scapy socket warnings during intentional shutdown
223 logging.getLogger("scapy.runtime").setLevel(logging.CRITICAL)
224 logging.getLogger("scapy").setLevel(logging.CRITICAL)
225 self._hopper.stop()
226 if self._capture_thread and self._capture_thread.is_alive():
227 self._capture_thread.join(timeout=5)
228 if self._capture_thread.is_alive():
229 logger.warning("Capture thread did not stop within timeout")
230
231 def clear(self) -> None:
232 """Clear all discovered APs."""
233 with self._lock:
234 self._aps.clear()
235
236 @staticmethod
237 def _pick_bpf_filter(interface: str) -> str:
238 """Try BPF filter candidates and return the first one that compiles.
239
240 Returns empty string if none work (safe fallback = no filter).
241 """
242 candidates = ["type mgt", ""]
243 for bpf in candidates:
244 if not bpf:
245 return ""
246 try:
247 # Dry-run: open a sniff socket with the filter to see if it compiles
248 from scapy.arch.common import compile_filter
249 compile_filter(bpf, interface)
250 return bpf
251 except Exception:
252 continue
253 return ""
254
255 def _capture_loop(self) -> None:
256 """Run Scapy sniff in the capture thread.
257
258 Tries to apply a BPF filter for 802.11 management frames only,
259 falling back to no filter + fast Python-level type check.
260 """
261 bpf = self._pick_bpf_filter(self.interface)
262 if bpf:
263 logger.debug("Using BPF filter: %s", bpf)
264 else:
265 logger.debug("BPF filter not available — using Python-level filtering")
266
267 try:
268 while not self._stop_event.is_set():
269 sniff(
270 iface=self.interface,
271 prn=self._on_packet,
272 store=False,
273 stop_filter=lambda _: self._stop_event.is_set(),
274 timeout=2,
275 monitor=True,
276 **({"filter": bpf} if bpf else {}),
277 )
278 except OSError as e:
279 if not self._stop_event.is_set():
280 logger.error("Capture error on %s: %s", self.interface, e)
281
282 def _maybe_cleanup(self) -> None:
283 """Periodically evict old entries from tracking maps to bound memory.
284
285 Uses a packet counter to avoid calling time.monotonic() on every packet.
286 """
287 self._pkt_count += 1
288 if self._pkt_count < self._CLEANUP_CHECK_INTERVAL:
289 return
290 self._pkt_count = 0
291
292 now = time.monotonic()
293 if now - self._last_cleanup < _CLEANUP_INTERVAL:
294 return
295 self._last_cleanup = now
296 with self._lock:
297 if len(self._client_ap_map) > MAX_CLIENTS:
298 keys = list(self._client_ap_map.keys())
299 for k in keys[: len(keys) // 2]:
300 del self._client_ap_map[k]
301 if len(self._probe_ssids) > MAX_PROBE_SSIDS:
302 keys = list(self._probe_ssids.keys())
303 for k in keys[: len(keys) // 2]:
304 del self._probe_ssids[k]
305
306 def _on_packet(self, pkt) -> None:
307 """Callback for each captured packet."""
308 # Fast reject: extract Dot11 layer once and reuse throughout.
309 dot11 = pkt.getlayer(Dot11)
310 if dot11 is None:
311 return
312 frame_type = dot11.type
313 if frame_type == 1: # Control frame — skip
314 return
315
316 self._maybe_cleanup()
317
318 # Track client-AP associations (Data/Auth frames only)
319 if frame_type in (0, 2):
320 self._track_client_ap(dot11)
321
322 # Try to resolve hidden SSIDs from Probe Requests
323 probe = parse_probe_request(pkt)
324 if probe:
325 ssid, client_mac = probe
326 with self._lock:
327 self._probe_ssids.setdefault(ssid, set()).add(client_mac)
328 self._try_resolve_hidden(ssid, client_mac)
329
330 ap = parse_packet(pkt)
331 if ap is None:
332 return
333
334 with self._lock:
335 existing = self._aps.get(ap.bssid)
336 if existing:
337 existing.last_seen = datetime.now()
338 existing.beacon_count += 1
339 if ap.rssi is not None:
340 existing.rssi = ap.rssi
341 if existing.is_hidden and not ap.is_hidden:
342 existing.ssid = ap.ssid
343 if ap.encryption != "OPN" and existing.encryption == "OPN":
344 existing.encryption = ap.encryption
345 existing.cipher = ap.cipher
346 existing.auth = ap.auth
347 if ap.wpa2 and not existing.wpa2:
348 existing.wpa2 = ap.wpa2
349 if ap.wpa3 and not existing.wpa3:
350 existing.wpa3 = ap.wpa3
351 if ap.wpa and not existing.wpa:
352 existing.wpa = ap.wpa
353 if ap.wep and not existing.wep:
354 existing.wep = ap.wep
355 if ap.wifi_standard and not existing.wifi_standard:
356 existing.wifi_standard = ap.wifi_standard
357 if ap.width > existing.width:
358 existing.width = ap.width
359 existing.channel_desc = ap.channel_desc
360 if ap.vendor and not existing.vendor:
361 existing.vendor = ap.vendor
362 if ap.wps and not existing.wps:
363 existing.wps = ap.wps
364 else:
365 self._aps[ap.bssid] = ap
366 if ap.is_hidden:
367 self._try_resolve_hidden_for_ap(ap.bssid)
368
369 def _track_client_ap(self, dot11) -> None:
370 """Track which clients are communicating with which APs.
371
372 Receives the Dot11 layer directly (already extracted by _on_packet).
373 """
374 addr1 = dot11.addr1 # receiver
375 addr2 = dot11.addr2 # transmitter
376 addr3 = dot11.addr3 # BSSID (in most cases)
377
378 if not addr2 or not addr3:
379 return
380
381 addr2_upper = addr2.upper()
382 addr3_upper = addr3.upper()
383
384 with self._lock:
385 if addr3_upper in self._aps:
386 if addr2_upper != addr3_upper:
387 self._client_ap_map.setdefault(addr2_upper, set()).add(addr3_upper)
388 elif addr2_upper in self._aps:
389 if addr1:
390 client = addr1.upper()
391 if client != addr2_upper:
392 self._client_ap_map.setdefault(client, set()).add(addr2_upper)
393
394 def _try_resolve_hidden(self, ssid: str, client_mac: str) -> None:
395 """Try to match a probe request SSID to a hidden AP via client association.
396
397 If this client is associated with a hidden AP, the probed SSID is likely
398 the hidden AP's real name.
399 """
400 bssids = self._client_ap_map.get(client_mac, set())
401 for bssid in bssids:
402 ap = self._aps.get(bssid)
403 if ap and ap.is_hidden:
404 ap.ssid = ssid
405
406 def _try_resolve_hidden_for_ap(self, bssid: str) -> None:
407 """When a new hidden AP appears, check if any known clients can reveal its SSID."""
408 # Find clients associated with this BSSID
409 for client_mac, associated_bssids in self._client_ap_map.items():
410 if bssid in associated_bssids:
411 # Check if this client has probed for any SSIDs
412 for ssid, probing_clients in self._probe_ssids.items():
413 if client_mac in probing_clients:
414 ap = self._aps.get(bssid)
415 if ap and ap.is_hidden:
416 ap.ssid = ssid
417 return