diff --git a/.gitignore b/.gitignore
@@ -0,0 +1,40 @@
+# Byte-compiled / optimized
+__pycache__/
+*.py[cod]
+*$py.class
+
+# Distribution / packaging
+dist/
+build/
+*.egg-info/
+*.egg
+
+# Virtual environments
+.venv/
+venv/
+env/
+
+# IDE
+.idea/
+.vscode/
+*.swp
+*.swo
+*~
+
+# Testing / Coverage
+.pytest_cache/
+htmlcov/
+.coverage
+coverage.xml
+
+# Scan output
+*.csv
+*.json
+result/
+
+# OS
+.DS_Store
+Thumbs.db
+
+# Logs
+*.log
diff --git a/LICENSE b/LICENSE
@@ -0,0 +1,17 @@
+AP Scanner - WiFi Access Point Scanner
+Copyright (C) 2026
+
+This program is free software; you can redistribute it and/or modify
+it under the terms of the GNU General Public License as published by
+the Free Software Foundation; either version 2 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License
+along with this program; if not, see <https://www.gnu.org/licenses/>.
+
+SPDX-License-Identifier: GPL-2.0-or-later
diff --git a/README.md b/README.md
@@ -0,0 +1,132 @@
+# AP Scanner
+
+Real-time wireless access point scanner with active reconnaissance for Linux.
+
+## Features
+
+### Passive Scanning
+- Real-time AP discovery with terminal display
+- Automatic channel hopping (2.4GHz / 5GHz / 6GHz)
+- Security detection: WEP, WPA, WPA2, WPA3
+- WiFi standard detection: 802.11 a/b/g/n/ac/ax
+- Channel width detection (20/40/80/160 MHz)
+- Hidden SSID detection and resolution
+- MAC vendor lookup
+- Band filtering and sort modes
+
+### Active Scanning (New!)
+- Interactive AP selection with arrow keys
+- Connect to open (OPN) networks for reconnaissance
+- DHCP lease acquisition and network info extraction
+- ARP scan to discover connected clients
+- Gateway port scanning (22, 53, 80, 443, 8080, 8443)
+- Client device enumeration with MAC vendor lookup
+- Results displayed in overlay panel
+
+### Export & Configuration
+- CSV / JSON export with active scan results
+- Configuration file support
+
+## Requirements
+
+- Python 3.10+
+- Linux with `iw` and `ip` commands
+- Root privileges
+- WiFi adapter that supports monitor mode
+
+## Installation
+
+### From PyPI (recommended)
+
+```bash
+# Using pipx (isolated environment)
+pipx install uaf
+
+# Or using pip
+pip install uaf
+```
+
+### From source
+
+```bash
+git clone https://github.com/devastator-x/uaf.git
+cd uaf
+pip install -e .
+```
+
+### Ubuntu/Debian (.deb package)
+
+```bash
+# Download from releases
+wget https://github.com/devastator-x/uaf/releases/download/v1.0.0/uaf_1.0.0_all.deb
+sudo dpkg -i uaf_1.0.0_all.deb
+sudo apt-get install -f # Install dependencies
+```
+
+## Usage
+
+```bash
+# Basic scan (auto-detect interface)
+sudo uaf
+
+# Specify interface
+sudo uaf -i wlan0
+
+# Scan specific channels
+sudo uaf -c 1,6,11
+
+# Band filter
+sudo uaf -b 2.4
+
+# Sort by channel
+sudo uaf --sort channel
+
+# Save to specific file
+sudo uaf -w output.csv
+
+# Use config file
+sudo uaf --config /path/to/config.ini
+```
+
+## Keybindings
+
+### Normal Mode
+| Key | Action |
+|-----|--------|
+| `q` | Quit |
+| `w` | Write CSV |
+| `j` | Write JSON |
+| `b` | Cycle band filter (All / 2.4GHz / 5GHz / 6GHz) |
+| `s` | Cycle sort mode (RSSI / Channel / ESSID / ENC) |
+| `Enter` | Enter AP selection mode |
+
+### Selection Mode
+| Key | Action |
+|-----|--------|
+| `↑` / `↓` | Navigate through AP list |
+| `Enter` | Run active scan on selected open AP |
+| `Esc` | Exit selection mode |
+
+## Configuration
+
+Default config path: `~/.config/ap-scanner/config.ini`
+
+```ini
+[scanner]
+interface = wlan0
+band = all
+sort = rssi
+no_color = false
+verbose = false
+```
+
+## Development
+
+```bash
+pip install -e ".[dev]"
+pytest tests/ -v
+```
+
+## License
+
+GPL-2.0-or-later. See [LICENSE](LICENSE) for details.
diff --git a/RELEASE.md b/RELEASE.md
@@ -0,0 +1,268 @@
+# Release Guide: PyPI & APT 패키지 배포
+
+## 1. PyPI 배포 (pipx 지원)
+
+### 1-1. PyPI 계정 생성
+1. https://pypi.org/account/register/ 방문
+2. 계정 생성 및 이메일 인증
+3. (선택) 2FA 활성화 (권장)
+
+### 1-2. API 토큰 생성
+1. https://pypi.org/manage/account/token/ 방문
+2. "Add API token" 클릭
+3. Token name: `uaf-upload`
+4. Scope: "Entire account" (첫 업로드) 또는 "Project: uaf"
+5. 생성된 토큰 복사 (다시 볼 수 없음!)
+
+### 1-3. TestPyPI에서 먼저 테스트 (권장)
+
+```bash
+# TestPyPI 계정 생성: https://test.pypi.org/account/register/
+# TestPyPI API 토큰 생성: https://test.pypi.org/manage/account/token/
+
+# 빌드
+python -m build
+
+# 테스트 업로드
+twine upload --repository testpypi dist/*
+
+# Username: __token__
+# Password: pypi-AgE... (복사한 토큰 전체)
+
+# 테스트 설치
+pipx install --index-url https://test.pypi.org/simple/ uaf
+
+# 테스트 실행
+sudo uaf --help
+```
+
+### 1-4. 정식 PyPI 업로드
+
+```bash
+# 정식 업로드
+twine upload dist/*
+
+# Username: __token__
+# Password: pypi-AgE... (복사한 토큰 전체)
+```
+
+업로드 성공 후 URL: https://pypi.org/project/uaf/
+
+### 1-5. 전 세계 사용자가 설치 가능!
+
+```bash
+# pipx로 설치 (권장)
+pipx install uaf
+
+# 또는 pip로 설치
+pip install uaf
+
+# 사용
+sudo uaf
+```
+
+---
+
+## 2. APT 패키지 배포 (.deb)
+
+### 2-1. 필요한 도구 설치
+
+```bash
+sudo apt install debhelper dh-python python3-all python3-setuptools
+```
+
+### 2-2. debian/ 디렉토리 구조 생성
+
+```bash
+mkdir -p debian
+cd debian
+```
+
+다음 파일들을 생성해야 합니다:
+
+#### `debian/control`
+
+```
+Source: uaf
+Section: net
+Priority: optional
+Maintainer: devastator-x
+Build-Depends: debhelper-compat (= 13),
+ dh-python,
+ python3-all,
+ python3-setuptools
+Standards-Version: 4.6.0
+Homepage: https://github.com/devastator-x/uaf
+Vcs-Git: https://github.com/devastator-x/uaf.git
+
+Package: uaf
+Architecture: all
+Depends: ${python3:Depends},
+ ${misc:Depends},
+ python3-scapy,
+ iw,
+ iproute2
+Recommends: wireless-tools
+Description: Real-time wireless access point scanner with active reconnaissance
+ UAF is a Linux CLI tool for discovering and analyzing WiFi access
+ points. It supports passive monitoring across 2.4/5/6 GHz bands and active
+ reconnaissance on open networks.
+ .
+ Features:
+ - Real-time AP discovery with channel hopping
+ - Security detection (WEP/WPA/WPA2/WPA3)
+ - Active scanning: DHCP, ARP, port scanning
+ - CSV/JSON export with scan results
+```
+
+#### `debian/rules`
+
+```makefile
+#!/usr/bin/make -f
+
+export PYBUILD_NAME=uaf
+
+%:
+ dh $@ --with python3 --buildsystem=pybuild
+```
+
+#### `debian/changelog`
+
+```
+uaf (1.0.0-1) unstable; urgency=medium
+
+ * Initial release
+ * Passive WiFi scanning with channel hopping
+ * Active reconnaissance on open networks
+ * CSV/JSON export support
+
+ -- devastator-x Tue, 11 Feb 2026 22:00:00 +0900
+```
+
+#### `debian/copyright`
+
+```
+Format: https://www.debian.org/doc/packaging-manuals/copyright-format/1.0/
+Upstream-Name: uaf
+Upstream-Contact: devastator-x
+Source: https://github.com/devastator-x/uaf
+
+Files: *
+Copyright: 2026 devastator-x
+License: GPL-2.0-or-later
+
+License: GPL-2.0-or-later
+ This program is free software; you can redistribute it and/or modify
+ it under the terms of the GNU General Public License as published by
+ the Free Software Foundation; either version 2 of the License, or
+ (at your option) any later version.
+ .
+ This program is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ GNU General Public License for more details.
+ .
+ On Debian systems, the complete text of the GNU General
+ Public License version 2 can be found in "/usr/share/common-licenses/GPL-2".
+```
+
+#### `debian/compat`
+
+```
+13
+```
+
+### 2-3. .deb 패키지 빌드
+
+```bash
+# 프로젝트 루트에서
+dpkg-buildpackage -us -uc -b
+
+# 또는 debuild 사용
+debuild -us -uc -b
+```
+
+빌드 성공 후 상위 디렉토리에 `.deb` 파일이 생성됩니다:
+```
+../uaf_1.0.0-1_all.deb
+```
+
+### 2-4. 로컬 테스트
+
+```bash
+# 설치
+sudo dpkg -i ../uaf_1.0.0-1_all.deb
+
+# 의존성 해결 (필요시)
+sudo apt-get install -f
+
+# 테스트
+sudo uaf --help
+
+# 제거
+sudo apt remove uaf
+```
+
+### 2-5. GitHub Releases에 업로드
+
+1. GitHub 저장소 → Releases → "Create a new release"
+2. Tag: `v1.0.0`
+3. Title: `UAF v1.0.0`
+4. Description: Release notes 작성
+5. `.deb` 파일 업로드 (Assets)
+6. "Publish release" 클릭
+
+사용자는 다음과 같이 설치:
+```bash
+wget https://github.com/devastator-x/uaf/releases/download/v1.0.0/uaf_1.0.0-1_all.deb
+sudo dpkg -i uaf_1.0.0-1_all.deb
+sudo apt-get install -f
+```
+
+### 2-6. PPA (Personal Package Archive) 생성 (고급)
+
+Ubuntu 공식 저장소처럼 `apt install uaf`를 지원하려면:
+
+1. Launchpad 계정 생성: https://launchpad.net/
+2. PPA 생성: https://launchpad.net/~/+activate-ppa
+3. GPG 키 생성 및 등록
+4. 소스 패키지 빌드 & 업로드
+
+자세한 가이드: https://help.launchpad.net/Packaging/PPA
+
+---
+
+## 3. 릴리스 체크리스트
+
+배포 전 확인사항:
+
+- [ ] 모든 테스트 통과 (`pytest tests/ -v`)
+- [ ] README.md 업데이트 (기능, 설치 방법)
+- [ ] CHANGELOG 작성
+- [ ] 버전 번호 업데이트 (`pyproject.toml`)
+- [ ] LICENSE 파일 확인
+- [ ] GitHub 저장소 public으로 설정
+- [ ] .gitignore 확인 (민감 정보 제외)
+
+---
+
+## 4. 문제 해결
+
+### PyPI 업로드 실패
+- 이름 충돌: 다른 이름 선택
+- 버전 충돌: 버전 번호 증가 후 재빌드
+- 인증 실패: API 토큰 재확인
+
+### .deb 빌드 실패
+- 의존성 부족: `debian/control` Dependencies 확인
+- 권한 문제: `debian/rules`에 실행 권한 (`chmod +x debian/rules`)
+
+---
+
+## 5. 다음 릴리스 (v1.1.0)
+
+향후 기능 추가 시:
+
+1. 버전 번호 업데이트 (`pyproject.toml`, `debian/changelog`)
+2. 패키지 재빌드 (`python -m build`, `dpkg-buildpackage`)
+3. 업로드 (`twine upload dist/*`, GitHub Release)
diff --git a/ap_scanner/__init__.py b/ap_scanner/__init__.py
@@ -0,0 +1,3 @@
+"""AP Scanner - airodump-ng style wireless AP scanner."""
+
+__version__ = "1.0.0"
diff --git a/ap_scanner/__main__.py b/ap_scanner/__main__.py
@@ -0,0 +1,435 @@
+"""Entry point: python -m ap_scanner"""
+
+import logging
+import os
+import select
+import signal
+import sys
+import termios
+import threading
+import time
+import tty
+
+from ap_scanner.cli import parse_args
+from ap_scanner.config import apply_config_to_args, load_config
+from ap_scanner.display import Display
+from ap_scanner.exporter import save_csv, save_json
+from ap_scanner.log import restore_stderr_after_scan, setup_logging, suppress_stderr_during_scan
+from ap_scanner.monitor import MonitorMode
+from ap_scanner.scanner import ScannerEngine
+from ap_scanner.utils import (
+ ALL_CHANNELS,
+ CHANNELS_2G,
+ CHANNELS_5G,
+ CHANNELS_6G,
+ check_ip,
+ check_iw,
+ check_root,
+ find_wireless_interfaces,
+ get_supported_channels,
+ parse_channels_arg,
+)
+
+logger = logging.getLogger("ap_scanner.main")
+
+# Key constants for escape sequence parsing
+KEY_UP = "UP"
+KEY_DOWN = "DOWN"
+KEY_ENTER = "ENTER"
+KEY_ESC = "ESC"
+
+
+def _read_key(timeout: float = 1.0) -> str | None:
+ """Read a single key or escape sequence from stdin.
+
+ Uses os.read() on the raw fd to avoid Python buffered IO issues where
+ sys.stdin.read(1) drains the entire escape sequence from the kernel buffer
+ into Python's internal buffer, causing select() to miss the continuation bytes.
+
+ Returns None on timeout.
+ """
+ fd = sys.stdin.fileno()
+
+ try:
+ readable, _, _ = select.select([fd], [], [], timeout)
+ except InterruptedError:
+ return None
+ if not readable:
+ return None
+
+ # Read all available bytes at once — escape sequences (e.g. \x1b[A)
+ # arrive as a single chunk in the kernel buffer
+ data = os.read(fd, 32)
+ if not data:
+ return None
+
+ if data[0] == 0x1B: # ESC
+ if len(data) >= 3 and data[1] == ord("["):
+ if data[2] == ord("A"):
+ return KEY_UP
+ if data[2] == ord("B"):
+ return KEY_DOWN
+ return None # unknown CSI sequence
+ if len(data) == 1:
+ return KEY_ESC
+ return None
+ if data[0] in (0x0D, 0x0A): # \r, \n
+ return KEY_ENTER
+ if data[0] == 0x03: # Ctrl+C
+ return "\x03"
+ return chr(data[0])
+
+
+def main(argv: list[str] | None = None) -> int:
+ args = parse_args(argv)
+
+ # Load config file and apply defaults (CLI args take precedence)
+ config = load_config(args.config)
+ apply_config_to_args(args, config)
+
+ # Setup logging
+ setup_logging(verbose=args.verbose, log_file=args.log_file)
+
+ # Check root
+ if not check_root():
+ logger.error("This tool requires root privileges.")
+ logger.error("Run with: sudo python -m ap_scanner")
+ return 1
+
+ # Check required system tools
+ if not check_iw():
+ logger.error("'iw' command not found. Install with: sudo apt install iw")
+ return 1
+ if not check_ip():
+ logger.error("'ip' command not found. Install with: sudo apt install iproute2")
+ return 1
+
+ # Resolve interface
+ interface = args.interface
+ if interface is None:
+ interfaces = find_wireless_interfaces()
+ if not interfaces:
+ logger.error("No wireless interfaces found.")
+ return 1
+ interface = interfaces[0]
+ logger.info("Auto-detected interface: %s", interface)
+
+ # Resolve channels
+ if args.channels:
+ channels = parse_channels_arg(args.channels)
+ elif args.band == "2.4":
+ channels = CHANNELS_2G
+ elif args.band == "5":
+ channels = CHANNELS_5G
+ elif args.band == "6":
+ channels = CHANNELS_6G
+ else:
+ channels = ALL_CHANNELS
+
+ if not channels:
+ logger.error("No valid channels specified.")
+ return 1
+
+ # Filter to only channels the adapter hardware supports
+ supported = get_supported_channels(interface)
+ supported_set = set(supported)
+ original_count = len(channels)
+ channels = [ch for ch in channels if ch in supported_set]
+ if not channels:
+ logger.warning(
+ "None of the requested channels are supported by %s. "
+ "Using adapter's supported channels instead.", interface,
+ )
+ channels = supported
+ elif len(channels) < original_count:
+ logger.info(
+ "Filtered to %d supported channels (adapter supports %d of %d requested)",
+ len(channels), len(channels), original_count,
+ )
+
+ logger.info("Interface: %s", interface)
+ logger.info("Channels: %d (%d-%d)", len(channels), channels[0], channels[-1])
+
+ # Resolve dwell time
+ dwell_time = args.dwell if args.dwell is not None else 0.5
+
+ # Setup display
+ display = Display(no_color=args.no_color)
+ display.set_sort(args.sort)
+ if args.band != "all":
+ display.band_filter = args.band + "GHz"
+
+ # Enable monitor mode
+ monitor = MonitorMode(interface)
+ try:
+ monitor.enable()
+ except RuntimeError as e:
+ logger.error("%s", e)
+ return 1
+
+ # Create scanner engine
+ engine = ScannerEngine(interface, channels, dwell_time=dwell_time)
+
+ # Signal handler for clean shutdown — use threading.Event for thread safety
+ shutdown_event = threading.Event()
+
+ def signal_handler(signum, frame):
+ shutdown_event.set()
+
+ signal.signal(signal.SIGINT, signal_handler)
+ signal.signal(signal.SIGTERM, signal_handler)
+
+ # Start scanning
+ engine.start()
+ logger.info("Scanning started (Ctrl+C or 'q' to stop)")
+ time.sleep(0.5)
+
+ # Suppress stderr logging during interactive display
+ suppress_stderr_during_scan()
+
+ # Setup terminal for non-blocking key input
+ old_settings = termios.tcgetattr(sys.stdin)
+ try:
+ tty.setcbreak(sys.stdin.fileno())
+ _main_loop(engine, display, monitor, interface, args.write, shutdown_event)
+ finally:
+ # Print newline to separate from display before cleanup logs
+ print()
+ # Re-enable stderr logging for shutdown messages
+ restore_stderr_after_scan()
+
+ # Each cleanup step is individually protected so a failure in one
+ # does not prevent the others from running.
+ try:
+ termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
+ except Exception:
+ logger.error("Failed to restore terminal settings")
+ try:
+ engine.stop()
+ except Exception:
+ logger.error("Failed to stop scanner engine")
+ try:
+ monitor.disable()
+ except Exception:
+ logger.error("Failed to restore interface to managed mode")
+
+ return 0
+
+
+def _main_loop(
+ engine: ScannerEngine,
+ display: Display,
+ monitor: MonitorMode,
+ interface: str,
+ csv_path: str | None,
+ shutdown_event: threading.Event,
+) -> None:
+ """Main display loop with 2-mode key dispatch (normal / selection)."""
+ frozen_aps = None # Snapshot of APs when entering selection mode
+
+ while not shutdown_event.is_set():
+ # In selection mode, use frozen snapshot; otherwise use live data
+ if display._selection_mode and frozen_aps is not None:
+ aps = frozen_aps
+ else:
+ aps = engine.access_points
+
+ display.render(aps, interface, engine.current_channel)
+
+ key = _read_key(timeout=1.0)
+ if key is None:
+ continue
+
+ if display._selection_mode:
+ sel_result = _handle_selection_key(
+ key, display, engine, monitor, interface, shutdown_event, frozen_aps
+ )
+ if sel_result == "exit_selection":
+ frozen_aps = None
+ else:
+ result = _handle_normal_key(key, display, engine, csv_path, shutdown_event)
+ # If we just entered selection mode, freeze the AP list
+ if result == "enter_selection":
+ frozen_aps = engine.access_points
+ elif result == "exit_selection":
+ frozen_aps = None
+
+
+
+def _handle_normal_key(
+ key: str,
+ display: Display,
+ engine: ScannerEngine,
+ csv_path: str | None,
+ shutdown_event: threading.Event,
+) -> str | None:
+ """Handle keys in normal (non-selection) mode. Returns action taken."""
+ if key == "q" or key == "\x03":
+ shutdown_event.set()
+ elif key == "w":
+ try:
+ filepath = save_csv(engine.access_points, csv_path)
+ display.show_message(f"Saved to {filepath}")
+ except (OSError, IOError) as e:
+ display.show_message(f"Save failed: {e}")
+ time.sleep(1)
+ elif key == "j":
+ try:
+ filepath = save_json(engine.access_points)
+ display.show_message(f"Saved JSON to {filepath}")
+ except (OSError, IOError) as e:
+ display.show_message(f"JSON save failed: {e}")
+ time.sleep(1)
+ elif key == "b":
+ band = display.cycle_band()
+ display.show_message(f"Band filter: {band}")
+ time.sleep(0.5)
+ elif key == "s":
+ sort_field = display.cycle_sort()
+ display.show_message(f"Sort by: {sort_field}")
+ time.sleep(0.5)
+ elif key == KEY_ENTER:
+ sorted_aps = display.get_sorted_filtered(engine.access_points)
+ if sorted_aps:
+ display.enter_selection_mode()
+ return "enter_selection"
+ return None
+
+
+def _handle_selection_key(
+ key: str,
+ display: Display,
+ engine: ScannerEngine,
+ monitor: MonitorMode,
+ interface: str,
+ shutdown_event: threading.Event,
+ frozen_aps: dict | None,
+) -> str | None:
+ """Handle keys in AP selection mode. Returns action taken."""
+ if key == KEY_ESC or key == "\x03":
+ display.exit_selection_mode()
+ return "exit_selection"
+ elif key == KEY_UP:
+ sorted_aps = display.get_sorted_filtered(frozen_aps or engine.access_points)
+ display.move_selection(-1, len(sorted_aps))
+ elif key == KEY_DOWN:
+ sorted_aps = display.get_sorted_filtered(frozen_aps or engine.access_points)
+ display.move_selection(1, len(sorted_aps))
+ elif key == KEY_ENTER:
+ sorted_aps = display.get_sorted_filtered(frozen_aps or engine.access_points)
+ if sorted_aps and display._selected_index < len(sorted_aps):
+ selected_ap = sorted_aps[display._selected_index]
+ display.exit_selection_mode()
+ if selected_ap.encryption == "OPN":
+ _run_active_scan(selected_ap, engine, display, monitor, interface)
+ return "exit_selection"
+ else:
+ display.show_message(
+ f"Active scan only available for open (OPN) APs — "
+ f"{selected_ap.display_ssid} uses {selected_ap.encryption}"
+ )
+ time.sleep(2)
+ return "exit_selection"
+ return None
+
+
+def _run_active_scan(ap, engine, display, monitor, interface):
+ """Orchestrate active scan: stop capture, switch to managed, scan, restore."""
+ from ap_scanner.active_scanner import ActiveScanner
+
+ display.set_status(f"Stopping capture for active scan on {ap.display_ssid}...")
+
+ # 1. Stop packet capture + channel hopping
+ engine.stop()
+
+ # 2. Switch to managed mode (fast, no NM restart)
+ display.set_status(f"Switching {interface} to managed mode...")
+ try:
+ monitor.temporarily_disable()
+ except Exception as e:
+ logger.error("Failed to switch to managed mode: %s", e)
+ display.set_status("")
+ engine.start()
+ return
+
+ try:
+ # 3. Run active scan
+ display.set_status(
+ f"Active scanning {ap.display_ssid} ({ap.bssid}) — "
+ f"scan/connect/DHCP/ARP/ports..."
+ )
+ scanner = ActiveScanner(interface)
+ result = scanner.scan(ap)
+
+ # 4. Store result on AP object
+ ap.active_scan_result = result
+
+ # 5. Show overlay with results
+ _show_scan_overlay(display, result)
+
+ # Wait for user to dismiss
+ _read_key(timeout=300)
+ finally:
+ display.set_status("Restoring monitor mode...")
+ # 6. Always restore monitor mode
+ try:
+ monitor.re_enable_monitor()
+ except RuntimeError:
+ logger.critical(
+ "CRITICAL: Failed to restore monitor mode on %s. "
+ "Run manually: sudo iw dev %s set type monitor",
+ interface, interface,
+ )
+ display.set_status("CRITICAL: Monitor mode restore failed!")
+ time.sleep(3)
+
+ # 7. Restart capture
+ display.set_status("")
+ engine.start()
+
+
+def _show_scan_overlay(display, result):
+ """Build and render the active scan results overlay."""
+ lines = []
+
+ if result.error:
+ lines.append(f"Error: {result.error}")
+ lines.append("")
+
+ if result.assigned_ip:
+ lines.append(f"Assigned IP: {result.assigned_ip}")
+ if result.subnet_mask:
+ lines.append(f"Subnet Mask: {result.subnet_mask}")
+ if result.gateway_ip:
+ gw = result.gateway_ip
+ if result.gateway_mac:
+ gw += f" ({result.gateway_mac})"
+ lines.append(f"Gateway: {gw}")
+ if result.dhcp_server:
+ lines.append(f"DHCP Server: {result.dhcp_server}")
+ if result.dns_servers:
+ lines.append(f"DNS Servers: {', '.join(result.dns_servers)}")
+ lines.append("")
+
+ if result.gateway_open_ports:
+ ports = ", ".join(str(p) for p in result.gateway_open_ports)
+ lines.append(f"Gateway Ports: {ports}")
+ lines.append("")
+
+ if result.clients:
+ lines.append(f"Connected Clients: {result.client_count}")
+ for client in result.clients[:10]:
+ vendor = f" {client.vendor}" if client.vendor else ""
+ lines.append(f" {client.ip:<16s} {client.mac}{vendor}")
+ if result.client_count > 10:
+ lines.append(f" ... +{result.client_count - 10} more")
+ lines.append("")
+
+ lines.append(f"Duration: {result.duration_secs}s")
+
+ title = f"ACTIVE SCAN: {result.ssid} ({result.bssid})"
+ display.render_overlay(lines, title)
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/ap_scanner/active_scanner.py b/ap_scanner/active_scanner.py
@@ -0,0 +1,375 @@
+"""Active scanner: connect to open APs and gather network intelligence."""
+
+import logging
+import re
+import socket
+import struct
+import subprocess
+import time
+from datetime import datetime
+from pathlib import Path
+
+from ap_scanner.models import AccessPoint, ActiveScanClient, ActiveScanResult
+
+logger = logging.getLogger("ap_scanner.active_scanner")
+
+# Common gateway ports to probe
+GATEWAY_PORTS = [22, 53, 80, 443, 8080, 8443]
+
+# Timeouts
+DHCP_TIMEOUT = 15
+CONNECT_TIMEOUT = 10
+PORT_SCAN_TIMEOUT = 1
+ARP_TIMEOUT = 3
+OVERALL_TIMEOUT = 90
+
+
+class ActiveScanner:
+ """Performs active reconnaissance on open (OPN) access points.
+
+ Sequence: connect -> DHCP -> parse network info -> ARP scan -> port scan -> disconnect.
+ """
+
+ def __init__(self, interface: str):
+ self.interface = interface
+
+ def scan(self, ap: AccessPoint) -> ActiveScanResult:
+ """Run full active scan on an open AP. Returns results even on partial failure."""
+ start = time.monotonic()
+ result = ActiveScanResult(bssid=ap.bssid, ssid=ap.display_ssid)
+
+ def _timed_out() -> bool:
+ return (time.monotonic() - start) >= OVERALL_TIMEOUT
+
+ try:
+ # Step 1: Connect
+ if not self._connect(ap):
+ result.error = "Failed to connect to AP"
+ return result
+ result.connected = True
+
+ if _timed_out():
+ result.error = "Overall timeout exceeded after connect"
+ return result
+
+ # Step 2: DHCP
+ if not self._acquire_dhcp():
+ result.error = "DHCP failed — no IP assigned"
+ self._disconnect()
+ return result
+
+ if _timed_out():
+ result.error = "Overall timeout exceeded after DHCP"
+ return result
+
+ # Step 3: Parse network info
+ self._parse_network_info(result)
+
+ # Step 4: Resolve gateway MAC
+ if result.gateway_ip and not _timed_out():
+ self._resolve_gateway_mac(result)
+
+ # Step 5: ARP scan (best-effort)
+ if result.assigned_ip and result.subnet_mask and not _timed_out():
+ try:
+ self._arp_scan(result)
+ except Exception as e:
+ logger.warning("ARP scan failed: %s", e)
+
+ # Step 6: Port scan gateway (best-effort)
+ if result.gateway_ip and not _timed_out():
+ try:
+ self._port_scan(result)
+ except Exception as e:
+ logger.warning("Port scan failed: %s", e)
+
+ except Exception as e:
+ result.error = str(e)
+ logger.error("Active scan error: %s", e)
+ finally:
+ result.duration_secs = round(time.monotonic() - start, 1)
+ self._disconnect()
+
+ return result
+
+ # --- Connection ---
+
+ def _connect(self, ap: AccessPoint) -> bool:
+ """Connect to an open AP using iw, falling back to wpa_supplicant."""
+ ssid = ap.ssid
+ if not ssid:
+ logger.warning("Cannot connect: AP has empty/hidden SSID")
+ return False
+
+ # Prevent NetworkManager from grabbing the interface while we connect.
+ self._run(
+ ["nmcli", "device", "set", self.interface, "managed", "no"],
+ timeout=5, ignore_errors=True,
+ )
+
+ # Trigger a scan so the AP appears in the driver's BSS list.
+ # Without this, 'iw connect' often fails with "not found" because
+ # the driver has no cached scan results after a mode switch.
+ logger.info("Scanning for %s on %s ...", ssid, self.interface)
+ scan_ret = self._run(
+ ["iw", "dev", self.interface, "scan"],
+ timeout=15, ignore_errors=True,
+ )
+ if ssid in scan_ret.stdout:
+ logger.info("AP '%s' found in scan results", ssid)
+ else:
+ logger.warning("AP '%s' not found in scan results — connecting anyway", ssid)
+
+ # Method 1: iw connect with BSSID + frequency for precise targeting
+ freq = str(ap.frequency) if ap.frequency else ""
+ iw_cmd = ["iw", "dev", self.interface, "connect", ssid]
+ if freq:
+ iw_cmd.append(freq)
+ iw_cmd.append(ap.bssid)
+
+ logger.info("Attempting: %s", " ".join(iw_cmd))
+ ret = self._run(iw_cmd, timeout=CONNECT_TIMEOUT)
+ if ret.returncode == 0:
+ for _ in range(5):
+ time.sleep(1)
+ if self._is_connected():
+ logger.info("Connected to %s via iw", ssid)
+ return True
+ logger.info("iw connect returned 0 but not associated after 5s")
+ else:
+ logger.info("iw connect failed: %s", ret.stderr.strip())
+
+ # Method 2: wpa_supplicant with key_mgmt=NONE + bssid
+ logger.info("Falling back to wpa_supplicant for '%s' ...", ssid)
+ # Sanitize SSID for wpa_supplicant config (escape \ and " to prevent injection)
+ safe_ssid = ssid.replace("\\", "\\\\").replace('"', '\\"')
+ conf = (
+ f'network={{\n'
+ f' ssid="{safe_ssid}"\n'
+ f' bssid={ap.bssid}\n'
+ f' key_mgmt=NONE\n'
+ f' scan_ssid=1\n'
+ f'}}\n'
+ )
+ conf_path = Path(f"/tmp/ap_scanner_wpa_{self.interface}.conf")
+ conf_path.write_text(conf)
+
+ # Kill any existing wpa_supplicant on this interface
+ self._run(["pkill", "-f", f"wpa_supplicant.*{self.interface}"],
+ timeout=3, ignore_errors=True)
+ time.sleep(0.5)
+
+ ret = self._run(
+ ["wpa_supplicant", "-B", "-i", self.interface,
+ "-c", str(conf_path), "-D", "nl80211"],
+ timeout=CONNECT_TIMEOUT, ignore_errors=True,
+ )
+ logger.info("wpa_supplicant launch: rc=%d stderr=%s",
+ ret.returncode, ret.stderr.strip())
+
+ for _ in range(8):
+ time.sleep(1)
+ if self._is_connected():
+ logger.info("Connected to %s via wpa_supplicant", ssid)
+ return True
+
+ # Log diagnostics on failure
+ link = self._run(["iw", "dev", self.interface, "link"],
+ timeout=5, ignore_errors=True)
+ logger.warning(
+ "Failed to connect to %s — link status: %s",
+ ssid, link.stdout.strip(),
+ )
+ return False
+
+ def _is_connected(self) -> bool:
+ """Check if interface is associated to an AP."""
+ ret = self._run(["iw", "dev", self.interface, "link"], timeout=5)
+ return ret.returncode == 0 and "Connected to" in ret.stdout
+
+ # --- DHCP ---
+
+ def _acquire_dhcp(self) -> bool:
+ """Acquire an IP via DHCP (dhclient, falling back to udhcpc)."""
+ # Try dhclient
+ ret = self._run(
+ ["dhclient", "-1", "-v", "-timeout", str(DHCP_TIMEOUT), self.interface],
+ timeout=DHCP_TIMEOUT + 5, ignore_errors=True,
+ )
+ if ret.returncode == 0 and self._has_ip():
+ logger.info("DHCP lease acquired via dhclient")
+ return True
+
+ # Fallback: udhcpc
+ ret = self._run(
+ ["udhcpc", "-i", self.interface, "-t", "5", "-T", "3", "-n", "-q"],
+ timeout=DHCP_TIMEOUT + 5, ignore_errors=True,
+ )
+ if ret.returncode == 0 and self._has_ip():
+ logger.info("DHCP lease acquired via udhcpc")
+ return True
+
+ logger.warning("DHCP failed on %s", self.interface)
+ return False
+
+ def _has_ip(self) -> bool:
+ """Check if the interface has an IPv4 address."""
+ ret = self._run(["ip", "-4", "addr", "show", self.interface], timeout=5)
+ return "inet " in ret.stdout
+
+ # --- Network info parsing ---
+
+ def _parse_network_info(self, result: ActiveScanResult) -> None:
+ """Parse IP, subnet, gateway, DNS from system commands."""
+ # IP and subnet
+ ret = self._run(["ip", "-4", "addr", "show", self.interface], timeout=5)
+ m = re.search(r"inet (\d+\.\d+\.\d+\.\d+)/(\d+)", ret.stdout)
+ if m:
+ result.assigned_ip = m.group(1)
+ result.subnet_mask = _cidr_to_netmask(int(m.group(2)))
+
+ # Default gateway
+ ret = self._run(["ip", "route", "show", "dev", self.interface], timeout=5)
+ m = re.search(r"default via (\d+\.\d+\.\d+\.\d+)", ret.stdout)
+ if m:
+ result.gateway_ip = m.group(1)
+
+ # DNS servers
+ try:
+ resolv = Path("/etc/resolv.conf").read_text()
+ result.dns_servers = re.findall(r"nameserver\s+(\d+\.\d+\.\d+\.\d+)", resolv)
+ except OSError:
+ pass
+
+ # DHCP server (from dhclient lease file)
+ for lease_path in Path("/var/lib/dhcp").glob("*.leases"):
+ try:
+ content = lease_path.read_text()
+ m = re.search(r"dhcp-server-identifier\s+(\d+\.\d+\.\d+\.\d+)", content)
+ if m:
+ result.dhcp_server = m.group(1)
+ break
+ except OSError:
+ continue
+
+ # --- ARP operations ---
+
+ def _resolve_gateway_mac(self, result: ActiveScanResult) -> None:
+ """Resolve gateway MAC via a single ARP request using scapy."""
+ try:
+ from scapy.all import ARP, Ether, sr1
+
+ pkt = Ether(dst="ff:ff:ff:ff:ff:ff") / ARP(
+ op="who-has", pdst=result.gateway_ip
+ )
+ resp = sr1(pkt, iface=self.interface, timeout=ARP_TIMEOUT, verbose=0)
+ if resp and resp.haslayer(ARP):
+ result.gateway_mac = resp[ARP].hwsrc
+ except Exception as e:
+ logger.warning("Gateway MAC resolution failed: %s", e)
+
+ def _arp_scan(self, result: ActiveScanResult) -> None:
+ """ARP scan the local subnet to discover connected clients."""
+ from scapy.all import ARP, Ether, srp
+
+ network = _calculate_network(result.assigned_ip, result.subnet_mask)
+ if not network:
+ return
+
+ pkt = Ether(dst="ff:ff:ff:ff:ff:ff") / ARP(op="who-has", pdst=network)
+ answered, _ = srp(pkt, iface=self.interface, timeout=ARP_TIMEOUT, verbose=0)
+
+ for _, recv in answered:
+ if recv.haslayer(ARP):
+ ip = recv[ARP].psrc
+ mac = recv[ARP].hwsrc
+ # Skip our own IP
+ if ip == result.assigned_ip:
+ continue
+ result.clients.append(ActiveScanClient(ip=ip, mac=mac))
+
+ # --- Port scan ---
+
+ def _port_scan(self, result: ActiveScanResult) -> None:
+ """TCP connect scan on common gateway ports."""
+ for port in GATEWAY_PORTS:
+ try:
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+ sock.settimeout(PORT_SCAN_TIMEOUT)
+ if sock.connect_ex((result.gateway_ip, port)) == 0:
+ result.gateway_open_ports.append(port)
+ except (OSError, socket.error):
+ pass
+
+ # --- Disconnect / Cleanup ---
+
+ def _disconnect(self) -> None:
+ """Release DHCP, kill wpa_supplicant, disconnect, flush IP."""
+ self._run(["dhclient", "-r", self.interface],
+ timeout=5, ignore_errors=True)
+ self._run(["pkill", "-f", f"wpa_supplicant.*{self.interface}"],
+ timeout=3, ignore_errors=True)
+ self._run(["iw", "dev", self.interface, "disconnect"],
+ timeout=5, ignore_errors=True)
+ self._run(["ip", "addr", "flush", "dev", self.interface],
+ timeout=5, ignore_errors=True)
+ # Re-enable NetworkManager management
+ self._run(["nmcli", "device", "set", self.interface, "managed", "yes"],
+ timeout=5, ignore_errors=True)
+ # Clean up temp config
+ conf_path = Path(f"/tmp/ap_scanner_wpa_{self.interface}.conf")
+ conf_path.unlink(missing_ok=True)
+
+ # --- Utilities ---
+
+ @staticmethod
+ def _run(
+ cmd: list[str],
+ timeout: int = 10,
+ ignore_errors: bool = False,
+ ) -> subprocess.CompletedProcess:
+ """Run a command, returning CompletedProcess."""
+ try:
+ return subprocess.run(
+ cmd, capture_output=True, text=True, timeout=timeout,
+ )
+ except FileNotFoundError:
+ if not ignore_errors:
+ raise
+ return subprocess.CompletedProcess(cmd, 127, "", f"{cmd[0]}: not found")
+ except subprocess.TimeoutExpired:
+ if not ignore_errors:
+ raise
+ return subprocess.CompletedProcess(cmd, 124, "", "timeout")
+
+
+# --- Module-level utility functions ---
+
+def _cidr_to_netmask(prefix: int) -> str:
+ """Convert CIDR prefix length to dotted netmask (e.g. 24 -> 255.255.255.0)."""
+ bits = (0xFFFFFFFF << (32 - prefix)) & 0xFFFFFFFF
+ return socket.inet_ntoa(struct.pack("!I", bits))
+
+
+def _calculate_network(ip: str, netmask: str) -> str:
+ """Calculate network address in CIDR notation for ARP scan target."""
+ try:
+ ip_int = struct.unpack("!I", socket.inet_aton(ip))[0]
+ mask_int = struct.unpack("!I", socket.inet_aton(netmask))[0]
+ net_int = ip_int & mask_int
+ # Count prefix bits
+ prefix = bin(mask_int).count("1")
+ net_addr = socket.inet_ntoa(struct.pack("!I", net_int))
+ return f"{net_addr}/{prefix}"
+ except (OSError, struct.error):
+ return ""
+
+
+def _is_ip(s: str) -> bool:
+ """Check if string is a valid IPv4 address."""
+ try:
+ socket.inet_aton(s)
+ parts = s.split(".")
+ return len(parts) == 4 and all(0 <= int(p) <= 255 for p in parts)
+ except (OSError, ValueError):
+ return False
diff --git a/ap_scanner/cli.py b/ap_scanner/cli.py
@@ -0,0 +1,81 @@
+"""Command-line argument parsing."""
+
+import argparse
+
+from ap_scanner import __version__
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+ """Parse command-line arguments."""
+ parser = argparse.ArgumentParser(
+ prog="ap-scanner",
+ description="Real-time wireless AP scanner for Linux",
+ )
+ parser.add_argument(
+ "-V", "--version",
+ action="version",
+ version=f"%(prog)s {__version__}",
+ )
+ parser.add_argument(
+ "-i", "--interface",
+ type=str,
+ default=None,
+ help="Wireless interface to use (default: auto-detect)",
+ )
+ parser.add_argument(
+ "-c", "--channels",
+ type=str,
+ default=None,
+ help="Channels to scan, e.g. '1,6,11' or '36-48' (default: all)",
+ )
+ parser.add_argument(
+ "-b", "--band",
+ type=str,
+ choices=["2.4", "5", "6", "all"],
+ default="all",
+ help="Band filter: 2.4, 5, 6 (6GHz), or all (default: all)",
+ )
+ parser.add_argument(
+ "-w", "--write",
+ type=str,
+ default=None,
+ help="CSV output file path (auto-saves on exit)",
+ )
+ parser.add_argument(
+ "--sort",
+ type=str,
+ choices=["rssi", "channel", "essid", "enc"],
+ default="rssi",
+ help="Sort by field (default: rssi)",
+ )
+ parser.add_argument(
+ "--no-color",
+ action="store_true",
+ default=False,
+ help="Disable colored output",
+ )
+ parser.add_argument(
+ "-v", "--verbose",
+ action="store_true",
+ default=False,
+ help="Enable verbose (DEBUG level) logging",
+ )
+ parser.add_argument(
+ "--log-file",
+ type=str,
+ default=None,
+ help="Write log output to file",
+ )
+ parser.add_argument(
+ "--dwell",
+ type=float,
+ default=None,
+ help="Channel dwell time in seconds (default: 0.5)",
+ )
+ parser.add_argument(
+ "--config",
+ type=str,
+ default=None,
+ help="Path to config file (default: ~/.config/ap-scanner/config.ini)",
+ )
+ return parser.parse_args(argv)
diff --git a/ap_scanner/config.py b/ap_scanner/config.py
@@ -0,0 +1,91 @@
+"""Configuration file support using configparser.
+
+Default path: ~/.config/ap-scanner/config.ini
+CLI arguments always take precedence over config file values.
+"""
+
+import configparser
+from pathlib import Path
+
+DEFAULT_CONFIG_PATH = Path.home() / ".config" / "ap-scanner" / "config.ini"
+
+# Defaults that match CLI defaults
+DEFAULTS = {
+ "interface": "",
+ "channels": "",
+ "band": "all",
+ "sort": "rssi",
+ "no_color": "false",
+ "verbose": "false",
+ "log_file": "",
+ "write": "",
+ "dwell": "",
+}
+
+
+def load_config(config_path: str | None = None) -> dict[str, str]:
+ """Load configuration from an INI file.
+
+ Returns a dict of setting name -> string value.
+ Missing keys are filled with defaults.
+ """
+ path = Path(config_path) if config_path else DEFAULT_CONFIG_PATH
+ result = dict(DEFAULTS)
+
+ if not path.is_file():
+ return result
+
+ parser = configparser.ConfigParser()
+ parser.read(path, encoding="utf-8")
+
+ if parser.has_section("scanner"):
+ for key in DEFAULTS:
+ if parser.has_option("scanner", key):
+ result[key] = parser.get("scanner", key)
+
+ return result
+
+
+def apply_config_to_args(args, config: dict[str, str]) -> None:
+ """Apply config file values to argparse namespace where CLI didn't override.
+
+ CLI arguments (non-default values) always take precedence.
+ """
+ # interface: CLI default is None
+ if args.interface is None and config["interface"]:
+ args.interface = config["interface"]
+
+ # channels: CLI default is None
+ if args.channels is None and config["channels"]:
+ args.channels = config["channels"]
+
+ # band: CLI default is "all"
+ if args.band == "all" and config["band"] != "all":
+ args.band = config["band"]
+
+ # sort: CLI default is "rssi"
+ if args.sort == "rssi" and config["sort"] != "rssi":
+ args.sort = config["sort"]
+
+ # no_color: CLI default is False
+ if not args.no_color and config["no_color"].lower() == "true":
+ args.no_color = True
+
+ # verbose: CLI default is False
+ if not args.verbose and config["verbose"].lower() == "true":
+ args.verbose = True
+
+ # log_file: CLI default is None
+ if args.log_file is None and config["log_file"]:
+ args.log_file = config["log_file"]
+
+ # write: CLI default is None
+ if args.write is None and config["write"]:
+ args.write = config["write"]
+
+ # dwell: CLI default is None
+ if getattr(args, "dwell", None) is None and config.get("dwell"):
+ try:
+ args.dwell = float(config["dwell"])
+ except ValueError:
+ pass
diff --git a/ap_scanner/display.py b/ap_scanner/display.py
@@ -0,0 +1,389 @@
+"""Terminal display: real-time AP listing with ANSI escape sequences."""
+
+import os
+import sys
+import unicodedata
+from datetime import datetime, timedelta
+
+from ap_scanner.models import AccessPoint
+
+
+def _display_width(s: str) -> int:
+ """Calculate terminal display width accounting for wide (CJK) characters."""
+ width = 0
+ for ch in s:
+ eaw = unicodedata.east_asian_width(ch)
+ width += 2 if eaw in ("W", "F") else 1
+ return width
+
+
+def _truncate_to_width(s: str, max_width: int) -> str:
+ """Truncate string to fit within max_width terminal cells, adding ellipsis if needed."""
+ width = 0
+ for i, ch in enumerate(s):
+ eaw = unicodedata.east_asian_width(ch)
+ char_width = 2 if eaw in ("W", "F") else 1
+ if width + char_width > max_width:
+ return s[:i] + "\u2026"
+ width += char_width
+ return s
+
+
+def _pad_to_width(s: str, target_width: int) -> str:
+ """Pad string with spaces to reach target terminal width."""
+ current = _display_width(s)
+ return s + " " * max(0, target_width - current)
+
+
+# ANSI color codes
+class Colors:
+ RESET = "\033[0m"
+ BOLD = "\033[1m"
+ DIM = "\033[2m"
+ RED = "\033[31m"
+ GREEN = "\033[32m"
+ YELLOW = "\033[33m"
+ BLUE = "\033[34m"
+ MAGENTA = "\033[35m"
+ CYAN = "\033[36m"
+ WHITE = "\033[37m"
+ BG_BLUE = "\033[44m"
+ REVERSE = "\033[7m"
+
+
+class NoColors:
+ """No-op colors for --no-color mode."""
+ RESET = BOLD = DIM = ""
+ RED = GREEN = YELLOW = BLUE = MAGENTA = CYAN = WHITE = ""
+ BG_BLUE = REVERSE = ""
+
+
+class Display:
+ """Renders AP scan results to the terminal in airodump-ng style."""
+
+ SORT_KEYS = {
+ "rssi": lambda ap: (ap.rssi if ap.rssi is not None else -999, ap.bssid),
+ "channel": lambda ap: (ap.channel, ap.bssid),
+ "essid": lambda ap: (ap.display_ssid.lower(), ap.bssid),
+ "enc": lambda ap: (ap.encryption, ap.bssid),
+ }
+
+ SORT_REVERSE = {
+ "rssi": True, # strongest first (-42 > -80)
+ "channel": False,
+ "essid": False,
+ "enc": False,
+ }
+
+ NEW_AP_HIGHLIGHT_SECS = 5
+
+ def __init__(self, no_color: bool = False):
+ self.c = NoColors if no_color else Colors
+ self.sort_field = "rssi"
+ self.band_filter: str | None = None # None = all, "2.4GHz", "5GHz"
+ self._start_time = datetime.now()
+ self._first_render = True
+ self._known_bssids: set[str] = set()
+ self._new_bssids: dict[str, datetime] = {} # bssid -> discovered time
+ self._selected_index: int = 0
+ self._selection_mode: bool = False
+ self._status_message: str = ""
+
+ def set_sort(self, field: str) -> None:
+ if field in self.SORT_KEYS:
+ self.sort_field = field
+
+ def cycle_sort(self) -> str:
+ """Cycle through sort fields. Returns the new sort field name."""
+ fields = list(self.SORT_KEYS.keys())
+ idx = fields.index(self.sort_field)
+ self.sort_field = fields[(idx + 1) % len(fields)]
+ return self.sort_field
+
+ def cycle_band(self) -> str:
+ """Cycle through band filters: all -> 2.4GHz -> 5GHz -> 6GHz -> all."""
+ cycle = [None, "2.4GHz", "5GHz", "6GHz"]
+ try:
+ idx = cycle.index(self.band_filter)
+ except ValueError:
+ idx = 0
+ self.band_filter = cycle[(idx + 1) % len(cycle)]
+ return self.band_filter or "all"
+
+ def enter_selection_mode(self) -> None:
+ """Enter AP selection mode for interactive navigation."""
+ self._selection_mode = True
+ self._selected_index = 0
+
+ def exit_selection_mode(self) -> None:
+ """Exit AP selection mode."""
+ self._selection_mode = False
+ self._selected_index = 0
+
+ def move_selection(self, delta: int, total: int) -> None:
+ """Move selection cursor by delta, clamping to valid range."""
+ if total <= 0:
+ self._selected_index = 0
+ return
+ self._selected_index = max(0, min(total - 1, self._selected_index + delta))
+
+ def get_sorted_filtered(self, aps: dict[str, "AccessPoint"]) -> list["AccessPoint"]:
+ """Return the currently filtered and sorted AP list (matches render order)."""
+ filtered = aps.values()
+ if self.band_filter:
+ filtered = [ap for ap in filtered if ap.band == self.band_filter]
+ else:
+ filtered = list(filtered)
+ sort_key = self.SORT_KEYS[self.sort_field]
+ reverse = self.SORT_REVERSE[self.sort_field]
+ return sorted(filtered, key=sort_key, reverse=reverse)
+
+ def set_status(self, msg: str) -> None:
+ """Set a persistent status message (cleared on next render if empty)."""
+ self._status_message = msg
+
+ def render(
+ self,
+ aps: dict[str, AccessPoint],
+ interface: str,
+ current_channel: int,
+ ) -> None:
+ """Clear screen and render the AP table."""
+ c = self.c
+ term_width, term_height = os.get_terminal_size()
+
+ # Reuse get_sorted_filtered to avoid duplicating filter+sort logic
+ sorted_aps = self.get_sorted_filtered(aps)
+
+ # Elapsed time
+ elapsed = datetime.now() - self._start_time
+ elapsed_str = str(timedelta(seconds=int(elapsed.total_seconds())))
+
+ # Count stats in single pass
+ total = len(aps)
+ count_2g = count_5g = count_6g = count_hidden = 0
+ for ap in aps.values():
+ band = ap.band
+ if band == "2.4GHz":
+ count_2g += 1
+ elif band == "5GHz":
+ count_5g += 1
+ elif band == "6GHz":
+ count_6g += 1
+ if ap.is_hidden:
+ count_hidden += 1
+
+ # Build output
+ lines: list[str] = []
+
+ # First render: clear entire screen; subsequent: just move cursor home
+ if self._first_render:
+ lines.append("\033[2J\033[H")
+ self._first_render = False
+ else:
+ lines.append("\033[H")
+
+ # Header
+ band_str = f"Band: {self.band_filter or 'all'}"
+ sort_str = f"Sort: {self.sort_field}"
+ header_text = (
+ f" AP Scanner v1.0 | {interface} | Monitor | "
+ f"CH: {current_channel:<3} | {elapsed_str} | "
+ f"{band_str} | {sort_str}"
+ )
+ header = f" {c.BOLD}{c.WHITE}{c.BG_BLUE}{header_text}"
+ padding = max(0, term_width - len(header_text) - 1)
+ header += " " * padding + c.RESET
+ lines.append(header)
+ lines.append("")
+
+ # Column headers
+ col_header = (
+ f" {c.BOLD}{c.CYAN}"
+ f"{'SSID':<24s} {'BSSID':<18s} {'RSSI':>5s} "
+ f"{'CH':<16s} {'Band':<6s} {'W':>3s} "
+ f"{'802.11':<14s} "
+ f"{'ENC':<10s} {'WPA2':<12s} {'WPA3':<12s} "
+ f"{'WPS':<4s} {'Vendor'}"
+ f"{c.RESET}"
+ )
+ lines.append(col_header)
+
+ # Separator
+ lines.append(f" {c.DIM}{'-' * min(term_width - 2, 150)}{c.RESET}")
+
+ # AP rows
+ max_rows = max(term_height - 9, 5)
+ now = datetime.now()
+
+ # Track newly discovered APs
+ for bssid in aps:
+ if bssid not in self._known_bssids:
+ self._known_bssids.add(bssid)
+ self._new_bssids[bssid] = now
+ # Expire old highlights
+ expired = [b for b, t in self._new_bssids.items()
+ if (now - t).total_seconds() > self.NEW_AP_HIGHLIGHT_SECS]
+ for b in expired:
+ del self._new_bssids[b]
+
+ for i, ap in enumerate(sorted_aps[:max_rows]):
+ rssi_str = f"{ap.rssi:>4d}" if ap.rssi is not None else " N/A"
+ rssi_color = self._rssi_color(ap.rssi)
+
+ is_open = ap.encryption == "OPN"
+ # Highlight newly discovered APs
+ is_new = ap.bssid in self._new_bssids
+ # Dim APs not seen in last 30 seconds
+ stale = (now - ap.last_seen) > timedelta(seconds=30)
+
+ # Selection mode: selected row gets REVERSE video
+ if self._selection_mode and i == self._selected_index:
+ row_prefix = c.REVERSE
+ elif is_new:
+ row_prefix = f"{c.BOLD}{c.GREEN}"
+ elif stale:
+ row_prefix = c.DIM
+ elif is_open:
+ row_prefix = f"{c.BOLD}{c.RED}"
+ else:
+ row_prefix = ""
+
+ ssid = _truncate_to_width(ap.display_ssid, 23)
+ ssid_padded = _pad_to_width(ssid, 24)
+ if ap.is_hidden and not is_new:
+ ssid_str = f"{c.RED}{ssid_padded}{c.RESET}{row_prefix}"
+ else:
+ ssid_str = ssid_padded
+
+ enc_color = self._enc_color(ap.encryption)
+ vendor = ap.vendor[:20] if ap.vendor else ""
+
+ line = (
+ f" {row_prefix}"
+ f"{ssid_str}"
+ f"{ap.bssid:<18s} "
+ f"{rssi_color}{rssi_str}{c.RESET}{row_prefix} "
+ f"{ap.channel_display:<16s} "
+ f"{ap.band:<6s} {ap.width:>3d} "
+ f"{ap.standard_display:<14s} "
+ f"{enc_color}{ap.encryption:<10s}{c.RESET}{row_prefix} "
+ f"{ap.wpa2:<12s} {ap.wpa3:<12s} "
+ f"{ap.wps:<4s} {vendor}"
+ f"{c.RESET}"
+ )
+ lines.append(line)
+
+ # Remaining count
+ if len(sorted_aps) > max_rows:
+ remaining = len(sorted_aps) - max_rows
+ lines.append(f" {c.DIM}... +{remaining} more APs{c.RESET}")
+
+ # Padding to push footer to bottom (status + stats + keys = 4 lines)
+ current_lines = len(lines)
+ for _ in range(max(0, term_height - current_lines - 4)):
+ lines.append("")
+
+ # Status message (e.g., active scan progress)
+ if self._status_message:
+ lines.append(f" {c.BOLD}{c.YELLOW}{self._status_message}{c.RESET}")
+ else:
+ lines.append("")
+
+ # Stats line
+ stats = (
+ f" {c.BOLD}"
+ f"APs: {c.WHITE}{total}{c.RESET}{c.BOLD} | "
+ f"2.4GHz: {c.GREEN}{count_2g}{c.RESET}{c.BOLD} | "
+ f"5GHz: {c.CYAN}{count_5g}{c.RESET}{c.BOLD} | "
+ f"6GHz: {c.MAGENTA}{count_6g}{c.RESET}{c.BOLD} | "
+ f"Hidden: {c.RED}{count_hidden}{c.RESET}"
+ )
+ lines.append(stats)
+
+ # Keybindings (mode-dependent)
+ if self._selection_mode:
+ keys = (
+ f" {c.DIM}"
+ f"[\u2191/\u2193] Navigate | [Enter] Active Scan | [Esc] Cancel"
+ f"{c.RESET}"
+ )
+ else:
+ keys = (
+ f" {c.DIM}"
+ f"[q] Quit | [w] CSV | [j] JSON | "
+ f"[b] Band | [s] Sort | [Enter] Select AP"
+ f"{c.RESET}"
+ )
+ lines.append(keys)
+
+ # Append clear-to-end-of-line on every line so leftover text is erased,
+ # then clear any remaining lines below the last one we wrote.
+ eol = "\033[K"
+ output = ("\n" + eol).join(lines)
+ # Erase from current position to bottom of screen
+ output += "\033[J"
+ sys.stdout.write(output)
+ sys.stdout.flush()
+
+ def _rssi_color(self, rssi: int | None) -> str:
+ c = self.c
+ if rssi is None:
+ return c.DIM
+ if rssi >= -50:
+ return c.GREEN
+ if rssi >= -70:
+ return c.YELLOW
+ return c.RED
+
+ def _enc_color(self, encryption: str) -> str:
+ c = self.c
+ if encryption == "OPN":
+ return c.RED
+ if encryption in ("WPA3", "WPA2/WPA3"):
+ return c.GREEN
+ if encryption == "WPA2":
+ return c.CYAN
+ if encryption == "WPA":
+ return c.YELLOW
+ if encryption == "WEP":
+ return c.RED
+ return ""
+
+ def render_overlay(self, lines_content: list[str], title: str = "") -> None:
+ """Render a centered overlay panel on top of the AP table."""
+ c = self.c
+ term_width, term_height = os.get_terminal_size()
+ box_width = min(term_width - 4, 72)
+ start_row = max(3, (term_height - len(lines_content) - 4) // 2)
+
+ border_h = "+" + "-" * (box_width - 2) + "+"
+
+ sys.stdout.write(f"\033[{start_row};1H")
+ sys.stdout.write(f" {c.BOLD}{c.CYAN}{border_h}{c.RESET}\033[K\n")
+
+ if title:
+ title_line = f"| {title}"
+ pad = box_width - 3 - _display_width(title)
+ title_line += " " * max(0, pad) + "|"
+ sys.stdout.write(f" {c.BOLD}{c.CYAN}{title_line}{c.RESET}\033[K\n")
+ sys.stdout.write(f" {c.BOLD}{c.CYAN}|{' ' * (box_width - 2)}|{c.RESET}\033[K\n")
+
+ for line in lines_content:
+ display_line = _truncate_to_width(line, box_width - 4)
+ pad = box_width - 2 - _display_width(display_line) - 1
+ row = f"| {display_line}" + " " * max(0, pad) + "|"
+ sys.stdout.write(f" {c.CYAN}{row}{c.RESET}\033[K\n")
+
+ sys.stdout.write(f" {c.BOLD}{c.CYAN}{border_h}{c.RESET}\033[K\n")
+ sys.stdout.write(
+ f" {c.DIM}Press any key to dismiss{c.RESET}\033[K"
+ )
+ sys.stdout.flush()
+
+ def show_message(self, msg: str) -> None:
+ """Show a temporary message at the bottom of the screen."""
+ c = self.c
+ _, term_height = os.get_terminal_size()
+ sys.stdout.write(f"\033[{term_height};0H\033[2K {c.BOLD}{c.GREEN}{msg}{c.RESET}")
+ sys.stdout.flush()
diff --git a/ap_scanner/exporter.py b/ap_scanner/exporter.py
@@ -0,0 +1,140 @@
+"""CSV and JSON export for scan results."""
+
+import csv
+import json
+from datetime import datetime
+from pathlib import Path
+
+from ap_scanner.models import AccessPoint
+
+
+def generate_filename(ext: str = "csv") -> str:
+ """Generate a default filename with timestamp."""
+ ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+ return f"ap_scan_{ts}.{ext}"
+
+
+def save_csv(aps: dict[str, AccessPoint], filepath: str | None = None) -> str:
+ """Save AP scan results to CSV file in wifi-inspector compatible format.
+
+ Returns the filepath that was written.
+ """
+ if filepath is None:
+ filepath = generate_filename()
+
+ path = Path(filepath)
+ path.parent.mkdir(parents=True, exist_ok=True)
+
+ fieldnames = [
+ "SSID", "MAC Address", "RSSI",
+ "Channel", "Band", "Width", "802.11",
+ "WEP", "WPA", "WPA2", "WPA3", "WPS",
+ "Vendor", "Beacon Count",
+ "First seen", "Last seen",
+ "Active Scan IP", "Active Scan Gateway",
+ "Active Scan Clients", "Active Scan Ports",
+ ]
+
+ sorted_aps = sorted(
+ aps.values(),
+ key=lambda ap: (ap.rssi if ap.rssi is not None else -999),
+ reverse=True,
+ )
+
+ with open(path, "w", newline="", encoding="utf-8") as f:
+ writer = csv.DictWriter(f, fieldnames=fieldnames, delimiter=";",
+ quoting=csv.QUOTE_ALL)
+ writer.writeheader()
+ for ap in sorted_aps:
+ asr = ap.active_scan_result
+ writer.writerow({
+ "SSID": ap.display_ssid,
+ "MAC Address": ap.bssid,
+ "RSSI": str(ap.rssi) if ap.rssi is not None else "",
+ "Channel": ap.channel_display,
+ "Band": ap.band,
+ "Width": str(ap.width),
+ "802.11": ap.wifi_standard,
+ "WEP": ap.wep,
+ "WPA": ap.wpa,
+ "WPA2": ap.wpa2,
+ "WPA3": ap.wpa3,
+ "WPS": ap.wps,
+ "Vendor": ap.vendor,
+ "Beacon Count": str(ap.beacon_count),
+ "First seen": ap.first_seen.strftime("%Y-%m-%d %H:%M:%S"),
+ "Last seen": ap.last_seen.strftime("%Y-%m-%d %H:%M:%S"),
+ "Active Scan IP": asr.assigned_ip if asr else "",
+ "Active Scan Gateway": asr.gateway_ip if asr else "",
+ "Active Scan Clients": str(asr.client_count) if asr else "",
+ "Active Scan Ports": ",".join(str(p) for p in asr.gateway_open_ports) if asr else "",
+ })
+
+ return str(path)
+
+
+def save_json(aps: dict[str, AccessPoint], filepath: str | None = None) -> str:
+ """Save AP scan results to JSON file.
+
+ Returns the filepath that was written.
+ """
+ if filepath is None:
+ filepath = generate_filename("json")
+
+ path = Path(filepath)
+ path.parent.mkdir(parents=True, exist_ok=True)
+
+ sorted_aps = sorted(
+ aps.values(),
+ key=lambda ap: (ap.rssi if ap.rssi is not None else -999),
+ reverse=True,
+ )
+
+ records = []
+ for ap in sorted_aps:
+ record = {
+ "ssid": ap.display_ssid,
+ "bssid": ap.bssid,
+ "rssi": ap.rssi,
+ "channel": ap.channel,
+ "channel_display": ap.channel_display,
+ "band": ap.band,
+ "width": ap.width,
+ "wifi_standard": ap.wifi_standard,
+ "encryption": ap.encryption,
+ "wep": ap.wep,
+ "wpa": ap.wpa,
+ "wpa2": ap.wpa2,
+ "wpa3": ap.wpa3,
+ "wps": ap.wps,
+ "vendor": ap.vendor,
+ "beacon_count": ap.beacon_count,
+ "first_seen": ap.first_seen.strftime("%Y-%m-%d %H:%M:%S"),
+ "last_seen": ap.last_seen.strftime("%Y-%m-%d %H:%M:%S"),
+ }
+ if ap.active_scan_result:
+ asr = ap.active_scan_result
+ record["active_scan"] = {
+ "scan_time": asr.scan_time.isoformat(),
+ "connected": asr.connected,
+ "assigned_ip": asr.assigned_ip,
+ "subnet_mask": asr.subnet_mask,
+ "gateway_ip": asr.gateway_ip,
+ "gateway_mac": asr.gateway_mac,
+ "dns_servers": asr.dns_servers,
+ "dhcp_server": asr.dhcp_server,
+ "gateway_open_ports": asr.gateway_open_ports,
+ "clients": [
+ {"ip": c.ip, "mac": c.mac, "vendor": c.vendor}
+ for c in asr.clients
+ ],
+ "duration_secs": asr.duration_secs,
+ "error": asr.error,
+ }
+ records.append(record)
+
+ with open(path, "w", encoding="utf-8") as f:
+ json.dump({"scan_time": datetime.now().isoformat(), "access_points": records},
+ f, ensure_ascii=False, indent=2)
+
+ return str(path)
diff --git a/ap_scanner/log.py b/ap_scanner/log.py
@@ -0,0 +1,52 @@
+"""Logging configuration for AP Scanner."""
+
+import logging
+import sys
+
+# Stores the stderr handler's original level so it can be restored after scan.
+_stderr_original_level: int = logging.INFO
+
+
+def setup_logging(verbose: bool = False, log_file: str | None = None) -> None:
+ """Configure logging for ap_scanner.
+
+ Logs go to stderr (not stdout, which is reserved for the ANSI display).
+ When a log_file is specified, DEBUG-level output is always written there.
+ """
+ global _stderr_original_level
+ level = logging.DEBUG if verbose else logging.INFO
+ _stderr_original_level = level
+ fmt = "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
+ datefmt = "%Y-%m-%d %H:%M:%S"
+
+ root = logging.getLogger("ap_scanner")
+ root.setLevel(logging.DEBUG)
+ root.handlers.clear()
+
+ # stderr handler — WARNING+ during scan to avoid ANSI display corruption
+ stderr_handler = logging.StreamHandler(sys.stderr)
+ stderr_handler.setLevel(level)
+ stderr_handler.setFormatter(logging.Formatter(fmt, datefmt=datefmt))
+ root.addHandler(stderr_handler)
+
+ if log_file:
+ file_handler = logging.FileHandler(log_file, encoding="utf-8")
+ file_handler.setLevel(logging.DEBUG)
+ file_handler.setFormatter(logging.Formatter(fmt, datefmt=datefmt))
+ root.addHandler(file_handler)
+
+
+def suppress_stderr_during_scan() -> None:
+ """Raise stderr handler level to avoid corrupting the ANSI display."""
+ root = logging.getLogger("ap_scanner")
+ for h in root.handlers:
+ if isinstance(h, logging.StreamHandler) and h.stream is sys.stderr:
+ h.setLevel(logging.CRITICAL)
+
+
+def restore_stderr_after_scan() -> None:
+ """Restore stderr handler to its original level."""
+ root = logging.getLogger("ap_scanner")
+ for h in root.handlers:
+ if isinstance(h, logging.StreamHandler) and h.stream is sys.stderr:
+ h.setLevel(_stderr_original_level)
diff --git a/ap_scanner/models.py b/ap_scanner/models.py
@@ -0,0 +1,103 @@
+"""Data models for AP Scanner."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+
+
+@dataclass
+class ActiveScanClient:
+ """A client discovered during active ARP scanning."""
+
+ ip: str
+ mac: str
+ vendor: str = ""
+
+
+@dataclass
+class ActiveScanResult:
+ """Results from an active scan of an open AP."""
+
+ bssid: str
+ ssid: str
+ scan_time: datetime = field(default_factory=datetime.now)
+ connected: bool = False
+ assigned_ip: str = ""
+ subnet_mask: str = ""
+ gateway_ip: str = ""
+ gateway_mac: str = ""
+ dns_servers: list[str] = field(default_factory=list)
+ dhcp_server: str = ""
+ clients: list[ActiveScanClient] = field(default_factory=list)
+ gateway_open_ports: list[int] = field(default_factory=list)
+ duration_secs: float = 0.0
+ error: str = ""
+
+ @property
+ def client_count(self) -> int:
+ return len(self.clients)
+
+ @property
+ def summary(self) -> str:
+ if self.error:
+ return f"Failed: {self.error}"
+ parts = []
+ if self.assigned_ip:
+ parts.append(f"IP: {self.assigned_ip}")
+ if self.gateway_ip:
+ parts.append(f"GW: {self.gateway_ip}")
+ if self.clients:
+ parts.append(f"Clients: {self.client_count}")
+ if self.gateway_open_ports:
+ parts.append(f"Ports: {','.join(str(p) for p in self.gateway_open_ports)}")
+ return " | ".join(parts) if parts else "No data collected"
+
+
+@dataclass
+class AccessPoint:
+ """Represents a discovered wireless access point."""
+
+ bssid: str
+ ssid: str # empty string for hidden networks
+ channel: int # primary channel
+ rssi: int | None # dBm (e.g., -42)
+ encryption: str # OPN, WEP, WPA, WPA2, WPA3, WPA2/WPA3
+ cipher: str # CCMP, TKIP, CCMP/TKIP, --
+ auth: str # PSK, SAE, MGT, PSK/SAE, --
+ band: str # "2.4GHz" or "5GHz"
+ wifi_standard: str # "b, g, n" or "a, n, ac, ax" (comma-separated, all supported)
+ width: int = 20 # Channel width: 20, 40, 80, 160 MHz
+ channel_desc: str = "" # Bonded channels: "36+40+44+48"
+ vendor: str = "" # OUI vendor name
+ wps: str = "" # WPS version: "1.0", "2.0", or ""
+ wep: str = "" # WEP security detail (auth-cipher) or ""
+ wpa: str = "" # WPA security detail (auth-cipher) or ""
+ wpa2: str = "" # WPA2 security detail (auth-cipher) or ""
+ wpa3: str = "" # WPA3 security detail (auth-cipher) or ""
+ first_seen: datetime = field(default_factory=datetime.now)
+ last_seen: datetime = field(default_factory=datetime.now)
+ beacon_count: int = 0
+ frequency: int = 0 # MHz
+ active_scan_result: ActiveScanResult | None = None
+
+ @property
+ def is_hidden(self) -> bool:
+ return len(self.ssid.strip("\x00").strip()) == 0
+
+ @property
+ def display_ssid(self) -> str:
+ return "[Hidden]" if self.is_hidden else self.ssid
+
+ @property
+ def rssi_display(self) -> str:
+ return str(self.rssi) if self.rssi is not None else "N/A"
+
+ @property
+ def standard_display(self) -> str:
+ return self.wifi_standard if self.wifi_standard else "--"
+
+ @property
+ def channel_display(self) -> str:
+ """Show bonded channels if available, otherwise primary channel."""
+ return self.channel_desc if self.channel_desc else str(self.channel)
diff --git a/ap_scanner/monitor.py b/ap_scanner/monitor.py
@@ -0,0 +1,207 @@
+"""Monitor mode management for wireless interfaces."""
+
+import atexit
+import logging
+import subprocess
+import sys
+import time
+
+from ap_scanner.utils import get_interface_mode
+
+logger = logging.getLogger("ap_scanner.monitor")
+
+
+class MonitorMode:
+ """Context manager for enabling/disabling monitor mode on a wireless interface.
+
+ Usage:
+ with MonitorMode("wlan0") as iface:
+ # iface is in monitor mode
+ ...
+ # iface is restored to managed mode
+ """
+
+ def __init__(self, interface: str):
+ self.interface = interface
+ self._original_mode: str | None = None
+ self._enabled = False
+ self._cleanup_registered = False
+
+ def enable(self) -> str:
+ """Enable monitor mode on the interface. Returns the interface name."""
+ self._original_mode = get_interface_mode(self.interface)
+
+ if self._original_mode == "monitor":
+ self._enabled = True
+ self._register_cleanup()
+ return self.interface
+
+ # Bring interface down
+ self._run_cmd(["ip", "link", "set", self.interface, "down"])
+
+ # Set monitor mode
+ self._run_cmd(["iw", "dev", self.interface, "set", "monitor", "none"])
+
+ # Bring interface up
+ self._run_cmd(["ip", "link", "set", self.interface, "up"])
+
+ # Verify
+ mode = get_interface_mode(self.interface)
+ if mode != "monitor":
+ # Try to restore
+ self._restore_managed()
+ raise RuntimeError(
+ f"Failed to set monitor mode on {self.interface}. "
+ f"Current mode: {mode}. "
+ f"Ensure the adapter supports monitor mode and the correct driver is loaded."
+ )
+
+ self._enabled = True
+ self._register_cleanup()
+ logger.info("%s: monitor mode enabled", self.interface)
+ return self.interface
+
+ def disable(self) -> None:
+ """Disable monitor mode and restore the interface to managed mode."""
+ if not self._enabled:
+ return
+
+ self._restore_managed()
+ self._enabled = False
+ logger.info("%s: managed mode restored", self.interface)
+
+ def _restore_managed(self) -> None:
+ """Restore the interface to managed mode."""
+ try:
+ self._run_cmd(["ip", "link", "set", self.interface, "down"],
+ ignore_errors=True)
+ self._run_cmd(["iw", "dev", self.interface, "set", "type", "managed"],
+ ignore_errors=True)
+ self._run_cmd(["ip", "link", "set", self.interface, "up"],
+ ignore_errors=True)
+ except Exception:
+ pass
+
+ # Restart NetworkManager to re-manage the interface
+ self._restart_network_manager()
+
+ # Verify restoration
+ mode = get_interface_mode(self.interface)
+ if mode and mode != "managed":
+ print(
+ f"[!] Warning: {self.interface} is in '{mode}' mode, expected 'managed'.\n"
+ f"[!] Run manually: sudo iw dev {self.interface} set type managed",
+ file=sys.stderr,
+ )
+
+ def _restart_network_manager(self) -> None:
+ """Re-manage only this interface in NetworkManager (without full restart).
+
+ Uses ``nmcli device set <iface> managed yes`` so that other
+ interfaces (e.g. internal WiFi) are not disrupted. Falls back to
+ a full ``systemctl restart NetworkManager`` only if nmcli fails.
+ """
+ try:
+ # Check if NM is running first
+ result = subprocess.run(
+ ["systemctl", "is-active", "--quiet", "NetworkManager"],
+ timeout=3,
+ )
+ if result.returncode != 0:
+ return # NM not running, nothing to do
+
+ # Targeted: re-manage only the scanned interface
+ result = subprocess.run(
+ ["nmcli", "device", "set", self.interface, "managed", "yes"],
+ capture_output=True, timeout=5,
+ )
+ if result.returncode == 0:
+ return # success — other interfaces untouched
+
+ # Fallback: full restart (last resort)
+ logger.debug("nmcli set managed failed, falling back to NM restart")
+ subprocess.run(
+ ["systemctl", "restart", "NetworkManager"],
+ capture_output=True, timeout=10,
+ )
+ except (subprocess.TimeoutExpired, FileNotFoundError):
+ pass
+
+ def temporarily_disable(self) -> None:
+ """Switch to managed mode for active scanning (skips NM restart for speed).
+
+ Raises RuntimeError if the interface cannot be switched to managed mode.
+ """
+ if not self._enabled:
+ return
+
+ self._run_cmd(["ip", "link", "set", self.interface, "down"])
+ self._run_cmd(["iw", "dev", self.interface, "set", "type", "managed"])
+ self._run_cmd(["ip", "link", "set", self.interface, "up"])
+
+ # Verify mode switch
+ time.sleep(0.5) # let the driver settle
+ mode = get_interface_mode(self.interface)
+ if mode != "managed":
+ raise RuntimeError(
+ f"Failed to switch {self.interface} to managed mode "
+ f"(current: {mode})"
+ )
+ logger.info("%s: temporarily switched to managed mode", self.interface)
+
+ def re_enable_monitor(self) -> None:
+ """Re-enable monitor mode after active scanning."""
+ try:
+ self._run_cmd(["ip", "link", "set", self.interface, "down"])
+ self._run_cmd(["iw", "dev", self.interface, "set", "monitor", "none"])
+ self._run_cmd(["ip", "link", "set", self.interface, "up"])
+ except RuntimeError as e:
+ logger.critical("Failed to restore monitor mode: %s", e)
+ raise
+
+ mode = get_interface_mode(self.interface)
+ if mode != "monitor":
+ logger.critical(
+ "%s is in '%s' mode after re_enable_monitor, expected 'monitor'",
+ self.interface, mode,
+ )
+ raise RuntimeError(
+ f"Failed to restore monitor mode on {self.interface}. "
+ f"Current mode: {mode}"
+ )
+ logger.info("%s: monitor mode re-enabled", self.interface)
+
+ def _register_cleanup(self) -> None:
+ """Register atexit handler for emergency cleanup."""
+ if not self._cleanup_registered:
+ atexit.register(self._emergency_cleanup)
+ self._cleanup_registered = True
+
+ def _emergency_cleanup(self) -> None:
+ """Best-effort cleanup on unexpected exit."""
+ if self._enabled:
+ try:
+ self._restore_managed()
+ except Exception:
+ logger.error(
+ "Could not restore %s to managed mode. "
+ "Run manually: sudo iw dev %s set type managed",
+ self.interface, self.interface,
+ )
+
+ @staticmethod
+ def _run_cmd(cmd: list[str], ignore_errors: bool = False) -> subprocess.CompletedProcess:
+ """Run a subprocess command."""
+ result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
+ if result.returncode != 0 and not ignore_errors:
+ raise RuntimeError(
+ f"Command failed: {' '.join(cmd)}\n"
+ f"stderr: {result.stderr.strip()}"
+ )
+ return result
+
+ def __enter__(self) -> str:
+ return self.enable()
+
+ def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+ self.disable()
diff --git a/ap_scanner/oui.py b/ap_scanner/oui.py
@@ -0,0 +1,172 @@
+"""OUI (Organizationally Unique Identifier) vendor lookup from MAC address."""
+
+from pathlib import Path
+
+# Common WiFi AP/router vendors - covers most consumer and enterprise APs
+# Format: first 3 bytes of MAC (uppercase, colon-separated) -> vendor name
+_BUILTIN_OUI: dict[str, str] = {
+ # TP-Link
+ "5C:A6:E6": "TP-Link", "E8:48:B8": "TP-Link", "50:C7:BF": "TP-Link",
+ "98:DA:C4": "TP-Link", "B0:BE:76": "TP-Link", "60:32:B1": "TP-Link",
+ "A4:2B:B0": "TP-Link", "30:DE:4B": "TP-Link", "C0:06:C3": "TP-Link",
+ "F4:EC:38": "TP-Link", "AC:84:C6": "TP-Link", "14:EB:B6": "TP-Link",
+ "EC:08:6B": "TP-Link", "78:8C:B5": "TP-Link", "B4:B0:24": "TP-Link",
+ # Netgear
+ "34:98:B5": "NETGEAR", "C4:04:15": "NETGEAR", "20:0C:C8": "NETGEAR",
+ "A0:40:A0": "NETGEAR", "E0:46:9A": "NETGEAR", "28:C6:8E": "NETGEAR",
+ "6C:B0:CE": "NETGEAR", "B0:7F:B9": "NETGEAR", "DC:EF:09": "NETGEAR",
+ # Cisco / Meraki
+ "00:1B:2A": "Cisco", "00:26:0B": "Cisco", "F8:C2:88": "Cisco Meraki",
+ "0C:8D:DB": "Cisco Meraki", "AC:17:C8": "Cisco Meraki", "34:56:FE": "Cisco",
+ "00:18:74": "Cisco", "00:1A:A1": "Cisco", "B0:AA:77": "Cisco",
+ # Aruba / HPE
+ "00:0B:86": "Aruba Networks", "24:DE:C6": "Aruba Networks",
+ "AC:A3:1E": "Aruba Networks", "D8:C7:C8": "Aruba Networks",
+ "9C:1C:12": "Aruba Networks", "70:3A:0E": "Aruba Networks",
+ # Ubiquiti
+ "24:A4:3C": "Ubiquiti", "FC:EC:DA": "Ubiquiti", "68:D7:9A": "Ubiquiti",
+ "80:2A:A8": "Ubiquiti", "F0:9F:C2": "Ubiquiti", "44:D9:E7": "Ubiquiti",
+ "78:8A:20": "Ubiquiti", "18:E8:29": "Ubiquiti", "74:83:C2": "Ubiquiti",
+ # Ruckus
+ "C4:10:8A": "Ruckus", "EC:58:EA": "Ruckus", "74:91:1A": "Ruckus",
+ "A4:12:42": "Ruckus", "58:B6:33": "Ruckus",
+ # Samsung
+ "A8:7C:01": "Samsung", "F4:42:8F": "Samsung", "C0:BD:D1": "Samsung",
+ "5C:3A:45": "Samsung", "B4:3A:28": "Samsung", "D0:22:BE": "Samsung",
+ "78:BD:BC": "Samsung", "CC:07:AB": "Samsung", "58:CB:52": "Samsung",
+ # Apple
+ "00:1C:B3": "Apple", "28:CF:E9": "Apple", "3C:06:30": "Apple",
+ "70:3E:AC": "Apple", "AC:BC:32": "Apple", "F0:D1:A9": "Apple",
+ "14:7D:DA": "Apple", "18:AF:61": "Apple", "D0:E1:40": "Apple",
+ "A8:8E:24": "Apple", "F8:FF:C2": "Apple", "DC:A9:04": "Apple",
+ # Intel
+ "8C:17:59": "Intel", "14:85:7F": "Intel", "A0:C5:89": "Intel",
+ "68:05:CA": "Intel", "34:13:E8": "Intel", "80:86:F2": "Intel",
+ "48:51:B7": "Intel", "7C:B2:7D": "Intel",
+ # Huawei / Honor
+ "00:E0:FC": "Huawei", "48:46:FB": "Huawei", "88:28:B3": "Huawei",
+ "C8:D1:5E": "Huawei", "04:F9:38": "Huawei", "4C:B1:6C": "Huawei",
+ "70:8C:B6": "Huawei", "AC:61:EA": "Huawei",
+ # ASUS
+ "F8:32:E4": "ASUS", "04:D9:F5": "ASUS", "2C:FD:A1": "ASUS",
+ "1C:87:2C": "ASUS", "38:D5:47": "ASUS", "60:45:CB": "ASUS",
+ "AC:22:0B": "ASUS", "B0:6E:BF": "ASUS",
+ # D-Link
+ "00:1C:F0": "D-Link", "1C:7E:E5": "D-Link", "28:10:7B": "D-Link",
+ "84:C9:B2": "D-Link", "C8:D3:A3": "D-Link", "FC:75:16": "D-Link",
+ "B8:A3:86": "D-Link",
+ # Linksys / Belkin
+ "C0:56:27": "Linksys", "20:AA:4B": "Linksys", "58:6D:8F": "Linksys",
+ "E8:9F:80": "Belkin", "94:44:52": "Belkin", "08:86:3B": "Belkin",
+ # EFM Networks (ipTIME)
+ "70:5D:CC": "EFM Networks", "58:86:94": "EFM Networks",
+ "88:36:6C": "EFM Networks", "00:08:9F": "EFM Networks",
+ "90:9F:33": "EFM Networks", "00:26:66": "EFM Networks",
+ "C8:3A:35": "EFM Networks", "04:BD:70": "EFM Networks",
+ # LG
+ "00:1E:75": "LG Electronics", "10:68:3F": "LG Electronics",
+ "64:99:68": "LG Electronics", "C4:36:6C": "LG Electronics",
+ "A8:23:FE": "LG Electronics",
+ # OHSUNG (LG IoT devices)
+ "1C:39:29": "OHSUNG",
+ # Fortinet
+ "E8:1C:BA": "Fortinet", "00:09:0F": "Fortinet", "08:5B:0E": "Fortinet",
+ # Xiaomi
+ "28:6C:07": "Xiaomi", "64:CC:2E": "Xiaomi", "78:11:DC": "Xiaomi",
+ "50:EC:50": "Xiaomi", "7C:1D:D9": "Xiaomi",
+ # Qualcomm / Mediatek (chipset vendors)
+ "00:03:7F": "Atheros", "00:1A:6B": "Atheros",
+ "00:0C:E7": "MediaTek", "00:0C:43": "Ralink",
+ # Realtek
+ "00:E0:4C": "Realtek", "48:5D:60": "Realtek", "52:54:00": "Realtek",
+ # Broadcom
+ "00:10:18": "Broadcom", "00:1A:2B": "Broadcom",
+ # Google
+ "F4:F5:D8": "Google", "54:60:09": "Google", "A4:77:33": "Google",
+ # Amazon (eero, Ring, etc.)
+ "F0:72:EA": "Amazon", "68:54:FD": "Amazon", "40:B4:CD": "Amazon",
+ # Seongji Industry
+ "88:57:1D": "Seongji Industry",
+ # SHENZHEN GONGJIN (ZTE ODM)
+ "80:CA:4B": "SHENZHEN GONGJIN ELECTRONICS",
+ # Alfa Network
+ "00:C0:CA": "Alfa Network",
+}
+
+# Loaded entries from system OUI file
+_loaded_oui: dict[str, str] = {}
+_oui_loaded = False
+
+
+def _load_system_oui() -> None:
+ """Try to load the system OUI database (e.g., from aircrack-ng or wireshark)."""
+ global _oui_loaded, _loaded_oui
+ if _oui_loaded:
+ return
+ _oui_loaded = True
+
+ oui_paths = [
+ Path("/usr/share/ieee-data/oui.txt"),
+ Path("/usr/share/misc/oui.txt"),
+ Path("/usr/share/wireshark/manuf"),
+ Path("/usr/share/aircrack-ng/airodump-ng-oui.txt"),
+ Path("/etc/aircrack-ng/airodump-ng-oui.txt"),
+ ]
+
+ for oui_path in oui_paths:
+ if oui_path.exists():
+ try:
+ _parse_oui_file(oui_path)
+ return
+ except Exception:
+ continue
+
+
+def _parse_oui_file(path: Path) -> None:
+ """Parse an OUI file (IEEE format or Wireshark manuf format)."""
+ with open(path, "r", encoding="utf-8", errors="replace") as f:
+ for line in f:
+ line = line.strip()
+ if not line or line.startswith("#"):
+ continue
+
+ # IEEE format: "AA-BB-CC (hex) Vendor Name"
+ if "(hex)" in line:
+ parts = line.split("(hex)")
+ if len(parts) == 2:
+ oui = parts[0].strip().replace("-", ":").upper()
+ vendor = parts[1].strip()
+ if len(oui) == 8 and vendor:
+ _loaded_oui[oui] = vendor
+
+ # Wireshark manuf format: "AA:BB:CC<tab>ShortName<tab>Long Name"
+ elif "\t" in line:
+ parts = line.split("\t")
+ if len(parts) >= 2:
+ oui = parts[0].strip().upper().replace("-", ":")
+ vendor = parts[-1].strip() if len(parts) >= 3 else parts[1].strip()
+ if len(oui) == 8 and vendor:
+ _loaded_oui[oui] = vendor
+
+
+def lookup_vendor(mac: str) -> str:
+ """Look up vendor name from MAC address.
+
+ Args:
+ mac: MAC address in format "AA:BB:CC:DD:EE:FF"
+
+ Returns:
+ Vendor name or empty string if not found.
+ """
+ oui = mac[:8].upper()
+
+ # Check builtin first (curated, clean names)
+ if oui in _BUILTIN_OUI:
+ return _BUILTIN_OUI[oui]
+
+ # Try system OUI database
+ _load_system_oui()
+ if oui in _loaded_oui:
+ return _loaded_oui[oui]
+
+ return ""
diff --git a/ap_scanner/parser.py b/ap_scanner/parser.py
@@ -0,0 +1,559 @@
+"""Parse 802.11 Beacon and Probe Response frames into AccessPoint objects."""
+
+from datetime import datetime
+
+from scapy.layers.dot11 import (
+ Dot11,
+ Dot11Beacon,
+ Dot11Elt,
+ Dot11ProbeReq,
+ Dot11ProbeResp,
+ RadioTap,
+)
+
+from ap_scanner.models import AccessPoint
+from ap_scanner.oui import lookup_vendor
+from ap_scanner.utils import CHANNELS_5G, channel_band, freq_to_channel
+
+
+def parse_packet(pkt) -> AccessPoint | None:
+ """Parse a Scapy packet into an AccessPoint. Returns None if not a beacon/probe response."""
+ if not (pkt.haslayer(Dot11Beacon) or pkt.haslayer(Dot11ProbeResp)):
+ return None
+
+ try:
+ bssid = pkt[Dot11].addr3
+ if not bssid:
+ return None
+ bssid = bssid.upper()
+ except (IndexError, AttributeError):
+ return None
+
+ # Collect all Information Elements in a single pass
+ ies = _collect_ies(pkt)
+
+ # SSID
+ ssid = _extract_ssid(ies)
+
+ # Channel and frequency
+ channel, frequency = _extract_channel_and_freq(pkt, ies)
+ if channel == 0:
+ return None
+
+ # RSSI
+ rssi = _extract_rssi(pkt)
+
+ # Channel width and bonded channels
+ width, secondary_channels = _extract_channel_width(ies, channel)
+ channel_desc = _build_channel_desc(channel, secondary_channels)
+
+ # Security: encryption, cipher, auth + per-protocol detail
+ security = _extract_security(pkt, ies)
+
+ # Band (use frequency to disambiguate 6GHz from 2.4/5GHz)
+ band = channel_band(channel, frequency) + "GHz"
+
+ # WiFi standards (all supported, comma-separated)
+ wifi_standard = _detect_all_standards(ies, channel)
+
+ # Vendor from OUI
+ vendor = lookup_vendor(bssid)
+
+ # WPS
+ wps = _extract_wps(ies)
+
+ now = datetime.now()
+ return AccessPoint(
+ bssid=bssid,
+ ssid=ssid,
+ channel=channel,
+ rssi=rssi,
+ encryption=security["encryption"],
+ cipher=security["cipher"],
+ auth=security["auth"],
+ band=band,
+ wifi_standard=wifi_standard,
+ width=width,
+ channel_desc=channel_desc,
+ vendor=vendor,
+ wps=wps,
+ wep=security["wep"],
+ wpa=security["wpa"],
+ wpa2=security["wpa2"],
+ wpa3=security["wpa3"],
+ first_seen=now,
+ last_seen=now,
+ beacon_count=1,
+ frequency=frequency,
+ )
+
+
+def parse_probe_request(pkt) -> tuple[str, str] | None:
+ """Parse a Probe Request frame to extract SSID and source MAC.
+
+ Clients looking for hidden networks send Probe Requests with the target SSID.
+ By capturing these, we can map SSIDs to hidden APs when the client is also
+ seen communicating with that AP.
+
+ Returns:
+ (ssid, client_mac) or None if not a probe request or SSID is empty/broadcast.
+ """
+ if not pkt.haslayer(Dot11ProbeReq):
+ return None
+
+ try:
+ client_mac = pkt[Dot11].addr2
+ if not client_mac:
+ return None
+ client_mac = client_mac.upper()
+ except (IndexError, AttributeError):
+ return None
+
+ # Extract SSID from IE
+ elt = pkt.getlayer(Dot11Elt)
+ while elt:
+ if elt.ID == 0 and elt.info:
+ ssid = elt.info.decode("utf-8", errors="replace").strip("\x00").strip()
+ if ssid: # Skip broadcast probes (empty SSID)
+ return ssid, client_mac
+ return None
+ elt = elt.payload.getlayer(Dot11Elt) if elt.payload else None
+
+ return None
+
+
+# ---------------------------------------------------------------------------
+# IE collection
+# ---------------------------------------------------------------------------
+
+def _collect_ies(pkt) -> dict[int, list[bytes]]:
+ """Walk all Dot11Elt layers and group by IE ID. Returns {id: [info_bytes, ...]}."""
+ ies: dict[int, list[bytes]] = {}
+ elt = pkt.getlayer(Dot11Elt)
+ while elt:
+ ie_id = elt.ID
+ info = elt.info if elt.info else b""
+ ies.setdefault(ie_id, []).append(info)
+ elt = elt.payload.getlayer(Dot11Elt) if elt.payload else None
+ return ies
+
+
+# ---------------------------------------------------------------------------
+# Field extraction
+# ---------------------------------------------------------------------------
+
+def _extract_ssid(ies: dict) -> str:
+ """Extract SSID from IE ID=0. Capped at 32 bytes per 802.11 spec."""
+ for info in ies.get(0, []):
+ try:
+ return info[:32].decode("utf-8", errors="replace")
+ except (UnicodeDecodeError, AttributeError):
+ return ""
+ return ""
+
+
+def _extract_channel_and_freq(pkt, ies: dict) -> tuple[int, int]:
+ """Extract primary channel and frequency."""
+ channel = 0
+ frequency = 0
+
+ # DS Parameter Set (IE 3)
+ for info in ies.get(3, []):
+ if info and len(info) >= 1:
+ channel = info[0]
+ break
+
+ # RadioTap frequency
+ if pkt.haslayer(RadioTap):
+ rt = pkt[RadioTap]
+ if hasattr(rt, "ChannelFrequency") and rt.ChannelFrequency:
+ frequency = rt.ChannelFrequency
+ if channel == 0:
+ channel = freq_to_channel(frequency)
+
+ return channel, frequency
+
+
+def _extract_rssi(pkt) -> int | None:
+ """Extract RSSI (dBm) from RadioTap header."""
+ if pkt.haslayer(RadioTap):
+ rt = pkt[RadioTap]
+ if hasattr(rt, "dBm_AntSignal") and rt.dBm_AntSignal is not None:
+ return rt.dBm_AntSignal
+ return None
+
+
+# ---------------------------------------------------------------------------
+# Channel width and bonded channels
+# ---------------------------------------------------------------------------
+
+# 5 GHz channel groupings for bonded channel calculation
+_5G_40MHZ_PAIRS = [
+ (36, 40), (44, 48), (52, 56), (60, 64),
+ (100, 104), (108, 112), (116, 120), (124, 128),
+ (132, 136), (140, 144), (149, 153), (157, 161),
+]
+_5G_80MHZ_GROUPS = [
+ (36, 40, 44, 48), (52, 56, 60, 64),
+ (100, 104, 108, 112), (116, 120, 124, 128),
+ (132, 136, 140, 144), (149, 153, 157, 161),
+]
+_5G_160MHZ_GROUPS = [
+ (36, 40, 44, 48, 52, 56, 60, 64),
+ (100, 104, 108, 112, 116, 120, 124, 128),
+ (132, 136, 140, 144, 149, 153, 157, 161),
+]
+
+# 2.4 GHz secondary channel offsets
+def _2g_40mhz_channels(primary: int, above: bool) -> list[int]:
+ """Get secondary channel for 2.4GHz 40MHz."""
+ if above:
+ secondary = primary + 4
+ else:
+ secondary = primary - 4
+ if 1 <= secondary <= 14:
+ return [secondary]
+ return []
+
+
+def _extract_channel_width(ies: dict, primary_channel: int) -> tuple[int, list[int]]:
+ """Extract channel width and secondary channel list from HT/VHT/HE Operation IEs.
+
+ Returns:
+ (width_mhz, [secondary_channels])
+ """
+ width = 20
+ secondary_channels: list[int] = []
+ is_5ghz = primary_channel in CHANNELS_5G
+
+ # HT Operation (IE 61): determines 20 vs 40 MHz
+ ht_secondary_offset = 0
+ for info in ies.get(61, []):
+ if len(info) >= 2:
+ # byte 0: primary channel
+ # byte 1, bits 0-1: secondary channel offset (0=none, 1=above, 3=below)
+ ht_secondary_offset = info[1] & 0x03
+ if ht_secondary_offset in (1, 3):
+ width = 40
+ above = (ht_secondary_offset == 1)
+ if is_5ghz:
+ # Find the 40MHz pair containing this channel
+ for pair in _5G_40MHZ_PAIRS:
+ if primary_channel in pair:
+ secondary_channels = [ch for ch in pair if ch != primary_channel]
+ break
+ else:
+ secondary_channels = _2g_40mhz_channels(primary_channel, above)
+ break
+
+ # VHT Operation (IE 192): determines 80/160 MHz
+ for info in ies.get(192, []):
+ if len(info) >= 1:
+ vht_width = info[0]
+ if vht_width == 1: # 80 MHz
+ width = 80
+ if is_5ghz:
+ for group in _5G_80MHZ_GROUPS:
+ if primary_channel in group:
+ secondary_channels = [ch for ch in group if ch != primary_channel]
+ break
+ elif vht_width in (2, 3): # 160 MHz or 80+80
+ width = 160
+ if is_5ghz:
+ for group in _5G_160MHZ_GROUPS:
+ if primary_channel in group:
+ secondary_channels = [ch for ch in group if ch != primary_channel]
+ break
+ break
+
+ # HE Operation (IE 255, ext ID=36): may indicate wider bandwidth
+ for info in ies.get(255, []):
+ if len(info) > 0 and info[0] == 36 and len(info) >= 7:
+ # HE Operation extension - check channel width
+ # byte 4 bits: channel width (bit 1 = 40MHz in 2.4, bit 2 = 40/80 in 5, bit 3 = 160 in 5)
+ he_params = info[1] if len(info) > 1 else 0
+ # This is complex; we rely on VHT/HT info primarily and only override if HE says wider
+ if len(info) >= 7:
+ he_ch_width = info[5] # Control channel width
+ if he_ch_width == 3 and width < 160:
+ width = 160
+ if is_5ghz:
+ for group in _5G_160MHZ_GROUPS:
+ if primary_channel in group:
+ secondary_channels = [ch for ch in group if ch != primary_channel]
+ break
+ break
+
+ return width, secondary_channels
+
+
+def _build_channel_desc(primary: int, secondary_channels: list[int]) -> str:
+ """Build channel description string like '36+40+44+48'."""
+ all_channels = sorted([primary] + secondary_channels)
+ return "+".join(str(ch) for ch in all_channels)
+
+
+# ---------------------------------------------------------------------------
+# Security extraction
+# ---------------------------------------------------------------------------
+
+def _extract_security(pkt, ies: dict) -> dict[str, str]:
+ """Extract all security information.
+
+ Returns dict with keys: encryption, cipher, auth, wep, wpa, wpa2, wpa3
+ """
+ result = {
+ "encryption": "OPN",
+ "cipher": "--",
+ "auth": "--",
+ "wep": "",
+ "wpa": "",
+ "wpa2": "",
+ "wpa3": "",
+ }
+
+ # Check privacy bit (bit 4 = 0x0010 in 802.11 Capability Info)
+ # Direct bit check instead of sprintf() which is ~10x slower
+ has_privacy = False
+ if pkt.haslayer(Dot11Beacon):
+ has_privacy = bool(pkt[Dot11Beacon].cap & 0x0010)
+ elif pkt.haslayer(Dot11ProbeResp):
+ has_privacy = bool(pkt[Dot11ProbeResp].cap & 0x0010)
+
+ # Parse RSN IE (ID=48) -> WPA2/WPA3
+ rsn_info = None
+ for info in ies.get(48, []):
+ rsn_info = _parse_rsn_wpa_ie(info, is_rsn=True)
+ break
+
+ # Parse WPA IE (Vendor IE ID=221, OUI 00:50:F2, type 1)
+ wpa_info = None
+ for info in ies.get(221, []):
+ if info[:4] == b"\x00\x50\xf2\x01":
+ wpa_info = _parse_rsn_wpa_ie(info, is_rsn=False)
+ break
+
+ # Build per-protocol detail strings (AUTH-CIPHER format, matching wifi-inspector)
+ if rsn_info:
+ rsn_auth = rsn_info["auth"]
+ rsn_cipher = rsn_info["cipher"]
+ auth_cipher = f"{rsn_auth}-{rsn_cipher}" if rsn_cipher != "--" else rsn_auth
+
+ if "SAE" in rsn_auth:
+ result["wpa3"] = auth_cipher
+ if "PSK" in rsn_auth:
+ # Transition mode: WPA2 PSK + WPA3 SAE
+ result["wpa2"] = f"PSK-{rsn_cipher}"
+ result["wpa3"] = f"SAE-{rsn_cipher}"
+ result["encryption"] = "WPA2/WPA3"
+ else:
+ result["encryption"] = "WPA3"
+ else:
+ result["wpa2"] = auth_cipher
+ result["encryption"] = "WPA2"
+
+ result["cipher"] = rsn_cipher
+ result["auth"] = rsn_auth
+
+ if wpa_info:
+ wpa_auth = wpa_info["auth"]
+ wpa_cipher = wpa_info["cipher"]
+ auth_cipher = f"{wpa_auth}-{wpa_cipher}" if wpa_cipher != "--" else wpa_auth
+ result["wpa"] = auth_cipher
+
+ if not rsn_info:
+ result["encryption"] = "WPA"
+ result["cipher"] = wpa_cipher
+ result["auth"] = wpa_auth
+
+ if not rsn_info and not wpa_info:
+ if has_privacy:
+ result["encryption"] = "WEP"
+ result["wep"] = "WEP"
+ result["cipher"] = "WEP"
+ else:
+ result["encryption"] = "OPN"
+
+ return result
+
+
+def _parse_rsn_wpa_ie(data: bytes, is_rsn: bool) -> dict[str, str]:
+ """Parse RSN (WPA2/WPA3) or WPA Information Element."""
+ result = {"cipher": "--", "auth": "--"}
+ try:
+ offset = 2 if is_rsn else 6 # RSN: skip version; WPA: skip OUI+type+version
+
+ # Group cipher suite (4 bytes)
+ offset += 4
+
+ # Pairwise cipher suites
+ if offset + 2 > len(data):
+ return result
+ pw_count = int.from_bytes(data[offset:offset + 2], "little")
+ offset += 2
+
+ ciphers = set()
+ for _ in range(pw_count):
+ if offset + 4 > len(data):
+ break
+ ciphers.add(_cipher_suite_name(data[offset + 3]))
+ offset += 4
+
+ result["cipher"] = "/".join(sorted(ciphers)) if ciphers else "--"
+
+ # AKM suites
+ if offset + 2 > len(data):
+ return result
+ akm_count = int.from_bytes(data[offset:offset + 2], "little")
+ offset += 2
+
+ auths = set()
+ for _ in range(akm_count):
+ if offset + 4 > len(data):
+ break
+ auths.add(_akm_suite_name(data[offset + 3]))
+ offset += 4
+
+ result["auth"] = "/".join(sorted(auths)) if auths else "--"
+
+ except (IndexError, ValueError):
+ pass
+
+ return result
+
+
+def _cipher_suite_name(cipher_type: int) -> str:
+ return {
+ 0: "GROUP", 1: "WEP40", 2: "TKIP", 3: "RESERVED",
+ 4: "CCMP", 5: "WEP104", 6: "BIP-CMAC",
+ 8: "GCMP", 9: "GCMP-256", 10: "CCMP-256",
+ }.get(cipher_type, f"CIPHER({cipher_type})")
+
+
+def _akm_suite_name(akm_type: int) -> str:
+ return {
+ 1: "MGT", 2: "PSK", 3: "FT-MGT", 4: "FT-PSK",
+ 6: "MGT-SHA256", 8: "SAE", 9: "FT-SAE",
+ 12: "OWE", 18: "OWE",
+ }.get(akm_type, f"AKM({akm_type})")
+
+
+# ---------------------------------------------------------------------------
+# WiFi standard detection
+# ---------------------------------------------------------------------------
+
+def _detect_all_standards(ies: dict, channel: int) -> str:
+ """Detect all supported WiFi standards from IEs.
+
+ Returns comma-separated string like "b, g, n" or "a, n, ac, ax".
+ """
+ standards: list[str] = []
+ is_5ghz = channel in CHANNELS_5G
+
+ # Base standard from band
+ if is_5ghz:
+ standards.append("a")
+ else:
+ # Check supported rates to determine b and/or g
+ supported_rates = _collect_rates(ies)
+ ofdm_rates = {6, 9, 12, 18, 24, 36, 48, 54}
+ b_rates = {1, 2, 5.5, 11}
+
+ has_b = any(r in b_rates for r in supported_rates)
+ has_g = any(r in ofdm_rates for r in supported_rates)
+
+ if has_b:
+ standards.append("b")
+ if has_g:
+ standards.append("g")
+ if not has_b and not has_g and supported_rates:
+ standards.append("b") # fallback
+
+ # HT Capabilities (IE 45) -> 802.11n
+ if 45 in ies:
+ standards.append("n")
+
+ # VHT Capabilities (IE 191) -> 802.11ac
+ if 191 in ies:
+ standards.append("ac")
+
+ # HE Capabilities (IE 255, ext ID 35) -> 802.11ax
+ for info in ies.get(255, []):
+ if info and len(info) > 0 and info[0] == 35:
+ standards.append("ax")
+ break
+
+ return ", ".join(standards)
+
+
+def _collect_rates(ies: dict) -> set[float]:
+ """Collect supported rates from IE 1 (Supported Rates) and IE 50 (Extended Rates)."""
+ rates: set[float] = set()
+ for ie_id in (1, 50):
+ for info in ies.get(ie_id, []):
+ for byte in info:
+ rate = (byte & 0x7F) * 0.5
+ rates.add(rate)
+ return rates
+
+
+# ---------------------------------------------------------------------------
+# WPS detection
+# ---------------------------------------------------------------------------
+
+def _extract_wps(ies: dict) -> str:
+ """Extract WPS version from Vendor Specific IEs.
+
+ WPS IE: Vendor IE (221) with OUI 00:50:F2, type 4.
+ """
+ for info in ies.get(221, []):
+ if len(info) >= 4 and info[:4] == b"\x00\x50\xf2\x04":
+ # WPS IE found - parse WPS attributes to find version
+ return _parse_wps_version(info[4:])
+ return ""
+
+
+def _parse_wps_version(data: bytes) -> str:
+ """Parse WPS attributes to extract version.
+
+ WPS TLV format: Type (2 bytes BE) | Length (2 bytes BE) | Value
+ Version attribute type = 0x104A
+ Version2 attribute type = 0x1032 (inside Vendor Extension 0x1049)
+ """
+ offset = 0
+ version = ""
+ while offset + 4 <= len(data):
+ attr_type = int.from_bytes(data[offset:offset + 2], "big")
+ attr_len = int.from_bytes(data[offset + 2:offset + 4], "big")
+ offset += 4
+
+ if offset + attr_len > len(data):
+ break
+
+ attr_data = data[offset:offset + attr_len]
+
+ if attr_type == 0x104A and attr_len >= 1:
+ # Version: single byte, major.minor (0x10 = 1.0, 0x20 = 2.0)
+ ver_byte = attr_data[0]
+ major = (ver_byte >> 4) & 0x0F
+ minor = ver_byte & 0x0F
+ version = f"{major}.{minor}"
+ elif attr_type == 0x1049:
+ # Vendor Extension - may contain Version2 (sub-element ID 0x00)
+ sub_offset = 3 # skip 3-byte WFA vendor ID
+ while sub_offset + 2 <= len(attr_data):
+ sub_id = attr_data[sub_offset]
+ sub_len = attr_data[sub_offset + 1]
+ sub_offset += 2
+ if sub_offset + sub_len > len(attr_data):
+ break # malformed sub-element
+ if sub_id == 0x00 and sub_len >= 1:
+ ver_byte = attr_data[sub_offset]
+ major = (ver_byte >> 4) & 0x0F
+ minor = ver_byte & 0x0F
+ version = f"{major}.{minor}"
+ sub_offset += sub_len
+
+ offset += attr_len
+
+ return version
diff --git a/ap_scanner/scanner.py b/ap_scanner/scanner.py
@@ -0,0 +1,417 @@
+"""Scanner engine: packet capture, channel hopping, AP management."""
+
+import logging
+import subprocess
+import threading
+import time
+from datetime import datetime
+
+from scapy.all import sendp, sniff
+from scapy.layers.dot11 import Dot11, Dot11Elt, Dot11ProbeReq, RadioTap
+
+from ap_scanner.models import AccessPoint
+from ap_scanner.parser import parse_packet, parse_probe_request
+
+logger = logging.getLogger("ap_scanner.scanner")
+
+# Memory bounds for tracking maps
+MAX_CLIENTS = 10000
+MAX_PROBE_SSIDS = 5000
+_CLEANUP_INTERVAL = 60 # seconds
+
+
+class ChannelHopper:
+ """Hops through WiFi channels in a background thread.
+
+ Uses a weighted schedule so primary 2.4 GHz channels (1, 6, 11) are
+ visited more frequently, improving discovery of the most common APs.
+ """
+
+ # Primary 2.4 GHz channels — visited twice per cycle for better coverage
+ _PRIMARY_24 = {1, 6, 11}
+
+ def __init__(
+ self,
+ interface: str,
+ channels: list[int],
+ dwell_time: float = 0.5,
+ ):
+ self.interface = interface
+ self.channels = channels
+ self.dwell_time = dwell_time
+ self._stop_event = threading.Event()
+ self._thread: threading.Thread | None = None
+ self._current_channel: int = 0
+ self._lock = threading.Lock()
+ self._fail_count: dict[int, int] = {} # ch -> consecutive failure count
+ self._failed_channels: set[int] = set() # permanently skipped after 3 failures
+ self._probe_src_mac = self._get_mac()
+ self._schedule = self._build_schedule()
+
+ def _get_mac(self) -> str:
+ """Read the interface MAC address for Probe Request source."""
+ try:
+ mac_path = f"/sys/class/net/{self.interface}/address"
+ with open(mac_path) as f:
+ return f.read().strip()
+ except (OSError, ValueError):
+ return "00:00:00:00:00:00"
+
+ def _build_schedule(self) -> list[int]:
+ """Build a weighted channel schedule.
+
+ Primary 2.4 GHz channels (1, 6, 11) are interleaved between other
+ channels so they get visited roughly twice as often.
+ """
+ primary = [ch for ch in self.channels if ch in self._PRIMARY_24]
+ others = [ch for ch in self.channels if ch not in self._PRIMARY_24]
+
+ if not primary:
+ return list(self.channels)
+
+ # Interleave: after every N other channels, insert a primary channel
+ # With ~36 other channels and 3 primary, insert one primary every 12 channels
+ schedule: list[int] = []
+ step = max(len(others) // len(primary), 3) if others else 1
+ pri_idx = 0
+
+ for i, ch in enumerate(others):
+ schedule.append(ch)
+ if (i + 1) % step == 0 and pri_idx < len(primary):
+ schedule.append(primary[pri_idx])
+ pri_idx += 1
+
+ # Append any remaining primary channels
+ while pri_idx < len(primary):
+ schedule.append(primary[pri_idx])
+ pri_idx += 1
+
+ # Prepend all primary channels at the start for quick first results
+ return primary + schedule
+
+ @property
+ def current_channel(self) -> int:
+ with self._lock:
+ return self._current_channel
+
+ def start(self) -> None:
+ self._stop_event.clear()
+ self._fail_count.clear()
+ self._failed_channels.clear()
+ self._thread = threading.Thread(target=self._hop_loop, daemon=True)
+ self._thread.start()
+
+ def stop(self) -> None:
+ self._stop_event.set()
+ if self._thread and self._thread.is_alive():
+ self._thread.join(timeout=3)
+ if self._thread.is_alive():
+ logger.warning("Channel hopper thread did not stop within timeout")
+
+ def _send_probe_request(self) -> None:
+ """Inject a broadcast Probe Request to trigger Probe Responses from APs.
+
+ This is the key difference between passive and active scanning.
+ APs respond to Probe Requests even when the next beacon is far away.
+ """
+ try:
+ probe = (
+ RadioTap()
+ / Dot11(
+ type=0,
+ subtype=4,
+ addr1="ff:ff:ff:ff:ff:ff",
+ addr2=self._probe_src_mac,
+ addr3="ff:ff:ff:ff:ff:ff",
+ )
+ / Dot11ProbeReq()
+ / Dot11Elt(ID=0, info=b"") # broadcast SSID
+ / Dot11Elt(ID=1, info=b"\x02\x04\x0b\x16\x0c\x12\x18\x24") # rates
+ )
+ sendp(probe, iface=self.interface, verbose=False, count=1)
+ except Exception:
+ pass # best-effort; injection may not be supported
+
+ def _hop_loop(self) -> None:
+ _BLACKLIST_THRESHOLD = 3 # failures before permanent skip
+
+ while not self._stop_event.is_set():
+ for ch in self._schedule:
+ if self._stop_event.is_set():
+ break
+ # Skip channels that have consistently failed
+ if ch in self._failed_channels:
+ continue
+ try:
+ result = subprocess.run(
+ ["iw", "dev", self.interface, "set", "channel", str(ch)],
+ capture_output=True,
+ timeout=2,
+ )
+ if result.returncode != 0:
+ count = self._fail_count.get(ch, 0) + 1
+ self._fail_count[ch] = count
+ if count >= _BLACKLIST_THRESHOLD:
+ self._failed_channels.add(ch)
+ logger.debug("Channel %d failed %d times — skipping", ch, count)
+ continue
+ # Success — reset failure counter
+ self._fail_count.pop(ch, None)
+ with self._lock:
+ self._current_channel = ch
+ except (subprocess.TimeoutExpired, FileNotFoundError):
+ count = self._fail_count.get(ch, 0) + 1
+ self._fail_count[ch] = count
+ if count >= _BLACKLIST_THRESHOLD:
+ self._failed_channels.add(ch)
+ continue
+
+ # Send Probe Request to trigger AP responses on this channel
+ self._send_probe_request()
+
+ self._stop_event.wait(self.dwell_time)
+
+
+class ScannerEngine:
+ """Orchestrates packet capture and channel hopping to discover APs."""
+
+ # Check time-based cleanup every N packets to avoid syscall on every packet
+ _CLEANUP_CHECK_INTERVAL = 5000
+
+ def __init__(
+ self,
+ interface: str,
+ channels: list[int],
+ dwell_time: float = 0.5,
+ ):
+ self.interface = interface
+ self.channels = channels
+
+ self._aps: dict[str, AccessPoint] = {}
+ # client_mac -> set of BSSIDs they communicate with
+ self._client_ap_map: dict[str, set[str]] = {}
+ # SSID -> set of client MACs that probed for it
+ self._probe_ssids: dict[str, set[str]] = {}
+ self._lock = threading.Lock()
+ self._stop_event = threading.Event()
+ self._hopper = ChannelHopper(interface, channels, dwell_time=dwell_time)
+ self._capture_thread: threading.Thread | None = None
+ self._last_cleanup = time.monotonic()
+ self._pkt_count = 0
+
+ @property
+ def access_points(self) -> dict[str, AccessPoint]:
+ """Return a copy of the current AP dictionary."""
+ with self._lock:
+ return dict(self._aps)
+
+ @property
+ def current_channel(self) -> int:
+ return self._hopper.current_channel
+
+ def start(self) -> None:
+ """Start scanning: channel hopper + packet capture."""
+ self._stop_event.clear()
+ self._hopper.start()
+ self._capture_thread = threading.Thread(target=self._capture_loop, daemon=True)
+ self._capture_thread.start()
+
+ def stop(self) -> None:
+ """Stop scanning and clean up threads."""
+ self._stop_event.set()
+ # Suppress Scapy socket warnings during intentional shutdown
+ logging.getLogger("scapy.runtime").setLevel(logging.CRITICAL)
+ logging.getLogger("scapy").setLevel(logging.CRITICAL)
+ self._hopper.stop()
+ if self._capture_thread and self._capture_thread.is_alive():
+ self._capture_thread.join(timeout=5)
+ if self._capture_thread.is_alive():
+ logger.warning("Capture thread did not stop within timeout")
+
+ def clear(self) -> None:
+ """Clear all discovered APs."""
+ with self._lock:
+ self._aps.clear()
+
+ @staticmethod
+ def _pick_bpf_filter(interface: str) -> str:
+ """Try BPF filter candidates and return the first one that compiles.
+
+ Returns empty string if none work (safe fallback = no filter).
+ """
+ candidates = ["type mgt", ""]
+ for bpf in candidates:
+ if not bpf:
+ return ""
+ try:
+ # Dry-run: open a sniff socket with the filter to see if it compiles
+ from scapy.arch.common import compile_filter
+ compile_filter(bpf, interface)
+ return bpf
+ except Exception:
+ continue
+ return ""
+
+ def _capture_loop(self) -> None:
+ """Run Scapy sniff in the capture thread.
+
+ Tries to apply a BPF filter for 802.11 management frames only,
+ falling back to no filter + fast Python-level type check.
+ """
+ bpf = self._pick_bpf_filter(self.interface)
+ if bpf:
+ logger.debug("Using BPF filter: %s", bpf)
+ else:
+ logger.debug("BPF filter not available — using Python-level filtering")
+
+ try:
+ while not self._stop_event.is_set():
+ sniff(
+ iface=self.interface,
+ prn=self._on_packet,
+ store=False,
+ stop_filter=lambda _: self._stop_event.is_set(),
+ timeout=2,
+ monitor=True,
+ **({"filter": bpf} if bpf else {}),
+ )
+ except OSError as e:
+ if not self._stop_event.is_set():
+ logger.error("Capture error on %s: %s", self.interface, e)
+
+ def _maybe_cleanup(self) -> None:
+ """Periodically evict old entries from tracking maps to bound memory.
+
+ Uses a packet counter to avoid calling time.monotonic() on every packet.
+ """
+ self._pkt_count += 1
+ if self._pkt_count < self._CLEANUP_CHECK_INTERVAL:
+ return
+ self._pkt_count = 0
+
+ now = time.monotonic()
+ if now - self._last_cleanup < _CLEANUP_INTERVAL:
+ return
+ self._last_cleanup = now
+ with self._lock:
+ if len(self._client_ap_map) > MAX_CLIENTS:
+ keys = list(self._client_ap_map.keys())
+ for k in keys[: len(keys) // 2]:
+ del self._client_ap_map[k]
+ if len(self._probe_ssids) > MAX_PROBE_SSIDS:
+ keys = list(self._probe_ssids.keys())
+ for k in keys[: len(keys) // 2]:
+ del self._probe_ssids[k]
+
+ def _on_packet(self, pkt) -> None:
+ """Callback for each captured packet."""
+ # Fast reject: extract Dot11 layer once and reuse throughout.
+ dot11 = pkt.getlayer(Dot11)
+ if dot11 is None:
+ return
+ frame_type = dot11.type
+ if frame_type == 1: # Control frame — skip
+ return
+
+ self._maybe_cleanup()
+
+ # Track client-AP associations (Data/Auth frames only)
+ if frame_type in (0, 2):
+ self._track_client_ap(dot11)
+
+ # Try to resolve hidden SSIDs from Probe Requests
+ probe = parse_probe_request(pkt)
+ if probe:
+ ssid, client_mac = probe
+ with self._lock:
+ self._probe_ssids.setdefault(ssid, set()).add(client_mac)
+ self._try_resolve_hidden(ssid, client_mac)
+
+ ap = parse_packet(pkt)
+ if ap is None:
+ return
+
+ with self._lock:
+ existing = self._aps.get(ap.bssid)
+ if existing:
+ existing.last_seen = datetime.now()
+ existing.beacon_count += 1
+ if ap.rssi is not None:
+ existing.rssi = ap.rssi
+ if existing.is_hidden and not ap.is_hidden:
+ existing.ssid = ap.ssid
+ if ap.encryption != "OPN" and existing.encryption == "OPN":
+ existing.encryption = ap.encryption
+ existing.cipher = ap.cipher
+ existing.auth = ap.auth
+ if ap.wpa2 and not existing.wpa2:
+ existing.wpa2 = ap.wpa2
+ if ap.wpa3 and not existing.wpa3:
+ existing.wpa3 = ap.wpa3
+ if ap.wpa and not existing.wpa:
+ existing.wpa = ap.wpa
+ if ap.wep and not existing.wep:
+ existing.wep = ap.wep
+ if ap.wifi_standard and not existing.wifi_standard:
+ existing.wifi_standard = ap.wifi_standard
+ if ap.width > existing.width:
+ existing.width = ap.width
+ existing.channel_desc = ap.channel_desc
+ if ap.vendor and not existing.vendor:
+ existing.vendor = ap.vendor
+ if ap.wps and not existing.wps:
+ existing.wps = ap.wps
+ else:
+ self._aps[ap.bssid] = ap
+ if ap.is_hidden:
+ self._try_resolve_hidden_for_ap(ap.bssid)
+
+ def _track_client_ap(self, dot11) -> None:
+ """Track which clients are communicating with which APs.
+
+ Receives the Dot11 layer directly (already extracted by _on_packet).
+ """
+ addr1 = dot11.addr1 # receiver
+ addr2 = dot11.addr2 # transmitter
+ addr3 = dot11.addr3 # BSSID (in most cases)
+
+ if not addr2 or not addr3:
+ return
+
+ addr2_upper = addr2.upper()
+ addr3_upper = addr3.upper()
+
+ with self._lock:
+ if addr3_upper in self._aps:
+ if addr2_upper != addr3_upper:
+ self._client_ap_map.setdefault(addr2_upper, set()).add(addr3_upper)
+ elif addr2_upper in self._aps:
+ if addr1:
+ client = addr1.upper()
+ if client != addr2_upper:
+ self._client_ap_map.setdefault(client, set()).add(addr2_upper)
+
+ def _try_resolve_hidden(self, ssid: str, client_mac: str) -> None:
+ """Try to match a probe request SSID to a hidden AP via client association.
+
+ If this client is associated with a hidden AP, the probed SSID is likely
+ the hidden AP's real name.
+ """
+ bssids = self._client_ap_map.get(client_mac, set())
+ for bssid in bssids:
+ ap = self._aps.get(bssid)
+ if ap and ap.is_hidden:
+ ap.ssid = ssid
+
+ def _try_resolve_hidden_for_ap(self, bssid: str) -> None:
+ """When a new hidden AP appears, check if any known clients can reveal its SSID."""
+ # Find clients associated with this BSSID
+ for client_mac, associated_bssids in self._client_ap_map.items():
+ if bssid in associated_bssids:
+ # Check if this client has probed for any SSIDs
+ for ssid, probing_clients in self._probe_ssids.items():
+ if client_mac in probing_clients:
+ ap = self._aps.get(bssid)
+ if ap and ap.is_hidden:
+ ap.ssid = ssid
+ return
diff --git a/ap_scanner/utils.py b/ap_scanner/utils.py
@@ -0,0 +1,213 @@
+"""Utility functions: channel/frequency mapping, privilege checks, interface detection."""
+
+import os
+import shutil
+import subprocess
+from pathlib import Path
+
+# 2.4 GHz channels (1-14)
+CHANNELS_2G = list(range(1, 15))
+
+# 5 GHz channels (common UNII bands)
+CHANNELS_5G = [
+ 36, 40, 44, 48, # UNII-1
+ 52, 56, 60, 64, # UNII-2
+ 100, 104, 108, 112, # UNII-2 Extended
+ 116, 120, 124, 128,
+ 132, 136, 140, 144,
+ 149, 153, 157, 161, # UNII-3
+ 165,
+]
+
+# 6 GHz channels (WiFi 6E / 7, UNII-5 through UNII-8)
+CHANNELS_6G = [
+ 1, 5, 9, 13, 17, 21, 25, 29, # UNII-5
+ 33, 37, 41, 45, 49, 53, 57, 61,
+ 65, 69, 73, 77, 81, 85, 89, 93, # UNII-6
+ 97, 101, 105, 109, 113, 117,
+ 121, 125, 129, 133, 137, 141, # UNII-7
+ 145, 149, 153, 157, 161, 165,
+ 169, 173, 177, 181, 185, 189, # UNII-8
+ 193, 197, 201, 205, 209, 213,
+ 217, 221, 225, 229, 233,
+]
+
+ALL_CHANNELS = CHANNELS_2G + CHANNELS_5G
+
+# Channel -> Center Frequency (MHz) mapping
+CHANNEL_TO_FREQ: dict[int, int] = {}
+# 2.4 GHz: channel 1 = 2412 MHz, each channel +5 MHz, channel 14 = 2484 MHz
+for ch in range(1, 14):
+ CHANNEL_TO_FREQ[ch] = 2407 + ch * 5
+CHANNEL_TO_FREQ[14] = 2484
+# 5 GHz: freq = 5000 + channel * 5
+for ch in CHANNELS_5G:
+ CHANNEL_TO_FREQ[ch] = 5000 + ch * 5
+# 6 GHz: freq = 5950 + channel * 5 (channel 1 = 5955 MHz)
+CHANNEL_TO_FREQ_6G: dict[int, int] = {}
+for ch in CHANNELS_6G:
+ CHANNEL_TO_FREQ_6G[ch] = 5950 + ch * 5
+
+# Reverse mapping (2.4G + 5G only; 6G channels overlap with 2.4G/5G numbers)
+FREQ_TO_CHANNEL: dict[int, int] = {freq: ch for ch, freq in CHANNEL_TO_FREQ.items()}
+# Add 6 GHz reverse mapping (no overlap since frequencies are unique)
+for ch, freq in CHANNEL_TO_FREQ_6G.items():
+ FREQ_TO_CHANNEL[freq] = ch
+
+
+def freq_to_channel(freq_mhz: int) -> int:
+ """Convert frequency in MHz to channel number."""
+ if freq_mhz in FREQ_TO_CHANNEL:
+ return FREQ_TO_CHANNEL[freq_mhz]
+ # Approximate: find closest frequency
+ if 2400 <= freq_mhz <= 2500:
+ return round((freq_mhz - 2407) / 5)
+ elif 5925 <= freq_mhz <= 7125:
+ return round((freq_mhz - 5950) / 5)
+ elif freq_mhz >= 5000:
+ return round((freq_mhz - 5000) / 5)
+ return 0
+
+
+def channel_to_freq(channel: int) -> int:
+ """Convert channel number to frequency in MHz."""
+ return CHANNEL_TO_FREQ.get(channel, 0)
+
+
+def is_5ghz(channel: int) -> bool:
+ """Check if a channel is in the 5 GHz band."""
+ return channel in CHANNELS_5G
+
+
+def channel_band(channel: int, freq: int = 0) -> str:
+ """Return '2.4', '5', or '6' for a given channel.
+
+ Uses frequency to disambiguate overlapping channel numbers between
+ 2.4GHz/5GHz and 6GHz bands.
+ """
+ if freq >= 5925:
+ return "6"
+ if freq >= 4900:
+ return "5"
+ if freq > 0:
+ return "2.4"
+ # Fallback: channel number only (no freq info, cannot detect 6GHz)
+ return "5" if is_5ghz(channel) else "2.4"
+
+
+def check_root() -> bool:
+ """Check if running as root."""
+ return os.geteuid() == 0
+
+
+def check_iw() -> bool:
+ """Check if 'iw' command is available."""
+ return shutil.which("iw") is not None
+
+
+def check_ip() -> bool:
+ """Check if 'ip' command is available."""
+ return shutil.which("ip") is not None
+
+
+def find_wireless_interfaces() -> list[str]:
+ """Find available wireless interfaces by checking /sys/class/net/*/wireless."""
+ interfaces = []
+ net_path = Path("/sys/class/net")
+ if not net_path.exists():
+ return interfaces
+ for iface_path in net_path.iterdir():
+ wireless_path = iface_path / "wireless"
+ if wireless_path.exists():
+ interfaces.append(iface_path.name)
+ return sorted(interfaces)
+
+
+def get_interface_mode(interface: str) -> str | None:
+ """Get the current mode of a wireless interface (managed, monitor, etc.)."""
+ try:
+ result = subprocess.run(
+ ["iw", "dev", interface, "info"],
+ capture_output=True, text=True, timeout=5,
+ )
+ if result.returncode != 0:
+ return None
+ for line in result.stdout.splitlines():
+ line = line.strip()
+ if line.startswith("type "):
+ return line.split("type ", 1)[1].strip()
+ except (subprocess.TimeoutExpired, FileNotFoundError):
+ return None
+ return None
+
+
+def get_interface_driver(interface: str) -> str:
+ """Get the driver name for a wireless interface."""
+ driver_link = Path(f"/sys/class/net/{interface}/device/driver")
+ try:
+ return driver_link.resolve().name
+ except (OSError, ValueError):
+ return "unknown"
+
+
+def get_supported_channels(interface: str) -> list[int]:
+ """Query the adapter's supported channels via 'iw phy'.
+
+ Returns a sorted list of channels the hardware can actually use.
+ Falls back to ALL_CHANNELS if the query fails.
+ """
+ # Get phy name for this interface
+ phy_path = Path(f"/sys/class/net/{interface}/phy80211/name")
+ try:
+ phy_name = phy_path.read_text().strip()
+ except (OSError, ValueError):
+ return ALL_CHANNELS
+
+ try:
+ result = subprocess.run(
+ ["iw", "phy", phy_name, "channels"],
+ capture_output=True, text=True, timeout=5,
+ )
+ if result.returncode != 0:
+ return ALL_CHANNELS
+ except (subprocess.TimeoutExpired, FileNotFoundError):
+ return ALL_CHANNELS
+
+ channels = []
+ for line in result.stdout.splitlines():
+ line = line.strip()
+ # Lines look like: "* 2412 MHz [1] (20.0 dBm)"
+ # or disabled: "* 5260 MHz [52] (disabled)"
+ if not line.startswith("*"):
+ continue
+ if "disabled" in line.lower():
+ continue
+ # Note: "no IR" channels are valid for monitor mode (receive-only)
+ # Extract channel number from [N]
+ bracket_start = line.find("[")
+ bracket_end = line.find("]")
+ if bracket_start == -1 or bracket_end == -1:
+ continue
+ try:
+ ch = int(line[bracket_start + 1 : bracket_end])
+ channels.append(ch)
+ except ValueError:
+ continue
+
+ return sorted(set(channels)) if channels else ALL_CHANNELS
+
+
+def parse_channels_arg(channels_str: str) -> list[int]:
+ """Parse channel argument string like '1,6,11' or '36-48' into a list of channels."""
+ channels = []
+ for part in channels_str.split(","):
+ part = part.strip()
+ if "-" in part:
+ start, end = part.split("-", 1)
+ start, end = int(start.strip()), int(end.strip())
+ channels.extend(ch for ch in ALL_CHANNELS if start <= ch <= end)
+ else:
+ ch = int(part)
+ if ch in ALL_CHANNELS:
+ channels.append(ch)
+ return sorted(set(channels)) if channels else ALL_CHANNELS
diff --git a/pyproject.toml b/pyproject.toml
@@ -0,0 +1,48 @@
+[project]
+name = "uaf"
+version = "1.0.0"
+description = "Real-time wireless AP scanner with active reconnaissance for Linux"
+requires-python = ">=3.10"
+license = {text = "GPL-2.0-or-later"}
+readme = "README.md"
+authors = [
+ {name = "devastator-x"},
+]
+classifiers = [
+ "Development Status :: 4 - Beta",
+ "Environment :: Console",
+ "Intended Audience :: System Administrators",
+ "Operating System :: POSIX :: Linux",
+ "Programming Language :: Python :: 3",
+ "Programming Language :: Python :: 3.10",
+ "Programming Language :: Python :: 3.11",
+ "Programming Language :: Python :: 3.12",
+ "Topic :: System :: Networking :: Monitoring",
+]
+dependencies = [
+ "scapy>=2.5.0",
+]
+
+[project.urls]
+Homepage = "https://github.com/devastator-x/uaf"
+Repository = "https://github.com/devastator-x/uaf"
+Issues = "https://github.com/devastator-x/uaf/issues"
+
+[project.optional-dependencies]
+dev = [
+ "pytest>=7.0",
+ "pytest-cov>=4.0",
+]
+
+[project.scripts]
+uaf = "ap_scanner.__main__:main"
+
+[build-system]
+requires = ["setuptools>=68.0"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools.packages.find]
+include = ["ap_scanner*"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
diff --git a/requirements-dev.txt b/requirements-dev.txt
@@ -0,0 +1,3 @@
+-r requirements.txt
+pytest>=7.0
+pytest-cov>=4.0
diff --git a/requirements.txt b/requirements.txt
@@ -0,0 +1 @@
+scapy>=2.5.0
diff --git a/tests/__init__.py b/tests/__init__.py
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -0,0 +1,67 @@
+"""Common test fixtures for AP Scanner tests."""
+
+import pytest
+from datetime import datetime
+
+from ap_scanner.models import AccessPoint
+
+
+@pytest.fixture
+def sample_ap():
+ return AccessPoint(
+ bssid="AA:BB:CC:DD:EE:FF",
+ ssid="TestNetwork",
+ channel=6,
+ rssi=-42,
+ encryption="WPA2",
+ cipher="CCMP",
+ auth="PSK",
+ band="2.4GHz",
+ wifi_standard="b, g, n",
+ width=40,
+ channel_desc="6",
+ vendor="TP-Link",
+ wps="2.0",
+ wpa2="PSK-CCMP",
+ first_seen=datetime(2026, 1, 1, 12, 0, 0),
+ last_seen=datetime(2026, 1, 1, 12, 5, 0),
+ beacon_count=100,
+ frequency=2437,
+ )
+
+
+@pytest.fixture
+def hidden_ap():
+ return AccessPoint(
+ bssid="11:22:33:44:55:66",
+ ssid="",
+ channel=36,
+ rssi=-65,
+ encryption="WPA3",
+ cipher="CCMP-256",
+ auth="SAE",
+ band="5GHz",
+ wifi_standard="a, n, ac, ax",
+ width=80,
+ channel_desc="36+40+44+48",
+ wpa3="SAE-CCMP-256",
+ first_seen=datetime(2026, 1, 1, 12, 0, 0),
+ last_seen=datetime(2026, 1, 1, 12, 5, 0),
+ beacon_count=50,
+ frequency=5180,
+ )
+
+
+@pytest.fixture
+def open_ap():
+ return AccessPoint(
+ bssid="AA:AA:AA:AA:AA:AA",
+ ssid="CafeGuest",
+ channel=11,
+ rssi=-73,
+ encryption="OPN",
+ cipher="--",
+ auth="--",
+ band="2.4GHz",
+ wifi_standard="b, g, n",
+ )
diff --git a/tests/test_active_scanner.py b/tests/test_active_scanner.py
@@ -0,0 +1,84 @@
+"""Tests for ap_scanner.active_scanner utility functions."""
+
+from ap_scanner.active_scanner import _calculate_network, _cidr_to_netmask, _is_ip
+
+
+# --- _cidr_to_netmask ---
+
+def test_cidr_to_netmask_24():
+ assert _cidr_to_netmask(24) == "255.255.255.0"
+
+
+def test_cidr_to_netmask_16():
+ assert _cidr_to_netmask(16) == "255.255.0.0"
+
+
+def test_cidr_to_netmask_32():
+ assert _cidr_to_netmask(32) == "255.255.255.255"
+
+
+def test_cidr_to_netmask_8():
+ assert _cidr_to_netmask(8) == "255.0.0.0"
+
+
+# --- _calculate_network ---
+
+def test_calculate_network_24():
+ result = _calculate_network("192.168.1.100", "255.255.255.0")
+ assert result == "192.168.1.0/24"
+
+
+def test_calculate_network_16():
+ result = _calculate_network("10.0.5.42", "255.255.0.0")
+ assert result == "10.0.0.0/16"
+
+
+def test_calculate_network_invalid():
+ result = _calculate_network("not_an_ip", "255.255.255.0")
+ assert result == ""
+
+
+# --- _is_ip ---
+
+def test_is_ip_valid():
+ assert _is_ip("192.168.1.1") is True
+
+
+def test_is_ip_valid_zeros():
+ assert _is_ip("0.0.0.0") is True
+
+
+def test_is_ip_invalid_string():
+ assert _is_ip("not_ip") is False
+
+
+def test_is_ip_invalid_overflow():
+ assert _is_ip("256.1.1.1") is False
+
+
+def test_is_ip_empty():
+ assert _is_ip("") is False
+
+
+# --- _cidr_to_netmask edge cases ---
+
+def test_cidr_to_netmask_0():
+ assert _cidr_to_netmask(0) == "0.0.0.0"
+
+
+def test_cidr_to_netmask_1():
+ assert _cidr_to_netmask(1) == "128.0.0.0"
+
+
+# --- _calculate_network edge cases ---
+
+def test_calculate_network_32():
+ """Host-only mask returns single host /32."""
+ result = _calculate_network("192.168.1.42", "255.255.255.255")
+ assert result == "192.168.1.42/32"
+
+
+def test_calculate_network_8():
+ """Class A mask."""
+ result = _calculate_network("10.200.50.1", "255.0.0.0")
+ assert result == "10.0.0.0/8"
diff --git a/tests/test_config.py b/tests/test_config.py
@@ -0,0 +1,76 @@
+"""Tests for ap_scanner.config."""
+
+import argparse
+
+from ap_scanner.config import apply_config_to_args, load_config
+
+
+def test_load_config_missing_file(tmp_path):
+ result = load_config(str(tmp_path / "nonexistent.ini"))
+ assert result["band"] == "all"
+ assert result["sort"] == "rssi"
+ assert result["interface"] == ""
+
+
+def test_load_config_valid_file(tmp_path):
+ cfg = tmp_path / "config.ini"
+ cfg.write_text("[scanner]\ninterface = wlan0\nband = 5\nsort = channel\n")
+ result = load_config(str(cfg))
+ assert result["interface"] == "wlan0"
+ assert result["band"] == "5"
+ assert result["sort"] == "channel"
+
+
+def test_load_config_partial_file(tmp_path):
+ cfg = tmp_path / "config.ini"
+ cfg.write_text("[scanner]\nband = 2.4\n")
+ result = load_config(str(cfg))
+ assert result["band"] == "2.4"
+ assert result["sort"] == "rssi" # default
+
+
+def test_apply_config_cli_overrides():
+ args = argparse.Namespace(
+ interface="wlan1", channels=None, band="all",
+ sort="rssi", no_color=False, verbose=False,
+ log_file=None, write=None,
+ )
+ config = {"interface": "wlan0", "channels": "1,6,11", "band": "5",
+ "sort": "channel", "no_color": "false", "verbose": "false",
+ "log_file": "", "write": ""}
+ apply_config_to_args(args, config)
+ # CLI set interface to wlan1, should NOT be overridden
+ assert args.interface == "wlan1"
+ # CLI didn't set channels, should pick up from config
+ assert args.channels == "1,6,11"
+ # CLI band is default "all", config says "5"
+ assert args.band == "5"
+
+
+def test_apply_config_all_defaults():
+ args = argparse.Namespace(
+ interface=None, channels=None, band="all",
+ sort="rssi", no_color=False, verbose=False,
+ log_file=None, write=None,
+ )
+ config = {"interface": "", "channels": "", "band": "all",
+ "sort": "rssi", "no_color": "false", "verbose": "false",
+ "log_file": "", "write": ""}
+ apply_config_to_args(args, config)
+ assert args.interface is None
+ assert args.band == "all"
+
+
+def test_apply_config_bool_options():
+ args = argparse.Namespace(
+ interface=None, channels=None, band="all",
+ sort="rssi", no_color=False, verbose=False,
+ log_file=None, write=None,
+ )
+ config = {"interface": "", "channels": "", "band": "all",
+ "sort": "rssi", "no_color": "true", "verbose": "true",
+ "log_file": "/tmp/test.log", "write": ""}
+ apply_config_to_args(args, config)
+ assert args.no_color is True
+ assert args.verbose is True
+ assert args.log_file == "/tmp/test.log"
diff --git a/tests/test_display.py b/tests/test_display.py
@@ -0,0 +1,209 @@
+"""Tests for ap_scanner.display."""
+
+from ap_scanner.display import Display, _display_width, _pad_to_width, _truncate_to_width
+
+
+# --- _display_width ---
+
+def test_display_width_ascii():
+ assert _display_width("hello") == 5
+
+
+def test_display_width_cjk():
+ # CJK characters are 2 cells wide each
+ assert _display_width("\u5317\u4eac") == 4 # 北京
+
+
+def test_display_width_mixed():
+ assert _display_width("AB\u5317\u4eac") == 6 # 2 ASCII + 2 CJK
+
+
+def test_display_width_empty():
+ assert _display_width("") == 0
+
+
+# --- _truncate_to_width ---
+
+def test_truncate_no_change():
+ assert _truncate_to_width("short", 10) == "short"
+
+
+def test_truncate_long_ascii():
+ result = _truncate_to_width("a" * 30, 10)
+ assert _display_width(result) <= 11 # 10 + ellipsis
+
+
+def test_truncate_cjk():
+ # 5 CJK chars = 10 cells, truncate to 7
+ result = _truncate_to_width("\u5317\u4eac\u4e0a\u6d77\u5e7f", 7)
+ assert _display_width(result) <= 8 # truncated + ellipsis
+
+
+# --- _pad_to_width ---
+
+def test_pad_to_width_shorter():
+ result = _pad_to_width("hi", 10)
+ assert _display_width(result) == 10
+
+
+def test_pad_to_width_exact():
+ result = _pad_to_width("12345", 5)
+ assert result == "12345"
+
+
+def test_pad_to_width_cjk():
+ # 2 CJK chars = 4 cells, pad to 10
+ result = _pad_to_width("\u5317\u4eac", 10)
+ assert _display_width(result) == 10
+ assert result.endswith(" " * 6)
+
+
+# --- Display class ---
+
+def test_cycle_sort():
+ d = Display()
+ assert d.sort_field == "rssi"
+ d.cycle_sort()
+ assert d.sort_field == "channel"
+ d.cycle_sort()
+ assert d.sort_field == "essid"
+ d.cycle_sort()
+ assert d.sort_field == "enc"
+ d.cycle_sort()
+ assert d.sort_field == "rssi" # wraps around
+
+
+def test_cycle_band():
+ d = Display()
+ assert d.band_filter is None
+
+ result = d.cycle_band()
+ assert result == "2.4GHz"
+ assert d.band_filter == "2.4GHz"
+
+ result = d.cycle_band()
+ assert result == "5GHz"
+
+ result = d.cycle_band()
+ assert result == "6GHz"
+ assert d.band_filter == "6GHz"
+
+ result = d.cycle_band()
+ assert result == "all"
+ assert d.band_filter is None
+
+
+def test_set_sort_valid():
+ d = Display()
+ d.set_sort("channel")
+ assert d.sort_field == "channel"
+
+
+def test_set_sort_invalid():
+ d = Display()
+ d.set_sort("nonexistent")
+ assert d.sort_field == "rssi" # unchanged
+
+
+# --- Selection mode ---
+
+def test_selection_mode_toggle():
+ d = Display()
+ assert d._selection_mode is False
+ d.enter_selection_mode()
+ assert d._selection_mode is True
+ assert d._selected_index == 0
+ d.exit_selection_mode()
+ assert d._selection_mode is False
+ assert d._selected_index == 0
+
+
+def test_move_selection_bounds():
+ d = Display()
+ d.enter_selection_mode()
+ # Move down within bounds
+ d.move_selection(1, 5)
+ assert d._selected_index == 1
+ d.move_selection(1, 5)
+ assert d._selected_index == 2
+ # Move past end — should clamp
+ d.move_selection(10, 5)
+ assert d._selected_index == 4
+ # Move past start — should clamp to 0
+ d.move_selection(-10, 5)
+ assert d._selected_index == 0
+
+
+def test_move_selection_empty():
+ d = Display()
+ d.enter_selection_mode()
+ d.move_selection(1, 0)
+ assert d._selected_index == 0
+
+
+def test_get_sorted_filtered(sample_ap, open_ap):
+ d = Display()
+ aps = {sample_ap.bssid: sample_ap, open_ap.bssid: open_ap}
+ result = d.get_sorted_filtered(aps)
+ assert len(result) == 2
+ # Default sort by rssi descending: -42 > -73
+ assert result[0].bssid == sample_ap.bssid
+
+
+def test_get_sorted_filtered_band(sample_ap, open_ap):
+ d = Display()
+ d.band_filter = "5GHz"
+ aps = {sample_ap.bssid: sample_ap, open_ap.bssid: open_ap}
+ result = d.get_sorted_filtered(aps)
+ # Both are 2.4GHz, so filtering to 5GHz returns nothing
+ assert len(result) == 0
+
+
+# --- Overlay title padding fix ---
+
+def test_overlay_title_width():
+ """Verify overlay title line does not exceed border width (bug fix)."""
+ from ap_scanner.display import _display_width
+
+ box_width = 72
+ title = "ACTIVE SCAN: TestNet (AA:BB:CC:DD:EE:FF)"
+ title_line = f"| {title}"
+ pad = box_width - 3 - _display_width(title)
+ title_line += " " * max(0, pad) + "|"
+
+ border = "+" + "-" * (box_width - 2) + "+"
+
+ assert _display_width(title_line) == _display_width(border), (
+ f"title_line width ({_display_width(title_line)}) != "
+ f"border width ({_display_width(border)})"
+ )
+
+
+def test_overlay_content_line_width():
+ """Verify overlay content lines match border width."""
+ from ap_scanner.display import _display_width, _truncate_to_width
+
+ box_width = 72
+ line = "Assigned IP: 192.168.1.100"
+ display_line = _truncate_to_width(line, box_width - 4)
+ pad = box_width - 2 - _display_width(display_line) - 1
+ row = f"| {display_line}" + " " * max(0, pad) + "|"
+
+ border = "+" + "-" * (box_width - 2) + "+"
+
+ assert _display_width(row) == _display_width(border)
+
+
+def test_overlay_title_width_cjk():
+ """Overlay title with CJK characters must not exceed border width."""
+ from ap_scanner.display import _display_width
+
+ box_width = 72
+ title = "\u5317\u4eac\u7f51\u7edc (AA:BB:CC:DD:EE:FF)" # 北京网络
+ title_line = f"| {title}"
+ pad = box_width - 3 - _display_width(title)
+ title_line += " " * max(0, pad) + "|"
+
+ border = "+" + "-" * (box_width - 2) + "+"
+
+ assert _display_width(title_line) == _display_width(border)
diff --git a/tests/test_exporter.py b/tests/test_exporter.py
@@ -0,0 +1,156 @@
+"""Tests for ap_scanner.exporter."""
+
+import csv
+import json
+from pathlib import Path
+
+from ap_scanner.exporter import generate_filename, save_csv, save_json
+
+
+def test_generate_filename_format():
+ name = generate_filename()
+ assert name.startswith("ap_scan_")
+ assert name.endswith(".csv")
+
+
+def test_save_csv_creates_file(sample_ap, tmp_path):
+ filepath = tmp_path / "test_output.csv"
+ aps = {sample_ap.bssid: sample_ap}
+ result = save_csv(aps, str(filepath))
+ assert Path(result).exists()
+
+
+def test_save_csv_field_count(sample_ap, tmp_path):
+ filepath = tmp_path / "test_fields.csv"
+ aps = {sample_ap.bssid: sample_ap}
+ save_csv(aps, str(filepath))
+
+ with open(filepath, "r", encoding="utf-8") as f:
+ reader = csv.reader(f, delimiter=";")
+ header = next(reader)
+ assert len(header) == 20
+ row = next(reader)
+ assert len(row) == 20
+
+
+def test_save_csv_semicolon_delimiter(sample_ap, tmp_path):
+ filepath = tmp_path / "test_delim.csv"
+ aps = {sample_ap.bssid: sample_ap}
+ save_csv(aps, str(filepath))
+
+ content = filepath.read_text(encoding="utf-8")
+ # Header should contain semicolons (semicolon-delimited)
+ assert ";" in content.splitlines()[0]
+
+
+def test_save_csv_auto_filename(sample_ap, tmp_path, monkeypatch):
+ monkeypatch.chdir(tmp_path)
+ aps = {sample_ap.bssid: sample_ap}
+ result = save_csv(aps)
+ assert result.startswith("ap_scan_")
+ assert Path(result).exists()
+
+
+# --- JSON export ---
+
+def test_generate_filename_json():
+ name = generate_filename("json")
+ assert name.startswith("ap_scan_")
+ assert name.endswith(".json")
+
+
+def test_save_json_creates_file(sample_ap, tmp_path):
+ filepath = tmp_path / "test_output.json"
+ aps = {sample_ap.bssid: sample_ap}
+ result = save_json(aps, str(filepath))
+ assert Path(result).exists()
+
+
+def test_save_json_valid_structure(sample_ap, tmp_path):
+ filepath = tmp_path / "test_struct.json"
+ aps = {sample_ap.bssid: sample_ap}
+ save_json(aps, str(filepath))
+
+ with open(filepath, "r", encoding="utf-8") as f:
+ data = json.load(f)
+
+ assert "scan_time" in data
+ assert "access_points" in data
+ assert len(data["access_points"]) == 1
+ ap = data["access_points"][0]
+ assert ap["bssid"] == sample_ap.bssid
+ assert ap["ssid"] == sample_ap.display_ssid
+ assert ap["channel"] == sample_ap.channel
+
+
+def test_save_json_auto_filename(sample_ap, tmp_path, monkeypatch):
+ monkeypatch.chdir(tmp_path)
+ aps = {sample_ap.bssid: sample_ap}
+ result = save_json(aps)
+ assert result.startswith("ap_scan_")
+ assert result.endswith(".json")
+ assert Path(result).exists()
+
+
+# --- Active scan export ---
+
+def test_save_json_with_active_scan(open_ap, tmp_path):
+ from ap_scanner.models import ActiveScanClient, ActiveScanResult
+
+ open_ap.active_scan_result = ActiveScanResult(
+ bssid=open_ap.bssid, ssid=open_ap.display_ssid,
+ connected=True, assigned_ip="192.168.1.100",
+ gateway_ip="192.168.1.1", gateway_mac="AA:BB:CC:DD:EE:01",
+ dns_servers=["8.8.8.8"], gateway_open_ports=[80, 443],
+ clients=[ActiveScanClient(ip="192.168.1.50", mac="AA:AA:AA:AA:AA:AA")],
+ duration_secs=12.3,
+ )
+ filepath = tmp_path / "active.json"
+ aps = {open_ap.bssid: open_ap}
+ save_json(aps, str(filepath))
+
+ with open(filepath, "r", encoding="utf-8") as f:
+ data = json.load(f)
+
+ ap_data = data["access_points"][0]
+ assert "active_scan" in ap_data
+ scan = ap_data["active_scan"]
+ assert scan["connected"] is True
+ assert scan["assigned_ip"] == "192.168.1.100"
+ assert scan["gateway_ip"] == "192.168.1.1"
+ assert scan["gateway_open_ports"] == [80, 443]
+ assert len(scan["clients"]) == 1
+ assert scan["clients"][0]["ip"] == "192.168.1.50"
+
+
+def test_save_csv_active_scan_columns(open_ap, tmp_path):
+ from ap_scanner.models import ActiveScanResult
+
+ open_ap.active_scan_result = ActiveScanResult(
+ bssid=open_ap.bssid, ssid=open_ap.display_ssid,
+ connected=True, assigned_ip="10.0.0.5",
+ gateway_ip="10.0.0.1", gateway_open_ports=[22, 80],
+ )
+ filepath = tmp_path / "active.csv"
+ aps = {open_ap.bssid: open_ap}
+ save_csv(aps, str(filepath))
+
+ with open(filepath, "r", encoding="utf-8") as f:
+ reader = csv.DictReader(f, delimiter=";")
+ row = next(reader)
+
+ assert row["Active Scan IP"] == "10.0.0.5"
+ assert row["Active Scan Gateway"] == "10.0.0.1"
+ assert row["Active Scan Ports"] == "22,80"
+
+
+def test_save_json_no_active_scan(sample_ap, tmp_path):
+ filepath = tmp_path / "no_active.json"
+ aps = {sample_ap.bssid: sample_ap}
+ save_json(aps, str(filepath))
+
+ with open(filepath, "r", encoding="utf-8") as f:
+ data = json.load(f)
+
+ ap_data = data["access_points"][0]
+ assert "active_scan" not in ap_data
diff --git a/tests/test_log.py b/tests/test_log.py
@@ -0,0 +1,43 @@
+"""Tests for ap_scanner.log — logging level save/restore."""
+
+import logging
+import sys
+
+from ap_scanner.log import (
+ restore_stderr_after_scan,
+ setup_logging,
+ suppress_stderr_during_scan,
+)
+
+
+def _get_stderr_handler_level() -> int | None:
+ """Return the level of the stderr handler on the ap_scanner logger."""
+ root = logging.getLogger("ap_scanner")
+ for h in root.handlers:
+ if isinstance(h, logging.StreamHandler) and h.stream is sys.stderr:
+ return h.level
+ return None
+
+
+def test_restore_level_info():
+ """After setup(verbose=False), restore returns to INFO (not DEBUG)."""
+ setup_logging(verbose=False, log_file=None)
+ assert _get_stderr_handler_level() == logging.INFO
+
+ suppress_stderr_during_scan()
+ assert _get_stderr_handler_level() == logging.CRITICAL
+
+ restore_stderr_after_scan()
+ assert _get_stderr_handler_level() == logging.INFO
+
+
+def test_restore_level_debug():
+ """After setup(verbose=True), restore returns to DEBUG."""
+ setup_logging(verbose=True, log_file=None)
+ assert _get_stderr_handler_level() == logging.DEBUG
+
+ suppress_stderr_during_scan()
+ assert _get_stderr_handler_level() == logging.CRITICAL
+
+ restore_stderr_after_scan()
+ assert _get_stderr_handler_level() == logging.DEBUG
diff --git a/tests/test_models.py b/tests/test_models.py
@@ -0,0 +1,120 @@
+"""Tests for ap_scanner.models."""
+
+from ap_scanner.models import AccessPoint, ActiveScanClient, ActiveScanResult
+
+
+def test_is_hidden_empty_ssid(hidden_ap):
+ assert hidden_ap.is_hidden is True
+
+
+def test_is_hidden_normal_ssid(sample_ap):
+ assert sample_ap.is_hidden is False
+
+
+def test_is_hidden_null_bytes():
+ ap = AccessPoint(
+ bssid="FF:FF:FF:FF:FF:FF", ssid="\x00\x00\x00",
+ channel=1, rssi=-50, encryption="OPN", cipher="--",
+ auth="--", band="2.4GHz", wifi_standard="b",
+ )
+ assert ap.is_hidden is True
+
+
+def test_is_hidden_whitespace_only():
+ ap = AccessPoint(
+ bssid="FF:FF:FF:FF:FF:FF", ssid=" ",
+ channel=1, rssi=-50, encryption="OPN", cipher="--",
+ auth="--", band="2.4GHz", wifi_standard="b",
+ )
+ assert ap.is_hidden is True
+
+
+def test_display_ssid_hidden(hidden_ap):
+ assert hidden_ap.display_ssid == "[Hidden]"
+
+
+def test_display_ssid_normal(sample_ap):
+ assert sample_ap.display_ssid == "TestNetwork"
+
+
+def test_rssi_display_none():
+ ap = AccessPoint(
+ bssid="FF:FF:FF:FF:FF:FF", ssid="Test",
+ channel=1, rssi=None, encryption="OPN", cipher="--",
+ auth="--", band="2.4GHz", wifi_standard="b",
+ )
+ assert ap.rssi_display == "N/A"
+
+
+def test_rssi_display_value(sample_ap):
+ assert sample_ap.rssi_display == "-42"
+
+
+def test_standard_display_empty():
+ ap = AccessPoint(
+ bssid="FF:FF:FF:FF:FF:FF", ssid="Test",
+ channel=1, rssi=-50, encryption="OPN", cipher="--",
+ auth="--", band="2.4GHz", wifi_standard="",
+ )
+ assert ap.standard_display == "--"
+
+
+def test_standard_display_value(sample_ap):
+ assert sample_ap.standard_display == "b, g, n"
+
+
+def test_channel_display_bonded(hidden_ap):
+ assert hidden_ap.channel_display == "36+40+44+48"
+
+
+def test_channel_display_primary():
+ ap = AccessPoint(
+ bssid="FF:FF:FF:FF:FF:FF", ssid="Test",
+ channel=6, rssi=-50, encryption="OPN", cipher="--",
+ auth="--", band="2.4GHz", wifi_standard="b",
+ )
+ assert ap.channel_display == "6"
+
+
+# --- ActiveScanResult ---
+
+def test_active_scan_result_defaults():
+ r = ActiveScanResult(bssid="AA:BB:CC:DD:EE:FF", ssid="Test")
+ assert r.connected is False
+ assert r.error == ""
+ assert r.clients == []
+ assert r.gateway_open_ports == []
+ assert r.client_count == 0
+
+
+def test_active_scan_result_summary_success():
+ r = ActiveScanResult(
+ bssid="AA:BB:CC:DD:EE:FF", ssid="Test",
+ connected=True, assigned_ip="192.168.1.100",
+ gateway_ip="192.168.1.1",
+ clients=[ActiveScanClient(ip="192.168.1.50", mac="AA:AA:AA:AA:AA:AA")],
+ gateway_open_ports=[80, 443],
+ )
+ s = r.summary
+ assert "IP: 192.168.1.100" in s
+ assert "GW: 192.168.1.1" in s
+ assert "Clients: 1" in s
+ assert "80,443" in s
+
+
+def test_active_scan_result_summary_failed():
+ r = ActiveScanResult(
+ bssid="AA:BB:CC:DD:EE:FF", ssid="Test",
+ error="Connection refused",
+ )
+ assert "Failed:" in r.summary
+ assert "Connection refused" in r.summary
+
+
+def test_active_scan_result_summary_empty():
+ r = ActiveScanResult(bssid="AA:BB:CC:DD:EE:FF", ssid="Test")
+ assert r.summary == "No data collected"
+
+
+def test_access_point_active_scan_none(sample_ap):
+ assert sample_ap.active_scan_result is None
diff --git a/tests/test_oui.py b/tests/test_oui.py
@@ -0,0 +1,25 @@
+"""Tests for ap_scanner.oui."""
+
+from ap_scanner.oui import lookup_vendor
+
+
+def test_lookup_known_vendor_tp_link():
+ result = lookup_vendor("5C:A6:E6:11:22:33")
+ assert result == "TP-Link"
+
+
+def test_lookup_known_vendor_cisco():
+ result = lookup_vendor("00:1B:2A:11:22:33")
+ assert result == "Cisco"
+
+
+def test_lookup_unknown_vendor():
+ result = lookup_vendor("FF:FF:FF:11:22:33")
+ # Should return empty string or some value from system OUI file
+ assert isinstance(result, str)
+
+
+def test_lookup_case_insensitive():
+ # The function should handle lowercase MAC addresses
+ result = lookup_vendor("5c:a6:e6:11:22:33")
+ assert result == "TP-Link"
diff --git a/tests/test_parser.py b/tests/test_parser.py
@@ -0,0 +1,220 @@
+"""Tests for ap_scanner.parser."""
+
+import pytest
+
+from scapy.layers.dot11 import (
+ Dot11,
+ Dot11Beacon,
+ Dot11Elt,
+ Dot11ProbeReq,
+ Dot11ProbeResp,
+ RadioTap,
+)
+
+from ap_scanner.parser import (
+ _extract_ssid,
+ _collect_ies,
+ _parse_wps_version,
+ parse_packet,
+ parse_probe_request,
+)
+
+
+def _make_beacon(ssid: str = "TestSSID", bssid: str = "AA:BB:CC:DD:EE:FF",
+ channel: int = 6, freq: int = 2437) -> object:
+ """Create a minimal beacon packet for testing."""
+ pkt = (
+ RadioTap(dBm_AntSignal=-42, ChannelFrequency=freq)
+ / Dot11(addr3=bssid)
+ / Dot11Beacon(cap="ESS+privacy")
+ / Dot11Elt(ID=0, info=ssid.encode())
+ / Dot11Elt(ID=3, info=bytes([channel]))
+ )
+ return pkt
+
+
+def _make_probe_request(ssid: str = "TargetNet",
+ client_mac: str = "11:22:33:44:55:66") -> object:
+ """Create a minimal probe request packet."""
+ pkt = (
+ RadioTap()
+ / Dot11(addr2=client_mac)
+ / Dot11ProbeReq()
+ / Dot11Elt(ID=0, info=ssid.encode())
+ )
+ return pkt
+
+
+# --- parse_packet ---
+
+def test_parse_beacon_basic():
+ pkt = _make_beacon(ssid="HomeNet", channel=6, freq=2437)
+ ap = parse_packet(pkt)
+ assert ap is not None
+ assert ap.ssid == "HomeNet"
+ assert ap.bssid == "AA:BB:CC:DD:EE:FF"
+ assert ap.channel == 6
+ # Note: RSSI may be None in mock packets depending on Scapy version
+ # The important thing is that the AP was parsed successfully
+
+
+def test_parse_hidden_ssid():
+ pkt = _make_beacon(ssid="", channel=1, freq=2412)
+ ap = parse_packet(pkt)
+ assert ap is not None
+ assert ap.is_hidden is True
+ assert ap.display_ssid == "[Hidden]"
+
+
+def test_parse_5ghz_beacon():
+ pkt = _make_beacon(ssid="Office5G", channel=36, freq=5180,
+ bssid="22:33:44:55:66:77")
+ ap = parse_packet(pkt)
+ assert ap is not None
+ assert ap.band == "5GHz"
+ assert ap.channel == 36
+
+
+def test_parse_non_beacon_returns_none():
+ # Data frame (not beacon/probe response)
+ pkt = RadioTap() / Dot11(type=2, addr3="AA:BB:CC:DD:EE:FF")
+ result = parse_packet(pkt)
+ assert result is None
+
+
+def test_parse_no_bssid_returns_none():
+ pkt = RadioTap() / Dot11(addr3=None) / Dot11Beacon() / Dot11Elt(ID=0, info=b"Test")
+ result = parse_packet(pkt)
+ assert result is None
+
+
+# --- parse_probe_request ---
+
+def test_parse_probe_request_basic():
+ pkt = _make_probe_request(ssid="TargetNet", client_mac="AA:BB:CC:DD:EE:01")
+ result = parse_probe_request(pkt)
+ assert result is not None
+ ssid, client = result
+ assert ssid == "TargetNet"
+ assert client == "AA:BB:CC:DD:EE:01"
+
+
+def test_parse_probe_request_empty_ssid():
+ pkt = _make_probe_request(ssid="")
+ result = parse_probe_request(pkt)
+ assert result is None
+
+
+def test_parse_probe_request_non_probe():
+ pkt = _make_beacon()
+ result = parse_probe_request(pkt)
+ assert result is None
+
+
+# --- _extract_ssid ---
+
+def test_extract_ssid_basic():
+ ies = {0: [b"MyNetwork"]}
+ assert _extract_ssid(ies) == "MyNetwork"
+
+
+def test_extract_ssid_empty():
+ ies = {0: [b""]}
+ assert _extract_ssid(ies) == ""
+
+
+def test_extract_ssid_missing():
+ ies = {}
+ assert _extract_ssid(ies) == ""
+
+
+def test_extract_ssid_capped_at_32_bytes():
+ long_ssid = b"A" * 50
+ ies = {0: [long_ssid]}
+ result = _extract_ssid(ies)
+ assert len(result) == 32
+
+
+# --- _parse_wps_version ---
+
+def test_parse_wps_version_basic():
+ # Version attribute: type=0x104A, length=1, value=0x10 (v1.0)
+ data = b"\x10\x4A\x00\x01\x10"
+ assert _parse_wps_version(data) == "1.0"
+
+
+def test_parse_wps_version_2():
+ # Version attribute: type=0x104A, length=1, value=0x20 (v2.0)
+ data = b"\x10\x4A\x00\x01\x20"
+ assert _parse_wps_version(data) == "2.0"
+
+
+def test_parse_wps_version_empty():
+ assert _parse_wps_version(b"") == ""
+
+
+def test_parse_wps_version_truncated():
+ # Truncated data should not crash
+ data = b"\x10\x4A\x00"
+ assert _parse_wps_version(data) == ""
+
+
+def test_parse_wps_version_malformed_vendor_ext():
+ # Vendor Extension (0x1049) with truncated sub-element
+ data = b"\x10\x49\x00\x04\x00\x37\x2A\x01" # WFA OUI + 1 byte
+ result = _parse_wps_version(data)
+ assert isinstance(result, str) # Should not crash
+
+
+# --- parse_packet edge cases ---
+
+def test_parse_beacon_channel_0_rejected():
+ """Beacons with channel 0 should be rejected (unreliable data)."""
+ pkt = (
+ RadioTap()
+ / Dot11(addr3="AA:BB:CC:DD:EE:FF")
+ / Dot11Beacon(cap="ESS+privacy")
+ / Dot11Elt(ID=0, info=b"TestSSID")
+ # No DS Parameter Set IE and no RadioTap frequency -> channel 0
+ )
+ ap = parse_packet(pkt)
+ assert ap is None
+
+
+def test_parse_probe_response():
+ """Probe Response frames should be parsed just like beacons."""
+ pkt = (
+ RadioTap(dBm_AntSignal=-55, ChannelFrequency=2437)
+ / Dot11(addr3="11:22:33:44:55:66")
+ / Dot11ProbeResp(cap="ESS+privacy")
+ / Dot11Elt(ID=0, info=b"ProbeNet")
+ / Dot11Elt(ID=3, info=bytes([6]))
+ )
+ ap = parse_packet(pkt)
+ assert ap is not None
+ assert ap.ssid == "ProbeNet"
+ assert ap.bssid == "11:22:33:44:55:66"
+
+
+def test_parse_ssid_utf8_decode():
+ """Non-ASCII SSIDs should be decoded with errors='replace'."""
+ pkt = _make_beacon(ssid="X", channel=1, freq=2412)
+ # Manually replace SSID with invalid UTF-8
+ elt = pkt.getlayer(Dot11Elt)
+ elt.info = b"\xc0\xc1\xff" # invalid UTF-8 bytes
+ ap = parse_packet(pkt)
+ assert ap is not None
+ # Should contain replacement characters, not crash
+ assert "\ufffd" in ap.ssid
+
+
+def test_parse_probe_request_null_addr2():
+ """Probe request with no source address should return None."""
+ pkt = (
+ RadioTap()
+ / Dot11(addr2=None)
+ / Dot11ProbeReq()
+ / Dot11Elt(ID=0, info=b"TestNet")
+ )
+ result = parse_probe_request(pkt)
+ assert result is None
diff --git a/tests/test_scanner.py b/tests/test_scanner.py
@@ -0,0 +1,108 @@
+"""Tests for ap_scanner.scanner — ChannelHopper schedule + ScannerEngine logic."""
+
+from unittest.mock import patch
+
+from ap_scanner.scanner import ChannelHopper, ScannerEngine
+
+
+# --- ChannelHopper._build_schedule ---
+
+class TestBuildSchedule:
+ """Tests for ChannelHopper._build_schedule weighted scheduling."""
+
+ def _make_hopper(self, channels):
+ """Create a ChannelHopper without starting any threads."""
+ with patch.object(ChannelHopper, "_get_mac", return_value="00:00:00:00:00:00"):
+ return ChannelHopper("wlan0", channels)
+
+ def test_primary_channels_appear_first(self):
+ """Channels 1, 6, 11 should be the first entries for quick discovery."""
+ hopper = self._make_hopper([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11])
+ sched = hopper._schedule
+ # First three entries should be the primary channels
+ assert sched[0] in {1, 6, 11}
+ assert sched[1] in {1, 6, 11}
+ assert sched[2] in {1, 6, 11}
+
+ def test_primary_channels_visited_more(self):
+ """Primary channels (1, 6, 11) should appear more than once."""
+ hopper = self._make_hopper(list(range(1, 14)))
+ sched = hopper._schedule
+ for ch in (1, 6, 11):
+ assert sched.count(ch) >= 2, f"Channel {ch} should appear >= 2 times"
+
+ def test_only_primary_channels(self):
+ """With only primary channels, schedule still works."""
+ hopper = self._make_hopper([1, 6, 11])
+ sched = hopper._schedule
+ assert len(sched) > 0
+ assert set(sched) == {1, 6, 11}
+
+ def test_no_primary_channels(self):
+ """With no primary channels (5GHz only), schedule is a plain list."""
+ hopper = self._make_hopper([36, 40, 44, 48])
+ sched = hopper._schedule
+ assert sched == [36, 40, 44, 48]
+
+ def test_empty_channels(self):
+ """Empty channel list produces empty schedule."""
+ hopper = self._make_hopper([])
+ assert hopper._schedule == []
+
+ def test_single_channel(self):
+ """Single non-primary channel produces single-entry schedule."""
+ hopper = self._make_hopper([36])
+ assert hopper._schedule == [36]
+
+ def test_single_primary_channel(self):
+ """Single primary channel appears twice (prepended + tail)."""
+ hopper = self._make_hopper([6])
+ sched = hopper._schedule
+ assert 6 in sched
+ assert len(sched) >= 1
+
+ def test_5ghz_schedule_unchanged(self):
+ """5GHz-only channels should maintain order (no reordering)."""
+ channels = [36, 40, 44, 48, 52, 56, 60, 64]
+ hopper = self._make_hopper(channels)
+ assert hopper._schedule == channels
+
+ def test_mixed_channels_all_present(self):
+ """All channels from input are present in the schedule."""
+ channels = [1, 6, 11, 36, 40, 44, 48]
+ hopper = self._make_hopper(channels)
+ sched = hopper._schedule
+ for ch in channels:
+ assert ch in sched, f"Channel {ch} missing from schedule"
+
+
+# --- ScannerEngine AP tracking ---
+
+class TestScannerEngineAPUpdate:
+ """Tests for AP update logic in _on_packet."""
+
+ def test_access_points_returns_copy(self):
+ """access_points property should return a copy, not the internal dict."""
+ with patch.object(ChannelHopper, "_get_mac", return_value="00:00:00:00:00:00"):
+ engine = ScannerEngine("wlan0", [1, 6, 11])
+ aps1 = engine.access_points
+ aps2 = engine.access_points
+ assert aps1 is not aps2
+
+ def test_clear_removes_all_aps(self):
+ """clear() should remove all discovered APs."""
+ from datetime import datetime
+ from ap_scanner.models import AccessPoint
+
+ with patch.object(ChannelHopper, "_get_mac", return_value="00:00:00:00:00:00"):
+ engine = ScannerEngine("wlan0", [1, 6, 11])
+ # Manually inject an AP
+ ap = AccessPoint(
+ bssid="AA:BB:CC:DD:EE:FF", ssid="Test", channel=6,
+ rssi=-50, encryption="WPA2", cipher="CCMP", auth="PSK",
+ band="2.4GHz", wifi_standard="n",
+ )
+ engine._aps["AA:BB:CC:DD:EE:FF"] = ap
+ assert len(engine.access_points) == 1
+ engine.clear()
+ assert len(engine.access_points) == 0
diff --git a/tests/test_utils.py b/tests/test_utils.py
@@ -0,0 +1,135 @@
+"""Tests for ap_scanner.utils."""
+
+from ap_scanner.utils import (
+ ALL_CHANNELS,
+ CHANNELS_2G,
+ CHANNELS_5G,
+ CHANNELS_6G,
+ channel_band,
+ channel_to_freq,
+ freq_to_channel,
+ parse_channels_arg,
+)
+
+
+# --- freq_to_channel ---
+
+def test_freq_to_channel_2g_ch1():
+ assert freq_to_channel(2412) == 1
+
+
+def test_freq_to_channel_2g_ch6():
+ assert freq_to_channel(2437) == 6
+
+
+def test_freq_to_channel_2g_ch11():
+ assert freq_to_channel(2462) == 11
+
+
+def test_freq_to_channel_5g_ch36():
+ assert freq_to_channel(5180) == 36
+
+
+def test_freq_to_channel_5g_ch48():
+ assert freq_to_channel(5240) == 48
+
+
+def test_freq_to_channel_5g_ch149():
+ assert freq_to_channel(5745) == 149
+
+
+def test_freq_to_channel_unknown():
+ assert freq_to_channel(0) == 0
+
+
+# --- channel_to_freq ---
+
+def test_channel_to_freq_2g():
+ assert channel_to_freq(1) == 2412
+ assert channel_to_freq(6) == 2437
+
+
+def test_channel_to_freq_5g():
+ assert channel_to_freq(36) == 5180
+ assert channel_to_freq(165) == 5825
+
+
+def test_channel_to_freq_invalid():
+ assert channel_to_freq(999) == 0
+
+
+# --- channel_band ---
+
+def test_channel_band_2g():
+ assert channel_band(1) == "2.4"
+ assert channel_band(6) == "2.4"
+ assert channel_band(11) == "2.4"
+
+
+def test_channel_band_5g():
+ assert channel_band(36) == "5"
+ assert channel_band(149) == "5"
+ assert channel_band(165) == "5"
+
+
+def test_channel_band_6g_with_freq():
+ # Channel 1 at 5955 MHz is 6GHz, not 2.4GHz
+ assert channel_band(1, freq=5955) == "6"
+ assert channel_band(37, freq=6135) == "6"
+
+
+def test_channel_band_disambiguate_by_freq():
+ # Channel 1 without freq defaults to 2.4GHz
+ assert channel_band(1) == "2.4"
+ # Channel 1 with 2.4GHz freq
+ assert channel_band(1, freq=2412) == "2.4"
+ # Channel 1 with 6GHz freq
+ assert channel_band(1, freq=5955) == "6"
+
+
+def test_freq_to_channel_6g():
+ assert freq_to_channel(5955) == 1
+ assert freq_to_channel(6135) == 37
+
+
+def test_channels_6g_contains_common():
+ for ch in [1, 5, 9, 13, 37, 53, 149, 233]:
+ assert ch in CHANNELS_6G
+
+
+# --- parse_channels_arg ---
+
+def test_parse_channels_single():
+ result = parse_channels_arg("6")
+ assert result == [6]
+
+
+def test_parse_channels_list():
+ result = parse_channels_arg("1,6,11")
+ assert result == [1, 6, 11]
+
+
+def test_parse_channels_range():
+ result = parse_channels_arg("36-48")
+ assert result == [36, 40, 44, 48]
+
+
+def test_parse_channels_mixed():
+ result = parse_channels_arg("1,6,36-48")
+ assert result == [1, 6, 36, 40, 44, 48]
+
+
+def test_parse_channels_invalid_falls_back():
+ result = parse_channels_arg("999")
+ assert result == ALL_CHANNELS
+
+
+# --- Channel lists ---
+
+def test_channels_2g_range():
+ assert CHANNELS_2G == list(range(1, 15))
+
+
+def test_channels_5g_contains_common():
+ for ch in [36, 40, 44, 48, 149, 153, 157, 161, 165]:
+ assert ch in CHANNELS_5G
|