ap_scanner/__main__.py (14056B)
1 """Entry point: python -m ap_scanner"""
2
3 import logging
4 import os
5 import select
6 import signal
7 import sys
8 import termios
9 import threading
10 import time
11 import tty
12
13 from ap_scanner.cli import parse_args
14 from ap_scanner.config import apply_config_to_args, load_config
15 from ap_scanner.display import Display
16 from ap_scanner.exporter import save_csv, save_json
17 from ap_scanner.log import restore_stderr_after_scan, setup_logging, suppress_stderr_during_scan
18 from ap_scanner.monitor import MonitorMode
19 from ap_scanner.scanner import ScannerEngine
20 from ap_scanner.utils import (
21 ALL_CHANNELS,
22 CHANNELS_2G,
23 CHANNELS_5G,
24 CHANNELS_6G,
25 check_ip,
26 check_iw,
27 check_root,
28 find_wireless_interfaces,
29 get_supported_channels,
30 parse_channels_arg,
31 )
32
33 logger = logging.getLogger("ap_scanner.main")
34
35 # Key constants for escape sequence parsing
36 KEY_UP = "UP"
37 KEY_DOWN = "DOWN"
38 KEY_ENTER = "ENTER"
39 KEY_ESC = "ESC"
40
41
42 def _read_key(timeout: float = 1.0) -> str | None:
43 """Read a single key or escape sequence from stdin.
44
45 Uses os.read() on the raw fd to avoid Python buffered IO issues where
46 sys.stdin.read(1) drains the entire escape sequence from the kernel buffer
47 into Python's internal buffer, causing select() to miss the continuation bytes.
48
49 Returns None on timeout.
50 """
51 fd = sys.stdin.fileno()
52
53 try:
54 readable, _, _ = select.select([fd], [], [], timeout)
55 except InterruptedError:
56 return None
57 if not readable:
58 return None
59
60 # Read all available bytes at once — escape sequences (e.g. \x1b[A)
61 # arrive as a single chunk in the kernel buffer
62 data = os.read(fd, 32)
63 if not data:
64 return None
65
66 if data[0] == 0x1B: # ESC
67 if len(data) >= 3 and data[1] == ord("["):
68 if data[2] == ord("A"):
69 return KEY_UP
70 if data[2] == ord("B"):
71 return KEY_DOWN
72 return None # unknown CSI sequence
73 if len(data) == 1:
74 return KEY_ESC
75 return None
76 if data[0] in (0x0D, 0x0A): # \r, \n
77 return KEY_ENTER
78 if data[0] == 0x03: # Ctrl+C
79 return "\x03"
80 return chr(data[0])
81
82
83 def main(argv: list[str] | None = None) -> int:
84 args = parse_args(argv)
85
86 # Load config file and apply defaults (CLI args take precedence)
87 config = load_config(args.config)
88 apply_config_to_args(args, config)
89
90 # Setup logging
91 setup_logging(verbose=args.verbose, log_file=args.log_file)
92
93 # Check root
94 if not check_root():
95 logger.error("This tool requires root privileges.")
96 logger.error("Run with: sudo python -m ap_scanner")
97 return 1
98
99 # Check required system tools
100 if not check_iw():
101 logger.error("'iw' command not found. Install with: sudo apt install iw")
102 return 1
103 if not check_ip():
104 logger.error("'ip' command not found. Install with: sudo apt install iproute2")
105 return 1
106
107 # Resolve interface
108 interface = args.interface
109 if interface is None:
110 interfaces = find_wireless_interfaces()
111 if not interfaces:
112 logger.error("No wireless interfaces found.")
113 return 1
114 interface = interfaces[0]
115 logger.info("Auto-detected interface: %s", interface)
116
117 # Resolve channels
118 if args.channels:
119 channels = parse_channels_arg(args.channels)
120 elif args.band == "2.4":
121 channels = CHANNELS_2G
122 elif args.band == "5":
123 channels = CHANNELS_5G
124 elif args.band == "6":
125 channels = CHANNELS_6G
126 else:
127 channels = ALL_CHANNELS
128
129 if not channels:
130 logger.error("No valid channels specified.")
131 return 1
132
133 # Filter to only channels the adapter hardware supports
134 supported = get_supported_channels(interface)
135 supported_set = set(supported)
136 original_count = len(channels)
137 channels = [ch for ch in channels if ch in supported_set]
138 if not channels:
139 logger.warning(
140 "None of the requested channels are supported by %s. "
141 "Using adapter's supported channels instead.", interface,
142 )
143 channels = supported
144 elif len(channels) < original_count:
145 logger.info(
146 "Filtered to %d supported channels (adapter supports %d of %d requested)",
147 len(channels), len(channels), original_count,
148 )
149
150 logger.info("Interface: %s", interface)
151 logger.info("Channels: %d (%d-%d)", len(channels), channels[0], channels[-1])
152
153 # Resolve dwell time
154 dwell_time = args.dwell if args.dwell is not None else 0.5
155
156 # Setup display
157 display = Display(no_color=args.no_color)
158 display.set_sort(args.sort)
159 if args.band != "all":
160 display.band_filter = args.band + "GHz"
161
162 # Enable monitor mode
163 monitor = MonitorMode(interface)
164 try:
165 monitor.enable()
166 except RuntimeError as e:
167 logger.error("%s", e)
168 return 1
169
170 # Create scanner engine
171 engine = ScannerEngine(interface, channels, dwell_time=dwell_time)
172
173 # Signal handler for clean shutdown — use threading.Event for thread safety
174 shutdown_event = threading.Event()
175
176 def signal_handler(signum, frame):
177 shutdown_event.set()
178
179 signal.signal(signal.SIGINT, signal_handler)
180 signal.signal(signal.SIGTERM, signal_handler)
181
182 # Start scanning
183 engine.start()
184 logger.info("Scanning started (Ctrl+C or 'q' to stop)")
185 time.sleep(0.5)
186
187 # Suppress stderr logging during interactive display
188 suppress_stderr_during_scan()
189
190 # Setup terminal for non-blocking key input
191 old_settings = termios.tcgetattr(sys.stdin)
192 try:
193 tty.setcbreak(sys.stdin.fileno())
194 _main_loop(engine, display, monitor, interface, args.write, shutdown_event)
195 finally:
196 # Print newline to separate from display before cleanup logs
197 print()
198 # Re-enable stderr logging for shutdown messages
199 restore_stderr_after_scan()
200
201 # Each cleanup step is individually protected so a failure in one
202 # does not prevent the others from running.
203 try:
204 termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
205 except Exception:
206 logger.error("Failed to restore terminal settings")
207 try:
208 engine.stop()
209 except Exception:
210 logger.error("Failed to stop scanner engine")
211 try:
212 monitor.disable()
213 except Exception:
214 logger.error("Failed to restore interface to managed mode")
215
216 return 0
217
218
219 def _main_loop(
220 engine: ScannerEngine,
221 display: Display,
222 monitor: MonitorMode,
223 interface: str,
224 csv_path: str | None,
225 shutdown_event: threading.Event,
226 ) -> None:
227 """Main display loop with 2-mode key dispatch (normal / selection)."""
228 frozen_aps = None # Snapshot of APs when entering selection mode
229
230 while not shutdown_event.is_set():
231 # In selection mode, use frozen snapshot; otherwise use live data
232 if display._selection_mode and frozen_aps is not None:
233 aps = frozen_aps
234 else:
235 aps = engine.access_points
236
237 display.render(aps, interface, engine.current_channel)
238
239 key = _read_key(timeout=1.0)
240 if key is None:
241 continue
242
243 if display._selection_mode:
244 sel_result = _handle_selection_key(
245 key, display, engine, monitor, interface, shutdown_event, frozen_aps
246 )
247 if sel_result == "exit_selection":
248 frozen_aps = None
249 else:
250 result = _handle_normal_key(key, display, engine, csv_path, shutdown_event)
251 # If we just entered selection mode, freeze the AP list
252 if result == "enter_selection":
253 frozen_aps = engine.access_points
254 elif result == "exit_selection":
255 frozen_aps = None
256
257
258
259 def _handle_normal_key(
260 key: str,
261 display: Display,
262 engine: ScannerEngine,
263 csv_path: str | None,
264 shutdown_event: threading.Event,
265 ) -> str | None:
266 """Handle keys in normal (non-selection) mode. Returns action taken."""
267 if key == "q" or key == "\x03":
268 shutdown_event.set()
269 elif key == "w":
270 try:
271 filepath = save_csv(engine.access_points, csv_path)
272 display.show_message(f"Saved to {filepath}")
273 except (OSError, IOError) as e:
274 display.show_message(f"Save failed: {e}")
275 time.sleep(1)
276 elif key == "j":
277 try:
278 filepath = save_json(engine.access_points)
279 display.show_message(f"Saved JSON to {filepath}")
280 except (OSError, IOError) as e:
281 display.show_message(f"JSON save failed: {e}")
282 time.sleep(1)
283 elif key == "b":
284 band = display.cycle_band()
285 display.show_message(f"Band filter: {band}")
286 time.sleep(0.5)
287 elif key == "s":
288 sort_field = display.cycle_sort()
289 display.show_message(f"Sort by: {sort_field}")
290 time.sleep(0.5)
291 elif key == KEY_ENTER:
292 sorted_aps = display.get_sorted_filtered(engine.access_points)
293 if sorted_aps:
294 display.enter_selection_mode()
295 return "enter_selection"
296 return None
297
298
299 def _handle_selection_key(
300 key: str,
301 display: Display,
302 engine: ScannerEngine,
303 monitor: MonitorMode,
304 interface: str,
305 shutdown_event: threading.Event,
306 frozen_aps: dict | None,
307 ) -> str | None:
308 """Handle keys in AP selection mode. Returns action taken."""
309 if key == KEY_ESC or key == "\x03":
310 display.exit_selection_mode()
311 return "exit_selection"
312 elif key == KEY_UP:
313 sorted_aps = display.get_sorted_filtered(frozen_aps or engine.access_points)
314 display.move_selection(-1, len(sorted_aps))
315 elif key == KEY_DOWN:
316 sorted_aps = display.get_sorted_filtered(frozen_aps or engine.access_points)
317 display.move_selection(1, len(sorted_aps))
318 elif key == KEY_ENTER:
319 sorted_aps = display.get_sorted_filtered(frozen_aps or engine.access_points)
320 if sorted_aps and display._selected_index < len(sorted_aps):
321 selected_ap = sorted_aps[display._selected_index]
322 display.exit_selection_mode()
323 if selected_ap.encryption == "OPN":
324 _run_active_scan(selected_ap, engine, display, monitor, interface)
325 return "exit_selection"
326 else:
327 display.show_message(
328 f"Active scan only available for open (OPN) APs — "
329 f"{selected_ap.display_ssid} uses {selected_ap.encryption}"
330 )
331 time.sleep(2)
332 return "exit_selection"
333 return None
334
335
336 def _run_active_scan(ap, engine, display, monitor, interface):
337 """Orchestrate active scan: stop capture, switch to managed, scan, restore."""
338 from ap_scanner.active_scanner import ActiveScanner
339
340 display.set_status(f"Stopping capture for active scan on {ap.display_ssid}...")
341
342 # 1. Stop packet capture + channel hopping
343 engine.stop()
344
345 # 2. Switch to managed mode (fast, no NM restart)
346 display.set_status(f"Switching {interface} to managed mode...")
347 try:
348 monitor.temporarily_disable()
349 except Exception as e:
350 logger.error("Failed to switch to managed mode: %s", e)
351 display.set_status("")
352 engine.start()
353 return
354
355 try:
356 # 3. Run active scan
357 display.set_status(
358 f"Active scanning {ap.display_ssid} ({ap.bssid}) — "
359 f"scan/connect/DHCP/ARP/ports..."
360 )
361 scanner = ActiveScanner(interface)
362 result = scanner.scan(ap)
363
364 # 4. Store result on AP object
365 ap.active_scan_result = result
366
367 # 5. Show overlay with results
368 _show_scan_overlay(display, result)
369
370 # Wait for user to dismiss
371 _read_key(timeout=300)
372 finally:
373 display.set_status("Restoring monitor mode...")
374 # 6. Always restore monitor mode
375 try:
376 monitor.re_enable_monitor()
377 except RuntimeError:
378 logger.critical(
379 "CRITICAL: Failed to restore monitor mode on %s. "
380 "Run manually: sudo iw dev %s set type monitor",
381 interface, interface,
382 )
383 display.set_status("CRITICAL: Monitor mode restore failed!")
384 time.sleep(3)
385
386 # 7. Restart capture
387 display.set_status("")
388 engine.start()
389
390
391 def _show_scan_overlay(display, result):
392 """Build and render the active scan results overlay."""
393 lines = []
394
395 if result.error:
396 lines.append(f"Error: {result.error}")
397 lines.append("")
398
399 if result.assigned_ip:
400 lines.append(f"Assigned IP: {result.assigned_ip}")
401 if result.subnet_mask:
402 lines.append(f"Subnet Mask: {result.subnet_mask}")
403 if result.gateway_ip:
404 gw = result.gateway_ip
405 if result.gateway_mac:
406 gw += f" ({result.gateway_mac})"
407 lines.append(f"Gateway: {gw}")
408 if result.dhcp_server:
409 lines.append(f"DHCP Server: {result.dhcp_server}")
410 if result.dns_servers:
411 lines.append(f"DNS Servers: {', '.join(result.dns_servers)}")
412 lines.append("")
413
414 if result.gateway_open_ports:
415 ports = ", ".join(str(p) for p in result.gateway_open_ports)
416 lines.append(f"Gateway Ports: {ports}")
417 lines.append("")
418
419 if result.clients:
420 lines.append(f"Connected Clients: {result.client_count}")
421 for client in result.clients[:10]:
422 vendor = f" {client.vendor}" if client.vendor else ""
423 lines.append(f" {client.ip:<16s} {client.mac}{vendor}")
424 if result.client_count > 10:
425 lines.append(f" ... +{result.client_count - 10} more")
426 lines.append("")
427
428 lines.append(f"Duration: {result.duration_secs}s")
429
430 title = f"ACTIVE SCAN: {result.ssid} ({result.bssid})"
431 display.render_overlay(lines, title)
432
433
434 if __name__ == "__main__":
435 sys.exit(main())