ap_scanner/monitor.py (7620B)
1 """Monitor mode management for wireless interfaces."""
2
3 import atexit
4 import logging
5 import subprocess
6 import sys
7 import time
8
9 from ap_scanner.utils import get_interface_mode
10
11 logger = logging.getLogger("ap_scanner.monitor")
12
13
14 class MonitorMode:
15 """Context manager for enabling/disabling monitor mode on a wireless interface.
16
17 Usage:
18 with MonitorMode("wlan0") as iface:
19 # iface is in monitor mode
20 ...
21 # iface is restored to managed mode
22 """
23
24 def __init__(self, interface: str):
25 self.interface = interface
26 self._original_mode: str | None = None
27 self._enabled = False
28 self._cleanup_registered = False
29
30 def enable(self) -> str:
31 """Enable monitor mode on the interface. Returns the interface name."""
32 self._original_mode = get_interface_mode(self.interface)
33
34 if self._original_mode == "monitor":
35 self._enabled = True
36 self._register_cleanup()
37 return self.interface
38
39 # Bring interface down
40 self._run_cmd(["ip", "link", "set", self.interface, "down"])
41
42 # Set monitor mode
43 self._run_cmd(["iw", "dev", self.interface, "set", "monitor", "none"])
44
45 # Bring interface up
46 self._run_cmd(["ip", "link", "set", self.interface, "up"])
47
48 # Verify
49 mode = get_interface_mode(self.interface)
50 if mode != "monitor":
51 # Try to restore
52 self._restore_managed()
53 raise RuntimeError(
54 f"Failed to set monitor mode on {self.interface}. "
55 f"Current mode: {mode}. "
56 f"Ensure the adapter supports monitor mode and the correct driver is loaded."
57 )
58
59 self._enabled = True
60 self._register_cleanup()
61 logger.info("%s: monitor mode enabled", self.interface)
62 return self.interface
63
64 def disable(self) -> None:
65 """Disable monitor mode and restore the interface to managed mode."""
66 if not self._enabled:
67 return
68
69 self._restore_managed()
70 self._enabled = False
71 logger.info("%s: managed mode restored", self.interface)
72
73 def _restore_managed(self) -> None:
74 """Restore the interface to managed mode."""
75 try:
76 self._run_cmd(["ip", "link", "set", self.interface, "down"],
77 ignore_errors=True)
78 self._run_cmd(["iw", "dev", self.interface, "set", "type", "managed"],
79 ignore_errors=True)
80 self._run_cmd(["ip", "link", "set", self.interface, "up"],
81 ignore_errors=True)
82 except Exception:
83 pass
84
85 # Restart NetworkManager to re-manage the interface
86 self._restart_network_manager()
87
88 # Verify restoration
89 mode = get_interface_mode(self.interface)
90 if mode and mode != "managed":
91 print(
92 f"[!] Warning: {self.interface} is in '{mode}' mode, expected 'managed'.\n"
93 f"[!] Run manually: sudo iw dev {self.interface} set type managed",
94 file=sys.stderr,
95 )
96
97 def _restart_network_manager(self) -> None:
98 """Re-manage only this interface in NetworkManager (without full restart).
99
100 Uses ``nmcli device set <iface> managed yes`` so that other
101 interfaces (e.g. internal WiFi) are not disrupted. Falls back to
102 a full ``systemctl restart NetworkManager`` only if nmcli fails.
103 """
104 try:
105 # Check if NM is running first
106 result = subprocess.run(
107 ["systemctl", "is-active", "--quiet", "NetworkManager"],
108 timeout=3,
109 )
110 if result.returncode != 0:
111 return # NM not running, nothing to do
112
113 # Targeted: re-manage only the scanned interface
114 result = subprocess.run(
115 ["nmcli", "device", "set", self.interface, "managed", "yes"],
116 capture_output=True, timeout=5,
117 )
118 if result.returncode == 0:
119 return # success — other interfaces untouched
120
121 # Fallback: full restart (last resort)
122 logger.debug("nmcli set managed failed, falling back to NM restart")
123 subprocess.run(
124 ["systemctl", "restart", "NetworkManager"],
125 capture_output=True, timeout=10,
126 )
127 except (subprocess.TimeoutExpired, FileNotFoundError):
128 pass
129
130 def temporarily_disable(self) -> None:
131 """Switch to managed mode for active scanning (skips NM restart for speed).
132
133 Raises RuntimeError if the interface cannot be switched to managed mode.
134 """
135 if not self._enabled:
136 return
137
138 self._run_cmd(["ip", "link", "set", self.interface, "down"])
139 self._run_cmd(["iw", "dev", self.interface, "set", "type", "managed"])
140 self._run_cmd(["ip", "link", "set", self.interface, "up"])
141
142 # Verify mode switch
143 time.sleep(0.5) # let the driver settle
144 mode = get_interface_mode(self.interface)
145 if mode != "managed":
146 raise RuntimeError(
147 f"Failed to switch {self.interface} to managed mode "
148 f"(current: {mode})"
149 )
150 logger.info("%s: temporarily switched to managed mode", self.interface)
151
152 def re_enable_monitor(self) -> None:
153 """Re-enable monitor mode after active scanning."""
154 try:
155 self._run_cmd(["ip", "link", "set", self.interface, "down"])
156 self._run_cmd(["iw", "dev", self.interface, "set", "monitor", "none"])
157 self._run_cmd(["ip", "link", "set", self.interface, "up"])
158 except RuntimeError as e:
159 logger.critical("Failed to restore monitor mode: %s", e)
160 raise
161
162 mode = get_interface_mode(self.interface)
163 if mode != "monitor":
164 logger.critical(
165 "%s is in '%s' mode after re_enable_monitor, expected 'monitor'",
166 self.interface, mode,
167 )
168 raise RuntimeError(
169 f"Failed to restore monitor mode on {self.interface}. "
170 f"Current mode: {mode}"
171 )
172 logger.info("%s: monitor mode re-enabled", self.interface)
173
174 def _register_cleanup(self) -> None:
175 """Register atexit handler for emergency cleanup."""
176 if not self._cleanup_registered:
177 atexit.register(self._emergency_cleanup)
178 self._cleanup_registered = True
179
180 def _emergency_cleanup(self) -> None:
181 """Best-effort cleanup on unexpected exit."""
182 if self._enabled:
183 try:
184 self._restore_managed()
185 except Exception:
186 logger.error(
187 "Could not restore %s to managed mode. "
188 "Run manually: sudo iw dev %s set type managed",
189 self.interface, self.interface,
190 )
191
192 @staticmethod
193 def _run_cmd(cmd: list[str], ignore_errors: bool = False) -> subprocess.CompletedProcess:
194 """Run a subprocess command."""
195 result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
196 if result.returncode != 0 and not ignore_errors:
197 raise RuntimeError(
198 f"Command failed: {' '.join(cmd)}\n"
199 f"stderr: {result.stderr.strip()}"
200 )
201 return result
202
203 def __enter__(self) -> str:
204 return self.enable()
205
206 def __exit__(self, exc_type, exc_val, exc_tb) -> None:
207 self.disable()