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

Add alerts/recording, WiFi/TSCM updates, optimize waterfall

Browse Source
This commit is contained in:
Smittix
2026-02-07 18:29:58 +00:00
parent 4bbc00b765
commit 86e4ba7e29
42 changed files with 2514 additions and 485 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
utils/tscm/advanced.py
View File

@@ -523,20 +523,22 @@ class BaselineDiff:
}
def calculate_baseline_diff(
baseline: dict,
current_wifi: list[dict],
current_bt: list[dict],
current_rf: list[dict],
sweep_id: int
) -> BaselineDiff:
def calculate_baseline_diff(
baseline: dict,
current_wifi: list[dict],
current_wifi_clients: list[dict],
current_bt: list[dict],
current_rf: list[dict],
sweep_id: int
) -> BaselineDiff:
"""
Calculate comprehensive diff between baseline and current scan.
Args:
baseline: Baseline dict from database
current_wifi: Current WiFi devices
current_bt: Current Bluetooth devices
current_wifi_clients: Current WiFi clients
current_bt: Current Bluetooth devices
current_rf: Current RF signals
sweep_id: Current sweep ID
@@ -564,11 +566,16 @@ def calculate_baseline_diff(
diff.is_stale = diff.baseline_age_hours > 72
# Build baseline lookup dicts
baseline_wifi = {
d.get('bssid', d.get('mac', '')).upper(): d
for d in baseline.get('wifi_networks', [])
if d.get('bssid') or d.get('mac')
}
baseline_wifi = {
d.get('bssid', d.get('mac', '')).upper(): d
for d in baseline.get('wifi_networks', [])
if d.get('bssid') or d.get('mac')
}
baseline_wifi_clients = {
d.get('mac', d.get('address', '')).upper(): d
for d in baseline.get('wifi_clients', [])
if d.get('mac') or d.get('address')
}
baseline_bt = {
d.get('mac', d.get('address', '')).upper(): d
for d in baseline.get('bt_devices', [])
@@ -580,8 +587,11 @@ def calculate_baseline_diff(
if d.get('frequency')
}
# Compare WiFi
_compare_wifi(diff, baseline_wifi, current_wifi)
# Compare WiFi
_compare_wifi(diff, baseline_wifi, current_wifi)
# Compare WiFi clients
_compare_wifi_clients(diff, baseline_wifi_clients, current_wifi_clients)
# Compare Bluetooth
_compare_bluetooth(diff, baseline_bt, current_bt)
@@ -607,7 +617,7 @@ def calculate_baseline_diff(
return diff
def _compare_wifi(diff: BaselineDiff, baseline: dict, current: list[dict]) -> None:
def _compare_wifi(diff: BaselineDiff, baseline: dict, current: list[dict]) -> None:
"""Compare WiFi devices between baseline and current."""
current_macs = {
d.get('bssid', d.get('mac', '')).upper(): d
@@ -630,7 +640,48 @@ def _compare_wifi(diff: BaselineDiff, baseline: dict, current: list[dict]) -> No
'channel': device.get('channel'),
'rssi': device.get('power', device.get('signal')),
}
))
))
def _compare_wifi_clients(diff: BaselineDiff, baseline: dict, current: list[dict]) -> None:
"""Compare WiFi clients between baseline and current."""
current_macs = {
d.get('mac', d.get('address', '')).upper(): d
for d in current
if d.get('mac') or d.get('address')
}
# Find new clients
for mac, device in current_macs.items():
if mac not in baseline:
name = device.get('vendor', 'WiFi Client')
diff.new_devices.append(DeviceChange(
identifier=mac,
protocol='wifi_client',
change_type='new',
description=f'New WiFi client: {name}',
expected=False,
details={
'vendor': name,
'rssi': device.get('rssi'),
'associated_bssid': device.get('associated_bssid'),
}
))
# Find missing clients
for mac, device in baseline.items():
if mac not in current_macs:
name = device.get('vendor', 'WiFi Client')
diff.missing_devices.append(DeviceChange(
identifier=mac,
protocol='wifi_client',
change_type='missing',
description=f'Missing WiFi client: {name}',
expected=True,
details={
'vendor': name,
}
))
else:
# Check for changes
baseline_dev = baseline[mac]
@@ -796,11 +847,12 @@ def _calculate_baseline_health(diff: BaselineDiff, baseline: dict) -> None:
reasons.append(f"Baseline is {diff.baseline_age_hours:.0f} hours old")
# Device churn penalty
total_baseline = (
len(baseline.get('wifi_networks', [])) +
len(baseline.get('bt_devices', [])) +
len(baseline.get('rf_frequencies', []))
)
total_baseline = (
len(baseline.get('wifi_networks', [])) +
len(baseline.get('wifi_clients', [])) +
len(baseline.get('bt_devices', [])) +
len(baseline.get('rf_frequencies', []))
)
if total_baseline > 0:
churn_rate = (diff.total_new + diff.total_missing) / total_baseline

245
utils/tscm/baseline.py
View File

@@ -26,12 +26,13 @@ class BaselineRecorder:
Records and manages TSCM environment baselines.
"""
def __init__(self):
self.recording = False
self.current_baseline_id: int | None = None
self.wifi_networks: dict[str, dict] = {} # BSSID -> network info
self.bt_devices: dict[str, dict] = {} # MAC -> device info
self.rf_frequencies: dict[float, dict] = {} # Frequency -> signal info
def __init__(self):
self.recording = False
self.current_baseline_id: int | None = None
self.wifi_networks: dict[str, dict] = {} # BSSID -> network info
self.wifi_clients: dict[str, dict] = {} # MAC -> client info
self.bt_devices: dict[str, dict] = {} # MAC -> device info
self.rf_frequencies: dict[float, dict] = {} # Frequency -> signal info
def start_recording(
self,
@@ -50,10 +51,11 @@ class BaselineRecorder:
Returns:
Baseline ID
"""
self.recording = True
self.wifi_networks = {}
self.bt_devices = {}
self.rf_frequencies = {}
self.recording = True
self.wifi_networks = {}
self.wifi_clients = {}
self.bt_devices = {}
self.rf_frequencies = {}
# Create baseline in database
self.current_baseline_id = create_tscm_baseline(
@@ -78,24 +80,27 @@ class BaselineRecorder:
self.recording = False
# Convert to lists for storage
wifi_list = list(self.wifi_networks.values())
bt_list = list(self.bt_devices.values())
rf_list = list(self.rf_frequencies.values())
wifi_list = list(self.wifi_networks.values())
wifi_client_list = list(self.wifi_clients.values())
bt_list = list(self.bt_devices.values())
rf_list = list(self.rf_frequencies.values())
# Update database
update_tscm_baseline(
self.current_baseline_id,
wifi_networks=wifi_list,
bt_devices=bt_list,
rf_frequencies=rf_list
)
update_tscm_baseline(
self.current_baseline_id,
wifi_networks=wifi_list,
wifi_clients=wifi_client_list,
bt_devices=bt_list,
rf_frequencies=rf_list
)
summary = {
'baseline_id': self.current_baseline_id,
'wifi_count': len(wifi_list),
'bt_count': len(bt_list),
'rf_count': len(rf_list),
}
summary = {
'baseline_id': self.current_baseline_id,
'wifi_count': len(wifi_list),
'wifi_client_count': len(wifi_client_list),
'bt_count': len(bt_list),
'rf_count': len(rf_list),
}
logger.info(
f"Baseline recording complete: {summary['wifi_count']} WiFi, "
@@ -135,8 +140,8 @@ class BaselineRecorder:
'last_seen': datetime.now().isoformat(),
}
def add_bt_device(self, device: dict) -> None:
"""Add a Bluetooth device to the current baseline."""
def add_bt_device(self, device: dict) -> None:
"""Add a Bluetooth device to the current baseline."""
if not self.recording:
return
@@ -150,7 +155,7 @@ class BaselineRecorder:
'rssi': device.get('rssi', self.bt_devices[mac].get('rssi')),
})
else:
self.bt_devices[mac] = {
self.bt_devices[mac] = {
'mac': mac,
'name': device.get('name', ''),
'rssi': device.get('rssi', device.get('signal')),
@@ -158,10 +163,37 @@ class BaselineRecorder:
'type': device.get('type', ''),
'first_seen': datetime.now().isoformat(),
'last_seen': datetime.now().isoformat(),
}
def add_rf_signal(self, signal: dict) -> None:
"""Add an RF signal to the current baseline."""
}
def add_wifi_client(self, client: dict) -> None:
"""Add a WiFi client to the current baseline."""
if not self.recording:
return
mac = client.get('mac', client.get('address', '')).upper()
if not mac:
return
if mac in self.wifi_clients:
self.wifi_clients[mac].update({
'last_seen': datetime.now().isoformat(),
'rssi': client.get('rssi', self.wifi_clients[mac].get('rssi')),
'associated_bssid': client.get('associated_bssid', self.wifi_clients[mac].get('associated_bssid')),
})
else:
self.wifi_clients[mac] = {
'mac': mac,
'vendor': client.get('vendor', ''),
'rssi': client.get('rssi'),
'associated_bssid': client.get('associated_bssid'),
'probed_ssids': client.get('probed_ssids', []),
'probe_count': client.get('probe_count', len(client.get('probed_ssids', []))),
'first_seen': datetime.now().isoformat(),
'last_seen': datetime.now().isoformat(),
}
def add_rf_signal(self, signal: dict) -> None:
"""Add an RF signal to the current baseline."""
if not self.recording:
return
@@ -191,15 +223,16 @@ class BaselineRecorder:
'hit_count': 1,
}
def get_recording_status(self) -> dict:
"""Get current recording status and counts."""
return {
'recording': self.recording,
'baseline_id': self.current_baseline_id,
'wifi_count': len(self.wifi_networks),
'bt_count': len(self.bt_devices),
'rf_count': len(self.rf_frequencies),
}
def get_recording_status(self) -> dict:
"""Get current recording status and counts."""
return {
'recording': self.recording,
'baseline_id': self.current_baseline_id,
'wifi_count': len(self.wifi_networks),
'wifi_client_count': len(self.wifi_clients),
'bt_count': len(self.bt_devices),
'rf_count': len(self.rf_frequencies),
}
class BaselineComparator:
@@ -220,11 +253,16 @@ class BaselineComparator:
for d in baseline.get('wifi_networks', [])
if d.get('bssid') or d.get('mac')
}
self.baseline_bt = {
d.get('mac', d.get('address', '')).upper(): d
for d in baseline.get('bt_devices', [])
if d.get('mac') or d.get('address')
}
self.baseline_bt = {
d.get('mac', d.get('address', '')).upper(): d
for d in baseline.get('bt_devices', [])
if d.get('mac') or d.get('address')
}
self.baseline_wifi_clients = {
d.get('mac', d.get('address', '')).upper(): d
for d in baseline.get('wifi_clients', [])
if d.get('mac') or d.get('address')
}
self.baseline_rf = {
round(d.get('frequency', 0), 1): d
for d in baseline.get('rf_frequencies', [])
@@ -269,8 +307,8 @@ class BaselineComparator:
'matching_count': len(matching_devices),
}
def compare_bluetooth(self, current_devices: list[dict]) -> dict:
"""Compare current Bluetooth devices against baseline."""
def compare_bluetooth(self, current_devices: list[dict]) -> dict:
"""Compare current Bluetooth devices against baseline."""
current_macs = {
d.get('mac', d.get('address', '')).upper(): d
for d in current_devices
@@ -291,14 +329,45 @@ class BaselineComparator:
if mac not in current_macs:
missing_devices.append(device)
return {
'new': new_devices,
'missing': missing_devices,
'matching': matching_devices,
'new_count': len(new_devices),
'missing_count': len(missing_devices),
'matching_count': len(matching_devices),
}
return {
'new': new_devices,
'missing': missing_devices,
'matching': matching_devices,
'new_count': len(new_devices),
'missing_count': len(missing_devices),
'matching_count': len(matching_devices),
}
def compare_wifi_clients(self, current_devices: list[dict]) -> dict:
"""Compare current WiFi clients against baseline."""
current_macs = {
d.get('mac', d.get('address', '')).upper(): d
for d in current_devices
if d.get('mac') or d.get('address')
}
new_devices = []
missing_devices = []
matching_devices = []
for mac, device in current_macs.items():
if mac not in self.baseline_wifi_clients:
new_devices.append(device)
else:
matching_devices.append(device)
for mac, device in self.baseline_wifi_clients.items():
if mac not in current_macs:
missing_devices.append(device)
return {
'new': new_devices,
'missing': missing_devices,
'matching': matching_devices,
'new_count': len(new_devices),
'missing_count': len(missing_devices),
'matching_count': len(matching_devices),
}
def compare_rf(self, current_signals: list[dict]) -> dict:
"""Compare current RF signals against baseline."""
@@ -331,35 +400,42 @@ class BaselineComparator:
'matching_count': len(matching_signals),
}
def compare_all(
self,
wifi_devices: list[dict] | None = None,
bt_devices: list[dict] | None = None,
rf_signals: list[dict] | None = None
) -> dict:
def compare_all(
self,
wifi_devices: list[dict] | None = None,
wifi_clients: list[dict] | None = None,
bt_devices: list[dict] | None = None,
rf_signals: list[dict] | None = None
) -> dict:
"""
Compare all current data against baseline.
Returns:
Dict with comparison results for each category
"""
results = {
'wifi': None,
'bluetooth': None,
'rf': None,
'total_new': 0,
'total_missing': 0,
}
results = {
'wifi': None,
'wifi_clients': None,
'bluetooth': None,
'rf': None,
'total_new': 0,
'total_missing': 0,
}
if wifi_devices is not None:
results['wifi'] = self.compare_wifi(wifi_devices)
results['total_new'] += results['wifi']['new_count']
results['total_missing'] += results['wifi']['missing_count']
if bt_devices is not None:
results['bluetooth'] = self.compare_bluetooth(bt_devices)
results['total_new'] += results['bluetooth']['new_count']
results['total_missing'] += results['bluetooth']['missing_count']
if wifi_devices is not None:
results['wifi'] = self.compare_wifi(wifi_devices)
results['total_new'] += results['wifi']['new_count']
results['total_missing'] += results['wifi']['missing_count']
if wifi_clients is not None:
results['wifi_clients'] = self.compare_wifi_clients(wifi_clients)
results['total_new'] += results['wifi_clients']['new_count']
results['total_missing'] += results['wifi_clients']['missing_count']
if bt_devices is not None:
results['bluetooth'] = self.compare_bluetooth(bt_devices)
results['total_new'] += results['bluetooth']['new_count']
results['total_missing'] += results['bluetooth']['missing_count']
if rf_signals is not None:
results['rf'] = self.compare_rf(rf_signals)
@@ -369,11 +445,12 @@ class BaselineComparator:
return results
def get_comparison_for_active_baseline(
wifi_devices: list[dict] | None = None,
bt_devices: list[dict] | None = None,
rf_signals: list[dict] | None = None
) -> dict | None:
def get_comparison_for_active_baseline(
wifi_devices: list[dict] | None = None,
wifi_clients: list[dict] | None = None,
bt_devices: list[dict] | None = None,
rf_signals: list[dict] | None = None
) -> dict | None:
"""
Convenience function to compare against the active baseline.
@@ -385,4 +462,4 @@ def get_comparison_for_active_baseline(
return None
comparator = BaselineComparator(baseline)
return comparator.compare_all(wifi_devices, bt_devices, rf_signals)
return comparator.compare_all(wifi_devices, wifi_clients, bt_devices, rf_signals)

20
utils/tscm/detector.py
View File

@@ -113,14 +113,18 @@ class ThreatDetector:
def _load_baseline(self, baseline: dict) -> None:
"""Load baseline device identifiers for comparison."""
# WiFi networks and clients
for network in baseline.get('wifi_networks', []):
if 'bssid' in network:
self.baseline_wifi_macs.add(network['bssid'].upper())
if 'clients' in network:
for client in network['clients']:
if 'mac' in client:
self.baseline_wifi_macs.add(client['mac'].upper())
# WiFi networks and clients
for network in baseline.get('wifi_networks', []):
if 'bssid' in network:
self.baseline_wifi_macs.add(network['bssid'].upper())
if 'clients' in network:
for client in network['clients']:
if 'mac' in client:
self.baseline_wifi_macs.add(client['mac'].upper())
for client in baseline.get('wifi_clients', []):
if 'mac' in client:
self.baseline_wifi_macs.add(client['mac'].upper())
# Bluetooth devices
for device in baseline.get('bt_devices', []):