ap_scanner/active_scanner.py (13808B)
1 """Active scanner: connect to open APs and gather network intelligence."""
2
3 import logging
4 import re
5 import socket
6 import struct
7 import subprocess
8 import time
9 from datetime import datetime
10 from pathlib import Path
11
12 from ap_scanner.models import AccessPoint, ActiveScanClient, ActiveScanResult
13
14 logger = logging.getLogger("ap_scanner.active_scanner")
15
16 # Common gateway ports to probe
17 GATEWAY_PORTS = [22, 53, 80, 443, 8080, 8443]
18
19 # Timeouts
20 DHCP_TIMEOUT = 15
21 CONNECT_TIMEOUT = 10
22 PORT_SCAN_TIMEOUT = 1
23 ARP_TIMEOUT = 3
24 OVERALL_TIMEOUT = 90
25
26
27 class ActiveScanner:
28 """Performs active reconnaissance on open (OPN) access points.
29
30 Sequence: connect -> DHCP -> parse network info -> ARP scan -> port scan -> disconnect.
31 """
32
33 def __init__(self, interface: str):
34 self.interface = interface
35
36 def scan(self, ap: AccessPoint) -> ActiveScanResult:
37 """Run full active scan on an open AP. Returns results even on partial failure."""
38 start = time.monotonic()
39 result = ActiveScanResult(bssid=ap.bssid, ssid=ap.display_ssid)
40
41 def _timed_out() -> bool:
42 return (time.monotonic() - start) >= OVERALL_TIMEOUT
43
44 try:
45 # Step 1: Connect
46 if not self._connect(ap):
47 result.error = "Failed to connect to AP"
48 return result
49 result.connected = True
50
51 if _timed_out():
52 result.error = "Overall timeout exceeded after connect"
53 return result
54
55 # Step 2: DHCP
56 if not self._acquire_dhcp():
57 result.error = "DHCP failed — no IP assigned"
58 self._disconnect()
59 return result
60
61 if _timed_out():
62 result.error = "Overall timeout exceeded after DHCP"
63 return result
64
65 # Step 3: Parse network info
66 self._parse_network_info(result)
67
68 # Step 4: Resolve gateway MAC
69 if result.gateway_ip and not _timed_out():
70 self._resolve_gateway_mac(result)
71
72 # Step 5: ARP scan (best-effort)
73 if result.assigned_ip and result.subnet_mask and not _timed_out():
74 try:
75 self._arp_scan(result)
76 except Exception as e:
77 logger.warning("ARP scan failed: %s", e)
78
79 # Step 6: Port scan gateway (best-effort)
80 if result.gateway_ip and not _timed_out():
81 try:
82 self._port_scan(result)
83 except Exception as e:
84 logger.warning("Port scan failed: %s", e)
85
86 except Exception as e:
87 result.error = str(e)
88 logger.error("Active scan error: %s", e)
89 finally:
90 result.duration_secs = round(time.monotonic() - start, 1)
91 self._disconnect()
92
93 return result
94
95 # --- Connection ---
96
97 def _connect(self, ap: AccessPoint) -> bool:
98 """Connect to an open AP using iw, falling back to wpa_supplicant."""
99 ssid = ap.ssid
100 if not ssid:
101 logger.warning("Cannot connect: AP has empty/hidden SSID")
102 return False
103
104 # Prevent NetworkManager from grabbing the interface while we connect.
105 self._run(
106 ["nmcli", "device", "set", self.interface, "managed", "no"],
107 timeout=5, ignore_errors=True,
108 )
109
110 # Trigger a scan so the AP appears in the driver's BSS list.
111 # Without this, 'iw connect' often fails with "not found" because
112 # the driver has no cached scan results after a mode switch.
113 logger.info("Scanning for %s on %s ...", ssid, self.interface)
114 scan_ret = self._run(
115 ["iw", "dev", self.interface, "scan"],
116 timeout=15, ignore_errors=True,
117 )
118 if ssid in scan_ret.stdout:
119 logger.info("AP '%s' found in scan results", ssid)
120 else:
121 logger.warning("AP '%s' not found in scan results — connecting anyway", ssid)
122
123 # Method 1: iw connect with BSSID + frequency for precise targeting
124 freq = str(ap.frequency) if ap.frequency else ""
125 iw_cmd = ["iw", "dev", self.interface, "connect", ssid]
126 if freq:
127 iw_cmd.append(freq)
128 iw_cmd.append(ap.bssid)
129
130 logger.info("Attempting: %s", " ".join(iw_cmd))
131 ret = self._run(iw_cmd, timeout=CONNECT_TIMEOUT)
132 if ret.returncode == 0:
133 for _ in range(5):
134 time.sleep(1)
135 if self._is_connected():
136 logger.info("Connected to %s via iw", ssid)
137 return True
138 logger.info("iw connect returned 0 but not associated after 5s")
139 else:
140 logger.info("iw connect failed: %s", ret.stderr.strip())
141
142 # Method 2: wpa_supplicant with key_mgmt=NONE + bssid
143 logger.info("Falling back to wpa_supplicant for '%s' ...", ssid)
144 # Sanitize SSID for wpa_supplicant config (escape \ and " to prevent injection)
145 safe_ssid = ssid.replace("\\", "\\\\").replace('"', '\\"')
146 conf = (
147 f'network={{\n'
148 f' ssid="{safe_ssid}"\n'
149 f' bssid={ap.bssid}\n'
150 f' key_mgmt=NONE\n'
151 f' scan_ssid=1\n'
152 f'}}\n'
153 )
154 conf_path = Path(f"/tmp/ap_scanner_wpa_{self.interface}.conf")
155 conf_path.write_text(conf)
156
157 # Kill any existing wpa_supplicant on this interface
158 self._run(["pkill", "-f", f"wpa_supplicant.*{self.interface}"],
159 timeout=3, ignore_errors=True)
160 time.sleep(0.5)
161
162 ret = self._run(
163 ["wpa_supplicant", "-B", "-i", self.interface,
164 "-c", str(conf_path), "-D", "nl80211"],
165 timeout=CONNECT_TIMEOUT, ignore_errors=True,
166 )
167 logger.info("wpa_supplicant launch: rc=%d stderr=%s",
168 ret.returncode, ret.stderr.strip())
169
170 for _ in range(8):
171 time.sleep(1)
172 if self._is_connected():
173 logger.info("Connected to %s via wpa_supplicant", ssid)
174 return True
175
176 # Log diagnostics on failure
177 link = self._run(["iw", "dev", self.interface, "link"],
178 timeout=5, ignore_errors=True)
179 logger.warning(
180 "Failed to connect to %s — link status: %s",
181 ssid, link.stdout.strip(),
182 )
183 return False
184
185 def _is_connected(self) -> bool:
186 """Check if interface is associated to an AP."""
187 ret = self._run(["iw", "dev", self.interface, "link"], timeout=5)
188 return ret.returncode == 0 and "Connected to" in ret.stdout
189
190 # --- DHCP ---
191
192 def _acquire_dhcp(self) -> bool:
193 """Acquire an IP via DHCP (dhclient, falling back to udhcpc)."""
194 # Try dhclient
195 ret = self._run(
196 ["dhclient", "-1", "-v", "-timeout", str(DHCP_TIMEOUT), self.interface],
197 timeout=DHCP_TIMEOUT + 5, ignore_errors=True,
198 )
199 if ret.returncode == 0 and self._has_ip():
200 logger.info("DHCP lease acquired via dhclient")
201 return True
202
203 # Fallback: udhcpc
204 ret = self._run(
205 ["udhcpc", "-i", self.interface, "-t", "5", "-T", "3", "-n", "-q"],
206 timeout=DHCP_TIMEOUT + 5, ignore_errors=True,
207 )
208 if ret.returncode == 0 and self._has_ip():
209 logger.info("DHCP lease acquired via udhcpc")
210 return True
211
212 logger.warning("DHCP failed on %s", self.interface)
213 return False
214
215 def _has_ip(self) -> bool:
216 """Check if the interface has an IPv4 address."""
217 ret = self._run(["ip", "-4", "addr", "show", self.interface], timeout=5)
218 return "inet " in ret.stdout
219
220 # --- Network info parsing ---
221
222 def _parse_network_info(self, result: ActiveScanResult) -> None:
223 """Parse IP, subnet, gateway, DNS from system commands."""
224 # IP and subnet
225 ret = self._run(["ip", "-4", "addr", "show", self.interface], timeout=5)
226 m = re.search(r"inet (\d+\.\d+\.\d+\.\d+)/(\d+)", ret.stdout)
227 if m:
228 result.assigned_ip = m.group(1)
229 result.subnet_mask = _cidr_to_netmask(int(m.group(2)))
230
231 # Default gateway
232 ret = self._run(["ip", "route", "show", "dev", self.interface], timeout=5)
233 m = re.search(r"default via (\d+\.\d+\.\d+\.\d+)", ret.stdout)
234 if m:
235 result.gateway_ip = m.group(1)
236
237 # DNS servers
238 try:
239 resolv = Path("/etc/resolv.conf").read_text()
240 result.dns_servers = re.findall(r"nameserver\s+(\d+\.\d+\.\d+\.\d+)", resolv)
241 except OSError:
242 pass
243
244 # DHCP server (from dhclient lease file)
245 for lease_path in Path("/var/lib/dhcp").glob("*.leases"):
246 try:
247 content = lease_path.read_text()
248 m = re.search(r"dhcp-server-identifier\s+(\d+\.\d+\.\d+\.\d+)", content)
249 if m:
250 result.dhcp_server = m.group(1)
251 break
252 except OSError:
253 continue
254
255 # --- ARP operations ---
256
257 def _resolve_gateway_mac(self, result: ActiveScanResult) -> None:
258 """Resolve gateway MAC via a single ARP request using scapy."""
259 try:
260 from scapy.all import ARP, Ether, sr1
261
262 pkt = Ether(dst="ff:ff:ff:ff:ff:ff") / ARP(
263 op="who-has", pdst=result.gateway_ip
264 )
265 resp = sr1(pkt, iface=self.interface, timeout=ARP_TIMEOUT, verbose=0)
266 if resp and resp.haslayer(ARP):
267 result.gateway_mac = resp[ARP].hwsrc
268 except Exception as e:
269 logger.warning("Gateway MAC resolution failed: %s", e)
270
271 def _arp_scan(self, result: ActiveScanResult) -> None:
272 """ARP scan the local subnet to discover connected clients."""
273 from scapy.all import ARP, Ether, srp
274
275 network = _calculate_network(result.assigned_ip, result.subnet_mask)
276 if not network:
277 return
278
279 pkt = Ether(dst="ff:ff:ff:ff:ff:ff") / ARP(op="who-has", pdst=network)
280 answered, _ = srp(pkt, iface=self.interface, timeout=ARP_TIMEOUT, verbose=0)
281
282 for _, recv in answered:
283 if recv.haslayer(ARP):
284 ip = recv[ARP].psrc
285 mac = recv[ARP].hwsrc
286 # Skip our own IP
287 if ip == result.assigned_ip:
288 continue
289 result.clients.append(ActiveScanClient(ip=ip, mac=mac))
290
291 # --- Port scan ---
292
293 def _port_scan(self, result: ActiveScanResult) -> None:
294 """TCP connect scan on common gateway ports."""
295 for port in GATEWAY_PORTS:
296 try:
297 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
298 sock.settimeout(PORT_SCAN_TIMEOUT)
299 if sock.connect_ex((result.gateway_ip, port)) == 0:
300 result.gateway_open_ports.append(port)
301 except (OSError, socket.error):
302 pass
303
304 # --- Disconnect / Cleanup ---
305
306 def _disconnect(self) -> None:
307 """Release DHCP, kill wpa_supplicant, disconnect, flush IP."""
308 self._run(["dhclient", "-r", self.interface],
309 timeout=5, ignore_errors=True)
310 self._run(["pkill", "-f", f"wpa_supplicant.*{self.interface}"],
311 timeout=3, ignore_errors=True)
312 self._run(["iw", "dev", self.interface, "disconnect"],
313 timeout=5, ignore_errors=True)
314 self._run(["ip", "addr", "flush", "dev", self.interface],
315 timeout=5, ignore_errors=True)
316 # Re-enable NetworkManager management
317 self._run(["nmcli", "device", "set", self.interface, "managed", "yes"],
318 timeout=5, ignore_errors=True)
319 # Clean up temp config
320 conf_path = Path(f"/tmp/ap_scanner_wpa_{self.interface}.conf")
321 conf_path.unlink(missing_ok=True)
322
323 # --- Utilities ---
324
325 @staticmethod
326 def _run(
327 cmd: list[str],
328 timeout: int = 10,
329 ignore_errors: bool = False,
330 ) -> subprocess.CompletedProcess:
331 """Run a command, returning CompletedProcess."""
332 try:
333 return subprocess.run(
334 cmd, capture_output=True, text=True, timeout=timeout,
335 )
336 except FileNotFoundError:
337 if not ignore_errors:
338 raise
339 return subprocess.CompletedProcess(cmd, 127, "", f"{cmd[0]}: not found")
340 except subprocess.TimeoutExpired:
341 if not ignore_errors:
342 raise
343 return subprocess.CompletedProcess(cmd, 124, "", "timeout")
344
345
346 # --- Module-level utility functions ---
347
348 def _cidr_to_netmask(prefix: int) -> str:
349 """Convert CIDR prefix length to dotted netmask (e.g. 24 -> 255.255.255.0)."""
350 bits = (0xFFFFFFFF << (32 - prefix)) & 0xFFFFFFFF
351 return socket.inet_ntoa(struct.pack("!I", bits))
352
353
354 def _calculate_network(ip: str, netmask: str) -> str:
355 """Calculate network address in CIDR notation for ARP scan target."""
356 try:
357 ip_int = struct.unpack("!I", socket.inet_aton(ip))[0]
358 mask_int = struct.unpack("!I", socket.inet_aton(netmask))[0]
359 net_int = ip_int & mask_int
360 # Count prefix bits
361 prefix = bin(mask_int).count("1")
362 net_addr = socket.inet_ntoa(struct.pack("!I", net_int))
363 return f"{net_addr}/{prefix}"
364 except (OSError, struct.error):
365 return ""
366
367
368 def _is_ip(s: str) -> bool:
369 """Check if string is a valid IPv4 address."""
370 try:
371 socket.inet_aton(s)
372 parts = s.split(".")
373 return len(parts) == 4 and all(0 <= int(p) <= 255 for p in parts)
374 except (OSError, ValueError):
375 return False