This class handles the entire network scanning process.
| 124 | return None |
| 125 | |
| 126 | class NetworkScanner: |
| 127 | """ |
| 128 | This class handles the entire network scanning process. |
| 129 | """ |
| 130 | def __init__(self, shared_data): |
| 131 | self.shared_data = shared_data |
| 132 | self.logger = logger |
| 133 | self.displaying_csv = shared_data.displaying_csv |
| 134 | self.blacklistcheck = shared_data.blacklistcheck |
| 135 | self.mac_scan_blacklist = shared_data.mac_scan_blacklist |
| 136 | self.ip_scan_blacklist = shared_data.ip_scan_blacklist |
| 137 | self.console = Console() if Console else None |
| 138 | self.lock = threading.Lock() |
| 139 | self.currentdir = shared_data.currentdir |
| 140 | # CRITICAL: Pi Zero W2 has limited resources - use conservative thread count |
| 141 | # 512MB RAM, 4 cores @ 1GHz can only handle a few concurrent operations |
| 142 | cpu_count = os.cpu_count() or 1 |
| 143 | # Limit concurrent socket operations aggressively on the Pi Zero 2 W |
| 144 | self.port_scan_workers = max(2, min(6, cpu_count)) |
| 145 | self.host_scan_workers = max(2, min(6, cpu_count)) |
| 146 | self.semaphore = threading.Semaphore(min(4, max(1, cpu_count // 2 or 1))) |
| 147 | self.nm = nmap.PortScanner() if nmap else None # Initialize nmap.PortScanner() |
| 148 | self.running = False |
| 149 | self.arp_scan_interface = self._detect_default_interface() |
| 150 | self._active_scan_network = None |
| 151 | # Initialize SQLite database manager |
| 152 | self.db = get_db(currentdir=self.currentdir) |
| 153 | |
| 154 | @staticmethod |
| 155 | def _detect_default_interface(): |
| 156 | """Detect the active network interface using ip route.""" |
| 157 | try: |
| 158 | result = subprocess.run( |
| 159 | ['ip', 'route', 'show', 'default'], |
| 160 | capture_output=True, text=True, timeout=5 |
| 161 | ) |
| 162 | # Parse: default via 172.16.52.1 dev br-lan ... |
| 163 | for line in result.stdout.strip().splitlines(): |
| 164 | parts = line.split() |
| 165 | if 'dev' in parts: |
| 166 | idx = parts.index('dev') |
| 167 | if idx + 1 < len(parts): |
| 168 | return parts[idx + 1] |
| 169 | except Exception: |
| 170 | pass |
| 171 | # Fallback: find first non-lo interface with an IP |
| 172 | try: |
| 173 | result = subprocess.run( |
| 174 | ['ip', '-o', '-4', 'addr', 'show'], |
| 175 | capture_output=True, text=True, timeout=5 |
| 176 | ) |
| 177 | for line in result.stdout.strip().splitlines(): |
| 178 | # Format: 3: br-lan inet 172.16.52.1/24 brd ... |
| 179 | parts = line.split() |
| 180 | if len(parts) >= 4 and '127.0.0.1' not in line: |
| 181 | iface = parts[1].rstrip(':') |
| 182 | if iface != 'lo': |
| 183 | return iface |
no outgoing calls
no test coverage detected