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

fix: SDR device registry collision with multiple SDR types

Browse Source 
The registry used plain int keys (device index), so HackRF at index 0
and RTL-SDR at index 0 would collide. Changed to composite string keys
("sdr_type:index") so each SDR type+index pair is tracked independently.
Updated all route callers, frontend device selectors, and session restore.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Smittix
2026-02-27 09:06:41 +00:00
parent 0d13638d70
commit 5aa68a49c6
17 changed files with 907 additions and 786 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+26 -20
View File
@@ -257,12 +257,12 @@ cleanup_manager.register(deauth_alerts)
# SDR DEVICE REGISTRY
# ============================================
# Tracks which mode is using which SDR device to prevent conflicts
# Key: device_index (int), Value: mode_name (str)
sdr_device_registry: dict[int, str] = {}
# Key: "sdr_type:device_index" (str), Value: mode_name (str)
sdr_device_registry: dict[str, str] = {}
sdr_device_registry_lock = threading.Lock()
def claim_sdr_device(device_index: int, mode_name: str) -> str | None:
def claim_sdr_device(device_index: int, mode_name: str, sdr_type: str = 'rtlsdr') -> str | None:
"""Claim an SDR device for a mode.
Checks the in-app registry first, then probes the USB device to
@@ -272,43 +272,48 @@ def claim_sdr_device(device_index: int, mode_name: str) -> str | None:
Args:
device_index: The SDR device index to claim
mode_name: Name of the mode claiming the device (e.g., 'sensor', 'rtlamr')
sdr_type: SDR type string (e.g., 'rtlsdr', 'hackrf', 'limesdr')
Returns:
Error message if device is in use, None if successfully claimed
"""
key = f"{sdr_type}:{device_index}"
with sdr_device_registry_lock:
if device_index in sdr_device_registry:
in_use_by = sdr_device_registry[device_index]
return f'SDR device {device_index} is in use by {in_use_by}. Stop {in_use_by} first or use a different device.'
if key in sdr_device_registry:
in_use_by = sdr_device_registry[key]
return f'SDR device {sdr_type}:{device_index} is in use by {in_use_by}. Stop {in_use_by} first or use a different device.'
# Probe the USB device to catch external processes holding the handle
try:
from utils.sdr.detection import probe_rtlsdr_device
usb_error = probe_rtlsdr_device(device_index)
if usb_error:
return usb_error
except Exception:
pass # If probe fails, let the caller proceed normally
if sdr_type == 'rtlsdr':
try:
from utils.sdr.detection import probe_rtlsdr_device
usb_error = probe_rtlsdr_device(device_index)
if usb_error:
return usb_error
except Exception:
pass # If probe fails, let the caller proceed normally
sdr_device_registry[device_index] = mode_name
sdr_device_registry[key] = mode_name
return None
def release_sdr_device(device_index: int) -> None:
def release_sdr_device(device_index: int, sdr_type: str = 'rtlsdr') -> None:
"""Release an SDR device from the registry.
Args:
device_index: The SDR device index to release
sdr_type: SDR type string (e.g., 'rtlsdr', 'hackrf', 'limesdr')
"""
key = f"{sdr_type}:{device_index}"
with sdr_device_registry_lock:
sdr_device_registry.pop(device_index, None)
sdr_device_registry.pop(key, None)
def get_sdr_device_status() -> dict[int, str]:
def get_sdr_device_status() -> dict[str, str]:
"""Get current SDR device allocations.
Returns:
Dictionary mapping device indices to mode names
Dictionary mapping 'sdr_type:device_index' keys to mode names
"""
with sdr_device_registry_lock:
return dict(sdr_device_registry)
@@ -429,8 +434,9 @@ def get_devices_status() -> Response:
result = []
for device in devices:
d = device.to_dict()
d['in_use'] = device.index in registry
d['used_by'] = registry.get(device.index)
key = f"{device.sdr_type.value}:{device.index}"
d['in_use'] = key in registry
d['used_by'] = registry.get(key)
result.append(d)
return jsonify(result)