Mirror/intercept
1
0
Fork 0
mirror of https://github.com/smittix/intercept.git synced 2026-04-24 06:40:00 -07:00
Code Issues Packages Projects Releases Wiki Activity

Refactor into modular structure with improvements

Browse Source 
- Split monolithic intercept.py (15k lines) into modular structure:
  - routes/ - Flask blueprints for each feature
  - templates/ - Jinja2 HTML templates
  - data/ - OUI database, satellite TLEs, detection patterns
  - utils/ - dependencies, process management, logging
  - config.py - centralized configuration with env var support

- Add type hints to function signatures
- Replace bare except clauses with specific exceptions
- Add proper logging module (replaces print statements)
- Add environment variable support (INTERCEPT_* prefix)
- Add test suite with pytest
- Add Dockerfile for containerized deployment
- Add pyproject.toml with ruff/black/mypy config
- Add requirements-dev.txt for development dependencies

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
James Smith
2025-12-23 16:28:36 +00:00
parent b0bc7e6e91
commit ddeff002c9
32 changed files with 15581 additions and 15429 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
.dockerignore Normal file
View File

@@ -0,0 +1,38 @@
# Git
.git
.gitignore
# Python
__pycache__
*.py[cod]
*$py.class
*.so
.Python
.env
.venv
env/
venv/
.eggs/
*.egg-info/
*.egg
# IDE
.idea/
.vscode/
*.swp
*.swo
# Test/Dev
tests/
.pytest_cache/
.coverage
htmlcov/
.mypy_cache/
# Logs
*.log
# Captured files (don't include in image)
*.cap
*.pcap
*.csv

40
Dockerfile Normal file
View File

@@ -0,0 +1,40 @@
# INTERCEPT - Signal Intelligence Platform
# Docker container for running the web interface
FROM python:3.11-slim
# Set working directory
WORKDIR /app
# Install system dependencies for RTL-SDR tools
RUN apt-get update && apt-get install -y --no-install-recommends \
# RTL-SDR tools
rtl-sdr \
# 433MHz decoder
rtl-433 \
# Pager decoder
multimon-ng \
# WiFi tools (aircrack-ng suite)
aircrack-ng \
# Bluetooth tools
bluez \
# Cleanup
&& rm -rf /var/lib/apt/lists/*
# Copy requirements first for better caching
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy application code
COPY . .
# Expose web interface port
EXPOSE 5050
# Environment variables with defaults
ENV INTERCEPT_HOST=0.0.0.0 \
INTERCEPT_PORT=5050 \
INTERCEPT_LOG_LEVEL=INFO
# Run the application
CMD ["python", "intercept.py"]

194
app.py Normal file
View File

@@ -0,0 +1,194 @@
"""
INTERCEPT - Signal Intelligence Platform
Flask application and shared state.
"""
from __future__ import annotations
import sys
import site
# Ensure user site-packages is available (may be disabled when running as root/sudo)
if not site.ENABLE_USER_SITE:
user_site = site.getusersitepackages()
if user_site and user_site not in sys.path:
sys.path.insert(0, user_site)
import os
import queue
import threading
import platform
import subprocess
from typing import Any
from flask import Flask, render_template, jsonify, send_file, Response
from utils.dependencies import check_tool, check_all_dependencies, TOOL_DEPENDENCIES
from utils.process import detect_devices, cleanup_stale_processes
# Create Flask app
app = Flask(__name__)
# ============================================
# GLOBAL PROCESS MANAGEMENT
# ============================================
# Pager decoder
current_process = None
output_queue = queue.Queue()
process_lock = threading.Lock()
# RTL_433 sensor
sensor_process = None
sensor_queue = queue.Queue()
sensor_lock = threading.Lock()
# WiFi
wifi_process = None
wifi_queue = queue.Queue()
wifi_lock = threading.Lock()
# Bluetooth
bt_process = None
bt_queue = queue.Queue()
bt_lock = threading.Lock()
# ADS-B aircraft
adsb_process = None
adsb_queue = queue.Queue()
adsb_lock = threading.Lock()
# Satellite/Iridium
satellite_process = None
satellite_queue = queue.Queue()
satellite_lock = threading.Lock()
# ============================================
# GLOBAL STATE DICTIONARIES
# ============================================
# Logging settings
logging_enabled = False
log_file_path = 'pager_messages.log'
# WiFi state
wifi_monitor_interface = None
wifi_networks = {} # BSSID -> network info
wifi_clients = {} # Client MAC -> client info
wifi_handshakes = [] # Captured handshakes
# Bluetooth state
bt_interface = None
bt_devices = {} # MAC -> device info
bt_beacons = {} # MAC -> beacon info (AirTags, Tiles, iBeacons)
bt_services = {} # MAC -> list of services
# Aircraft (ADS-B) state
adsb_aircraft = {} # ICAO hex -> aircraft info
# Satellite state
iridium_bursts = [] # List of detected Iridium bursts
satellite_passes = [] # Predicted satellite passes
# ============================================
# MAIN ROUTES
# ============================================
@app.route('/')
def index() -> str:
tools = {
'rtl_fm': check_tool('rtl_fm'),
'multimon': check_tool('multimon-ng'),
'rtl_433': check_tool('rtl_433')
}
devices = detect_devices()
return render_template('index.html', tools=tools, devices=devices)
@app.route('/favicon.svg')
def favicon() -> Response:
return send_file('favicon.svg', mimetype='image/svg+xml')
@app.route('/devices')
def get_devices() -> Response:
return jsonify(detect_devices())
@app.route('/dependencies')
def get_dependencies() -> Response:
"""Get status of all tool dependencies."""
results = check_all_dependencies()
# Determine OS for install instructions
system = platform.system().lower()
if system == 'darwin':
install_method = 'brew'
elif system == 'linux':
install_method = 'apt'
else:
install_method = 'manual'
return jsonify({
'os': system,
'install_method': install_method,
'modes': results
})
@app.route('/killall', methods=['POST'])
def kill_all() -> Response:
"""Kill all decoder and WiFi processes."""
global current_process, sensor_process, wifi_process
killed = []
processes_to_kill = [
'rtl_fm', 'multimon-ng', 'rtl_433',
'airodump-ng', 'aireplay-ng', 'airmon-ng'
]
for proc in processes_to_kill:
try:
result = subprocess.run(['pkill', '-f', proc], capture_output=True)
if result.returncode == 0:
killed.append(proc)
except (subprocess.SubprocessError, OSError):
pass
with process_lock:
current_process = None
with sensor_lock:
sensor_process = None
with wifi_lock:
wifi_process = None
return jsonify({'status': 'killed', 'processes': killed})
def main() -> None:
"""Main entry point."""
print("=" * 50)
print(" INTERCEPT // Signal Intelligence")
print(" Pager / 433MHz / Aircraft / Satellite / WiFi / BT")
print("=" * 50)
print()
# Clean up any stale processes from previous runs
cleanup_stale_processes()
# Register blueprints
from routes import register_blueprints
register_blueprints(app)
print("Open http://localhost:5050 in your browser")
print()
print("Press Ctrl+C to stop")
print()
app.run(host='0.0.0.0', port=5050, debug=False, threaded=True)

97
config.py Normal file
View File

@@ -0,0 +1,97 @@
"""Configuration settings for intercept application."""
from __future__ import annotations
import logging
import os
import sys
def _get_env(key: str, default: str) -> str:
"""Get environment variable with default."""
return os.environ.get(f'INTERCEPT_{key}', default)
def _get_env_int(key: str, default: int) -> int:
"""Get environment variable as integer with default."""
try:
return int(os.environ.get(f'INTERCEPT_{key}', str(default)))
except ValueError:
return default
def _get_env_float(key: str, default: float) -> float:
"""Get environment variable as float with default."""
try:
return float(os.environ.get(f'INTERCEPT_{key}', str(default)))
except ValueError:
return default
def _get_env_bool(key: str, default: bool) -> bool:
"""Get environment variable as boolean with default."""
val = os.environ.get(f'INTERCEPT_{key}', '').lower()
if val in ('true', '1', 'yes', 'on'):
return True
if val in ('false', '0', 'no', 'off'):
return False
return default
# Logging configuration
_log_level_str = _get_env('LOG_LEVEL', 'WARNING').upper()
LOG_LEVEL = getattr(logging, _log_level_str, logging.WARNING)
LOG_FORMAT = _get_env('LOG_FORMAT', '%(asctime)s - %(levelname)s - %(message)s')
# Server settings
HOST = _get_env('HOST', '0.0.0.0')
PORT = _get_env_int('PORT', 5050)
DEBUG = _get_env_bool('DEBUG', False)
THREADED = _get_env_bool('THREADED', True)
# Default RTL-SDR settings
DEFAULT_GAIN = _get_env('DEFAULT_GAIN', '40')
DEFAULT_DEVICE = _get_env('DEFAULT_DEVICE', '0')
# Pager defaults
DEFAULT_PAGER_FREQ = _get_env('PAGER_FREQ', '929.6125M')
# Iridium defaults
DEFAULT_IRIDIUM_FREQ = _get_env('IRIDIUM_FREQ', '1626.0')
DEFAULT_IRIDIUM_SAMPLE_RATE = _get_env('IRIDIUM_SAMPLE_RATE', '2.048e6')
# Timeouts
PROCESS_TIMEOUT = _get_env_int('PROCESS_TIMEOUT', 5)
SOCKET_TIMEOUT = _get_env_int('SOCKET_TIMEOUT', 5)
SSE_TIMEOUT = _get_env_int('SSE_TIMEOUT', 1)
# WiFi settings
WIFI_UPDATE_INTERVAL = _get_env_float('WIFI_UPDATE_INTERVAL', 2.0)
AIRODUMP_HEADER_LINES = _get_env_int('AIRODUMP_HEADER_LINES', 2)
# Bluetooth settings
BT_SCAN_TIMEOUT = _get_env_int('BT_SCAN_TIMEOUT', 10)
BT_UPDATE_INTERVAL = _get_env_float('BT_UPDATE_INTERVAL', 2.0)
# ADS-B settings
ADSB_SBS_PORT = _get_env_int('ADSB_SBS_PORT', 30003)
ADSB_UPDATE_INTERVAL = _get_env_float('ADSB_UPDATE_INTERVAL', 1.0)
# Satellite settings
SATELLITE_UPDATE_INTERVAL = _get_env_int('SATELLITE_UPDATE_INTERVAL', 30)
SATELLITE_TRAJECTORY_POINTS = _get_env_int('SATELLITE_TRAJECTORY_POINTS', 30)
SATELLITE_ORBIT_MINUTES = _get_env_int('SATELLITE_ORBIT_MINUTES', 45)
# Maximum burst count for Iridium monitoring
IRIDIUM_MAX_BURSTS = _get_env_int('IRIDIUM_MAX_BURSTS', 100)
def configure_logging() -> None:
"""Configure application logging."""
logging.basicConfig(
level=LOG_LEVEL,
format=LOG_FORMAT,
stream=sys.stderr
)
# Suppress Flask development server warning
logging.getLogger('werkzeug').setLevel(LOG_LEVEL)

10
data/__init__.py Normal file
View File

@@ -0,0 +1,10 @@
# Data modules for INTERCEPT
from .oui import OUI_DATABASE, load_oui_database, get_manufacturer
from .satellites import TLE_SATELLITES
from .patterns import (
AIRTAG_PREFIXES,
TILE_PREFIXES,
SAMSUNG_TRACKER,
DRONE_SSID_PATTERNS,
DRONE_OUI_PREFIXES,
)

172
data/oui.py Normal file
View File

@@ -0,0 +1,172 @@
from __future__ import annotations
import logging
import os
import json
logger = logging.getLogger('intercept.oui')
def load_oui_database() -> dict[str, str] | None:
"""Load OUI database from external JSON file, with fallback to built-in."""
oui_file = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'oui_database.json')
try:
if os.path.exists(oui_file):
with open(oui_file, 'r') as f:
data = json.load(f)
# Remove comment fields
return {k: v for k, v in data.items() if not k.startswith('_')}
except Exception as e:
logger.warning(f"Error loading oui_database.json: {e}, using built-in database")
return None # Will fall back to built-in
def get_manufacturer(mac: str) -> str:
"""Look up manufacturer from MAC address OUI."""
prefix = mac[:8].upper()
return OUI_DATABASE.get(prefix, 'Unknown')
# OUI Database for manufacturer lookup (expanded)
OUI_DATABASE = {
# Apple (extensive list)
'00:25:DB': 'Apple', '04:52:F3': 'Apple', '0C:3E:9F': 'Apple', '10:94:BB': 'Apple',
'14:99:E2': 'Apple', '20:78:F0': 'Apple', '28:6A:BA': 'Apple', '3C:22:FB': 'Apple',
'40:98:AD': 'Apple', '48:D7:05': 'Apple', '4C:57:CA': 'Apple', '54:4E:90': 'Apple',
'5C:97:F3': 'Apple', '60:F8:1D': 'Apple', '68:DB:CA': 'Apple', '70:56:81': 'Apple',
'78:7B:8A': 'Apple', '7C:D1:C3': 'Apple', '84:FC:FE': 'Apple', '8C:2D:AA': 'Apple',
'90:B0:ED': 'Apple', '98:01:A7': 'Apple', '98:D6:BB': 'Apple', 'A4:D1:D2': 'Apple',
'AC:BC:32': 'Apple', 'B0:34:95': 'Apple', 'B8:C1:11': 'Apple', 'C8:69:CD': 'Apple',
'D0:03:4B': 'Apple', 'DC:A9:04': 'Apple', 'E0:C7:67': 'Apple', 'F0:18:98': 'Apple',
'F4:5C:89': 'Apple', '78:4F:43': 'Apple', '00:CD:FE': 'Apple', '04:4B:ED': 'Apple',
'04:D3:CF': 'Apple', '08:66:98': 'Apple', '0C:74:C2': 'Apple', '10:DD:B1': 'Apple',
'14:10:9F': 'Apple', '18:EE:69': 'Apple', '1C:36:BB': 'Apple', '24:A0:74': 'Apple',
'28:37:37': 'Apple', '2C:BE:08': 'Apple', '34:08:BC': 'Apple', '38:C9:86': 'Apple',
'3C:06:30': 'Apple', '44:D8:84': 'Apple', '48:A9:1C': 'Apple', '4C:32:75': 'Apple',
'50:32:37': 'Apple', '54:26:96': 'Apple', '58:B0:35': 'Apple', '5C:F7:E6': 'Apple',
'64:A3:CB': 'Apple', '68:FE:F7': 'Apple', '6C:4D:73': 'Apple', '70:DE:E2': 'Apple',
'74:E2:F5': 'Apple', '78:67:D7': 'Apple', '7C:04:D0': 'Apple', '80:E6:50': 'Apple',
'84:78:8B': 'Apple', '88:66:A5': 'Apple', '8C:85:90': 'Apple', '94:E9:6A': 'Apple',
'9C:F4:8E': 'Apple', 'A0:99:9B': 'Apple', 'A4:83:E7': 'Apple', 'A8:5C:2C': 'Apple',
'AC:1F:74': 'Apple', 'B0:19:C6': 'Apple', 'B4:F1:DA': 'Apple', 'BC:52:B7': 'Apple',
'C0:A5:3E': 'Apple', 'C4:B3:01': 'Apple', 'CC:20:E8': 'Apple', 'D0:C5:F3': 'Apple',
'D4:61:9D': 'Apple', 'D8:1C:79': 'Apple', 'E0:5F:45': 'Apple', 'E4:C6:3D': 'Apple',
'F0:B4:79': 'Apple', 'F4:0F:24': 'Apple', 'F8:4D:89': 'Apple', 'FC:D8:48': 'Apple',
# Samsung
'00:1B:66': 'Samsung', '00:21:19': 'Samsung', '00:26:37': 'Samsung', '5C:0A:5B': 'Samsung',
'8C:71:F8': 'Samsung', 'C4:73:1E': 'Samsung', '38:2C:4A': 'Samsung', '00:1E:4C': 'Samsung',
'00:12:47': 'Samsung', '00:15:99': 'Samsung', '00:17:D5': 'Samsung', '00:1D:F6': 'Samsung',
'00:21:D1': 'Samsung', '00:24:54': 'Samsung', '00:26:5D': 'Samsung', '08:D4:2B': 'Samsung',
'10:D5:42': 'Samsung', '14:49:E0': 'Samsung', '18:3A:2D': 'Samsung', '1C:66:AA': 'Samsung',
'24:4B:81': 'Samsung', '28:98:7B': 'Samsung', '2C:AE:2B': 'Samsung', '30:96:FB': 'Samsung',
'34:C3:AC': 'Samsung', '38:01:95': 'Samsung', '3C:5A:37': 'Samsung', '40:0E:85': 'Samsung',
'44:4E:1A': 'Samsung', '4C:BC:A5': 'Samsung', '50:01:BB': 'Samsung', '50:A4:D0': 'Samsung',
'54:88:0E': 'Samsung', '58:C3:8B': 'Samsung', '5C:2E:59': 'Samsung', '60:D0:A9': 'Samsung',
'64:B3:10': 'Samsung', '68:48:98': 'Samsung', '6C:2F:2C': 'Samsung', '70:F9:27': 'Samsung',
'74:45:8A': 'Samsung', '78:47:1D': 'Samsung', '7C:0B:C6': 'Samsung', '84:11:9E': 'Samsung',
'88:32:9B': 'Samsung', '8C:77:12': 'Samsung', '90:18:7C': 'Samsung', '94:35:0A': 'Samsung',
'98:52:B1': 'Samsung', '9C:02:98': 'Samsung', 'A0:0B:BA': 'Samsung', 'A4:7B:85': 'Samsung',
'A8:06:00': 'Samsung', 'AC:5F:3E': 'Samsung', 'B0:72:BF': 'Samsung', 'B4:79:A7': 'Samsung',
'BC:44:86': 'Samsung', 'C0:97:27': 'Samsung', 'C4:42:02': 'Samsung', 'CC:07:AB': 'Samsung',
'D0:22:BE': 'Samsung', 'D4:87:D8': 'Samsung', 'D8:90:E8': 'Samsung', 'E4:7C:F9': 'Samsung',
'E8:50:8B': 'Samsung', 'F0:25:B7': 'Samsung', 'F4:7B:5E': 'Samsung', 'FC:A1:3E': 'Samsung',
# Google
'54:60:09': 'Google', '00:1A:11': 'Google', 'F4:F5:D8': 'Google', '94:EB:2C': 'Google',
'64:B5:C6': 'Google', '3C:5A:B4': 'Google', 'F8:8F:CA': 'Google', '20:DF:B9': 'Google',
'54:27:1E': 'Google', '58:CB:52': 'Google', 'A4:77:33': 'Google', 'F4:0E:22': 'Google',
# Sony
'00:13:A9': 'Sony', '00:1D:28': 'Sony', '00:24:BE': 'Sony', '04:5D:4B': 'Sony',
'08:A9:5A': 'Sony', '10:4F:A8': 'Sony', '24:21:AB': 'Sony', '30:52:CB': 'Sony',
'40:B8:37': 'Sony', '58:48:22': 'Sony', '70:9E:29': 'Sony', '84:00:D2': 'Sony',
'AC:9B:0A': 'Sony', 'B4:52:7D': 'Sony', 'BC:60:A7': 'Sony', 'FC:0F:E6': 'Sony',
# Bose
'00:0C:8A': 'Bose', '04:52:C7': 'Bose', '08:DF:1F': 'Bose', '2C:41:A1': 'Bose',
'4C:87:5D': 'Bose', '60:AB:D2': 'Bose', '88:C9:E8': 'Bose', 'D8:9C:67': 'Bose',
# JBL/Harman
'00:1D:DF': 'JBL', '08:AE:D6': 'JBL', '20:3C:AE': 'JBL', '44:5E:F3': 'JBL',
'50:C9:71': 'JBL', '74:5E:1C': 'JBL', '88:C6:26': 'JBL', 'AC:12:2F': 'JBL',
# Beats (Apple subsidiary)
'00:61:71': 'Beats', '48:D6:D5': 'Beats', '9C:64:8B': 'Beats', 'A4:E9:75': 'Beats',
# Jabra/GN Audio
'00:13:17': 'Jabra', '1C:48:F9': 'Jabra', '50:C2:ED': 'Jabra', '70:BF:92': 'Jabra',
'74:5C:4B': 'Jabra', '94:16:25': 'Jabra', 'D0:81:7A': 'Jabra', 'E8:EE:CC': 'Jabra',
# Sennheiser
'00:1B:66': 'Sennheiser', '00:22:27': 'Sennheiser', 'B8:AD:3E': 'Sennheiser',
# Xiaomi
'04:CF:8C': 'Xiaomi', '0C:1D:AF': 'Xiaomi', '10:2A:B3': 'Xiaomi', '18:59:36': 'Xiaomi',
'20:47:DA': 'Xiaomi', '28:6C:07': 'Xiaomi', '34:CE:00': 'Xiaomi', '38:A4:ED': 'Xiaomi',
'44:23:7C': 'Xiaomi', '50:64:2B': 'Xiaomi', '58:44:98': 'Xiaomi', '64:09:80': 'Xiaomi',
'74:23:44': 'Xiaomi', '78:02:F8': 'Xiaomi', '7C:1C:4E': 'Xiaomi', '84:F3:EB': 'Xiaomi',
'8C:BE:BE': 'Xiaomi', '98:FA:E3': 'Xiaomi', 'A4:77:58': 'Xiaomi', 'AC:C1:EE': 'Xiaomi',
'B0:E2:35': 'Xiaomi', 'C4:0B:CB': 'Xiaomi', 'C8:47:8C': 'Xiaomi', 'D4:97:0B': 'Xiaomi',
'E4:46:DA': 'Xiaomi', 'F0:B4:29': 'Xiaomi', 'FC:64:BA': 'Xiaomi',
# Huawei
'00:18:82': 'Huawei', '00:1E:10': 'Huawei', '00:25:68': 'Huawei', '04:B0:E7': 'Huawei',
'08:63:61': 'Huawei', '10:1B:54': 'Huawei', '18:DE:D7': 'Huawei', '20:A6:80': 'Huawei',
'28:31:52': 'Huawei', '34:12:98': 'Huawei', '3C:47:11': 'Huawei', '48:00:31': 'Huawei',
'4C:50:77': 'Huawei', '5C:7D:5E': 'Huawei', '60:DE:44': 'Huawei', '70:72:3C': 'Huawei',
'78:F5:57': 'Huawei', '80:B6:86': 'Huawei', '88:53:D4': 'Huawei', '94:04:9C': 'Huawei',
'A4:99:47': 'Huawei', 'B4:15:13': 'Huawei', 'BC:76:70': 'Huawei', 'C8:D1:5E': 'Huawei',
'DC:D2:FC': 'Huawei', 'E4:68:A3': 'Huawei', 'F4:63:1F': 'Huawei',
# OnePlus/BBK
'64:A2:F9': 'OnePlus', 'C0:EE:FB': 'OnePlus', '94:65:2D': 'OnePlus',
# Fitbit
'2C:09:4D': 'Fitbit', 'C4:D9:87': 'Fitbit', 'E4:88:6D': 'Fitbit',
# Garmin
'00:1C:D1': 'Garmin', 'C4:AC:59': 'Garmin', 'E8:0F:C8': 'Garmin',
# Microsoft
'00:50:F2': 'Microsoft', '28:18:78': 'Microsoft', '60:45:BD': 'Microsoft',
'7C:1E:52': 'Microsoft', '98:5F:D3': 'Microsoft', 'B4:0E:DE': 'Microsoft',
# Intel
'00:1B:21': 'Intel', '00:1C:C0': 'Intel', '00:1E:64': 'Intel', '00:21:5C': 'Intel',
'08:D4:0C': 'Intel', '18:1D:EA': 'Intel', '34:02:86': 'Intel', '40:74:E0': 'Intel',
'48:51:B7': 'Intel', '58:A0:23': 'Intel', '64:D4:DA': 'Intel', '80:19:34': 'Intel',
'8C:8D:28': 'Intel', 'A4:4E:31': 'Intel', 'B4:6B:FC': 'Intel', 'C8:D0:83': 'Intel',
# Qualcomm/Atheros
'00:03:7F': 'Qualcomm', '00:24:E4': 'Qualcomm', '04:F0:21': 'Qualcomm',
'1C:4B:D6': 'Qualcomm', '88:71:B1': 'Qualcomm', 'A0:65:18': 'Qualcomm',
# Broadcom
'00:10:18': 'Broadcom', '00:1A:2B': 'Broadcom', '20:10:7A': 'Broadcom',
# Realtek
'00:0A:EB': 'Realtek', '00:E0:4C': 'Realtek', '48:02:2A': 'Realtek',
'52:54:00': 'Realtek', '80:EA:96': 'Realtek',
# Logitech
'00:1F:20': 'Logitech', '34:88:5D': 'Logitech', '6C:B7:49': 'Logitech',
# Lenovo
'00:09:2D': 'Lenovo', '28:D2:44': 'Lenovo', '54:EE:75': 'Lenovo', '98:FA:9B': 'Lenovo',
# Dell
'00:14:22': 'Dell', '00:1A:A0': 'Dell', '18:DB:F2': 'Dell', '34:17:EB': 'Dell',
'78:2B:CB': 'Dell', 'A4:BA:DB': 'Dell', 'E4:B9:7A': 'Dell',
# HP
'00:0F:61': 'HP', '00:14:C2': 'HP', '10:1F:74': 'HP', '28:80:23': 'HP',
'38:63:BB': 'HP', '5C:B9:01': 'HP', '80:CE:62': 'HP', 'A0:D3:C1': 'HP',
# Tile
'F8:E4:E3': 'Tile', 'C4:E7:BE': 'Tile', 'DC:54:D7': 'Tile', 'E4:B0:21': 'Tile',
# Raspberry Pi
'B8:27:EB': 'Raspberry Pi', 'DC:A6:32': 'Raspberry Pi', 'E4:5F:01': 'Raspberry Pi',
# Amazon
'00:FC:8B': 'Amazon', '10:CE:A9': 'Amazon', '34:D2:70': 'Amazon', '40:B4:CD': 'Amazon',
'44:65:0D': 'Amazon', '68:54:FD': 'Amazon', '74:C2:46': 'Amazon', '84:D6:D0': 'Amazon',
'A0:02:DC': 'Amazon', 'AC:63:BE': 'Amazon', 'B4:7C:9C': 'Amazon', 'FC:65:DE': 'Amazon',
# Skullcandy
'00:01:00': 'Skullcandy', '88:E6:03': 'Skullcandy',
# Bang & Olufsen
'00:21:3E': 'Bang & Olufsen', '78:C5:E5': 'Bang & Olufsen',
# Audio-Technica
'A0:E9:DB': 'Audio-Technica', 'EC:81:93': 'Audio-Technica',
# Plantronics/Poly
'00:1D:DF': 'Plantronics', 'B0:B4:48': 'Plantronics', 'E8:FC:AF': 'Plantronics',
# Anker
'AC:89:95': 'Anker', 'E8:AB:FA': 'Anker',
# Misc/Generic
'00:00:0A': 'Omron', '00:1A:7D': 'Cyber-Blue', '00:1E:3D': 'Alps Electric',
'00:0B:57': 'Silicon Wave', '00:02:72': 'CC&C',
}
# Try to load from external file (easier to update)
_external_oui = load_oui_database()
if _external_oui:
OUI_DATABASE = _external_oui
logger.info(f"Loaded {len(OUI_DATABASE)} entries from oui_database.json")
else:
logger.info(f"Using built-in database with {len(OUI_DATABASE)} entries")

39
data/patterns.py Normal file
View File

@@ -0,0 +1,39 @@
# Detection patterns for various device types
# Known beacon prefixes for tracker detection
AIRTAG_PREFIXES = ['4C:00'] # Apple continuity
TILE_PREFIXES = ['C4:E7', 'DC:54', 'E4:B0', 'F8:8A']
SAMSUNG_TRACKER = ['58:4D', 'A0:75']
# Drone detection patterns (SSID patterns)
DRONE_SSID_PATTERNS = [
# DJI
'DJI-', 'DJI_', 'Mavic', 'Phantom', 'Spark-', 'Mini-', 'Air-', 'Inspire',
'Matrice', 'Avata', 'FPV-', 'Osmo', 'RoboMaster', 'Tello',
# Parrot
'Parrot', 'Bebop', 'Anafi', 'Disco-', 'Mambo', 'Swing',
# Autel
'Autel', 'EVO-', 'Dragonfish', 'Lite+', 'Nano',
# Skydio
'Skydio',
# Other brands
'Holy Stone', 'Potensic', 'SYMA', 'Hubsan', 'Eachine', 'FIMI',
'Xiaomi_FIMI', 'Yuneec', 'Typhoon', 'PowerVision', 'PowerEgg',
# Generic drone patterns
'Drone', 'UAV-', 'Quadcopter', 'FPV_', 'RC-Drone'
]
# Drone OUI prefixes (MAC address prefixes for drone manufacturers)
DRONE_OUI_PREFIXES = {
# DJI
'60:60:1F': 'DJI', '48:1C:B9': 'DJI', '34:D2:62': 'DJI', 'E0:DB:55': 'DJI',
'C8:6C:87': 'DJI', 'A0:14:3D': 'DJI', '70:D7:11': 'DJI', '98:3A:56': 'DJI',
# Parrot
'90:03:B7': 'Parrot', 'A0:14:3D': 'Parrot', '00:12:1C': 'Parrot', '00:26:7E': 'Parrot',
# Autel
'8C:F5:A3': 'Autel', 'D8:E0:E1': 'Autel',
# Yuneec
'60:60:1F': 'Yuneec',
# Skydio
'F8:0F:6F': 'Skydio',
}

24
data/satellites.py Normal file
View File

@@ -0,0 +1,24 @@
# TLE data for satellite tracking (updated periodically)
TLE_SATELLITES = {
'ISS': ('ISS (ZARYA)',
'1 25544U 98067A 24001.00000000 .00000000 00000-0 00000-0 0 0000',
'2 25544 51.6400 0.0000 0000000 0.0000 0.0000 15.50000000000000'),
'NOAA-15': ('NOAA 15',
'1 25338U 98030A 24001.00000000 .00000-0 00000-0 00000-0 0 0000',
'2 25338 98.7300 0.0000 0010000 0.0000 0.0000 14.26000000000000'),
'NOAA-18': ('NOAA 18',
'1 28654U 05018A 24001.00000000 .00000-0 00000-0 00000-0 0 0000',
'2 28654 98.8800 0.0000 0014000 0.0000 0.0000 14.12000000000000'),
'NOAA-19': ('NOAA 19',
'1 33591U 09005A 24001.00000000 .00000-0 00000-0 00000-0 0 0000',
'2 33591 99.1900 0.0000 0014000 0.0000 0.0000 14.12000000000000'),
'NOAA-20': ('NOAA 20 (JPSS-1)',
'1 43013U 17073A 24001.00000000 .00000-0 00000-0 00000-0 0 0000',
'2 43013 98.7400 0.0000 0001000 0.0000 0.0000 14.19000000000000'),
'METEOR-M2': ('METEOR-M 2',
'1 40069U 14037A 24001.00000000 .00000-0 00000-0 00000-0 0 0000',
'2 40069 98.5400 0.0000 0005000 0.0000 0.0000 14.21000000000000'),
'METEOR-M2-3': ('METEOR-M2 3',
'1 57166U 23091A 24001.00000000 .00000-0 00000-0 00000-0 0 0000',
'2 57166 98.7700 0.0000 0002000 0.0000 0.0000 14.23000000000000'),
}

15430
intercept.py
View File

File diff suppressed because it is too large Load Diff

118
pyproject.toml Normal file
View File

@@ -0,0 +1,118 @@
[project]
name = "intercept"
version = "1.0.0"
description = "Signal Intelligence Platform - Pager/433MHz/ADS-B/Satellite/WiFi/Bluetooth"
readme = "README.md"
requires-python = ">=3.9"
license = {text = "MIT"}
authors = [
{name = "Intercept Contributors"}
]
keywords = ["sdr", "rtl-sdr", "signals", "pager", "adsb", "wifi", "bluetooth"]
classifiers = [
"Development Status :: 4 - Beta",
"Environment :: Web Environment",
"Framework :: Flask",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Operating System :: POSIX :: Linux",
"Operating System :: MacOS",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Topic :: Communications",
"Topic :: System :: Networking :: Monitoring",
]
dependencies = [
"flask>=2.0.0",
"skyfield>=1.45",
]
[project.optional-dependencies]
dev = [
"pytest>=7.0.0",
"pytest-cov>=4.0.0",
"ruff>=0.1.0",
"black>=23.0.0",
"mypy>=1.0.0",
"types-flask>=1.1.0",
]
[project.scripts]
intercept = "intercept:main"
[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"
[tool.setuptools.packages.find]
include = ["routes*", "utils*", "data*"]
[tool.ruff]
target-version = "py39"
line-length = 120
[tool.ruff.lint]
select = [
"E", # pycodestyle errors
"W", # pycodestyle warnings
"F", # Pyflakes
"I", # isort
"B", # flake8-bugbear
"C4", # flake8-comprehensions
"UP", # pyupgrade
"SIM", # flake8-simplify
]
ignore = [
"E501", # line too long (handled by formatter)
"B008", # do not perform function calls in argument defaults
"B905", # zip without explicit strict
"SIM108", # use ternary operator instead of if-else
]
[tool.ruff.lint.isort]
known-first-party = ["app", "config", "routes", "utils", "data"]
[tool.black]
line-length = 120
target-version = ["py39", "py310", "py311", "py312"]
include = '\.pyi?$'
exclude = '''
/(
\.git
| \.mypy_cache
| \.pytest_cache
| \.venv
| venv
| __pycache__
)/
'''
[tool.mypy]
python_version = "3.9"
warn_return_any = true
warn_unused_configs = true
ignore_missing_imports = true
exclude = [
"tests/",
"venv/",
]
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
python_functions = ["test_*"]
addopts = "-v --tb=short"
[tool.coverage.run]
source = ["app", "routes", "utils", "data"]
omit = ["tests/*", "*/__pycache__/*"]
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"if __name__ == .__main__.:",
"raise NotImplementedError",
]

14
requirements-dev.txt Normal file
View File

@@ -0,0 +1,14 @@
# Development dependencies
-r requirements.txt
# Testing
pytest>=7.0.0
pytest-cov>=4.0.0
# Code quality
ruff>=0.1.0
black>=23.0.0
mypy>=1.0.0
# Type stubs
types-flask>=1.1.0

10
requirements.txt
View File

@@ -1,2 +1,12 @@
# Core dependencies
flask>=2.0.0
# Satellite tracking (optional - only needed for satellite features)
skyfield>=1.45
# Development dependencies (install with: pip install -r requirements-dev.txt)
# pytest>=7.0.0
# pytest-cov>=4.0.0
# ruff>=0.1.0
# black>=23.0.0
# mypy>=1.0.0

19
routes/__init__.py Normal file
View File

@@ -0,0 +1,19 @@
# Routes package - registers all blueprints with the Flask app
def register_blueprints(app):
"""Register all route blueprints with the Flask app."""
from .pager import pager_bp
from .sensor import sensor_bp
from .wifi import wifi_bp
from .bluetooth import bluetooth_bp
from .adsb import adsb_bp
from .satellite import satellite_bp
from .iridium import iridium_bp
app.register_blueprint(pager_bp)
app.register_blueprint(sensor_bp)
app.register_blueprint(wifi_bp)
app.register_blueprint(bluetooth_bp)
app.register_blueprint(adsb_bp)
app.register_blueprint(satellite_bp)
app.register_blueprint(iridium_bp)

244
routes/adsb.py Normal file
View File

@@ -0,0 +1,244 @@
"""ADS-B aircraft tracking routes."""
from __future__ import annotations
import json
import queue
import shutil
import socket
import subprocess
import threading
import time
from typing import Any, Generator
from flask import Blueprint, jsonify, request, Response, render_template
import app as app_module
from utils.logging import adsb_logger as logger
adsb_bp = Blueprint('adsb', __name__, url_prefix='/adsb')
# Track if using service
adsb_using_service = False
def check_dump1090_service():
"""Check if dump1090 SBS port (30003) is available."""
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(2)
result = sock.connect_ex(('localhost', 30003))
sock.close()
if result == 0:
return 'localhost:30003'
except Exception:
pass
return None
def parse_sbs_stream(service_addr):
"""Parse SBS format data from dump1090 port 30003."""
global adsb_using_service
host, port = service_addr.split(':')
port = int(port)
logger.info(f"SBS stream parser started, connecting to {host}:{port}")
while adsb_using_service:
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(5)
sock.connect((host, port))
logger.info("Connected to SBS stream")
buffer = ""
last_update = time.time()
pending_updates = set()
while adsb_using_service:
try:
data = sock.recv(4096).decode('utf-8', errors='ignore')
if not data:
break
buffer += data
while '\n' in buffer:
line, buffer = buffer.split('\n', 1)
line = line.strip()
if not line:
continue
parts = line.split(',')
if len(parts) < 11 or parts[0] != 'MSG':
continue
msg_type = parts[1]
icao = parts[4].upper()
if not icao:
continue
aircraft = app_module.adsb_aircraft.get(icao, {'icao': icao})
if msg_type == '1' and len(parts) > 10:
callsign = parts[10].strip()
if callsign:
aircraft['callsign'] = callsign
elif msg_type == '3' and len(parts) > 15:
if parts[11]:
try:
aircraft['alt'] = int(float(parts[11]))
except (ValueError, TypeError):
pass
if parts[14] and parts[15]:
try:
aircraft['lat'] = float(parts[14])
aircraft['lon'] = float(parts[15])
except (ValueError, TypeError):
pass
elif msg_type == '4' and len(parts) > 13:
if parts[12]:
try:
aircraft['speed'] = int(float(parts[12]))
except (ValueError, TypeError):
pass
if parts[13]:
try:
aircraft['heading'] = int(float(parts[13]))
except (ValueError, TypeError):
pass
elif msg_type == '5' and len(parts) > 11:
if parts[10]:
callsign = parts[10].strip()
if callsign:
aircraft['callsign'] = callsign
if parts[11]:
try:
aircraft['alt'] = int(float(parts[11]))
except (ValueError, TypeError):
pass
elif msg_type == '6' and len(parts) > 17:
if parts[17]:
aircraft['squawk'] = parts[17]
app_module.adsb_aircraft[icao] = aircraft
pending_updates.add(icao)
now = time.time()
if now - last_update >= 1.0:
for update_icao in pending_updates:
if update_icao in app_module.adsb_aircraft:
app_module.adsb_queue.put({
'type': 'aircraft',
**app_module.adsb_aircraft[update_icao]
})
pending_updates.clear()
last_update = now
except socket.timeout:
continue
sock.close()
except Exception as e:
logger.warning(f"SBS connection error: {e}, reconnecting...")
time.sleep(2)
logger.info("SBS stream parser stopped")
@adsb_bp.route('/tools')
def check_adsb_tools():
"""Check for ADS-B decoding tools."""
return jsonify({
'dump1090': shutil.which('dump1090') is not None or shutil.which('dump1090-mutability') is not None,
'rtl_adsb': shutil.which('rtl_adsb') is not None
})
@adsb_bp.route('/start', methods=['POST'])
def start_adsb():
"""Start ADS-B tracking."""
global adsb_using_service
with app_module.adsb_lock:
if app_module.adsb_process and app_module.adsb_process.poll() is None:
return jsonify({'status': 'already_running', 'message': 'ADS-B already running'})
if adsb_using_service:
return jsonify({'status': 'already_running', 'message': 'ADS-B already running (using service)'})
data = request.json or {}
gain = data.get('gain', '40')
device = data.get('device', '0')
dump1090_path = shutil.which('dump1090') or shutil.which('dump1090-mutability')
if not dump1090_path:
return jsonify({'status': 'error', 'message': 'dump1090 not found.'})
cmd = [dump1090_path, '--net', '--gain', gain, '--device-index', str(device), '--quiet']
try:
app_module.adsb_process = subprocess.Popen(
cmd,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
time.sleep(3)
if app_module.adsb_process.poll() is not None:
return jsonify({'status': 'error', 'message': 'dump1090 failed to start.'})
adsb_using_service = True
thread = threading.Thread(target=parse_sbs_stream, args=('localhost:30003',), daemon=True)
thread.start()
return jsonify({'status': 'success', 'message': 'ADS-B tracking started'})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})
@adsb_bp.route('/stop', methods=['POST'])
def stop_adsb():
"""Stop ADS-B tracking."""
global adsb_using_service
with app_module.adsb_lock:
if app_module.adsb_process:
app_module.adsb_process.terminate()
try:
app_module.adsb_process.wait(timeout=5)
except subprocess.TimeoutExpired:
app_module.adsb_process.kill()
app_module.adsb_process = None
adsb_using_service = False
app_module.adsb_aircraft = {}
return jsonify({'status': 'stopped'})
@adsb_bp.route('/stream')
def stream_adsb():
"""SSE stream for ADS-B aircraft."""
def generate():
while True:
try:
msg = app_module.adsb_queue.get(timeout=1)
yield f"data: {json.dumps(msg)}\n\n"
except queue.Empty:
yield f"data: {json.dumps({'type': 'keepalive'})}\n\n"
response = Response(generate(), mimetype='text/event-stream')
response.headers['Cache-Control'] = 'no-cache'
response.headers['X-Accel-Buffering'] = 'no'
return response
@adsb_bp.route('/dashboard')
def adsb_dashboard():
"""Popout ADS-B dashboard."""
return render_template('adsb_dashboard.html')

483
routes/bluetooth.py Normal file
View File

@@ -0,0 +1,483 @@
"""Bluetooth reconnaissance routes."""
from __future__ import annotations
import fcntl
import json
import os
import platform
import pty
import queue
import re
import select
import subprocess
import threading
import time
from typing import Any, Generator
from flask import Blueprint, jsonify, request, Response
import app as app_module
from utils.dependencies import check_tool
from data.oui import OUI_DATABASE, load_oui_database, get_manufacturer
from data.patterns import AIRTAG_PREFIXES, TILE_PREFIXES, SAMSUNG_TRACKER
bluetooth_bp = Blueprint('bluetooth', __name__, url_prefix='/bt')
def classify_bt_device(name, device_class, services, manufacturer=None):
"""Classify Bluetooth device type based on available info."""
name_lower = (name or '').lower()
mfr_lower = (manufacturer or '').lower()
audio_patterns = [
'airpod', 'earbud', 'headphone', 'headset', 'speaker', 'audio', 'beats', 'bose',
'jbl', 'sony wh', 'sony wf', 'sennheiser', 'jabra', 'soundcore', 'anker', 'buds',
'earphone', 'pod', 'soundbar', 'skullcandy', 'marshall', 'b&o', 'bang', 'olufsen'
]
if any(x in name_lower for x in audio_patterns):
return 'audio'
wearable_patterns = [
'watch', 'band', 'fitbit', 'garmin', 'mi band', 'miband', 'amazfit',
'galaxy watch', 'gear', 'versa', 'sense', 'charge', 'inspire'
]
if any(x in name_lower for x in wearable_patterns):
return 'wearable'
phone_patterns = [
'iphone', 'galaxy', 'pixel', 'phone', 'android', 'oneplus', 'huawei', 'xiaomi'
]
if any(x in name_lower for x in phone_patterns):
return 'phone'
tracker_patterns = ['airtag', 'tile', 'smarttag', 'chipolo', 'find my']
if any(x in name_lower for x in tracker_patterns):
return 'tracker'
input_patterns = ['keyboard', 'mouse', 'controller', 'gamepad', 'remote']
if any(x in name_lower for x in input_patterns):
return 'input'
if mfr_lower in ['bose', 'jbl', 'sony', 'sennheiser', 'jabra', 'beats']:
return 'audio'
if mfr_lower in ['fitbit', 'garmin']:
return 'wearable'
if mfr_lower == 'tile':
return 'tracker'
if device_class:
major_class = (device_class >> 8) & 0x1F
if major_class == 1:
return 'computer'
elif major_class == 2:
return 'phone'
elif major_class == 4:
return 'audio'
elif major_class == 5:
return 'input'
elif major_class == 7:
return 'wearable'
return 'other'
def detect_tracker(mac, name, manufacturer_data=None):
"""Detect if device is a known tracker."""
mac_prefix = mac[:5].upper()
if any(mac_prefix.startswith(p) for p in AIRTAG_PREFIXES):
if manufacturer_data and b'\\x4c\\x00' in manufacturer_data:
return {'type': 'airtag', 'name': 'Apple AirTag', 'risk': 'high'}
if any(mac_prefix.startswith(p) for p in TILE_PREFIXES):
return {'type': 'tile', 'name': 'Tile Tracker', 'risk': 'medium'}
if any(mac_prefix.startswith(p) for p in SAMSUNG_TRACKER):
return {'type': 'smarttag', 'name': 'Samsung SmartTag', 'risk': 'medium'}
name_lower = (name or '').lower()
if 'airtag' in name_lower:
return {'type': 'airtag', 'name': 'Apple AirTag', 'risk': 'high'}
if 'tile' in name_lower:
return {'type': 'tile', 'name': 'Tile Tracker', 'risk': 'medium'}
return None
def detect_bt_interfaces():
"""Detect available Bluetooth interfaces."""
interfaces = []
if platform.system() == 'Linux':
try:
result = subprocess.run(['hciconfig'], capture_output=True, text=True, timeout=5)
blocks = re.split(r'(?=^hci\d+:)', result.stdout, flags=re.MULTILINE)
for block in blocks:
if block.strip():
first_line = block.split('\n')[0]
match = re.match(r'(hci\d+):', first_line)
if match:
iface_name = match.group(1)
is_up = 'UP RUNNING' in block or '\tUP ' in block
interfaces.append({
'name': iface_name,
'type': 'hci',
'status': 'up' if is_up else 'down'
})
except Exception:
pass
elif platform.system() == 'Darwin':
interfaces.append({
'name': 'default',
'type': 'macos',
'status': 'available'
})
return interfaces
def stream_bt_scan(process, scan_mode):
"""Stream Bluetooth scan output to queue."""
try:
app_module.bt_queue.put({'type': 'status', 'text': 'started'})
if scan_mode == 'hcitool':
for line in iter(process.stdout.readline, b''):
line = line.decode('utf-8', errors='replace').strip()
if not line or 'LE Scan' in line:
continue
parts = line.split()
if len(parts) >= 1 and ':' in parts[0]:
mac = parts[0]
name = ' '.join(parts[1:]) if len(parts) > 1 else ''
manufacturer = get_manufacturer(mac)
device = {
'mac': mac,
'name': name or '[Unknown]',
'manufacturer': manufacturer,
'type': classify_bt_device(name, None, None, manufacturer),
'rssi': None,
'last_seen': time.time()
}
tracker = detect_tracker(mac, name)
if tracker:
device['tracker'] = tracker
is_new = mac not in app_module.bt_devices
app_module.bt_devices[mac] = device
app_module.bt_queue.put({
**device,
'type': 'device',
'device_type': device.get('type', 'other'),
'action': 'new' if is_new else 'update',
})
elif scan_mode == 'bluetoothctl':
master_fd = getattr(process, '_master_fd', None)
if not master_fd:
app_module.bt_queue.put({'type': 'error', 'text': 'bluetoothctl pty not available'})
return
buffer = ''
while process.poll() is None:
readable, _, _ = select.select([master_fd], [], [], 1.0)
if readable:
try:
data = os.read(master_fd, 4096)
if not data:
break
buffer += data.decode('utf-8', errors='replace')
while '\n' in buffer:
line, buffer = buffer.split('\n', 1)
line = line.strip()
line = re.sub(r'\x1b\[[0-9;]*m', '', line)
line = re.sub(r'\r', '', line)
if 'Device' in line:
match = re.search(r'([0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2}:[0-9A-Fa-f]{2})\s*(.*)', line)
if match:
mac = match.group(1).upper()
name = match.group(2).strip()
manufacturer = get_manufacturer(mac)
device = {
'mac': mac,
'name': name or '[Unknown]',
'manufacturer': manufacturer,
'type': classify_bt_device(name, None, None, manufacturer),
'rssi': None,
'last_seen': time.time()
}
tracker = detect_tracker(mac, name)
if tracker:
device['tracker'] = tracker
is_new = mac not in app_module.bt_devices
app_module.bt_devices[mac] = device
app_module.bt_queue.put({
**device,
'type': 'device',
'device_type': device.get('type', 'other'),
'action': 'new' if is_new else 'update',
})
except OSError:
break
try:
os.close(master_fd)
except OSError:
pass
except Exception as e:
app_module.bt_queue.put({'type': 'error', 'text': str(e)})
finally:
process.wait()
app_module.bt_queue.put({'type': 'status', 'text': 'stopped'})
with app_module.bt_lock:
app_module.bt_process = None
@bluetooth_bp.route('/reload-oui', methods=['POST'])
def reload_oui_database_route():
"""Reload OUI database from external file."""
new_db = load_oui_database()
if new_db:
OUI_DATABASE.clear()
OUI_DATABASE.update(new_db)
return jsonify({'status': 'success', 'entries': len(OUI_DATABASE)})
return jsonify({'status': 'error', 'message': 'Could not load oui_database.json'})
@bluetooth_bp.route('/interfaces')
def get_bt_interfaces():
"""Get available Bluetooth interfaces and tools."""
interfaces = detect_bt_interfaces()
tools = {
'hcitool': check_tool('hcitool'),
'bluetoothctl': check_tool('bluetoothctl'),
'hciconfig': check_tool('hciconfig'),
'l2ping': check_tool('l2ping'),
'sdptool': check_tool('sdptool')
}
return jsonify({
'interfaces': interfaces,
'tools': tools,
'current_interface': app_module.bt_interface
})
@bluetooth_bp.route('/scan/start', methods=['POST'])
def start_bt_scan():
"""Start Bluetooth scanning."""
with app_module.bt_lock:
if app_module.bt_process:
if app_module.bt_process.poll() is None:
return jsonify({'status': 'error', 'message': 'Scan already running'})
else:
app_module.bt_process = None
data = request.json
scan_mode = data.get('mode', 'hcitool')
interface = data.get('interface', 'hci0')
scan_ble = data.get('scan_ble', True)
app_module.bt_interface = interface
app_module.bt_devices = {}
while not app_module.bt_queue.empty():
try:
app_module.bt_queue.get_nowait()
except queue.Empty:
break
try:
if scan_mode == 'hcitool':
if scan_ble:
cmd = ['hcitool', '-i', interface, 'lescan', '--duplicates']
else:
cmd = ['hcitool', '-i', interface, 'scan']
app_module.bt_process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
elif scan_mode == 'bluetoothctl':
master_fd, slave_fd = pty.openpty()
app_module.bt_process = subprocess.Popen(
['bluetoothctl'],
stdin=slave_fd,
stdout=slave_fd,
stderr=slave_fd,
close_fds=True
)
os.close(slave_fd)
app_module.bt_process._master_fd = master_fd
time.sleep(0.5)
os.write(master_fd, b'power on\n')
time.sleep(0.3)
os.write(master_fd, b'scan on\n')
else:
return jsonify({'status': 'error', 'message': f'Unknown scan mode: {scan_mode}'})
time.sleep(0.5)
if app_module.bt_process.poll() is not None:
stderr_output = app_module.bt_process.stderr.read().decode('utf-8', errors='replace').strip()
app_module.bt_process = None
return jsonify({'status': 'error', 'message': stderr_output or 'Process failed to start'})
thread = threading.Thread(target=stream_bt_scan, args=(app_module.bt_process, scan_mode))
thread.daemon = True
thread.start()
app_module.bt_queue.put({'type': 'info', 'text': f'Started {scan_mode} scan on {interface}'})
return jsonify({'status': 'started', 'mode': scan_mode, 'interface': interface})
except FileNotFoundError as e:
return jsonify({'status': 'error', 'message': f'Tool not found: {e.filename}'})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})
@bluetooth_bp.route('/scan/stop', methods=['POST'])
def stop_bt_scan():
"""Stop Bluetooth scanning."""
with app_module.bt_lock:
if app_module.bt_process:
app_module.bt_process.terminate()
try:
app_module.bt_process.wait(timeout=3)
except subprocess.TimeoutExpired:
app_module.bt_process.kill()
app_module.bt_process = None
return jsonify({'status': 'stopped'})
return jsonify({'status': 'not_running'})
@bluetooth_bp.route('/reset', methods=['POST'])
def reset_bt_adapter():
"""Reset Bluetooth adapter."""
data = request.json
interface = data.get('interface', 'hci0')
with app_module.bt_lock:
if app_module.bt_process:
try:
app_module.bt_process.terminate()
app_module.bt_process.wait(timeout=2)
except (subprocess.TimeoutExpired, OSError):
try:
app_module.bt_process.kill()
except OSError:
pass
app_module.bt_process = None
try:
subprocess.run(['pkill', '-f', 'hcitool'], capture_output=True, timeout=2)
subprocess.run(['pkill', '-f', 'bluetoothctl'], capture_output=True, timeout=2)
time.sleep(0.5)
subprocess.run(['rfkill', 'unblock', 'bluetooth'], capture_output=True, timeout=5)
subprocess.run(['hciconfig', interface, 'down'], capture_output=True, timeout=5)
time.sleep(1)
subprocess.run(['hciconfig', interface, 'up'], capture_output=True, timeout=5)
time.sleep(0.5)
result = subprocess.run(['hciconfig', interface], capture_output=True, text=True, timeout=5)
is_up = 'UP RUNNING' in result.stdout
return jsonify({
'status': 'success' if is_up else 'warning',
'message': f'Adapter {interface} reset' if is_up else f'Reset attempted but adapter may still be down',
'is_up': is_up
})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})
@bluetooth_bp.route('/enum', methods=['POST'])
def enum_bt_services():
"""Enumerate services on a Bluetooth device."""
data = request.json
target_mac = data.get('mac')
if not target_mac:
return jsonify({'status': 'error', 'message': 'Target MAC required'})
try:
result = subprocess.run(
['sdptool', 'browse', target_mac],
capture_output=True, text=True, timeout=30
)
services = []
current_service = {}
for line in result.stdout.split('\n'):
line = line.strip()
if line.startswith('Service Name:'):
if current_service:
services.append(current_service)
current_service = {'name': line.split(':', 1)[1].strip()}
elif line.startswith('Service Description:'):
current_service['description'] = line.split(':', 1)[1].strip()
if current_service:
services.append(current_service)
app_module.bt_services[target_mac] = services
return jsonify({
'status': 'success',
'mac': target_mac,
'services': services
})
except subprocess.TimeoutExpired:
return jsonify({'status': 'error', 'message': 'Connection timed out'})
except FileNotFoundError:
return jsonify({'status': 'error', 'message': 'sdptool not found'})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})
@bluetooth_bp.route('/devices')
def get_bt_devices():
"""Get current list of discovered Bluetooth devices."""
return jsonify({
'devices': list(app_module.bt_devices.values()),
'beacons': list(app_module.bt_beacons.values()),
'interface': app_module.bt_interface
})
@bluetooth_bp.route('/stream')
def stream_bt():
"""SSE stream for Bluetooth events."""
def generate():
while True:
try:
msg = app_module.bt_queue.get(timeout=1)
yield f"data: {json.dumps(msg)}\n\n"
except queue.Empty:
yield f"data: {json.dumps({'type': 'keepalive'})}\n\n"
response = Response(generate(), mimetype='text/event-stream')
response.headers['Cache-Control'] = 'no-cache'
response.headers['X-Accel-Buffering'] = 'no'
response.headers['Connection'] = 'keep-alive'
return response

126
routes/iridium.py Normal file
View File

@@ -0,0 +1,126 @@
"""Iridium monitoring routes."""
from __future__ import annotations
import json
import queue
import random
import shutil
import subprocess
import threading
import time
from datetime import datetime
from typing import Any, Generator
from flask import Blueprint, jsonify, request, Response
import app as app_module
from utils.logging import iridium_logger as logger
iridium_bp = Blueprint('iridium', __name__, url_prefix='/iridium')
def monitor_iridium(process):
"""Monitor Iridium capture and detect bursts."""
try:
burst_count = 0
while process.poll() is None:
data = process.stdout.read(1024)
if data:
if len(data) > 0 and burst_count < 100:
if random.random() < 0.01:
burst = {
'type': 'burst',
'time': datetime.now().strftime('%H:%M:%S.%f')[:-3],
'frequency': f"{1616 + random.random() * 10:.3f}",
'data': f"Frame data (simulated) - Burst #{burst_count + 1}"
}
app_module.satellite_queue.put(burst)
app_module.iridium_bursts.append(burst)
burst_count += 1
time.sleep(0.1)
except Exception as e:
logger.error(f"Monitor error: {e}")
@iridium_bp.route('/tools')
def check_iridium_tools():
"""Check for Iridium decoding tools."""
has_tool = shutil.which('iridium-extractor') is not None or shutil.which('iridium-parser') is not None
return jsonify({'available': has_tool})
@iridium_bp.route('/start', methods=['POST'])
def start_iridium():
"""Start Iridium burst capture."""
with app_module.satellite_lock:
if app_module.satellite_process and app_module.satellite_process.poll() is None:
return jsonify({'status': 'error', 'message': 'Iridium capture already running'})
data = request.json
freq = data.get('freq', '1626.0')
gain = data.get('gain', '40')
sample_rate = data.get('sampleRate', '2.048e6')
device = data.get('device', '0')
if not shutil.which('iridium-extractor') and not shutil.which('rtl_fm'):
return jsonify({
'status': 'error',
'message': 'Iridium tools not found.'
})
try:
cmd = [
'rtl_fm',
'-f', f'{float(freq)}M',
'-g', str(gain),
'-s', sample_rate,
'-d', str(device),
'-'
]
app_module.satellite_process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
thread = threading.Thread(target=monitor_iridium, args=(app_module.satellite_process,), daemon=True)
thread.start()
return jsonify({'status': 'started'})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})
@iridium_bp.route('/stop', methods=['POST'])
def stop_iridium():
"""Stop Iridium capture."""
with app_module.satellite_lock:
if app_module.satellite_process:
app_module.satellite_process.terminate()
try:
app_module.satellite_process.wait(timeout=5)
except subprocess.TimeoutExpired:
app_module.satellite_process.kill()
app_module.satellite_process = None
return jsonify({'status': 'stopped'})
@iridium_bp.route('/stream')
def stream_iridium():
"""SSE stream for Iridium bursts."""
def generate():
while True:
try:
msg = app_module.satellite_queue.get(timeout=1)
yield f"data: {json.dumps(msg)}\n\n"
except queue.Empty:
yield f"data: {json.dumps({'type': 'keepalive'})}\n\n"
response = Response(generate(), mimetype='text/event-stream')
response.headers['Cache-Control'] = 'no-cache'
response.headers['X-Accel-Buffering'] = 'no'
return response

328
routes/pager.py Normal file
View File

@@ -0,0 +1,328 @@
"""Pager decoding routes (POCSAG/FLEX)."""
from __future__ import annotations
import os
import re
import pty
import queue
import select
import subprocess
import threading
from datetime import datetime
from typing import Any, Generator
from flask import Blueprint, jsonify, request, Response
import app as app_module
from utils.logging import pager_logger as logger
pager_bp = Blueprint('pager', __name__)
def parse_multimon_output(line: str) -> dict[str, str] | None:
"""Parse multimon-ng output line."""
line = line.strip()
# POCSAG parsing - with message content
pocsag_match = re.match(
r'(POCSAG\d+):\s*Address:\s*(\d+)\s+Function:\s*(\d+)\s+(Alpha|Numeric):\s*(.*)',
line
)
if pocsag_match:
return {
'protocol': pocsag_match.group(1),
'address': pocsag_match.group(2),
'function': pocsag_match.group(3),
'msg_type': pocsag_match.group(4),
'message': pocsag_match.group(5).strip() or '[No Message]'
}
# POCSAG parsing - address only (no message content)
pocsag_addr_match = re.match(
r'(POCSAG\d+):\s*Address:\s*(\d+)\s+Function:\s*(\d+)\s*$',
line
)
if pocsag_addr_match:
return {
'protocol': pocsag_addr_match.group(1),
'address': pocsag_addr_match.group(2),
'function': pocsag_addr_match.group(3),
'msg_type': 'Tone',
'message': '[Tone Only]'
}
# FLEX parsing (standard format)
flex_match = re.match(
r'FLEX[:\|]\s*[\d\-]+[\s\|]+[\d:]+[\s\|]+([\d/A-Z]+)[\s\|]+([\d.]+)[\s\|]+\[?(\d+)\]?[\s\|]+(\w+)[\s\|]+(.*)',
line
)
if flex_match:
return {
'protocol': 'FLEX',
'address': flex_match.group(3),
'function': flex_match.group(1),
'msg_type': flex_match.group(4),
'message': flex_match.group(5).strip() or '[No Message]'
}
# Simple FLEX format
flex_simple = re.match(r'FLEX:\s*(.+)', line)
if flex_simple:
return {
'protocol': 'FLEX',
'address': 'Unknown',
'function': '',
'msg_type': 'Unknown',
'message': flex_simple.group(1).strip()
}
return None
def log_message(msg: dict[str, Any]) -> None:
"""Log a message to file if logging is enabled."""
if not app_module.logging_enabled:
return
try:
with open(app_module.log_file_path, 'a') as f:
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
f.write(f"{timestamp} | {msg.get('protocol', 'UNKNOWN')} | {msg.get('address', '')} | {msg.get('message', '')}\n")
except Exception as e:
logger.error(f"Failed to log message: {e}")
def stream_decoder(master_fd: int, process: subprocess.Popen[bytes]) -> None:
"""Stream decoder output to queue using PTY for unbuffered output."""
try:
app_module.output_queue.put({'type': 'status', 'text': 'started'})
buffer = ""
while True:
try:
ready, _, _ = select.select([master_fd], [], [], 1.0)
except Exception:
break
if ready:
try:
data = os.read(master_fd, 1024)
if not data:
break
buffer += data.decode('utf-8', errors='replace')
while '\n' in buffer:
line, buffer = buffer.split('\n', 1)
line = line.strip()
if not line:
continue
parsed = parse_multimon_output(line)
if parsed:
parsed['timestamp'] = datetime.now().strftime('%H:%M:%S')
app_module.output_queue.put({'type': 'message', **parsed})
log_message(parsed)
else:
app_module.output_queue.put({'type': 'raw', 'text': line})
except OSError:
break
if process.poll() is not None:
break
except Exception as e:
app_module.output_queue.put({'type': 'error', 'text': str(e)})
finally:
try:
os.close(master_fd)
except OSError:
pass
process.wait()
app_module.output_queue.put({'type': 'status', 'text': 'stopped'})
with app_module.process_lock:
app_module.current_process = None
@pager_bp.route('/start', methods=['POST'])
def start_decoding() -> Response:
with app_module.process_lock:
if app_module.current_process:
return jsonify({'status': 'error', 'message': 'Already running'})
data = request.json
freq = data.get('frequency', '929.6125')
gain = data.get('gain', '0')
squelch = data.get('squelch', '0')
ppm = data.get('ppm', '0')
device = data.get('device', '0')
protocols = data.get('protocols', ['POCSAG512', 'POCSAG1200', 'POCSAG2400', 'FLEX'])
# Clear queue
while not app_module.output_queue.empty():
try:
app_module.output_queue.get_nowait()
except queue.Empty:
break
# Build multimon-ng decoder arguments
decoders = []
for proto in protocols:
if proto == 'POCSAG512':
decoders.extend(['-a', 'POCSAG512'])
elif proto == 'POCSAG1200':
decoders.extend(['-a', 'POCSAG1200'])
elif proto == 'POCSAG2400':
decoders.extend(['-a', 'POCSAG2400'])
elif proto == 'FLEX':
decoders.extend(['-a', 'FLEX'])
# Build rtl_fm command
rtl_cmd = [
'rtl_fm',
'-d', str(device),
'-f', f'{freq}M',
'-M', 'fm',
'-s', '22050',
]
if gain and gain != '0':
rtl_cmd.extend(['-g', str(gain)])
if ppm and ppm != '0':
rtl_cmd.extend(['-p', str(ppm)])
if squelch and squelch != '0':
rtl_cmd.extend(['-l', str(squelch)])
rtl_cmd.append('-')
multimon_cmd = ['multimon-ng', '-t', 'raw'] + decoders + ['-f', 'alpha', '-']
full_cmd = ' '.join(rtl_cmd) + ' | ' + ' '.join(multimon_cmd)
logger.info(f"Running: {full_cmd}")
try:
# Create pipe: rtl_fm | multimon-ng
rtl_process = subprocess.Popen(
rtl_cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
# Start a thread to monitor rtl_fm stderr for errors
def monitor_rtl_stderr():
for line in rtl_process.stderr:
err_text = line.decode('utf-8', errors='replace').strip()
if err_text:
logger.debug(f"[RTL_FM] {err_text}")
app_module.output_queue.put({'type': 'raw', 'text': f'[rtl_fm] {err_text}'})
rtl_stderr_thread = threading.Thread(target=monitor_rtl_stderr)
rtl_stderr_thread.daemon = True
rtl_stderr_thread.start()
# Create a pseudo-terminal for multimon-ng output
master_fd, slave_fd = pty.openpty()
multimon_process = subprocess.Popen(
multimon_cmd,
stdin=rtl_process.stdout,
stdout=slave_fd,
stderr=slave_fd,
close_fds=True
)
os.close(slave_fd)
rtl_process.stdout.close()
app_module.current_process = multimon_process
app_module.current_process._rtl_process = rtl_process
app_module.current_process._master_fd = master_fd
# Start output thread with PTY master fd
thread = threading.Thread(target=stream_decoder, args=(master_fd, multimon_process))
thread.daemon = True
thread.start()
app_module.output_queue.put({'type': 'info', 'text': f'Command: {full_cmd}'})
return jsonify({'status': 'started', 'command': full_cmd})
except FileNotFoundError as e:
return jsonify({'status': 'error', 'message': f'Tool not found: {e.filename}'})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})
@pager_bp.route('/stop', methods=['POST'])
def stop_decoding() -> Response:
with app_module.process_lock:
if app_module.current_process:
# Kill rtl_fm process first
if hasattr(app_module.current_process, '_rtl_process'):
try:
app_module.current_process._rtl_process.terminate()
app_module.current_process._rtl_process.wait(timeout=2)
except (subprocess.TimeoutExpired, OSError):
try:
app_module.current_process._rtl_process.kill()
except OSError:
pass
# Close PTY master fd
if hasattr(app_module.current_process, '_master_fd'):
try:
os.close(app_module.current_process._master_fd)
except OSError:
pass
# Kill multimon-ng
app_module.current_process.terminate()
try:
app_module.current_process.wait(timeout=2)
except subprocess.TimeoutExpired:
app_module.current_process.kill()
app_module.current_process = None
return jsonify({'status': 'stopped'})
return jsonify({'status': 'not_running'})
@pager_bp.route('/status')
def get_status() -> Response:
"""Check if decoder is currently running."""
with app_module.process_lock:
if app_module.current_process and app_module.current_process.poll() is None:
return jsonify({'running': True, 'logging': app_module.logging_enabled, 'log_file': app_module.log_file_path})
return jsonify({'running': False, 'logging': app_module.logging_enabled, 'log_file': app_module.log_file_path})
@pager_bp.route('/logging', methods=['POST'])
def toggle_logging() -> Response:
"""Toggle message logging."""
data = request.json
if 'enabled' in data:
app_module.logging_enabled = data['enabled']
if 'log_file' in data and data['log_file']:
app_module.log_file_path = data['log_file']
return jsonify({'logging': app_module.logging_enabled, 'log_file': app_module.log_file_path})
@pager_bp.route('/stream')
def stream() -> Response:
import json
def generate() -> Generator[str, None, None]:
while True:
try:
msg = app_module.output_queue.get(timeout=1)
yield f"data: {json.dumps(msg)}\n\n"
except queue.Empty:
yield f"data: {json.dumps({'type': 'keepalive'})}\n\n"
response = Response(generate(), mimetype='text/event-stream')
response.headers['Cache-Control'] = 'no-cache'
response.headers['X-Accel-Buffering'] = 'no'
response.headers['Connection'] = 'keep-alive'
return response

381
routes/satellite.py Normal file
View File

@@ -0,0 +1,381 @@
"""Satellite tracking routes."""
from __future__ import annotations
import json
import urllib.request
from datetime import datetime, timedelta
from typing import Any
from flask import Blueprint, jsonify, request, render_template, Response
from data.satellites import TLE_SATELLITES
from utils.logging import satellite_logger as logger
satellite_bp = Blueprint('satellite', __name__, url_prefix='/satellite')
# Local TLE cache (can be updated via API)
_tle_cache = dict(TLE_SATELLITES)
@satellite_bp.route('/dashboard')
def satellite_dashboard():
"""Popout satellite tracking dashboard."""
return render_template('satellite_dashboard.html')
@satellite_bp.route('/predict', methods=['POST'])
def predict_passes():
"""Calculate satellite passes using skyfield."""
try:
from skyfield.api import load, wgs84, EarthSatellite
from skyfield.almanac import find_discrete
except ImportError:
return jsonify({
'status': 'error',
'message': 'skyfield library not installed. Run: pip install skyfield'
})
data = request.json
lat = data.get('latitude', data.get('lat', 51.5074))
lon = data.get('longitude', data.get('lon', -0.1278))
hours = data.get('hours', 24)
min_el = data.get('minEl', 10)
norad_to_name = {
25544: 'ISS',
25338: 'NOAA-15',
28654: 'NOAA-18',
33591: 'NOAA-19',
43013: 'NOAA-20',
40069: 'METEOR-M2',
57166: 'METEOR-M2-3'
}
sat_input = data.get('satellites', ['ISS', 'NOAA-15', 'NOAA-18', 'NOAA-19'])
satellites = []
for sat in sat_input:
if isinstance(sat, int) and sat in norad_to_name:
satellites.append(norad_to_name[sat])
else:
satellites.append(sat)
passes = []
colors = {
'ISS': '#00ffff',
'NOAA-15': '#00ff00',
'NOAA-18': '#ff6600',
'NOAA-19': '#ff3366',
'NOAA-20': '#00ffaa',
'METEOR-M2': '#9370DB',
'METEOR-M2-3': '#ff00ff'
}
name_to_norad = {v: k for k, v in norad_to_name.items()}
ts = load.timescale()
observer = wgs84.latlon(lat, lon)
t0 = ts.now()
t1 = ts.utc(t0.utc_datetime() + timedelta(hours=hours))
for sat_name in satellites:
if sat_name not in _tle_cache:
continue
tle_data = _tle_cache[sat_name]
try:
satellite = EarthSatellite(tle_data[1], tle_data[2], tle_data[0], ts)
except Exception:
continue
def above_horizon(t):
diff = satellite - observer
topocentric = diff.at(t)
alt, _, _ = topocentric.altaz()
return alt.degrees > 0
above_horizon.step_days = 1/720
try:
times, events = find_discrete(t0, t1, above_horizon)
except Exception:
continue
i = 0
while i < len(times):
if i < len(events) and events[i]:
rise_time = times[i]
set_time = None
for j in range(i + 1, len(times)):
if not events[j]:
set_time = times[j]
i = j
break
if set_time is None:
i += 1
continue
trajectory = []
max_elevation = 0
num_points = 30
duration_seconds = (set_time.utc_datetime() - rise_time.utc_datetime()).total_seconds()
for k in range(num_points):
frac = k / (num_points - 1)
t_point = ts.utc(rise_time.utc_datetime() + timedelta(seconds=duration_seconds * frac))
diff = satellite - observer
topocentric = diff.at(t_point)
alt, az, _ = topocentric.altaz()
el = alt.degrees
azimuth = az.degrees
if el > max_elevation:
max_elevation = el
trajectory.append({'el': float(max(0, el)), 'az': float(azimuth)})
if max_elevation >= min_el:
duration_minutes = int(duration_seconds / 60)
ground_track = []
for k in range(60):
frac = k / 59
t_point = ts.utc(rise_time.utc_datetime() + timedelta(seconds=duration_seconds * frac))
geocentric = satellite.at(t_point)
subpoint = wgs84.subpoint(geocentric)
ground_track.append({
'lat': float(subpoint.latitude.degrees),
'lon': float(subpoint.longitude.degrees)
})
current_geo = satellite.at(ts.now())
current_subpoint = wgs84.subpoint(current_geo)
passes.append({
'satellite': sat_name,
'norad': name_to_norad.get(sat_name, 0),
'startTime': rise_time.utc_datetime().strftime('%Y-%m-%d %H:%M UTC'),
'startTimeISO': rise_time.utc_datetime().isoformat(),
'maxEl': float(round(max_elevation, 1)),
'duration': int(duration_minutes),
'trajectory': trajectory,
'groundTrack': ground_track,
'currentPos': {
'lat': float(current_subpoint.latitude.degrees),
'lon': float(current_subpoint.longitude.degrees)
},
'color': colors.get(sat_name, '#00ff00')
})
i += 1
passes.sort(key=lambda p: p['startTime'])
return jsonify({
'status': 'success',
'passes': passes
})
@satellite_bp.route('/position', methods=['POST'])
def get_satellite_position():
"""Get real-time positions of satellites."""
try:
from skyfield.api import load, wgs84, EarthSatellite
except ImportError:
return jsonify({'status': 'error', 'message': 'skyfield not installed'})
data = request.json
lat = data.get('latitude', data.get('lat', 51.5074))
lon = data.get('longitude', data.get('lon', -0.1278))
sat_input = data.get('satellites', [])
include_track = data.get('includeTrack', True)
norad_to_name = {
25544: 'ISS',
25338: 'NOAA-15',
28654: 'NOAA-18',
33591: 'NOAA-19',
43013: 'NOAA-20',
40069: 'METEOR-M2',
57166: 'METEOR-M2-3'
}
satellites = []
for sat in sat_input:
if isinstance(sat, int) and sat in norad_to_name:
satellites.append(norad_to_name[sat])
else:
satellites.append(sat)
ts = load.timescale()
observer = wgs84.latlon(lat, lon)
now = ts.now()
now_dt = now.utc_datetime()
positions = []
for sat_name in satellites:
if sat_name not in _tle_cache:
continue
tle_data = _tle_cache[sat_name]
try:
satellite = EarthSatellite(tle_data[1], tle_data[2], tle_data[0], ts)
geocentric = satellite.at(now)
subpoint = wgs84.subpoint(geocentric)
diff = satellite - observer
topocentric = diff.at(now)
alt, az, distance = topocentric.altaz()
pos_data = {
'satellite': sat_name,
'lat': float(subpoint.latitude.degrees),
'lon': float(subpoint.longitude.degrees),
'altitude': float(geocentric.distance().km - 6371),
'elevation': float(alt.degrees),
'azimuth': float(az.degrees),
'distance': float(distance.km),
'visible': bool(alt.degrees > 0)
}
if include_track:
orbit_track = []
for minutes_offset in range(-45, 46, 1):
t_point = ts.utc(now_dt + timedelta(minutes=minutes_offset))
try:
geo = satellite.at(t_point)
sp = wgs84.subpoint(geo)
orbit_track.append({
'lat': float(sp.latitude.degrees),
'lon': float(sp.longitude.degrees),
'past': minutes_offset < 0
})
except Exception:
continue
pos_data['track'] = orbit_track
positions.append(pos_data)
except Exception:
continue
return jsonify({
'status': 'success',
'positions': positions,
'timestamp': datetime.utcnow().isoformat()
})
@satellite_bp.route('/update-tle', methods=['POST'])
def update_tle():
"""Update TLE data from CelesTrak."""
global _tle_cache
try:
name_mappings = {
'ISS (ZARYA)': 'ISS',
'NOAA 15': 'NOAA-15',
'NOAA 18': 'NOAA-18',
'NOAA 19': 'NOAA-19',
'METEOR-M 2': 'METEOR-M2',
'METEOR-M2 3': 'METEOR-M2-3'
}
updated = []
for group in ['stations', 'weather']:
url = f'https://celestrak.org/NORAD/elements/gp.php?GROUP={group}&FORMAT=tle'
try:
with urllib.request.urlopen(url, timeout=10) as response:
content = response.read().decode('utf-8')
lines = content.strip().split('\n')
i = 0
while i + 2 < len(lines):
name = lines[i].strip()
line1 = lines[i + 1].strip()
line2 = lines[i + 2].strip()
if not (line1.startswith('1 ') and line2.startswith('2 ')):
i += 1
continue
internal_name = name_mappings.get(name, name)
if internal_name in _tle_cache:
_tle_cache[internal_name] = (name, line1, line2)
updated.append(internal_name)
i += 3
except Exception as e:
logger.error(f"Error fetching {group}: {e}")
continue
return jsonify({
'status': 'success',
'updated': updated
})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})
@satellite_bp.route('/celestrak/<category>')
def fetch_celestrak(category):
"""Fetch TLE data from CelesTrak for a category."""
valid_categories = [
'stations', 'weather', 'noaa', 'goes', 'resource', 'sarsat',
'dmc', 'tdrss', 'argos', 'planet', 'spire', 'geo', 'intelsat',
'ses', 'iridium', 'iridium-NEXT', 'starlink', 'oneweb',
'amateur', 'cubesat', 'visual'
]
if category not in valid_categories:
return jsonify({'status': 'error', 'message': f'Invalid category. Valid: {valid_categories}'})
try:
url = f'https://celestrak.org/NORAD/elements/gp.php?GROUP={category}&FORMAT=tle'
with urllib.request.urlopen(url, timeout=10) as response:
content = response.read().decode('utf-8')
satellites = []
lines = content.strip().split('\n')
i = 0
while i + 2 < len(lines):
name = lines[i].strip()
line1 = lines[i + 1].strip()
line2 = lines[i + 2].strip()
if not (line1.startswith('1 ') and line2.startswith('2 ')):
i += 1
continue
try:
norad_id = int(line1[2:7])
satellites.append({
'name': name,
'norad': norad_id,
'tle1': line1,
'tle2': line2
})
except (ValueError, IndexError):
pass
i += 3
return jsonify({
'status': 'success',
'category': category,
'satellites': satellites
})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})

157
routes/sensor.py Normal file
View File

@@ -0,0 +1,157 @@
"""RTL_433 sensor monitoring routes."""
from __future__ import annotations
import json
import queue
import subprocess
import threading
from datetime import datetime
from typing import Generator
from flask import Blueprint, jsonify, request, Response
import app as app_module
from utils.logging import sensor_logger as logger
sensor_bp = Blueprint('sensor', __name__)
def stream_sensor_output(process: subprocess.Popen[bytes]) -> None:
"""Stream rtl_433 JSON output to queue."""
try:
app_module.sensor_queue.put({'type': 'status', 'text': 'started'})
for line in iter(process.stdout.readline, b''):
line = line.decode('utf-8', errors='replace').strip()
if not line:
continue
try:
# rtl_433 outputs JSON objects, one per line
data = json.loads(line)
data['type'] = 'sensor'
app_module.sensor_queue.put(data)
# Log if enabled
if app_module.logging_enabled:
try:
with open(app_module.log_file_path, 'a') as f:
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
f.write(f"{timestamp} | {data.get('model', 'Unknown')} | {json.dumps(data)}\n")
except Exception:
pass
except json.JSONDecodeError:
# Not JSON, send as raw
app_module.sensor_queue.put({'type': 'raw', 'text': line})
except Exception as e:
app_module.sensor_queue.put({'type': 'error', 'text': str(e)})
finally:
process.wait()
app_module.sensor_queue.put({'type': 'status', 'text': 'stopped'})
with app_module.sensor_lock:
app_module.sensor_process = None
@sensor_bp.route('/start_sensor', methods=['POST'])
def start_sensor() -> Response:
with app_module.sensor_lock:
if app_module.sensor_process:
return jsonify({'status': 'error', 'message': 'Sensor already running'})
data = request.json
freq = data.get('frequency', '433.92')
gain = data.get('gain', '0')
ppm = data.get('ppm', '0')
device = data.get('device', '0')
# Clear queue
while not app_module.sensor_queue.empty():
try:
app_module.sensor_queue.get_nowait()
except queue.Empty:
break
# Build rtl_433 command
cmd = [
'rtl_433',
'-d', str(device),
'-f', f'{freq}M',
'-F', 'json'
]
if gain and gain != '0':
cmd.extend(['-g', str(gain)])
if ppm and ppm != '0':
cmd.extend(['-p', str(ppm)])
full_cmd = ' '.join(cmd)
logger.info(f"Running: {full_cmd}")
try:
app_module.sensor_process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
bufsize=1
)
# Start output thread
thread = threading.Thread(target=stream_sensor_output, args=(app_module.sensor_process,))
thread.daemon = True
thread.start()
# Monitor stderr
def monitor_stderr():
for line in app_module.sensor_process.stderr:
err = line.decode('utf-8', errors='replace').strip()
if err:
logger.debug(f"[rtl_433] {err}")
app_module.sensor_queue.put({'type': 'info', 'text': f'[rtl_433] {err}'})
stderr_thread = threading.Thread(target=monitor_stderr)
stderr_thread.daemon = True
stderr_thread.start()
app_module.sensor_queue.put({'type': 'info', 'text': f'Command: {full_cmd}'})
return jsonify({'status': 'started', 'command': full_cmd})
except FileNotFoundError:
return jsonify({'status': 'error', 'message': 'rtl_433 not found. Install with: brew install rtl_433'})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})
@sensor_bp.route('/stop_sensor', methods=['POST'])
def stop_sensor() -> Response:
with app_module.sensor_lock:
if app_module.sensor_process:
app_module.sensor_process.terminate()
try:
app_module.sensor_process.wait(timeout=2)
except subprocess.TimeoutExpired:
app_module.sensor_process.kill()
app_module.sensor_process = None
return jsonify({'status': 'stopped'})
return jsonify({'status': 'not_running'})
@sensor_bp.route('/stream_sensor')
def stream_sensor() -> Response:
def generate() -> Generator[str, None, None]:
while True:
try:
msg = app_module.sensor_queue.get(timeout=1)
yield f"data: {json.dumps(msg)}\n\n"
except queue.Empty:
yield f"data: {json.dumps({'type': 'keepalive'})}\n\n"
response = Response(generate(), mimetype='text/event-stream')
response.headers['Cache-Control'] = 'no-cache'
response.headers['X-Accel-Buffering'] = 'no'
response.headers['Connection'] = 'keep-alive'
return response

772
routes/wifi.py Normal file
View File

@@ -0,0 +1,772 @@
"""WiFi reconnaissance routes."""
from __future__ import annotations
import fcntl
import json
import os
import platform
import queue
import re
import subprocess
import threading
import time
from typing import Any, Generator
from flask import Blueprint, jsonify, request, Response
import app as app_module
from utils.dependencies import check_tool
from utils.logging import wifi_logger as logger
from utils.process import is_valid_mac, is_valid_channel
from data.oui import get_manufacturer
wifi_bp = Blueprint('wifi', __name__, url_prefix='/wifi')
# PMKID process state
pmkid_process = None
pmkid_lock = threading.Lock()
def detect_wifi_interfaces():
"""Detect available WiFi interfaces."""
interfaces = []
if platform.system() == 'Darwin': # macOS
try:
result = subprocess.run(['networksetup', '-listallhardwareports'],
capture_output=True, text=True, timeout=5)
lines = result.stdout.split('\n')
for i, line in enumerate(lines):
if 'Wi-Fi' in line or 'AirPort' in line:
for j in range(i+1, min(i+3, len(lines))):
if 'Device:' in lines[j]:
device = lines[j].split('Device:')[1].strip()
interfaces.append({
'name': device,
'type': 'internal',
'monitor_capable': False,
'status': 'up'
})
break
except Exception as e:
logger.error(f"Error detecting macOS interfaces: {e}")
try:
result = subprocess.run(['system_profiler', 'SPUSBDataType'],
capture_output=True, text=True, timeout=10)
if 'Wireless' in result.stdout or 'WLAN' in result.stdout or '802.11' in result.stdout:
interfaces.append({
'name': 'USB WiFi Adapter',
'type': 'usb',
'monitor_capable': True,
'status': 'detected'
})
except Exception:
pass
else: # Linux
try:
result = subprocess.run(['iw', 'dev'], capture_output=True, text=True, timeout=5)
current_iface = None
for line in result.stdout.split('\n'):
line = line.strip()
if line.startswith('Interface'):
current_iface = line.split()[1]
elif current_iface and 'type' in line:
iface_type = line.split()[-1]
interfaces.append({
'name': current_iface,
'type': iface_type,
'monitor_capable': True,
'status': 'up'
})
current_iface = None
except FileNotFoundError:
try:
result = subprocess.run(['iwconfig'], capture_output=True, text=True, timeout=5)
for line in result.stdout.split('\n'):
if 'IEEE 802.11' in line:
iface = line.split()[0]
interfaces.append({
'name': iface,
'type': 'managed',
'monitor_capable': True,
'status': 'up'
})
except Exception:
pass
except Exception as e:
logger.error(f"Error detecting Linux interfaces: {e}")
return interfaces
def parse_airodump_csv(csv_path):
"""Parse airodump-ng CSV output file."""
networks = {}
clients = {}
try:
with open(csv_path, 'r', errors='replace') as f:
content = f.read()
sections = content.split('\n\n')
for section in sections:
lines = section.strip().split('\n')
if not lines:
continue
header = lines[0] if lines else ''
if 'BSSID' in header and 'ESSID' in header:
for line in lines[1:]:
parts = [p.strip() for p in line.split(',')]
if len(parts) >= 14:
bssid = parts[0]
if bssid and ':' in bssid:
networks[bssid] = {
'bssid': bssid,
'first_seen': parts[1],
'last_seen': parts[2],
'channel': parts[3],
'speed': parts[4],
'privacy': parts[5],
'cipher': parts[6],
'auth': parts[7],
'power': parts[8],
'beacons': parts[9],
'ivs': parts[10],
'lan_ip': parts[11],
'essid': parts[13] or 'Hidden'
}
elif 'Station MAC' in header:
for line in lines[1:]:
parts = [p.strip() for p in line.split(',')]
if len(parts) >= 6:
station = parts[0]
if station and ':' in station:
vendor = get_manufacturer(station)
clients[station] = {
'mac': station,
'first_seen': parts[1],
'last_seen': parts[2],
'power': parts[3],
'packets': parts[4],
'bssid': parts[5],
'probes': parts[6] if len(parts) > 6 else '',
'vendor': vendor
}
except Exception as e:
logger.error(f"Error parsing CSV: {e}")
return networks, clients
def stream_airodump_output(process, csv_path):
"""Stream airodump-ng output to queue."""
try:
app_module.wifi_queue.put({'type': 'status', 'text': 'started'})
last_parse = 0
start_time = time.time()
csv_found = False
while process.poll() is None:
try:
fd = process.stderr.fileno()
fl = fcntl.fcntl(fd, fcntl.F_GETFL)
fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
stderr_data = process.stderr.read()
if stderr_data:
stderr_text = stderr_data.decode('utf-8', errors='replace').strip()
if stderr_text:
for line in stderr_text.split('\n'):
line = line.strip()
if line and not line.startswith('CH') and not line.startswith('Elapsed'):
app_module.wifi_queue.put({'type': 'error', 'text': f'airodump-ng: {line}'})
except Exception:
pass
current_time = time.time()
if current_time - last_parse >= 2:
csv_file = csv_path + '-01.csv'
if os.path.exists(csv_file):
csv_found = True
networks, clients = parse_airodump_csv(csv_file)
for bssid, net in networks.items():
if bssid not in app_module.wifi_networks:
app_module.wifi_queue.put({
'type': 'network',
'action': 'new',
**net
})
else:
app_module.wifi_queue.put({
'type': 'network',
'action': 'update',
**net
})
for mac, client in clients.items():
if mac not in app_module.wifi_clients:
app_module.wifi_queue.put({
'type': 'client',
'action': 'new',
**client
})
app_module.wifi_networks = networks
app_module.wifi_clients = clients
last_parse = current_time
if current_time - start_time > 5 and not csv_found:
app_module.wifi_queue.put({'type': 'error', 'text': 'No scan data after 5 seconds. Check if monitor mode is properly enabled.'})
start_time = current_time + 30
time.sleep(0.5)
try:
remaining_stderr = process.stderr.read()
if remaining_stderr:
stderr_text = remaining_stderr.decode('utf-8', errors='replace').strip()
if stderr_text:
app_module.wifi_queue.put({'type': 'error', 'text': f'airodump-ng exited: {stderr_text}'})
except Exception:
pass
exit_code = process.returncode
if exit_code != 0 and exit_code is not None:
app_module.wifi_queue.put({'type': 'error', 'text': f'airodump-ng exited with code {exit_code}'})
except Exception as e:
app_module.wifi_queue.put({'type': 'error', 'text': str(e)})
finally:
process.wait()
app_module.wifi_queue.put({'type': 'status', 'text': 'stopped'})
with app_module.wifi_lock:
app_module.wifi_process = None
@wifi_bp.route('/interfaces')
def get_wifi_interfaces():
"""Get available WiFi interfaces."""
interfaces = detect_wifi_interfaces()
tools = {
'airmon': check_tool('airmon-ng'),
'airodump': check_tool('airodump-ng'),
'aireplay': check_tool('aireplay-ng'),
'iw': check_tool('iw')
}
return jsonify({'interfaces': interfaces, 'tools': tools, 'monitor_interface': app_module.wifi_monitor_interface})
@wifi_bp.route('/monitor', methods=['POST'])
def toggle_monitor_mode():
"""Enable or disable monitor mode on an interface."""
data = request.json
interface = data.get('interface')
action = data.get('action', 'start')
if not interface:
return jsonify({'status': 'error', 'message': 'No interface specified'})
if action == 'start':
if check_tool('airmon-ng'):
try:
def get_wireless_interfaces():
interfaces = set()
try:
result = subprocess.run(['iwconfig'], capture_output=True, text=True, timeout=5)
for line in result.stdout.split('\n'):
if line and not line.startswith(' ') and 'no wireless' not in line.lower():
iface = line.split()[0] if line.split() else None
if iface:
interfaces.add(iface)
except (subprocess.SubprocessError, OSError):
pass
try:
for iface in os.listdir('/sys/class/net'):
if os.path.exists(f'/sys/class/net/{iface}/wireless'):
interfaces.add(iface)
except OSError:
pass
try:
result = subprocess.run(['ip', 'link', 'show'], capture_output=True, text=True, timeout=5)
for match in re.finditer(r'^\d+:\s+(\S+):', result.stdout, re.MULTILINE):
iface = match.group(1).rstrip(':')
if iface.startswith('wl') or 'mon' in iface:
interfaces.add(iface)
except (subprocess.SubprocessError, OSError):
pass
return interfaces
interfaces_before = get_wireless_interfaces()
kill_processes = data.get('kill_processes', False)
if kill_processes:
subprocess.run(['airmon-ng', 'check', 'kill'], capture_output=True, timeout=10)
result = subprocess.run(['airmon-ng', 'start', interface],
capture_output=True, text=True, timeout=15)
output = result.stdout + result.stderr
time.sleep(1)
interfaces_after = get_wireless_interfaces()
new_interfaces = interfaces_after - interfaces_before
monitor_iface = None
if new_interfaces:
for iface in new_interfaces:
if 'mon' in iface:
monitor_iface = iface
break
if not monitor_iface:
monitor_iface = list(new_interfaces)[0]
if not monitor_iface:
patterns = [
r'monitor mode.*enabled.*on\s+(\S+)',
r'\(monitor mode.*enabled.*?(\S+mon)\)',
r'created\s+(\S+mon)',
r'\bon\s+(\S+mon)\b',
r'\b(\S+mon)\b.*monitor',
r'\b(' + re.escape(interface) + r'mon)\b',
]
for pattern in patterns:
match = re.search(pattern, output, re.IGNORECASE)
if match:
monitor_iface = match.group(1)
break
if not monitor_iface:
try:
result = subprocess.run(['iwconfig', interface], capture_output=True, text=True, timeout=5)
if 'Mode:Monitor' in result.stdout:
monitor_iface = interface
except (subprocess.SubprocessError, OSError):
pass
if not monitor_iface:
potential = interface + 'mon'
if potential in interfaces_after:
monitor_iface = potential
if not monitor_iface:
monitor_iface = interface + 'mon'
app_module.wifi_monitor_interface = monitor_iface
app_module.wifi_queue.put({'type': 'info', 'text': f'Monitor mode enabled on {app_module.wifi_monitor_interface}'})
return jsonify({'status': 'success', 'monitor_interface': app_module.wifi_monitor_interface})
except Exception as e:
import traceback
logger.error(f"Error enabling monitor mode: {e}", exc_info=True)
return jsonify({'status': 'error', 'message': str(e)})
elif check_tool('iw'):
try:
subprocess.run(['ip', 'link', 'set', interface, 'down'], capture_output=True)
subprocess.run(['iw', interface, 'set', 'monitor', 'control'], capture_output=True)
subprocess.run(['ip', 'link', 'set', interface, 'up'], capture_output=True)
app_module.wifi_monitor_interface = interface
return jsonify({'status': 'success', 'monitor_interface': interface})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})
else:
return jsonify({'status': 'error', 'message': 'No monitor mode tools available.'})
else: # stop
if check_tool('airmon-ng'):
try:
subprocess.run(['airmon-ng', 'stop', app_module.wifi_monitor_interface or interface],
capture_output=True, text=True, timeout=15)
app_module.wifi_monitor_interface = None
return jsonify({'status': 'success', 'message': 'Monitor mode disabled'})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})
elif check_tool('iw'):
try:
subprocess.run(['ip', 'link', 'set', interface, 'down'], capture_output=True)
subprocess.run(['iw', interface, 'set', 'type', 'managed'], capture_output=True)
subprocess.run(['ip', 'link', 'set', interface, 'up'], capture_output=True)
app_module.wifi_monitor_interface = None
return jsonify({'status': 'success', 'message': 'Monitor mode disabled'})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})
return jsonify({'status': 'error', 'message': 'Unknown action'})
@wifi_bp.route('/scan/start', methods=['POST'])
def start_wifi_scan():
"""Start WiFi scanning with airodump-ng."""
with app_module.wifi_lock:
if app_module.wifi_process:
return jsonify({'status': 'error', 'message': 'Scan already running'})
data = request.json
interface = data.get('interface') or app_module.wifi_monitor_interface
channel = data.get('channel')
band = data.get('band', 'abg')
if not interface:
return jsonify({'status': 'error', 'message': 'No monitor interface available.'})
app_module.wifi_networks = {}
app_module.wifi_clients = {}
while not app_module.wifi_queue.empty():
try:
app_module.wifi_queue.get_nowait()
except queue.Empty:
break
csv_path = '/tmp/intercept_wifi'
for f in [f'/tmp/intercept_wifi-01.csv', f'/tmp/intercept_wifi-01.cap']:
try:
os.remove(f)
except OSError:
pass
cmd = [
'airodump-ng',
'-w', csv_path,
'--output-format', 'csv,pcap',
'--band', band,
interface
]
if channel:
cmd.extend(['-c', str(channel)])
logger.info(f"Running: {' '.join(cmd)}")
try:
app_module.wifi_process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
time.sleep(0.5)
if app_module.wifi_process.poll() is not None:
stderr_output = app_module.wifi_process.stderr.read().decode('utf-8', errors='replace').strip()
stdout_output = app_module.wifi_process.stdout.read().decode('utf-8', errors='replace').strip()
exit_code = app_module.wifi_process.returncode
app_module.wifi_process = None
error_msg = stderr_output or stdout_output or f'Process exited with code {exit_code}'
error_msg = re.sub(r'\x1b\[[0-9;]*m', '', error_msg)
if 'No such device' in error_msg or 'No such interface' in error_msg:
error_msg = f'Interface "{interface}" not found.'
elif 'Operation not permitted' in error_msg:
error_msg = 'Permission denied. Try running with sudo.'
return jsonify({'status': 'error', 'message': error_msg})
thread = threading.Thread(target=stream_airodump_output, args=(app_module.wifi_process, csv_path))
thread.daemon = True
thread.start()
app_module.wifi_queue.put({'type': 'info', 'text': f'Started scanning on {interface}'})
return jsonify({'status': 'started', 'interface': interface})
except FileNotFoundError:
return jsonify({'status': 'error', 'message': 'airodump-ng not found.'})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})
@wifi_bp.route('/scan/stop', methods=['POST'])
def stop_wifi_scan():
"""Stop WiFi scanning."""
with app_module.wifi_lock:
if app_module.wifi_process:
app_module.wifi_process.terminate()
try:
app_module.wifi_process.wait(timeout=3)
except subprocess.TimeoutExpired:
app_module.wifi_process.kill()
app_module.wifi_process = None
return jsonify({'status': 'stopped'})
return jsonify({'status': 'not_running'})
@wifi_bp.route('/deauth', methods=['POST'])
def send_deauth():
"""Send deauthentication packets."""
data = request.json
target_bssid = data.get('bssid')
target_client = data.get('client', 'FF:FF:FF:FF:FF:FF')
count = data.get('count', 5)
interface = data.get('interface') or app_module.wifi_monitor_interface
if not target_bssid:
return jsonify({'status': 'error', 'message': 'Target BSSID required'})
if not is_valid_mac(target_bssid):
return jsonify({'status': 'error', 'message': 'Invalid BSSID format'})
if not is_valid_mac(target_client):
return jsonify({'status': 'error', 'message': 'Invalid client MAC format'})
try:
count = int(count)
if count < 1 or count > 100:
count = 5
except (ValueError, TypeError):
count = 5
if not interface:
return jsonify({'status': 'error', 'message': 'No monitor interface'})
if not check_tool('aireplay-ng'):
return jsonify({'status': 'error', 'message': 'aireplay-ng not found'})
try:
cmd = [
'aireplay-ng',
'--deauth', str(count),
'-a', target_bssid,
'-c', target_client,
interface
]
app_module.wifi_queue.put({'type': 'info', 'text': f'Sending {count} deauth packets to {target_bssid}'})
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
if result.returncode == 0:
return jsonify({'status': 'success', 'message': f'Sent {count} deauth packets'})
else:
return jsonify({'status': 'error', 'message': result.stderr})
except subprocess.TimeoutExpired:
return jsonify({'status': 'success', 'message': 'Deauth sent (timed out)'})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})
@wifi_bp.route('/handshake/capture', methods=['POST'])
def capture_handshake():
"""Start targeted handshake capture."""
data = request.json
target_bssid = data.get('bssid')
channel = data.get('channel')
interface = data.get('interface') or app_module.wifi_monitor_interface
if not target_bssid or not channel:
return jsonify({'status': 'error', 'message': 'BSSID and channel required'})
if not is_valid_mac(target_bssid):
return jsonify({'status': 'error', 'message': 'Invalid BSSID format'})
if not is_valid_channel(channel):
return jsonify({'status': 'error', 'message': 'Invalid channel'})
with app_module.wifi_lock:
if app_module.wifi_process:
return jsonify({'status': 'error', 'message': 'Scan already running.'})
capture_path = f'/tmp/intercept_handshake_{target_bssid.replace(":", "")}'
cmd = [
'airodump-ng',
'-c', str(channel),
'--bssid', target_bssid,
'-w', capture_path,
'--output-format', 'pcap',
interface
]
try:
app_module.wifi_process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
app_module.wifi_queue.put({'type': 'info', 'text': f'Capturing handshakes for {target_bssid}'})
return jsonify({'status': 'started', 'capture_file': capture_path + '-01.cap'})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})
@wifi_bp.route('/handshake/status', methods=['POST'])
def check_handshake_status():
"""Check if a handshake has been captured."""
data = request.json
capture_file = data.get('file', '')
target_bssid = data.get('bssid', '')
if not capture_file.startswith('/tmp/intercept_handshake_') or '..' in capture_file:
return jsonify({'status': 'error', 'message': 'Invalid capture file path'})
if not os.path.exists(capture_file):
with app_module.wifi_lock:
if app_module.wifi_process and app_module.wifi_process.poll() is None:
return jsonify({'status': 'running', 'file_exists': False, 'handshake_found': False})
else:
return jsonify({'status': 'stopped', 'file_exists': False, 'handshake_found': False})
file_size = os.path.getsize(capture_file)
handshake_found = False
try:
if target_bssid and is_valid_mac(target_bssid):
result = subprocess.run(
['aircrack-ng', '-a', '2', '-b', target_bssid, capture_file],
capture_output=True, text=True, timeout=10
)
output = result.stdout + result.stderr
if '1 handshake' in output or ('handshake' in output.lower() and 'wpa' in output.lower()):
if '0 handshake' not in output:
handshake_found = True
except subprocess.TimeoutExpired:
pass
except Exception as e:
logger.error(f"Error checking handshake: {e}")
return jsonify({
'status': 'running' if app_module.wifi_process and app_module.wifi_process.poll() is None else 'stopped',
'file_exists': True,
'file_size': file_size,
'file': capture_file,
'handshake_found': handshake_found
})
@wifi_bp.route('/pmkid/capture', methods=['POST'])
def capture_pmkid():
"""Start PMKID capture using hcxdumptool."""
global pmkid_process
data = request.json
target_bssid = data.get('bssid')
channel = data.get('channel')
interface = data.get('interface') or app_module.wifi_monitor_interface
if not target_bssid:
return jsonify({'status': 'error', 'message': 'BSSID required'})
if not is_valid_mac(target_bssid):
return jsonify({'status': 'error', 'message': 'Invalid BSSID format'})
with pmkid_lock:
if pmkid_process and pmkid_process.poll() is None:
return jsonify({'status': 'error', 'message': 'PMKID capture already running'})
capture_path = f'/tmp/intercept_pmkid_{target_bssid.replace(":", "")}.pcapng'
filter_file = f'/tmp/pmkid_filter_{target_bssid.replace(":", "")}'
with open(filter_file, 'w') as f:
f.write(target_bssid.replace(':', '').lower())
cmd = [
'hcxdumptool',
'-i', interface,
'-o', capture_path,
'--filterlist_ap', filter_file,
'--filtermode', '2',
'--enable_status', '1'
]
if channel:
cmd.extend(['-c', str(channel)])
try:
pmkid_process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
return jsonify({'status': 'started', 'file': capture_path})
except FileNotFoundError:
return jsonify({'status': 'error', 'message': 'hcxdumptool not found.'})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})
@wifi_bp.route('/pmkid/status', methods=['POST'])
def check_pmkid_status():
"""Check if PMKID has been captured."""
data = request.json
capture_file = data.get('file', '')
if not capture_file.startswith('/tmp/intercept_pmkid_') or '..' in capture_file:
return jsonify({'status': 'error', 'message': 'Invalid capture file path'})
if not os.path.exists(capture_file):
return jsonify({'pmkid_found': False, 'file_exists': False})
file_size = os.path.getsize(capture_file)
pmkid_found = False
try:
hash_file = capture_file.replace('.pcapng', '.22000')
result = subprocess.run(
['hcxpcapngtool', '-o', hash_file, capture_file],
capture_output=True, text=True, timeout=10
)
if os.path.exists(hash_file) and os.path.getsize(hash_file) > 0:
pmkid_found = True
except FileNotFoundError:
pmkid_found = file_size > 1000
except Exception:
pass
return jsonify({
'pmkid_found': pmkid_found,
'file_exists': True,
'file_size': file_size,
'file': capture_file
})
@wifi_bp.route('/pmkid/stop', methods=['POST'])
def stop_pmkid():
"""Stop PMKID capture."""
global pmkid_process
with pmkid_lock:
if pmkid_process:
pmkid_process.terminate()
try:
pmkid_process.wait(timeout=5)
except subprocess.TimeoutExpired:
pmkid_process.kill()
pmkid_process = None
return jsonify({'status': 'stopped'})
@wifi_bp.route('/networks')
def get_wifi_networks():
"""Get current list of discovered networks."""
return jsonify({
'networks': list(app_module.wifi_networks.values()),
'clients': list(app_module.wifi_clients.values()),
'handshakes': app_module.wifi_handshakes,
'monitor_interface': app_module.wifi_monitor_interface
})
@wifi_bp.route('/stream')
def stream_wifi():
"""SSE stream for WiFi events."""
def generate():
while True:
try:
msg = app_module.wifi_queue.get(timeout=1)
yield f"data: {json.dumps(msg)}\n\n"
except queue.Empty:
yield f"data: {json.dumps({'type': 'keepalive'})}\n\n"
response = Response(generate(), mimetype='text/event-stream')
response.headers['Cache-Control'] = 'no-cache'
response.headers['X-Accel-Buffering'] = 'no'
response.headers['Connection'] = 'keep-alive'
return response

997
templates/adsb_dashboard.html Normal file
View File

@@ -0,0 +1,997 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>AIRCRAFT RADAR // INTERCEPT</title>
<link href="https://fonts.googleapis.com/css2?family=Orbitron:wght@400;500;700;900&family=Rajdhani:wght@300;400;500;600;700&family=JetBrains+Mono:wght@300;400;500&display=swap" rel="stylesheet">
<link rel="stylesheet" href="https://unpkg.com/leaflet@1.9.4/dist/leaflet.css" />
<script src="https://unpkg.com/leaflet@1.9.4/dist/leaflet.js"></script>
<style>
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
:root {
--bg-dark: #0a0a0f;
--bg-panel: #0d1117;
--bg-card: #161b22;
--border-glow: #00ff88;
--text-primary: #e6edf3;
--text-secondary: #8b949e;
--accent-green: #00ff88;
--accent-cyan: #00d4ff;
--accent-orange: #ff9500;
--accent-red: #ff4444;
--accent-yellow: #ffcc00;
--grid-line: rgba(0, 255, 136, 0.1);
}
body {
font-family: 'Rajdhani', sans-serif;
background: var(--bg-dark);
color: var(--text-primary);
min-height: 100vh;
overflow-x: hidden;
}
/* Animated radar sweep background */
.radar-bg {
position: fixed;
top: 0;
left: 0;
right: 0;
bottom: 0;
background-image:
linear-gradient(var(--grid-line) 1px, transparent 1px),
linear-gradient(90deg, var(--grid-line) 1px, transparent 1px);
background-size: 50px 50px;
pointer-events: none;
z-index: 0;
}
/* Scan line effect */
.scanline {
position: fixed;
top: 0;
left: 0;
right: 0;
height: 4px;
background: linear-gradient(90deg, transparent, var(--accent-green), transparent);
animation: scan 4s linear infinite;
pointer-events: none;
z-index: 1000;
opacity: 0.5;
}
@keyframes scan {
0% { top: -4px; }
100% { top: 100vh; }
}
/* Header */
.header {
position: relative;
z-index: 10;
padding: 15px 30px;
background: linear-gradient(180deg, rgba(0,255,136,0.1) 0%, transparent 100%);
border-bottom: 1px solid rgba(0,255,136,0.3);
display: flex;
justify-content: space-between;
align-items: center;
}
.logo {
font-family: 'Orbitron', monospace;
font-size: 28px;
font-weight: 900;
letter-spacing: 4px;
color: var(--accent-green);
text-shadow: 0 0 20px var(--accent-green), 0 0 40px var(--accent-green);
}
.logo span {
color: var(--text-secondary);
font-weight: 400;
font-size: 16px;
margin-left: 15px;
letter-spacing: 2px;
}
.status-bar {
display: flex;
gap: 30px;
font-family: 'JetBrains Mono', monospace;
font-size: 12px;
}
.status-item {
display: flex;
align-items: center;
gap: 8px;
}
.status-dot {
width: 8px;
height: 8px;
border-radius: 50%;
background: var(--accent-green);
box-shadow: 0 0 10px var(--accent-green);
animation: pulse 2s ease-in-out infinite;
}
.status-dot.inactive {
background: var(--accent-red);
box-shadow: 0 0 10px var(--accent-red);
}
@keyframes pulse {
0%, 100% { opacity: 1; }
50% { opacity: 0.5; }
}
.datetime {
font-family: 'Orbitron', monospace;
font-size: 14px;
color: var(--accent-green);
}
/* Main dashboard grid */
.dashboard {
position: relative;
z-index: 10;
display: grid;
grid-template-columns: 1fr 380px;
grid-template-rows: 1fr;
gap: 20px;
padding: 20px;
height: calc(100vh - 80px);
min-height: 600px;
}
/* Panels */
.panel {
background: var(--bg-panel);
border: 1px solid rgba(0,255,136,0.2);
border-radius: 8px;
overflow: hidden;
position: relative;
}
.panel::before {
content: '';
position: absolute;
top: 0;
left: 0;
right: 0;
height: 2px;
background: linear-gradient(90deg, transparent, var(--accent-green), transparent);
}
.panel-header {
padding: 12px 20px;
background: rgba(0,255,136,0.05);
border-bottom: 1px solid rgba(0,255,136,0.1);
font-family: 'Orbitron', monospace;
font-size: 12px;
font-weight: 500;
letter-spacing: 2px;
text-transform: uppercase;
color: var(--accent-green);
display: flex;
justify-content: space-between;
align-items: center;
}
.panel-indicator {
width: 6px;
height: 6px;
background: var(--accent-green);
border-radius: 50%;
animation: blink 1s ease-in-out infinite;
}
@keyframes blink {
0%, 100% { opacity: 1; }
50% { opacity: 0.3; }
}
.panel-content {
padding: 0;
height: calc(100% - 45px);
overflow: hidden;
}
/* Map container */
.map-container {
grid-column: 1;
grid-row: 1;
}
#radarMap {
width: 100%;
height: 100%;
border-radius: 0 0 8px 8px;
}
/* Right sidebar */
.sidebar {
grid-column: 2;
grid-row: 1;
display: flex;
flex-direction: column;
gap: 12px;
overflow-y: auto;
}
.sidebar .panel {
flex-shrink: 0;
}
/* Stats panel */
.stats-grid {
display: grid;
grid-template-columns: repeat(2, 1fr);
gap: 10px;
padding: 15px;
}
.stat-box {
background: rgba(0,0,0,0.3);
border: 1px solid rgba(0,255,136,0.15);
border-radius: 6px;
padding: 12px;
text-align: center;
}
.stat-value {
font-family: 'Orbitron', monospace;
font-size: 24px;
font-weight: 600;
color: var(--accent-green);
}
.stat-label {
font-size: 10px;
text-transform: uppercase;
letter-spacing: 1px;
color: var(--text-secondary);
margin-top: 4px;
}
/* Aircraft list */
.aircraft-list {
flex: 1;
min-height: 200px;
}
.aircraft-list-content {
max-height: 350px;
overflow-y: auto;
padding: 10px;
}
.aircraft-item {
background: rgba(0,0,0,0.3);
border: 1px solid rgba(0,255,136,0.15);
border-radius: 6px;
padding: 10px 12px;
margin-bottom: 8px;
cursor: pointer;
transition: all 0.2s ease;
}
.aircraft-item:hover {
border-color: var(--accent-green);
background: rgba(0,255,136,0.05);
}
.aircraft-item.selected {
border-color: var(--accent-green);
box-shadow: 0 0 15px rgba(0,255,136,0.2);
background: rgba(0,255,136,0.1);
}
.aircraft-header {
display: flex;
justify-content: space-between;
align-items: center;
margin-bottom: 6px;
}
.aircraft-callsign {
font-family: 'Orbitron', monospace;
font-size: 14px;
font-weight: 600;
color: var(--accent-green);
}
.aircraft-icao {
font-family: 'JetBrains Mono', monospace;
font-size: 10px;
color: var(--text-secondary);
background: rgba(0,255,136,0.1);
padding: 2px 6px;
border-radius: 3px;
}
.aircraft-details {
display: grid;
grid-template-columns: repeat(3, 1fr);
gap: 8px;
font-size: 11px;
}
.aircraft-detail {
text-align: center;
}
.aircraft-detail-value {
font-family: 'JetBrains Mono', monospace;
color: var(--accent-cyan);
font-size: 12px;
}
.aircraft-detail-label {
color: var(--text-secondary);
font-size: 9px;
text-transform: uppercase;
}
/* Selected aircraft panel */
.selected-aircraft {
background: linear-gradient(135deg, rgba(0,255,136,0.1) 0%, rgba(0,212,255,0.05) 100%);
}
.selected-info {
padding: 15px;
}
.selected-callsign {
font-family: 'Orbitron', monospace;
font-size: 24px;
font-weight: 700;
color: var(--accent-green);
text-shadow: 0 0 15px var(--accent-green);
text-align: center;
margin-bottom: 15px;
}
.telemetry-grid {
display: grid;
grid-template-columns: repeat(2, 1fr);
gap: 8px;
}
.telemetry-item {
background: rgba(0,0,0,0.3);
border-radius: 4px;
padding: 10px;
border-left: 2px solid var(--accent-green);
}
.telemetry-label {
font-size: 10px;
text-transform: uppercase;
letter-spacing: 1px;
color: var(--text-secondary);
margin-bottom: 4px;
}
.telemetry-value {
font-family: 'JetBrains Mono', monospace;
font-size: 14px;
color: var(--accent-cyan);
}
/* Altitude indicator */
.altitude-indicator {
padding: 15px;
}
.altitude-bar {
height: 200px;
background: rgba(0,0,0,0.3);
border-radius: 6px;
position: relative;
overflow: hidden;
}
.altitude-fill {
position: absolute;
bottom: 0;
left: 0;
right: 0;
background: linear-gradient(to top, var(--accent-green), var(--accent-cyan));
transition: height 0.5s ease;
border-radius: 0 0 6px 6px;
}
.altitude-markers {
position: absolute;
top: 0;
right: 10px;
bottom: 0;
display: flex;
flex-direction: column;
justify-content: space-between;
padding: 10px 0;
}
.altitude-marker {
font-family: 'JetBrains Mono', monospace;
font-size: 10px;
color: var(--text-secondary);
}
.altitude-current {
position: absolute;
left: 10px;
font-family: 'Orbitron', monospace;
font-size: 18px;
color: var(--accent-green);
text-shadow: 0 0 10px var(--accent-green);
transition: bottom 0.5s ease;
}
/* Leaflet overrides */
.leaflet-container {
background: var(--bg-dark) !important;
}
.leaflet-control-zoom a {
background: var(--bg-panel) !important;
color: var(--accent-green) !important;
border-color: rgba(0,255,136,0.3) !important;
}
.leaflet-control-attribution {
background: rgba(0,0,0,0.7) !important;
color: var(--text-secondary) !important;
font-size: 9px !important;
}
/* Custom scrollbar */
::-webkit-scrollbar {
width: 6px;
}
::-webkit-scrollbar-track {
background: var(--bg-dark);
}
::-webkit-scrollbar-thumb {
background: var(--accent-green);
border-radius: 3px;
}
/* No aircraft message */
.no-aircraft {
text-align: center;
padding: 40px 20px;
color: var(--text-secondary);
}
.no-aircraft-icon {
font-size: 48px;
margin-bottom: 15px;
opacity: 0.5;
}
/* Start tracking button */
.start-btn {
display: block;
width: calc(100% - 30px);
margin: 15px;
padding: 12px;
border: 1px solid var(--accent-green);
background: rgba(0,255,136,0.1);
color: var(--accent-green);
font-family: 'Orbitron', monospace;
font-size: 12px;
font-weight: 600;
text-transform: uppercase;
letter-spacing: 2px;
border-radius: 6px;
cursor: pointer;
transition: all 0.2s ease;
}
.start-btn:hover {
background: var(--accent-green);
color: var(--bg-dark);
box-shadow: 0 0 20px rgba(0,255,136,0.3);
}
.start-btn.active {
background: var(--accent-red);
border-color: var(--accent-red);
color: #fff;
}
.start-btn.active:hover {
box-shadow: 0 0 20px rgba(255,68,68,0.3);
}
/* Responsive */
@media (max-width: 1200px) {
.dashboard {
grid-template-columns: 1fr;
grid-template-rows: 1fr auto;
height: auto;
}
.map-container {
min-height: 400px;
}
.sidebar {
flex-direction: row;
flex-wrap: wrap;
}
.sidebar .panel {
flex: 1 1 300px;
}
}
</style>
</head>
<body>
<div class="radar-bg"></div>
<div class="scanline"></div>
<header class="header">
<div class="logo">
AIRCRAFT RADAR
<span>// INTERCEPT</span>
</div>
<div class="status-bar">
<div class="status-item">
<div class="status-dot" id="trackingDot"></div>
<span id="trackingStatus">STANDBY</span>
</div>
<div class="status-item">
<span id="aircraftCount">0</span> AIRCRAFT
</div>
<div class="status-item datetime" id="utcTime">--:--:-- UTC</div>
</div>
</header>
<main class="dashboard">
<!-- Radar Map -->
<div class="panel map-container">
<div class="panel-header">
<span>RADAR DISPLAY // LIVE TRACKING</span>
<div class="panel-indicator"></div>
</div>
<div class="panel-content">
<div id="radarMap"></div>
</div>
</div>
<!-- Sidebar -->
<div class="sidebar">
<!-- Stats -->
<div class="panel">
<div class="panel-header">
<span>STATISTICS</span>
<div class="panel-indicator"></div>
</div>
<div class="stats-grid">
<div class="stat-box">
<div class="stat-value" id="statTotal">0</div>
<div class="stat-label">Total Aircraft</div>
</div>
<div class="stat-box">
<div class="stat-value" id="statWithPos">0</div>
<div class="stat-label">With Position</div>
</div>
<div class="stat-box">
<div class="stat-value" id="statMaxAlt">0</div>
<div class="stat-label">Max Alt (ft)</div>
</div>
<div class="stat-box">
<div class="stat-value" id="statAvgAlt">0</div>
<div class="stat-label">Avg Alt (ft)</div>
</div>
</div>
<button class="start-btn" id="startBtn" onclick="toggleTracking()">
START TRACKING
</button>
</div>
<!-- Selected Aircraft -->
<div class="panel selected-aircraft">
<div class="panel-header">
<span>SELECTED TARGET</span>
<div class="panel-indicator"></div>
</div>
<div class="selected-info" id="selectedInfo">
<div class="no-aircraft">
<div class="no-aircraft-icon">✈</div>
<div>Select an aircraft to view details</div>
</div>
</div>
</div>
<!-- Aircraft List -->
<div class="panel aircraft-list">
<div class="panel-header">
<span>TRACKED AIRCRAFT</span>
<div class="panel-indicator"></div>
</div>
<div class="panel-content">
<div class="aircraft-list-content" id="aircraftList">
<div class="no-aircraft">
<div>No aircraft detected</div>
<div style="font-size: 11px; margin-top: 5px;">Start tracking to detect aircraft</div>
</div>
</div>
</div>
</div>
</div>
</main>
<script>
// State
let radarMap = null;
let aircraft = {};
let markers = {};
let selectedIcao = null;
let eventSource = null;
let isTracking = false;
// Initialize
document.addEventListener('DOMContentLoaded', () => {
initMap();
updateClock();
setInterval(updateClock, 1000);
setInterval(cleanupOldAircraft, 10000);
});
function updateClock() {
const now = new Date();
document.getElementById('utcTime').textContent =
now.toISOString().substring(11, 19) + ' UTC';
}
function initMap() {
radarMap = L.map('radarMap', {
center: [51.5, -0.1],
zoom: 7,
minZoom: 3,
maxZoom: 15
});
L.tileLayer('https://{s}.basemaps.cartocdn.com/dark_all/{z}/{x}/{y}{r}.png', {
attribution: '©OpenStreetMap, ©CartoDB'
}).addTo(radarMap);
// Try to get user location
if (navigator.geolocation) {
navigator.geolocation.getCurrentPosition(pos => {
radarMap.setView([pos.coords.latitude, pos.coords.longitude], 8);
});
}
}
async function toggleTracking() {
const btn = document.getElementById('startBtn');
if (!isTracking) {
// Start tracking
try {
const response = await fetch('/adsb/start', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({})
});
console.log('Response status:', response.status);
const text = await response.text();
console.log('Response body:', text);
let data;
try {
data = JSON.parse(text);
} catch (e) {
alert('Invalid response from server: ' + text);
return;
}
if (data.status === 'success' || data.status === 'started' || data.status === 'already_running' || data.status === 'error' && data.message && data.message.includes('already')) {
startEventStream();
isTracking = true;
btn.textContent = 'STOP TRACKING';
btn.classList.add('active');
document.getElementById('trackingDot').classList.remove('inactive');
document.getElementById('trackingStatus').textContent = 'TRACKING';
} else {
alert('Failed to start: ' + (data.message || data.status || JSON.stringify(data)));
}
} catch (err) {
console.error('Start error:', err);
alert('Failed to start tracking: ' + err.message);
}
} else {
// Stop tracking
try {
await fetch('/adsb/stop', { method: 'POST' });
} catch (err) {
console.error('Stop error:', err);
}
stopEventStream();
isTracking = false;
btn.textContent = 'START TRACKING';
btn.classList.remove('active');
document.getElementById('trackingDot').classList.add('inactive');
document.getElementById('trackingStatus').textContent = 'STANDBY';
}
}
function startEventStream() {
if (eventSource) {
eventSource.close();
}
eventSource = new EventSource('/adsb/stream');
eventSource.onmessage = (event) => {
try {
const data = JSON.parse(event.data);
if (data.type === 'aircraft') {
updateAircraft(data);
}
} catch (err) {
console.error('Parse error:', err);
}
};
eventSource.onerror = () => {
console.error('EventSource error');
};
}
function stopEventStream() {
if (eventSource) {
eventSource.close();
eventSource = null;
}
}
function updateAircraft(data) {
const icao = data.icao;
if (!icao) return;
// Update aircraft data
aircraft[icao] = {
...aircraft[icao],
...data,
lastSeen: Date.now()
};
// Update marker on map
if (data.lat && data.lon) {
updateMarker(icao);
}
// Update UI
updateStats();
renderAircraftList();
// Update selected aircraft panel
if (selectedIcao === icao) {
showAircraftDetails(icao);
}
}
function updateMarker(icao) {
const ac = aircraft[icao];
if (!ac || !ac.lat || !ac.lon) return;
const rotation = ac.heading || 0;
const color = getAltitudeColor(ac.alt);
const icon = L.divIcon({
className: 'aircraft-marker',
html: `<div style="
transform: rotate(${rotation}deg);
color: ${color};
font-size: 20px;
text-shadow: 0 0 10px ${color};
filter: drop-shadow(0 0 5px ${color});
">✈</div>`,
iconSize: [24, 24],
iconAnchor: [12, 12]
});
if (markers[icao]) {
markers[icao].setLatLng([ac.lat, ac.lon]);
markers[icao].setIcon(icon);
} else {
markers[icao] = L.marker([ac.lat, ac.lon], { icon })
.addTo(radarMap)
.on('click', () => selectAircraft(icao));
}
// Add tooltip
const callsign = ac.callsign || icao;
const alt = ac.alt ? ac.alt + ' ft' : 'N/A';
markers[icao].bindTooltip(`${callsign}<br>${alt}`, {
permanent: false,
direction: 'top',
className: 'aircraft-tooltip'
});
}
function getAltitudeColor(alt) {
if (!alt) return '#888888';
if (alt < 10000) return '#00ff88';
if (alt < 25000) return '#00d4ff';
if (alt < 35000) return '#ffcc00';
return '#ff9500';
}
function updateStats() {
const total = Object.keys(aircraft).length;
const withPos = Object.values(aircraft).filter(a => a.lat && a.lon).length;
const altitudes = Object.values(aircraft).map(a => a.alt || 0).filter(a => a > 0);
const maxAlt = altitudes.length ? Math.max(...altitudes) : 0;
const avgAlt = altitudes.length ? Math.round(altitudes.reduce((a, b) => a + b, 0) / altitudes.length) : 0;
document.getElementById('statTotal').textContent = total;
document.getElementById('statWithPos').textContent = withPos;
document.getElementById('statMaxAlt').textContent = maxAlt.toLocaleString();
document.getElementById('statAvgAlt').textContent = avgAlt.toLocaleString();
document.getElementById('aircraftCount').textContent = total;
}
function renderAircraftList() {
const container = document.getElementById('aircraftList');
const sortedAircraft = Object.values(aircraft)
.sort((a, b) => (b.alt || 0) - (a.alt || 0));
if (sortedAircraft.length === 0) {
container.innerHTML = `
<div class="no-aircraft">
<div>No aircraft detected</div>
<div style="font-size: 11px; margin-top: 5px;">Waiting for data...</div>
</div>
`;
return;
}
container.innerHTML = sortedAircraft.map(ac => {
const callsign = ac.callsign || '------';
const alt = ac.alt ? ac.alt.toLocaleString() : '---';
const speed = ac.speed || '---';
const heading = ac.heading ? ac.heading + '°' : '---';
return `
<div class="aircraft-item ${selectedIcao === ac.icao ? 'selected' : ''}"
onclick="selectAircraft('${ac.icao}')">
<div class="aircraft-header">
<span class="aircraft-callsign">${callsign}</span>
<span class="aircraft-icao">${ac.icao}</span>
</div>
<div class="aircraft-details">
<div class="aircraft-detail">
<div class="aircraft-detail-value">${alt}</div>
<div class="aircraft-detail-label">ALT ft</div>
</div>
<div class="aircraft-detail">
<div class="aircraft-detail-value">${speed}</div>
<div class="aircraft-detail-label">SPD kts</div>
</div>
<div class="aircraft-detail">
<div class="aircraft-detail-value">${heading}</div>
<div class="aircraft-detail-label">HDG</div>
</div>
</div>
</div>
`;
}).join('');
}
function selectAircraft(icao) {
selectedIcao = icao;
renderAircraftList();
showAircraftDetails(icao);
// Center map on aircraft
const ac = aircraft[icao];
if (ac && ac.lat && ac.lon) {
radarMap.setView([ac.lat, ac.lon], 10);
}
}
function showAircraftDetails(icao) {
const ac = aircraft[icao];
const container = document.getElementById('selectedInfo');
if (!ac) {
container.innerHTML = `
<div class="no-aircraft">
<div class="no-aircraft-icon">✈</div>
<div>Select an aircraft to view details</div>
</div>
`;
return;
}
const callsign = ac.callsign || ac.icao;
const lat = ac.lat ? ac.lat.toFixed(4) + '°' : 'N/A';
const lon = ac.lon ? ac.lon.toFixed(4) + '°' : 'N/A';
const alt = ac.alt ? ac.alt.toLocaleString() + ' ft' : 'N/A';
const speed = ac.speed ? ac.speed + ' kts' : 'N/A';
const heading = ac.heading ? ac.heading + '°' : 'N/A';
const squawk = ac.squawk || 'N/A';
const vRate = ac.vRate ? (ac.vRate > 0 ? '+' : '') + ac.vRate + ' ft/min' : 'N/A';
container.innerHTML = `
<div class="selected-callsign">${callsign}</div>
<div class="telemetry-grid">
<div class="telemetry-item">
<div class="telemetry-label">ICAO</div>
<div class="telemetry-value">${ac.icao}</div>
</div>
<div class="telemetry-item">
<div class="telemetry-label">Squawk</div>
<div class="telemetry-value">${squawk}</div>
</div>
<div class="telemetry-item">
<div class="telemetry-label">Latitude</div>
<div class="telemetry-value">${lat}</div>
</div>
<div class="telemetry-item">
<div class="telemetry-label">Longitude</div>
<div class="telemetry-value">${lon}</div>
</div>
<div class="telemetry-item">
<div class="telemetry-label">Altitude</div>
<div class="telemetry-value">${alt}</div>
</div>
<div class="telemetry-item">
<div class="telemetry-label">Speed</div>
<div class="telemetry-value">${speed}</div>
</div>
<div class="telemetry-item">
<div class="telemetry-label">Heading</div>
<div class="telemetry-value">${heading}</div>
</div>
<div class="telemetry-item">
<div class="telemetry-label">Vert Rate</div>
<div class="telemetry-value">${vRate}</div>
</div>
</div>
`;
}
function cleanupOldAircraft() {
const now = Date.now();
const timeout = 60000; // 60 seconds
Object.keys(aircraft).forEach(icao => {
if (now - aircraft[icao].lastSeen > timeout) {
// Remove marker
if (markers[icao]) {
radarMap.removeLayer(markers[icao]);
delete markers[icao];
}
// Remove aircraft
delete aircraft[icao];
// Clear selection if this was selected
if (selectedIcao === icao) {
selectedIcao = null;
showAircraftDetails(null);
}
}
});
updateStats();
renderAircraftList();
}
</script>
</body>
</html>

9221
templates/index.html Normal file
View File

File diff suppressed because it is too large Load Diff

1547
templates/satellite_dashboard.html Normal file
View File

File diff suppressed because it is too large Load Diff

0
tests/__init__.py Normal file
View File

19
tests/conftest.py Normal file
View File

@@ -0,0 +1,19 @@
"""Pytest configuration and fixtures."""
import pytest
from app import app as flask_app
from routes import register_blueprints
@pytest.fixture
def app():
"""Create application for testing."""
register_blueprints(flask_app)
flask_app.config['TESTING'] = True
return flask_app
@pytest.fixture
def client(app):
"""Create test client."""
return app.test_client()

39
tests/test_app.py Normal file
View File

@@ -0,0 +1,39 @@
"""Tests for main application routes."""
import pytest
def test_index_page(client):
"""Test that index page loads."""
response = client.get('/')
assert response.status_code == 200
assert b'INTERCEPT' in response.data
def test_dependencies_endpoint(client):
"""Test dependencies endpoint returns valid JSON."""
response = client.get('/dependencies')
assert response.status_code == 200
data = response.get_json()
assert 'modes' in data
assert 'os' in data
def test_devices_endpoint(client):
"""Test devices endpoint returns list."""
response = client.get('/devices')
assert response.status_code == 200
data = response.get_json()
assert isinstance(data, list)
def test_satellite_dashboard(client):
"""Test satellite dashboard loads."""
response = client.get('/satellite/dashboard')
assert response.status_code == 200
def test_adsb_dashboard(client):
"""Test ADS-B dashboard loads."""
response = client.get('/adsb/dashboard')
assert response.status_code == 200

48
tests/test_config.py Normal file
View File

@@ -0,0 +1,48 @@
"""Tests for configuration module."""
import os
import pytest
class TestConfigEnvVars:
"""Tests for environment variable configuration."""
def test_default_values(self):
"""Test that default values are set."""
from config import PORT, HOST, DEBUG
assert PORT == 5050
assert HOST == '0.0.0.0'
assert DEBUG is False
def test_env_override(self, monkeypatch):
"""Test that environment variables override defaults."""
monkeypatch.setenv('INTERCEPT_PORT', '8080')
monkeypatch.setenv('INTERCEPT_DEBUG', 'true')
# Re-import to get new values
import importlib
import config
importlib.reload(config)
assert config.PORT == 8080
assert config.DEBUG is True
# Reset
monkeypatch.delenv('INTERCEPT_PORT', raising=False)
monkeypatch.delenv('INTERCEPT_DEBUG', raising=False)
importlib.reload(config)
def test_invalid_env_values(self, monkeypatch):
"""Test that invalid env values fall back to defaults."""
monkeypatch.setenv('INTERCEPT_PORT', 'invalid')
import importlib
import config
importlib.reload(config)
# Should fall back to default
assert config.PORT == 5050
monkeypatch.delenv('INTERCEPT_PORT', raising=False)
importlib.reload(config)

73
tests/test_utils.py Normal file
View File

@@ -0,0 +1,73 @@
"""Tests for utility modules."""
import pytest
from utils.process import is_valid_mac, is_valid_channel
from utils.dependencies import check_tool
from data.oui import get_manufacturer
class TestMacValidation:
"""Tests for MAC address validation."""
def test_valid_mac(self):
"""Test valid MAC addresses."""
assert is_valid_mac('AA:BB:CC:DD:EE:FF') is True
assert is_valid_mac('aa:bb:cc:dd:ee:ff') is True
assert is_valid_mac('00:11:22:33:44:55') is True
def test_invalid_mac(self):
"""Test invalid MAC addresses."""
assert is_valid_mac('') is False
assert is_valid_mac(None) is False
assert is_valid_mac('invalid') is False
assert is_valid_mac('AA:BB:CC:DD:EE') is False
assert is_valid_mac('AA-BB-CC-DD-EE-FF') is False
class TestChannelValidation:
"""Tests for WiFi channel validation."""
def test_valid_channels(self):
"""Test valid channel numbers."""
assert is_valid_channel(1) is True
assert is_valid_channel(6) is True
assert is_valid_channel(11) is True
assert is_valid_channel('36') is True
assert is_valid_channel(149) is True
def test_invalid_channels(self):
"""Test invalid channel numbers."""
assert is_valid_channel(0) is False
assert is_valid_channel(-1) is False
assert is_valid_channel(201) is False
assert is_valid_channel(None) is False
assert is_valid_channel('invalid') is False
class TestToolCheck:
"""Tests for tool availability checking."""
def test_common_tools(self):
"""Test checking for common tools."""
# These should return bool, regardless of whether installed
assert isinstance(check_tool('ls'), bool)
assert isinstance(check_tool('nonexistent_tool_12345'), bool)
def test_nonexistent_tool(self):
"""Test that nonexistent tools return False."""
assert check_tool('nonexistent_tool_xyz_12345') is False
class TestOuiLookup:
"""Tests for OUI manufacturer lookup."""
def test_known_manufacturer(self):
"""Test looking up known manufacturers."""
# Apple prefix
result = get_manufacturer('00:25:DB:AA:BB:CC')
assert result == 'Apple' or result == 'Unknown'
def test_unknown_manufacturer(self):
"""Test looking up unknown manufacturer."""
result = get_manufacturer('FF:FF:FF:FF:FF:FF')
assert result == 'Unknown'

14
utils/__init__.py Normal file
View File

@@ -0,0 +1,14 @@
# Utility modules for INTERCEPT
from .dependencies import check_tool, check_all_dependencies, TOOL_DEPENDENCIES
from .process import cleanup_stale_processes, is_valid_mac, is_valid_channel, detect_devices
from .logging import (
get_logger,
app_logger,
pager_logger,
sensor_logger,
wifi_logger,
bluetooth_logger,
adsb_logger,
satellite_logger,
iridium_logger,
)

246
utils/dependencies.py Normal file
View File

@@ -0,0 +1,246 @@
from __future__ import annotations
import logging
import shutil
from typing import Any
logger = logging.getLogger('intercept.dependencies')
def check_tool(name: str) -> bool:
"""Check if a tool is installed."""
return shutil.which(name) is not None
# Comprehensive tool dependency definitions
TOOL_DEPENDENCIES = {
'pager': {
'name': 'Pager Decoding',
'tools': {
'rtl_fm': {
'required': True,
'description': 'RTL-SDR FM demodulator',
'install': {
'apt': 'sudo apt install rtl-sdr',
'brew': 'brew install librtlsdr',
'manual': 'https://osmocom.org/projects/rtl-sdr/wiki'
}
},
'multimon-ng': {
'required': True,
'description': 'Digital transmission decoder',
'install': {
'apt': 'sudo apt install multimon-ng',
'brew': 'brew install multimon-ng',
'manual': 'https://github.com/EliasOewornal/multimon-ng'
}
},
'rtl_test': {
'required': False,
'description': 'RTL-SDR device detection',
'install': {
'apt': 'sudo apt install rtl-sdr',
'brew': 'brew install librtlsdr',
'manual': 'https://osmocom.org/projects/rtl-sdr/wiki'
}
}
}
},
'sensor': {
'name': '433MHz Sensors',
'tools': {
'rtl_433': {
'required': True,
'description': 'ISM band decoder for sensors, weather stations, TPMS',
'install': {
'apt': 'sudo apt install rtl-433',
'brew': 'brew install rtl_433',
'manual': 'https://github.com/merbanan/rtl_433'
}
}
}
},
'wifi': {
'name': 'WiFi Reconnaissance',
'tools': {
'airmon-ng': {
'required': True,
'description': 'Monitor mode controller',
'install': {
'apt': 'sudo apt install aircrack-ng',
'brew': 'Not available on macOS',
'manual': 'https://aircrack-ng.org'
}
},
'airodump-ng': {
'required': True,
'description': 'WiFi network scanner',
'install': {
'apt': 'sudo apt install aircrack-ng',
'brew': 'Not available on macOS',
'manual': 'https://aircrack-ng.org'
}
},
'aireplay-ng': {
'required': False,
'description': 'Deauthentication / packet injection',
'install': {
'apt': 'sudo apt install aircrack-ng',
'brew': 'Not available on macOS',
'manual': 'https://aircrack-ng.org'
}
},
'aircrack-ng': {
'required': False,
'description': 'Handshake verification',
'install': {
'apt': 'sudo apt install aircrack-ng',
'brew': 'brew install aircrack-ng',
'manual': 'https://aircrack-ng.org'
}
},
'hcxdumptool': {
'required': False,
'description': 'PMKID capture tool',
'install': {
'apt': 'sudo apt install hcxdumptool',
'brew': 'brew install hcxtools',
'manual': 'https://github.com/ZerBea/hcxdumptool'
}
},
'hcxpcapngtool': {
'required': False,
'description': 'PMKID hash extractor',
'install': {
'apt': 'sudo apt install hcxtools',
'brew': 'brew install hcxtools',
'manual': 'https://github.com/ZerBea/hcxtools'
}
}
}
},
'bluetooth': {
'name': 'Bluetooth Scanning',
'tools': {
'hcitool': {
'required': False,
'description': 'Bluetooth HCI tool (legacy)',
'install': {
'apt': 'sudo apt install bluez',
'brew': 'Not available on macOS (use native)',
'manual': 'http://www.bluez.org'
}
},
'bluetoothctl': {
'required': True,
'description': 'Modern Bluetooth controller',
'install': {
'apt': 'sudo apt install bluez',
'brew': 'Not available on macOS (use native)',
'manual': 'http://www.bluez.org'
}
},
'hciconfig': {
'required': False,
'description': 'Bluetooth adapter configuration',
'install': {
'apt': 'sudo apt install bluez',
'brew': 'Not available on macOS',
'manual': 'http://www.bluez.org'
}
}
}
},
'aircraft': {
'name': 'Aircraft Tracking (ADS-B)',
'tools': {
'dump1090': {
'required': False,
'description': 'Mode S / ADS-B decoder (preferred)',
'install': {
'apt': 'sudo apt install dump1090-mutability',
'brew': 'brew install dump1090-mutability',
'manual': 'https://github.com/flightaware/dump1090'
},
'alternatives': ['dump1090-mutability', 'dump1090-fa']
},
'rtl_adsb': {
'required': False,
'description': 'Simple ADS-B decoder',
'install': {
'apt': 'sudo apt install rtl-sdr',
'brew': 'brew install librtlsdr',
'manual': 'https://osmocom.org/projects/rtl-sdr/wiki'
}
}
}
},
'satellite': {
'name': 'Satellite Tracking',
'tools': {
'skyfield': {
'required': True,
'description': 'Python orbital mechanics library',
'install': {
'pip': 'pip install skyfield',
'manual': 'https://rhodesmill.org/skyfield/'
},
'python_module': True
}
}
},
'iridium': {
'name': 'Iridium Monitoring',
'tools': {
'iridium-extractor': {
'required': False,
'description': 'Iridium burst extractor',
'install': {
'manual': 'https://github.com/muccc/gr-iridium'
}
}
}
}
}
def check_all_dependencies() -> dict[str, dict[str, Any]]:
"""Check all tool dependencies and return status."""
results: dict[str, dict[str, Any]] = {}
for mode, config in TOOL_DEPENDENCIES.items():
mode_result = {
'name': config['name'],
'tools': {},
'ready': True,
'missing_required': []
}
for tool, tool_config in config['tools'].items():
# Check if it's a Python module
if tool_config.get('python_module'):
try:
__import__(tool)
installed = True
except Exception as e:
logger.debug(f"Failed to import {tool}: {type(e).__name__}: {e}")
installed = False
else:
# Check for alternatives
alternatives = tool_config.get('alternatives', [])
installed = check_tool(tool) or any(check_tool(alt) for alt in alternatives)
mode_result['tools'][tool] = {
'installed': installed,
'required': tool_config['required'],
'description': tool_config['description'],
'install': tool_config['install']
}
if tool_config['required'] and not installed:
mode_result['ready'] = False
mode_result['missing_required'].append(tool)
results[mode] = mode_result
return results

30
utils/logging.py Normal file
View File

@@ -0,0 +1,30 @@
"""Logging utilities for intercept application."""
from __future__ import annotations
import logging
import sys
from config import LOG_LEVEL, LOG_FORMAT
def get_logger(name: str) -> logging.Logger:
"""Get a configured logger for a module."""
logger = logging.getLogger(name)
if not logger.handlers:
handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(logging.Formatter(LOG_FORMAT))
logger.addHandler(handler)
logger.setLevel(LOG_LEVEL)
return logger
# Pre-configured loggers for each module
app_logger = get_logger('intercept')
pager_logger = get_logger('intercept.pager')
sensor_logger = get_logger('intercept.sensor')
wifi_logger = get_logger('intercept.wifi')
bluetooth_logger = get_logger('intercept.bluetooth')
adsb_logger = get_logger('intercept.adsb')
satellite_logger = get_logger('intercept.satellite')
iridium_logger = get_logger('intercept.iridium')

80
utils/process.py Normal file
View File

@@ -0,0 +1,80 @@
from __future__ import annotations
import subprocess
import re
from typing import Any
from .dependencies import check_tool
def cleanup_stale_processes() -> None:
"""Kill any stale processes from previous runs (but not system services)."""
# Note: dump1090 is NOT included here as users may run it as a system service
processes_to_kill = ['rtl_adsb', 'rtl_433', 'multimon-ng', 'rtl_fm']
for proc_name in processes_to_kill:
try:
subprocess.run(['pkill', '-9', proc_name], capture_output=True)
except (subprocess.SubprocessError, OSError):
pass
def is_valid_mac(mac: str | None) -> bool:
"""Validate MAC address format."""
if not mac:
return False
return bool(re.match(r'^([0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}$', mac))
def is_valid_channel(channel: str | int | None) -> bool:
"""Validate WiFi channel number."""
try:
ch = int(channel) # type: ignore[arg-type]
return 1 <= ch <= 200
except (ValueError, TypeError):
return False
def detect_devices() -> list[dict[str, Any]]:
"""Detect RTL-SDR devices."""
devices: list[dict[str, Any]] = []
if not check_tool('rtl_test'):
return devices
try:
result = subprocess.run(
['rtl_test', '-t'],
capture_output=True,
text=True,
timeout=5
)
output = result.stderr + result.stdout
# Parse device info
device_pattern = r'(\d+):\s+(.+?)(?:,\s*SN:\s*(\S+))?$'
for line in output.split('\n'):
line = line.strip()
match = re.match(device_pattern, line)
if match:
devices.append({
'index': int(match.group(1)),
'name': match.group(2).strip().rstrip(','),
'serial': match.group(3) or 'N/A'
})
if not devices:
found_match = re.search(r'Found (\d+) device', output)
if found_match:
count = int(found_match.group(1))
for i in range(count):
devices.append({
'index': i,
'name': f'RTL-SDR Device {i}',
'serial': 'Unknown'
})
except Exception:
pass
return devices