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

Remove GSM spy functionality for legal compliance

Browse Source 
Remove all GSM cellular intelligence features including tower scanning,
signal monitoring, rogue detection, crowd density analysis, and
OpenCellID integration across routes, templates, utils, tests, and
build configuration.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Smittix
2026-02-08 22:04:12 +00:00
parent 2bed35dd64
commit c2891938ab
17 changed files with 21 additions and 6516 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/constants.py
-10
View File
@@ -275,13 +275,3 @@ MAX_DEAUTH_ALERTS_AGE_SECONDS = 300 # 5 minutes
# Deauth detector sniff timeout (seconds)
DEAUTH_SNIFF_TIMEOUT = 0.5
# =============================================================================
# GSM SPY (Cellular Intelligence)
# =============================================================================
# Maximum age for GSM tower/device data in DataStore (seconds)
MAX_GSM_AGE_SECONDS = 300 # 5 minutes
# Timing Advance conversion to meters
GSM_TA_METERS_PER_UNIT = 554
utils/database.py
-185
View File
@@ -453,134 +453,6 @@ def init_db() -> None:
ON tscm_cases(status, created_at)
''')
# =====================================================================
# GSM (Global System for Mobile) Intelligence Tables
# =====================================================================
# gsm_cells - Known cell towers (OpenCellID cache)
conn.execute('''
CREATE TABLE IF NOT EXISTS gsm_cells (
id INTEGER PRIMARY KEY AUTOINCREMENT,
mcc INTEGER NOT NULL,
mnc INTEGER NOT NULL,
lac INTEGER NOT NULL,
cid INTEGER NOT NULL,
lat REAL,
lon REAL,
azimuth INTEGER,
range_meters INTEGER,
samples INTEGER,
radio TEXT,
operator TEXT,
first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_verified TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
metadata TEXT,
UNIQUE(mcc, mnc, lac, cid)
)
''')
# gsm_rogues - Detected rogue towers / IMSI catchers
conn.execute('''
CREATE TABLE IF NOT EXISTS gsm_rogues (
id INTEGER PRIMARY KEY AUTOINCREMENT,
arfcn INTEGER NOT NULL,
mcc INTEGER,
mnc INTEGER,
lac INTEGER,
cid INTEGER,
signal_strength REAL,
reason TEXT NOT NULL,
threat_level TEXT DEFAULT 'medium',
detected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
location_lat REAL,
location_lon REAL,
acknowledged BOOLEAN DEFAULT 0,
notes TEXT,
metadata TEXT
)
''')
# gsm_signals - 60-day archive of signal observations
conn.execute('''
CREATE TABLE IF NOT EXISTS gsm_signals (
id INTEGER PRIMARY KEY AUTOINCREMENT,
imsi TEXT,
tmsi TEXT,
mcc INTEGER,
mnc INTEGER,
lac INTEGER,
cid INTEGER,
ta_value INTEGER,
signal_strength REAL,
arfcn INTEGER,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
metadata TEXT
)
''')
# gsm_tmsi_log - 24-hour raw pings for crowd density
conn.execute('''
CREATE TABLE IF NOT EXISTS gsm_tmsi_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
tmsi TEXT NOT NULL,
lac INTEGER,
cid INTEGER,
ta_value INTEGER,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# gsm_velocity_log - 1-hour buffer for movement tracking
conn.execute('''
CREATE TABLE IF NOT EXISTS gsm_velocity_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
device_id TEXT NOT NULL,
prev_ta INTEGER,
curr_ta INTEGER,
prev_cid INTEGER,
curr_cid INTEGER,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
estimated_velocity REAL,
metadata TEXT
)
''')
# GSM indexes for performance
conn.execute('''
CREATE INDEX IF NOT EXISTS idx_gsm_cells_location
ON gsm_cells(lat, lon)
''')
conn.execute('''
CREATE INDEX IF NOT EXISTS idx_gsm_cells_identity
ON gsm_cells(mcc, mnc, lac, cid)
''')
conn.execute('''
CREATE INDEX IF NOT EXISTS idx_gsm_rogues_severity
ON gsm_rogues(threat_level, detected_at)
''')
conn.execute('''
CREATE INDEX IF NOT EXISTS idx_gsm_signals_cell_time
ON gsm_signals(cid, lac, timestamp)
''')
conn.execute('''
CREATE INDEX IF NOT EXISTS idx_gsm_signals_device
ON gsm_signals(imsi, tmsi, timestamp)
''')
conn.execute('''
CREATE INDEX IF NOT EXISTS idx_gsm_tmsi_log_time
ON gsm_tmsi_log(timestamp)
''')
conn.execute('''
CREATE INDEX IF NOT EXISTS idx_gsm_velocity_log_device
ON gsm_velocity_log(device_id, timestamp)
''')
# =====================================================================
# DSC (Digital Selective Calling) Tables
# =====================================================================
@@ -2298,60 +2170,3 @@ def cleanup_old_payloads(max_age_hours: int = 24) -> int:
''', (f'-{max_age_hours} hours',))
return cursor.rowcount
# =============================================================================
# GSM Cleanup Functions
# =============================================================================
def cleanup_old_gsm_signals(max_age_days: int = 60) -> int:
"""
Remove old GSM signal observations (60-day archive).
Args:
max_age_days: Maximum age in days (default: 60)
Returns:
Number of deleted entries
"""
with get_db() as conn:
cursor = conn.execute('''
DELETE FROM gsm_signals
WHERE timestamp < datetime('now', ?)
''', (f'-{max_age_days} days',))
return cursor.rowcount
def cleanup_old_gsm_tmsi_log(max_age_hours: int = 24) -> int:
"""
Remove old TMSI log entries (24-hour buffer for crowd density).
Args:
max_age_hours: Maximum age in hours (default: 24)
Returns:
Number of deleted entries
"""
with get_db() as conn:
cursor = conn.execute('''
DELETE FROM gsm_tmsi_log
WHERE timestamp < datetime('now', ?)
''', (f'-{max_age_hours} hours',))
return cursor.rowcount
def cleanup_old_gsm_velocity_log(max_age_hours: int = 1) -> int:
"""
Remove old velocity log entries (1-hour buffer for movement tracking).
Args:
max_age_hours: Maximum age in hours (default: 1)
Returns:
Number of deleted entries
"""
with get_db() as conn:
cursor = conn.execute('''
DELETE FROM gsm_velocity_log
WHERE timestamp < datetime('now', ?)
''', (f'-{max_age_hours} hours',))
return cursor.rowcount
utils/dependencies.py
-32
View File
@@ -444,38 +444,6 @@ TOOL_DEPENDENCIES = {
}
}
},
'gsm': {
'name': 'GSM Intelligence',
'tools': {
'grgsm_scanner': {
'required': True,
'description': 'gr-gsm scanner for finding GSM towers',
'install': {
'apt': 'Build gr-gsm from source: https://github.com/bkerler/gr-gsm',
'brew': 'brew install gr-gsm (may require manual build)',
'manual': 'https://github.com/bkerler/gr-gsm'
}
},
'grgsm_livemon': {
'required': True,
'description': 'gr-gsm live monitor for decoding GSM signals',
'install': {
'apt': 'Included with gr-gsm package',
'brew': 'Included with gr-gsm',
'manual': 'Included with gr-gsm'
}
},
'tshark': {
'required': True,
'description': 'Wireshark CLI for parsing GSM packets',
'install': {
'apt': 'sudo apt-get install tshark',
'brew': 'brew install wireshark',
'manual': 'https://www.wireshark.org/download.html'
}
}
}
}
}
utils/gsm_geocoding.py
-226
View File
@@ -1,226 +0,0 @@
"""GSM Cell Tower Geocoding Service.
Provides hybrid cache-first geocoding with async API fallback for cell towers.
"""
from __future__ import annotations
import logging
import queue
from typing import Any
import requests
import config
from utils.database import get_db
logger = logging.getLogger('intercept.gsm_geocoding')
# Queue for pending geocoding requests
_geocoding_queue = queue.Queue(maxsize=100)
def lookup_cell_coordinates(mcc: int, mnc: int, lac: int, cid: int) -> dict[str, Any] | None:
"""
Lookup cell tower coordinates with cache-first strategy.
Strategy:
1. Check gsm_cells table (cache) - fast synchronous lookup
2. If not found, return None (caller decides whether to use API)
Args:
mcc: Mobile Country Code
mnc: Mobile Network Code
lac: Location Area Code
cid: Cell ID
Returns:
dict with keys: lat, lon, source='cache', azimuth (optional),
range_meters (optional), operator (optional), radio (optional)
Returns None if not found in cache.
"""
try:
with get_db() as conn:
result = conn.execute('''
SELECT lat, lon, azimuth, range_meters, operator, radio
FROM gsm_cells
WHERE mcc = ? AND mnc = ? AND lac = ? AND cid = ?
''', (mcc, mnc, lac, cid)).fetchone()
if result and result['lat'] is not None and result['lon'] is not None:
return {
'lat': result['lat'],
'lon': result['lon'],
'source': 'cache',
'azimuth': result['azimuth'],
'range_meters': result['range_meters'],
'operator': result['operator'],
'radio': result['radio']
}
return None
except Exception as e:
logger.error(f"Error looking up coordinates from cache: {e}")
return None
def _get_api_key() -> str:
"""Get OpenCellID API key at runtime (env var first, then database)."""
env_key = config.GSM_OPENCELLID_API_KEY
if env_key:
return env_key
from utils.database import get_setting
return get_setting('gsm.opencellid.api_key', '')
def lookup_cell_from_api(mcc: int, mnc: int, lac: int, cid: int) -> dict[str, Any] | None:
"""
Lookup cell tower from OpenCellID API and cache result.
Args:
mcc: Mobile Country Code
mnc: Mobile Network Code
lac: Location Area Code
cid: Cell ID
Returns:
dict with keys: lat, lon, source='api', azimuth (optional),
range_meters (optional), operator (optional), radio (optional)
Returns None if API call fails or cell not found.
"""
try:
api_key = _get_api_key()
if not api_key:
logger.warning("OpenCellID API key not configured")
return None
api_url = config.GSM_OPENCELLID_API_URL
params = {
'key': api_key,
'mcc': mcc,
'mnc': mnc,
'lac': lac,
'cellid': cid,
'format': 'json'
}
response = requests.get(api_url, params=params, timeout=10)
if response.status_code == 200:
cell_data = response.json()
lat = cell_data.get('lat')
lon = cell_data.get('lon')
# Validate response has actual coordinates
if lat is None or lon is None:
logger.warning(
f"OpenCellID API returned 200 but no coordinates for "
f"MCC={mcc} MNC={mnc} LAC={lac} CID={cid}: {cell_data}"
)
return None
# Cache the result
with get_db() as conn:
conn.execute('''
INSERT OR REPLACE INTO gsm_cells
(mcc, mnc, lac, cid, lat, lon, azimuth, range_meters, samples, radio, operator, last_verified)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
''', (
mcc, mnc, lac, cid,
lat, lon,
cell_data.get('azimuth'),
cell_data.get('range'),
cell_data.get('samples'),
cell_data.get('radio'),
cell_data.get('operator')
))
conn.commit()
logger.info(f"Cached cell tower from API: MCC={mcc} MNC={mnc} LAC={lac} CID={cid} -> ({lat}, {lon})")
return {
'lat': lat,
'lon': lon,
'source': 'api',
'azimuth': cell_data.get('azimuth'),
'range_meters': cell_data.get('range'),
'operator': cell_data.get('operator'),
'radio': cell_data.get('radio')
}
else:
logger.warning(
f"OpenCellID API returned {response.status_code} for "
f"MCC={mcc} MNC={mnc} LAC={lac} CID={cid}: {response.text[:200]}"
)
return None
except Exception as e:
logger.error(f"Error calling OpenCellID API: {e}")
return None
def enrich_tower_data(tower_data: dict[str, Any]) -> dict[str, Any]:
"""
Enrich tower data with coordinates using cache-first strategy.
If coordinates found in cache, adds them immediately.
If not found, marks as 'pending' and queues for background API lookup.
Args:
tower_data: Dictionary with keys mcc, mnc, lac, cid (and other tower data)
Returns:
Enriched tower_data dict with added fields:
- lat, lon (if found in cache)
- status='pending' (if needs API lookup)
- source='cache' (if from cache)
"""
mcc = tower_data.get('mcc')
mnc = tower_data.get('mnc')
lac = tower_data.get('lac')
cid = tower_data.get('cid')
# Validate required fields
if not all([mcc is not None, mnc is not None, lac is not None, cid is not None]):
logger.warning(f"Tower data missing required fields: {tower_data}")
return tower_data
# Try cache lookup
coords = lookup_cell_coordinates(mcc, mnc, lac, cid)
if coords:
# Found in cache - add coordinates immediately
tower_data['lat'] = coords['lat']
tower_data['lon'] = coords['lon']
tower_data['source'] = 'cache'
# Add optional fields if available
if coords.get('azimuth') is not None:
tower_data['azimuth'] = coords['azimuth']
if coords.get('range_meters') is not None:
tower_data['range_meters'] = coords['range_meters']
if coords.get('operator'):
tower_data['operator'] = coords['operator']
if coords.get('radio'):
tower_data['radio'] = coords['radio']
logger.debug(f"Cache hit for tower: MCC={mcc} MNC={mnc} LAC={lac} CID={cid}")
else:
# Not in cache - mark as pending and queue for API lookup
tower_data['status'] = 'pending'
tower_data['source'] = 'unknown'
# Queue for background geocoding (non-blocking)
try:
_geocoding_queue.put_nowait(tower_data.copy())
logger.debug(f"Queued tower for geocoding: MCC={mcc} MNC={mnc} LAC={lac} CID={cid}")
except queue.Full:
logger.warning("Geocoding queue full, dropping tower")
return tower_data
def get_geocoding_queue() -> queue.Queue:
"""Get the geocoding queue for the background worker."""
return _geocoding_queue
utils/logging.py
-1
View File
@@ -28,4 +28,3 @@ wifi_logger = get_logger('intercept.wifi')
bluetooth_logger = get_logger('intercept.bluetooth')
adsb_logger = get_logger('intercept.adsb')
satellite_logger = get_logger('intercept.satellite')
gsm_spy_logger = get_logger('intercept.gsm_spy')