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

perf: skip tracker signature scan when BLE payload fingerprint is unchanged

Browse Source
This commit is contained in:
James Smith
2026-05-19 13:14:16 +01:00
parent 0e0e17b089
commit 260240728a
2 changed files with 85 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_bluetooth_aggregator.py
+33
View File
@@ -1,6 +1,8 @@
"""Unit tests for Bluetooth device aggregation."""
import dataclasses
from datetime import datetime, timedelta
from unittest.mock import patch
import pytest
@@ -554,3 +556,34 @@ class TestDeviceFiltering:
devices = aggregator.get_all_devices(sort_by="rssi")
rssi_values = [d.rssi_current for d in devices]
assert rssi_values == [-50, -60, -70, -90]
class TestTrackerDetectionOptimization:
"""Tests for tracker detection payload fingerprint optimization."""
def test_tracker_detection_skipped_when_payload_unchanged(self, aggregator, sample_observation):
"""detect_tracker must not be called on the second observation if payload is identical."""
aggregator.ingest(sample_observation)
with patch.object(
aggregator._tracker_engine, "detect_tracker", wraps=aggregator._tracker_engine.detect_tracker
) as mock_detect:
aggregator.ingest(sample_observation)
assert mock_detect.call_count == 0, (
"detect_tracker should not be called when the device payload fingerprint is unchanged"
)
def test_tracker_detection_runs_when_payload_changes(self, aggregator, sample_observation):
"""detect_tracker must be called again when the device's manufacturer data changes."""
aggregator.ingest(sample_observation)
changed = dataclasses.replace(
sample_observation,
manufacturer_data=bytes([0xDE, 0xAD, 0xBE, 0xEF]),
)
with patch.object(
aggregator._tracker_engine, "detect_tracker", wraps=aggregator._tracker_engine.detect_tracker
) as mock_detect:
aggregator.ingest(changed)
assert mock_detect.call_count == 1, "detect_tracker must be called when manufacturer data changes"
utils/bluetooth/aggregator.py
+52 -52
View File
@@ -114,7 +114,7 @@ class DeviceAggregator:
device.rssi_samples.append((observation.timestamp, observation.rssi))
# Prune old samples
if len(device.rssi_samples) > self._max_rssi_samples:
device.rssi_samples = device.rssi_samples[-self._max_rssi_samples:]
device.rssi_samples = device.rssi_samples[-self._max_rssi_samples :]
# Recalculate RSSI statistics
self._update_rssi_stats(device)
@@ -189,7 +189,7 @@ class DeviceAggregator:
return PROTOCOL_CLASSIC
# If address type is anything other than public, likely BLE
if observation.address_type != 'public':
if observation.address_type != "public":
return PROTOCOL_BLE
# If service UUIDs are present with 16-bit format, likely BLE
@@ -283,7 +283,7 @@ class DeviceAggregator:
def _update_proximity(self, device: BTDeviceAggregate) -> None:
"""Update proximity estimation for a device."""
if device.rssi_ema is None:
device.proximity_band = 'unknown'
device.proximity_band = "unknown"
device.estimated_distance_m = None
device.distance_confidence = 0.0
return
@@ -311,14 +311,33 @@ class DeviceAggregator:
observation: BTObservation,
) -> None:
"""Run tracker signature detection on a device."""
# Prepare service data from observation if available
service_data = observation.service_data if observation.service_data else {}
# Store service data on device for investigation
for uuid, data in service_data.items():
device.service_data[uuid] = data
# Run tracker detection
# Generate fingerprint first — cheap hash of stable payload features.
fingerprint = self._tracker_engine.generate_device_fingerprint(
manufacturer_id=device.manufacturer_id,
manufacturer_data=device.manufacturer_bytes,
service_uuids=device.service_uuids,
service_data=service_data,
tx_power=device.tx_power,
name=device.name,
)
# Track fingerprint → device mapping regardless of whether we re-scan.
if fingerprint.fingerprint_id not in self._fingerprint_to_devices:
self._fingerprint_to_devices[fingerprint.fingerprint_id] = set()
self._fingerprint_to_devices[fingerprint.fingerprint_id].add(device.device_id)
# Record sighting for persistence tracking.
self._tracker_engine.record_sighting(fingerprint.fingerprint_id)
# Only re-run the expensive signature scan when the payload has changed.
if fingerprint.fingerprint_id == device.payload_fingerprint_id:
return
result = self._tracker_engine.detect_tracker(
address=device.address,
address_type=device.address_type,
@@ -330,34 +349,15 @@ class DeviceAggregator:
tx_power=device.tx_power,
)
# Update device with detection results
device.is_tracker = result.is_tracker
device.tracker_type = result.tracker_type.value if result.tracker_type else None
device.tracker_name = result.tracker_name
device.tracker_confidence = result.confidence.value if result.confidence else None
device.tracker_confidence_score = result.confidence_score
device.tracker_evidence = result.evidence
# Generate and store payload fingerprint
fingerprint = self._tracker_engine.generate_device_fingerprint(
manufacturer_id=device.manufacturer_id,
manufacturer_data=device.manufacturer_bytes,
service_uuids=device.service_uuids,
service_data=service_data,
tx_power=device.tx_power,
name=device.name,
)
device.payload_fingerprint_id = fingerprint.fingerprint_id
device.payload_fingerprint_stability = fingerprint.stability_confidence
# Track fingerprint to device mapping
if fingerprint.fingerprint_id not in self._fingerprint_to_devices:
self._fingerprint_to_devices[fingerprint.fingerprint_id] = set()
self._fingerprint_to_devices[fingerprint.fingerprint_id].add(device.device_id)
# Record sighting for persistence tracking
self._tracker_engine.record_sighting(fingerprint.fingerprint_id)
def _update_risk_analysis(self, device: BTDeviceAggregate) -> None:
"""Evaluate suspicious presence heuristics for a device."""
if not device.payload_fingerprint_id:
@@ -386,8 +386,7 @@ class DeviceAggregator:
if observation.manufacturer_id is not None:
device.manufacturer_id = observation.manufacturer_id
device.manufacturer_name = MANUFACTURER_NAMES.get(
observation.manufacturer_id,
f"Unknown (0x{observation.manufacturer_id:04X})"
observation.manufacturer_id, f"Unknown (0x{observation.manufacturer_id:04X})"
)
if observation.manufacturer_data:
device.manufacturer_bytes = observation.manufacturer_data
@@ -437,10 +436,7 @@ class DeviceAggregator:
"""
cutoff = datetime.now() - timedelta(seconds=max_age_seconds)
with self._lock:
stale_ids = [
device_id for device_id, device in self._devices.items()
if device.last_seen < cutoff
]
stale_ids = [device_id for device_id, device in self._devices.items() if device.last_seen < cutoff]
for device_id in stale_ids:
del self._devices[device_id]
return len(stale_ids)
@@ -544,7 +540,7 @@ class DeviceAggregator:
top_n: int = 20,
window_minutes: int = 10,
bucket_seconds: int = 10,
sort_by: str = 'recency',
sort_by: str = "recency",
) -> dict:
"""
Get heatmap data for visualization.
@@ -568,37 +564,41 @@ class DeviceAggregator:
# Enrich with device metadata
result = {
'window_minutes': window_minutes,
'bucket_seconds': bucket_seconds,
'devices': [],
"window_minutes": window_minutes,
"bucket_seconds": bucket_seconds,
"devices": [],
}
with self._lock:
for device_key, ts_data in timeseries.items():
device = self.get_device_by_key(device_key)
device_info = {
'device_key': device_key,
'timeseries': ts_data,
"device_key": device_key,
"timeseries": ts_data,
}
if device:
device_info.update({
'name': device.name,
'address': device.address,
'rssi_current': device.rssi_current,
'rssi_ema': round(device.rssi_ema, 1) if device.rssi_ema else None,
'proximity_band': device.proximity_band,
})
device_info.update(
{
"name": device.name,
"address": device.address,
"rssi_current": device.rssi_current,
"rssi_ema": round(device.rssi_ema, 1) if device.rssi_ema else None,
"proximity_band": device.proximity_band,
}
)
else:
device_info.update({
'name': None,
'address': None,
'rssi_current': None,
'rssi_ema': None,
'proximity_band': 'unknown',
})
device_info.update(
{
"name": None,
"address": None,
"rssi_current": None,
"rssi_ema": None,
"proximity_band": "unknown",
}
)
result['devices'].append(device_info)
result["devices"].append(device_info)
return result