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

Add comprehensive TSCM advanced features

Browse Source 
Implement 9 major TSCM feature enhancements:

1. Capability & Coverage Reality Panel - Exposes what sweeps can/cannot
   detect based on OS, privileges, adapters, and SDR limits

2. Baseline Diff & Health - Shows changes vs baseline with health scoring
   (healthy/noisy/stale) based on age and device churn

3. Per-Device Timelines - Time-bucketed observations with RSSI stability,
   movement patterns, and meeting correlation

4. Whitelist/Known-Good Registry + Case Grouping - Global and per-location
   device registry with case management for sweeps/threats/notes

5. Meeting-Window Summary Enhancements - Tracks devices first seen during
   meetings with scoring modifiers

6. Client-Ready PDF Report + Technical Annex - Executive summary, findings
   by risk tier, JSON/CSV annex export

7. WiFi Advanced Indicators - Evil twin detection, probe request tracking,
   deauth burst detection (auto-disables without monitor mode)

8. Bluetooth Risk Explainability - Proximity estimates, tracker brand
   explanations, human-readable risk descriptions

9. Operator Playbooks - Procedural guidance by risk level with steps,
   safety notes, and documentation requirements

All features include mandatory disclaimers, preserve existing architecture,
and follow TSCM best practices (no packet capture, no surveillance claims).

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Smittix
2026-01-16 16:06:18 +00:00
parent 3210fc0d20
commit 234f254f4f
4 changed files with 4560 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
routes/tscm.py
+959
View File
@@ -2274,3 +2274,962 @@ def get_cluster_detail(cluster_id: str):
except Exception as e:
logger.error(f"Get cluster detail error: {e}")
return jsonify({'status': 'error', 'message': str(e)}), 500
# =============================================================================
# Capabilities & Coverage Endpoints
# =============================================================================
@tscm_bp.route('/capabilities')
def get_capabilities():
"""
Get current system capabilities for TSCM sweeping.
Returns what the system CAN and CANNOT detect based on OS,
privileges, adapters, and SDR hardware.
"""
try:
from utils.tscm.advanced import detect_sweep_capabilities
wifi_interface = request.args.get('wifi_interface', '')
bt_adapter = request.args.get('bt_adapter', '')
caps = detect_sweep_capabilities(
wifi_interface=wifi_interface,
bt_adapter=bt_adapter
)
return jsonify({
'status': 'success',
'capabilities': caps.to_dict()
})
except Exception as e:
logger.error(f"Get capabilities error: {e}")
return jsonify({'status': 'error', 'message': str(e)}), 500
@tscm_bp.route('/sweep/<int:sweep_id>/capabilities')
def get_sweep_stored_capabilities(sweep_id: int):
"""Get stored capabilities for a specific sweep."""
from utils.database import get_sweep_capabilities
caps = get_sweep_capabilities(sweep_id)
if not caps:
return jsonify({'status': 'error', 'message': 'No capabilities stored for this sweep'}), 404
return jsonify({
'status': 'success',
'capabilities': caps
})
# =============================================================================
# Baseline Diff & Health Endpoints
# =============================================================================
@tscm_bp.route('/baseline/diff/<int:baseline_id>/<int:sweep_id>')
def get_baseline_diff(baseline_id: int, sweep_id: int):
"""
Get comprehensive diff between a baseline and a sweep.
Shows new devices, missing devices, changed characteristics,
and baseline health assessment.
"""
try:
from utils.tscm.advanced import calculate_baseline_diff
baseline = get_tscm_baseline(baseline_id)
if not baseline:
return jsonify({'status': 'error', 'message': 'Baseline not found'}), 404
sweep = get_tscm_sweep(sweep_id)
if not sweep:
return jsonify({'status': 'error', 'message': 'Sweep not found'}), 404
# Get current devices from sweep results
results = sweep.get('results', {})
if isinstance(results, str):
import json
results = json.loads(results)
current_wifi = results.get('wifi', [])
current_bt = results.get('bluetooth', [])
current_rf = results.get('rf', [])
diff = calculate_baseline_diff(
baseline=baseline,
current_wifi=current_wifi,
current_bt=current_bt,
current_rf=current_rf,
sweep_id=sweep_id
)
return jsonify({
'status': 'success',
'diff': diff.to_dict()
})
except Exception as e:
logger.error(f"Get baseline diff error: {e}")
return jsonify({'status': 'error', 'message': str(e)}), 500
@tscm_bp.route('/baseline/<int:baseline_id>/health')
def get_baseline_health(baseline_id: int):
"""Get health assessment for a baseline."""
try:
from utils.tscm.advanced import BaselineHealth
from datetime import datetime
baseline = get_tscm_baseline(baseline_id)
if not baseline:
return jsonify({'status': 'error', 'message': 'Baseline not found'}), 404
# Calculate age
created_at = baseline.get('created_at')
age_hours = 0
if created_at:
if isinstance(created_at, str):
created = datetime.fromisoformat(created_at.replace('Z', '+00:00'))
age_hours = (datetime.now() - created.replace(tzinfo=None)).total_seconds() / 3600
elif isinstance(created_at, datetime):
age_hours = (datetime.now() - created_at).total_seconds() / 3600
# Count devices
total_devices = (
len(baseline.get('wifi_networks', [])) +
len(baseline.get('bt_devices', [])) +
len(baseline.get('rf_frequencies', []))
)
# Determine health
health = 'healthy'
score = 1.0
reasons = []
if age_hours > 168:
health = 'stale'
score = 0.3
reasons.append(f'Baseline is {age_hours:.0f} hours old (over 1 week)')
elif age_hours > 72:
health = 'noisy'
score = 0.6
reasons.append(f'Baseline is {age_hours:.0f} hours old (over 3 days)')
if total_devices < 3:
score -= 0.2
reasons.append(f'Baseline has few devices ({total_devices})')
if health == 'healthy':
health = 'noisy'
return jsonify({
'status': 'success',
'health': {
'status': health,
'score': round(max(0, score), 2),
'age_hours': round(age_hours, 1),
'total_devices': total_devices,
'reasons': reasons,
}
})
except Exception as e:
logger.error(f"Get baseline health error: {e}")
return jsonify({'status': 'error', 'message': str(e)}), 500
# =============================================================================
# Device Timeline Endpoints
# =============================================================================
@tscm_bp.route('/device/<identifier>/timeline')
def get_device_timeline_endpoint(identifier: str):
"""
Get timeline of observations for a device.
Shows behavior over time including RSSI stability, presence,
and meeting window correlation.
"""
try:
from utils.tscm.advanced import get_timeline_manager
from utils.database import get_device_timeline
protocol = request.args.get('protocol', 'bluetooth')
since_hours = request.args.get('since_hours', 24, type=int)
# Try in-memory timeline first
manager = get_timeline_manager()
timeline = manager.get_timeline(identifier, protocol)
# Also get stored timeline from database
stored = get_device_timeline(identifier, since_hours=since_hours)
result = {
'identifier': identifier,
'protocol': protocol,
'observations': stored,
}
if timeline:
result['metrics'] = {
'first_seen': timeline.first_seen.isoformat() if timeline.first_seen else None,
'last_seen': timeline.last_seen.isoformat() if timeline.last_seen else None,
'total_observations': timeline.total_observations,
'presence_ratio': round(timeline.presence_ratio, 2),
}
result['signal'] = {
'rssi_min': timeline.rssi_min,
'rssi_max': timeline.rssi_max,
'rssi_mean': round(timeline.rssi_mean, 1) if timeline.rssi_mean else None,
'stability': round(timeline.rssi_stability, 2),
}
result['movement'] = {
'appears_stationary': timeline.appears_stationary,
'pattern': timeline.movement_pattern,
}
result['meeting_correlation'] = {
'correlated': timeline.meeting_correlated,
'observations_during_meeting': timeline.meeting_observations,
}
return jsonify({
'status': 'success',
'timeline': result
})
except Exception as e:
logger.error(f"Get device timeline error: {e}")
return jsonify({'status': 'error', 'message': str(e)}), 500
@tscm_bp.route('/timelines')
def get_all_device_timelines():
"""Get all device timelines."""
try:
from utils.tscm.advanced import get_timeline_manager
manager = get_timeline_manager()
timelines = manager.get_all_timelines()
return jsonify({
'status': 'success',
'count': len(timelines),
'timelines': [t.to_dict() for t in timelines]
})
except Exception as e:
logger.error(f"Get all timelines error: {e}")
return jsonify({'status': 'error', 'message': str(e)}), 500
# =============================================================================
# Known-Good Registry (Whitelist) Endpoints
# =============================================================================
@tscm_bp.route('/known-devices', methods=['GET'])
def list_known_devices():
"""List all known-good devices."""
from utils.database import get_all_known_devices
location = request.args.get('location')
scope = request.args.get('scope')
devices = get_all_known_devices(location=location, scope=scope)
return jsonify({
'status': 'success',
'count': len(devices),
'devices': devices
})
@tscm_bp.route('/known-devices', methods=['POST'])
def add_known_device_endpoint():
"""
Add a device to the known-good registry.
Known devices remain visible but receive reduced risk scores.
They are NOT suppressed from reports (preserves audit trail).
"""
from utils.database import add_known_device
data = request.get_json() or {}
identifier = data.get('identifier')
protocol = data.get('protocol')
if not identifier or not protocol:
return jsonify({
'status': 'error',
'message': 'identifier and protocol are required'
}), 400
device_id = add_known_device(
identifier=identifier,
protocol=protocol,
name=data.get('name'),
description=data.get('description'),
location=data.get('location'),
scope=data.get('scope', 'global'),
added_by=data.get('added_by'),
score_modifier=data.get('score_modifier', -2),
metadata=data.get('metadata')
)
return jsonify({
'status': 'success',
'message': 'Device added to known-good registry',
'device_id': device_id
})
@tscm_bp.route('/known-devices/<identifier>', methods=['GET'])
def get_known_device_endpoint(identifier: str):
"""Get a known device by identifier."""
from utils.database import get_known_device
device = get_known_device(identifier)
if not device:
return jsonify({'status': 'error', 'message': 'Device not found'}), 404
return jsonify({
'status': 'success',
'device': device
})
@tscm_bp.route('/known-devices/<identifier>', methods=['DELETE'])
def delete_known_device_endpoint(identifier: str):
"""Remove a device from the known-good registry."""
from utils.database import delete_known_device
success = delete_known_device(identifier)
if not success:
return jsonify({'status': 'error', 'message': 'Device not found'}), 404
return jsonify({
'status': 'success',
'message': 'Device removed from known-good registry'
})
@tscm_bp.route('/known-devices/check/<identifier>')
def check_known_device(identifier: str):
"""Check if a device is in the known-good registry."""
from utils.database import is_known_good_device
location = request.args.get('location')
result = is_known_good_device(identifier, location=location)
return jsonify({
'status': 'success',
'is_known': result is not None,
'details': result
})
# =============================================================================
# Case Management Endpoints
# =============================================================================
@tscm_bp.route('/cases', methods=['GET'])
def list_cases():
"""List all TSCM cases."""
from utils.database import get_all_tscm_cases
status = request.args.get('status')
limit = request.args.get('limit', 50, type=int)
cases = get_all_tscm_cases(status=status, limit=limit)
return jsonify({
'status': 'success',
'count': len(cases),
'cases': cases
})
@tscm_bp.route('/cases', methods=['POST'])
def create_case():
"""Create a new TSCM case."""
from utils.database import create_tscm_case
data = request.get_json() or {}
name = data.get('name')
if not name:
return jsonify({'status': 'error', 'message': 'name is required'}), 400
case_id = create_tscm_case(
name=name,
description=data.get('description'),
location=data.get('location'),
priority=data.get('priority', 'normal'),
created_by=data.get('created_by'),
metadata=data.get('metadata')
)
return jsonify({
'status': 'success',
'message': 'Case created',
'case_id': case_id
})
@tscm_bp.route('/cases/<int:case_id>', methods=['GET'])
def get_case(case_id: int):
"""Get a TSCM case with all linked sweeps, threats, and notes."""
from utils.database import get_tscm_case
case = get_tscm_case(case_id)
if not case:
return jsonify({'status': 'error', 'message': 'Case not found'}), 404
return jsonify({
'status': 'success',
'case': case
})
@tscm_bp.route('/cases/<int:case_id>', methods=['PUT'])
def update_case(case_id: int):
"""Update a TSCM case."""
from utils.database import update_tscm_case
data = request.get_json() or {}
success = update_tscm_case(
case_id=case_id,
status=data.get('status'),
priority=data.get('priority'),
assigned_to=data.get('assigned_to'),
notes=data.get('notes')
)
if not success:
return jsonify({'status': 'error', 'message': 'Case not found'}), 404
return jsonify({
'status': 'success',
'message': 'Case updated'
})
@tscm_bp.route('/cases/<int:case_id>/sweeps/<int:sweep_id>', methods=['POST'])
def link_sweep_to_case(case_id: int, sweep_id: int):
"""Link a sweep to a case."""
from utils.database import add_sweep_to_case
success = add_sweep_to_case(case_id, sweep_id)
return jsonify({
'status': 'success' if success else 'error',
'message': 'Sweep linked to case' if success else 'Already linked or not found'
})
@tscm_bp.route('/cases/<int:case_id>/threats/<int:threat_id>', methods=['POST'])
def link_threat_to_case(case_id: int, threat_id: int):
"""Link a threat to a case."""
from utils.database import add_threat_to_case
success = add_threat_to_case(case_id, threat_id)
return jsonify({
'status': 'success' if success else 'error',
'message': 'Threat linked to case' if success else 'Already linked or not found'
})
@tscm_bp.route('/cases/<int:case_id>/notes', methods=['POST'])
def add_note_to_case(case_id: int):
"""Add a note to a case."""
from utils.database import add_case_note
data = request.get_json() or {}
content = data.get('content')
if not content:
return jsonify({'status': 'error', 'message': 'content is required'}), 400
note_id = add_case_note(
case_id=case_id,
content=content,
note_type=data.get('note_type', 'general'),
created_by=data.get('created_by')
)
return jsonify({
'status': 'success',
'message': 'Note added',
'note_id': note_id
})
# =============================================================================
# Meeting Window Enhanced Endpoints
# =============================================================================
@tscm_bp.route('/meeting/start-tracked', methods=['POST'])
def start_tracked_meeting():
"""
Start a tracked meeting window with database persistence.
Tracks devices first seen during meeting and behavior changes.
"""
from utils.database import start_meeting_window
from utils.tscm.advanced import get_timeline_manager
data = request.get_json() or {}
meeting_id = start_meeting_window(
sweep_id=_current_sweep_id,
name=data.get('name'),
location=data.get('location'),
notes=data.get('notes')
)
# Start meeting in correlation engine
correlation = get_correlation_engine()
correlation.start_meeting_window()
# Start in timeline manager
manager = get_timeline_manager()
manager.start_meeting_window()
_emit_event('meeting_started', {
'meeting_id': meeting_id,
'timestamp': datetime.now().isoformat(),
'name': data.get('name'),
})
return jsonify({
'status': 'success',
'message': 'Tracked meeting window started',
'meeting_id': meeting_id
})
@tscm_bp.route('/meeting/<int:meeting_id>/end', methods=['POST'])
def end_tracked_meeting(meeting_id: int):
"""End a tracked meeting window."""
from utils.database import end_meeting_window
from utils.tscm.advanced import get_timeline_manager
success = end_meeting_window(meeting_id)
if not success:
return jsonify({'status': 'error', 'message': 'Meeting not found or already ended'}), 404
# End in correlation engine
correlation = get_correlation_engine()
correlation.end_meeting_window()
# End in timeline manager
manager = get_timeline_manager()
manager.end_meeting_window()
_emit_event('meeting_ended', {
'meeting_id': meeting_id,
'timestamp': datetime.now().isoformat()
})
return jsonify({
'status': 'success',
'message': 'Meeting window ended'
})
@tscm_bp.route('/meeting/<int:meeting_id>/summary')
def get_meeting_summary_endpoint(meeting_id: int):
"""Get detailed summary of device activity during a meeting."""
try:
from utils.database import get_meeting_windows
from utils.tscm.advanced import generate_meeting_summary, get_timeline_manager
# Get meeting window
windows = get_meeting_windows(_current_sweep_id or 0)
meeting = None
for w in windows:
if w.get('id') == meeting_id:
meeting = w
break
if not meeting:
return jsonify({'status': 'error', 'message': 'Meeting not found'}), 404
# Get timelines and profiles
manager = get_timeline_manager()
timelines = manager.get_all_timelines()
correlation = get_correlation_engine()
profiles = [p.to_dict() for p in correlation.device_profiles.values()]
summary = generate_meeting_summary(meeting, timelines, profiles)
return jsonify({
'status': 'success',
'summary': summary.to_dict()
})
except Exception as e:
logger.error(f"Get meeting summary error: {e}")
return jsonify({'status': 'error', 'message': str(e)}), 500
@tscm_bp.route('/meeting/active')
def get_active_meeting():
"""Get currently active meeting window."""
from utils.database import get_active_meeting_window
meeting = get_active_meeting_window(_current_sweep_id)
return jsonify({
'status': 'success',
'meeting': meeting,
'is_active': meeting is not None
})
# =============================================================================
# PDF Report & Technical Annex Endpoints
# =============================================================================
@tscm_bp.route('/report/pdf')
def get_pdf_report():
"""
Generate client-safe PDF report.
Contains executive summary, findings by risk tier, meeting window
summary, and mandatory disclaimers.
"""
try:
from utils.tscm.reports import generate_report, get_pdf_report
from utils.tscm.advanced import detect_sweep_capabilities, get_timeline_manager
sweep_id = request.args.get('sweep_id', _current_sweep_id, type=int)
if not sweep_id:
return jsonify({'status': 'error', 'message': 'No sweep specified'}), 400
sweep = get_tscm_sweep(sweep_id)
if not sweep:
return jsonify({'status': 'error', 'message': 'Sweep not found'}), 404
# Get data for report
correlation = get_correlation_engine()
profiles = [p.to_dict() for p in correlation.device_profiles.values()]
caps = detect_sweep_capabilities().to_dict()
manager = get_timeline_manager()
timelines = [t.to_dict() for t in manager.get_all_timelines()]
# Generate report
report = generate_report(
sweep_id=sweep_id,
sweep_data=sweep,
device_profiles=profiles,
capabilities=caps,
timelines=timelines
)
pdf_content = get_pdf_report(report)
return Response(
pdf_content,
mimetype='text/plain',
headers={
'Content-Disposition': f'attachment; filename=tscm_report_{sweep_id}.txt'
}
)
except Exception as e:
logger.error(f"Generate PDF report error: {e}")
return jsonify({'status': 'error', 'message': str(e)}), 500
@tscm_bp.route('/report/annex')
def get_technical_annex():
"""
Generate technical annex (JSON + CSV).
Contains device timelines, all indicators, and detailed data
for audit purposes. No packet data included.
"""
try:
from utils.tscm.reports import generate_report, get_json_annex, get_csv_annex
from utils.tscm.advanced import detect_sweep_capabilities, get_timeline_manager
sweep_id = request.args.get('sweep_id', _current_sweep_id, type=int)
format_type = request.args.get('format', 'json')
if not sweep_id:
return jsonify({'status': 'error', 'message': 'No sweep specified'}), 400
sweep = get_tscm_sweep(sweep_id)
if not sweep:
return jsonify({'status': 'error', 'message': 'Sweep not found'}), 404
# Get data for report
correlation = get_correlation_engine()
profiles = [p.to_dict() for p in correlation.device_profiles.values()]
caps = detect_sweep_capabilities().to_dict()
manager = get_timeline_manager()
timelines = [t.to_dict() for t in manager.get_all_timelines()]
# Generate report
report = generate_report(
sweep_id=sweep_id,
sweep_data=sweep,
device_profiles=profiles,
capabilities=caps,
timelines=timelines
)
if format_type == 'csv':
csv_content = get_csv_annex(report)
return Response(
csv_content,
mimetype='text/csv',
headers={
'Content-Disposition': f'attachment; filename=tscm_annex_{sweep_id}.csv'
}
)
else:
annex = get_json_annex(report)
return jsonify({
'status': 'success',
'annex': annex
})
except Exception as e:
logger.error(f"Generate technical annex error: {e}")
return jsonify({'status': 'error', 'message': str(e)}), 500
# =============================================================================
# WiFi Advanced Indicators Endpoints
# =============================================================================
@tscm_bp.route('/wifi/advanced-indicators')
def get_wifi_advanced_indicators():
"""
Get advanced WiFi indicators (Evil Twin, Probes, Deauth).
These indicators require analysis of WiFi patterns.
Some features require monitor mode.
"""
try:
from utils.tscm.advanced import get_wifi_detector
detector = get_wifi_detector()
return jsonify({
'status': 'success',
'indicators': detector.get_all_indicators(),
'unavailable_features': detector.get_unavailable_features(),
'disclaimer': (
"All indicators represent pattern detections, NOT confirmed attacks. "
"Further investigation is required."
)
})
except Exception as e:
logger.error(f"Get WiFi indicators error: {e}")
return jsonify({'status': 'error', 'message': str(e)}), 500
@tscm_bp.route('/wifi/analyze-network', methods=['POST'])
def analyze_wifi_network():
"""
Analyze a WiFi network for evil twin patterns.
Compares against known networks to detect SSID spoofing.
"""
try:
from utils.tscm.advanced import get_wifi_detector
data = request.get_json() or {}
detector = get_wifi_detector()
# Set known networks from baseline if available
baseline = get_active_tscm_baseline()
if baseline:
detector.set_known_networks(baseline.get('wifi_networks', []))
indicators = detector.analyze_network(data)
return jsonify({
'status': 'success',
'indicators': [i.to_dict() for i in indicators]
})
except Exception as e:
logger.error(f"Analyze WiFi network error: {e}")
return jsonify({'status': 'error', 'message': str(e)}), 500
# =============================================================================
# Bluetooth Risk Explainability Endpoints
# =============================================================================
@tscm_bp.route('/bluetooth/<identifier>/explain')
def explain_bluetooth_risk(identifier: str):
"""
Get human-readable risk explanation for a BLE device.
Includes proximity estimate, tracker explanation, and
recommended actions.
"""
try:
from utils.tscm.advanced import generate_ble_risk_explanation
# Get device from correlation engine
correlation = get_correlation_engine()
profile = None
key = f"bluetooth:{identifier.upper()}"
if key in correlation.device_profiles:
profile = correlation.device_profiles[key].to_dict()
# Try to find device info
device = {'mac': identifier}
if profile:
device['name'] = profile.get('name')
device['rssi'] = profile.get('rssi_samples', [None])[-1] if profile.get('rssi_samples') else None
# Check meeting status
is_meeting = correlation.is_during_meeting()
explanation = generate_ble_risk_explanation(device, profile, is_meeting)
return jsonify({
'status': 'success',
'explanation': explanation.to_dict()
})
except Exception as e:
logger.error(f"Explain BLE risk error: {e}")
return jsonify({'status': 'error', 'message': str(e)}), 500
@tscm_bp.route('/bluetooth/<identifier>/proximity')
def get_bluetooth_proximity(identifier: str):
"""Get proximity estimate for a BLE device."""
try:
from utils.tscm.advanced import estimate_ble_proximity
rssi = request.args.get('rssi', type=int)
if rssi is None:
# Try to get from correlation engine
correlation = get_correlation_engine()
key = f"bluetooth:{identifier.upper()}"
if key in correlation.device_profiles:
profile = correlation.device_profiles[key]
if profile.rssi_samples:
rssi = profile.rssi_samples[-1]
if rssi is None:
return jsonify({
'status': 'error',
'message': 'RSSI value required'
}), 400
proximity, explanation, distance = estimate_ble_proximity(rssi)
return jsonify({
'status': 'success',
'proximity': {
'estimate': proximity.value,
'explanation': explanation,
'estimated_distance': distance,
'rssi_used': rssi,
},
'disclaimer': (
"Proximity estimates are approximate and affected by "
"environment, obstacles, and device characteristics."
)
})
except Exception as e:
logger.error(f"Get BLE proximity error: {e}")
return jsonify({'status': 'error', 'message': str(e)}), 500
# =============================================================================
# Operator Playbook Endpoints
# =============================================================================
@tscm_bp.route('/playbooks')
def list_playbooks():
"""List all available operator playbooks."""
try:
from utils.tscm.advanced import PLAYBOOKS
return jsonify({
'status': 'success',
'playbooks': {
pid: pb.to_dict()
for pid, pb in PLAYBOOKS.items()
}
})
except Exception as e:
logger.error(f"List playbooks error: {e}")
return jsonify({'status': 'error', 'message': str(e)}), 500
@tscm_bp.route('/playbooks/<playbook_id>')
def get_playbook(playbook_id: str):
"""Get a specific playbook."""
try:
from utils.tscm.advanced import PLAYBOOKS
if playbook_id not in PLAYBOOKS:
return jsonify({'status': 'error', 'message': 'Playbook not found'}), 404
return jsonify({
'status': 'success',
'playbook': PLAYBOOKS[playbook_id].to_dict()
})
except Exception as e:
logger.error(f"Get playbook error: {e}")
return jsonify({'status': 'error', 'message': str(e)}), 500
@tscm_bp.route('/findings/<identifier>/playbook')
def get_finding_playbook(identifier: str):
"""Get recommended playbook for a specific finding."""
try:
from utils.tscm.advanced import get_playbook_for_finding
# Get profile
correlation = get_correlation_engine()
profile = None
for protocol in ['bluetooth', 'wifi', 'rf']:
key = f"{protocol}:{identifier.upper()}"
if key in correlation.device_profiles:
profile = correlation.device_profiles[key].to_dict()
break
if not profile:
return jsonify({'status': 'error', 'message': 'Finding not found'}), 404
playbook = get_playbook_for_finding(
risk_level=profile.get('risk_level', 'informational'),
indicators=profile.get('indicators', [])
)
return jsonify({
'status': 'success',
'playbook': playbook.to_dict(),
'suggested_next_steps': [
f"Step {s.step_number}: {s.action}"
for s in playbook.steps[:3]
]
})
except Exception as e:
logger.error(f"Get finding playbook error: {e}")
return jsonify({'status': 'error', 'message': str(e)}), 500
utils/database.py
+636
View File
@@ -178,6 +178,123 @@ def init_db() -> None:
)
''')
# TSCM Device Timelines - Periodic observations per device
conn.execute('''
CREATE TABLE IF NOT EXISTS tscm_device_timelines (
id INTEGER PRIMARY KEY AUTOINCREMENT,
device_identifier TEXT NOT NULL,
protocol TEXT NOT NULL,
sweep_id INTEGER,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
rssi INTEGER,
presence BOOLEAN DEFAULT 1,
channel INTEGER,
frequency REAL,
attributes TEXT,
FOREIGN KEY (sweep_id) REFERENCES tscm_sweeps(id)
)
''')
# TSCM Known-Good Registry - Whitelist of expected devices
conn.execute('''
CREATE TABLE IF NOT EXISTS tscm_known_devices (
id INTEGER PRIMARY KEY AUTOINCREMENT,
identifier TEXT NOT NULL UNIQUE,
protocol TEXT NOT NULL,
name TEXT,
description TEXT,
location TEXT,
scope TEXT DEFAULT 'global',
added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
added_by TEXT,
last_verified TIMESTAMP,
score_modifier INTEGER DEFAULT -2,
metadata TEXT
)
''')
# TSCM Cases - Grouping sweeps, threats, and notes
conn.execute('''
CREATE TABLE IF NOT EXISTS tscm_cases (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
description TEXT,
location TEXT,
status TEXT DEFAULT 'open',
priority TEXT DEFAULT 'normal',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
closed_at TIMESTAMP,
created_by TEXT,
assigned_to TEXT,
notes TEXT,
metadata TEXT
)
''')
# TSCM Case Sweeps - Link sweeps to cases
conn.execute('''
CREATE TABLE IF NOT EXISTS tscm_case_sweeps (
case_id INTEGER,
sweep_id INTEGER,
added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (case_id, sweep_id),
FOREIGN KEY (case_id) REFERENCES tscm_cases(id),
FOREIGN KEY (sweep_id) REFERENCES tscm_sweeps(id)
)
''')
# TSCM Case Threats - Link threats to cases
conn.execute('''
CREATE TABLE IF NOT EXISTS tscm_case_threats (
case_id INTEGER,
threat_id INTEGER,
added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (case_id, threat_id),
FOREIGN KEY (case_id) REFERENCES tscm_cases(id),
FOREIGN KEY (threat_id) REFERENCES tscm_threats(id)
)
''')
# TSCM Case Notes - Notes attached to cases
conn.execute('''
CREATE TABLE IF NOT EXISTS tscm_case_notes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
case_id INTEGER,
content TEXT NOT NULL,
note_type TEXT DEFAULT 'general',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
created_by TEXT,
FOREIGN KEY (case_id) REFERENCES tscm_cases(id)
)
''')
# TSCM Meeting Windows - Track sensitive periods
conn.execute('''
CREATE TABLE IF NOT EXISTS tscm_meeting_windows (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sweep_id INTEGER,
name TEXT,
start_time TIMESTAMP NOT NULL,
end_time TIMESTAMP,
location TEXT,
notes TEXT,
FOREIGN KEY (sweep_id) REFERENCES tscm_sweeps(id)
)
''')
# TSCM Sweep Capabilities - Store sweep capability snapshot
conn.execute('''
CREATE TABLE IF NOT EXISTS tscm_sweep_capabilities (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sweep_id INTEGER UNIQUE,
capabilities TEXT NOT NULL,
limitations TEXT,
recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (sweep_id) REFERENCES tscm_sweeps(id)
)
''')
# TSCM indexes for performance
conn.execute('''
CREATE INDEX IF NOT EXISTS idx_tscm_threats_sweep
@@ -194,6 +311,21 @@ def init_db() -> None:
ON tscm_sweeps(baseline_id)
''')
conn.execute('''
CREATE INDEX IF NOT EXISTS idx_tscm_timelines_device
ON tscm_device_timelines(device_identifier, timestamp)
''')
conn.execute('''
CREATE INDEX IF NOT EXISTS idx_tscm_known_devices_identifier
ON tscm_known_devices(identifier)
''')
conn.execute('''
CREATE INDEX IF NOT EXISTS idx_tscm_cases_status
ON tscm_cases(status, created_at)
''')
logger.info("Database initialized successfully")
@@ -793,3 +925,507 @@ def get_tscm_threat_summary() -> dict:
summary['total'] += row['count']
return summary
# =============================================================================
# TSCM Device Timeline Functions
# =============================================================================
def add_device_timeline_entry(
device_identifier: str,
protocol: str,
sweep_id: int | None = None,
rssi: int | None = None,
presence: bool = True,
channel: int | None = None,
frequency: float | None = None,
attributes: dict | None = None
) -> int:
"""Add a device timeline observation entry."""
with get_db() as conn:
cursor = conn.execute('''
INSERT INTO tscm_device_timelines
(device_identifier, protocol, sweep_id, rssi, presence, channel, frequency, attributes)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
''', (
device_identifier, protocol, sweep_id, rssi, presence,
channel, frequency, json.dumps(attributes) if attributes else None
))
return cursor.lastrowid
def get_device_timeline(
device_identifier: str,
limit: int = 100,
since_hours: int = 24
) -> list[dict]:
"""Get timeline entries for a device."""
with get_db() as conn:
cursor = conn.execute('''
SELECT * FROM tscm_device_timelines
WHERE device_identifier = ?
AND timestamp > datetime('now', ?)
ORDER BY timestamp DESC
LIMIT ?
''', (device_identifier, f'-{since_hours} hours', limit))
results = []
for row in cursor:
results.append({
'id': row['id'],
'device_identifier': row['device_identifier'],
'protocol': row['protocol'],
'sweep_id': row['sweep_id'],
'timestamp': row['timestamp'],
'rssi': row['rssi'],
'presence': bool(row['presence']),
'channel': row['channel'],
'frequency': row['frequency'],
'attributes': json.loads(row['attributes']) if row['attributes'] else None
})
return list(reversed(results))
def cleanup_old_timeline_entries(max_age_hours: int = 72) -> int:
"""Remove old timeline entries."""
with get_db() as conn:
cursor = conn.execute('''
DELETE FROM tscm_device_timelines
WHERE timestamp < datetime('now', ?)
''', (f'-{max_age_hours} hours',))
return cursor.rowcount
# =============================================================================
# TSCM Known-Good Registry Functions
# =============================================================================
def add_known_device(
identifier: str,
protocol: str,
name: str | None = None,
description: str | None = None,
location: str | None = None,
scope: str = 'global',
added_by: str | None = None,
score_modifier: int = -2,
metadata: dict | None = None
) -> int:
"""Add a device to the known-good registry."""
with get_db() as conn:
cursor = conn.execute('''
INSERT INTO tscm_known_devices
(identifier, protocol, name, description, location, scope, added_by, score_modifier, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(identifier) DO UPDATE SET
name = excluded.name,
description = excluded.description,
location = excluded.location,
scope = excluded.scope,
score_modifier = excluded.score_modifier,
metadata = excluded.metadata,
last_verified = CURRENT_TIMESTAMP
''', (
identifier.upper(), protocol, name, description, location,
scope, added_by, score_modifier, json.dumps(metadata) if metadata else None
))
return cursor.lastrowid
def get_known_device(identifier: str) -> dict | None:
"""Get a known device by identifier."""
with get_db() as conn:
cursor = conn.execute(
'SELECT * FROM tscm_known_devices WHERE identifier = ?',
(identifier.upper(),)
)
row = cursor.fetchone()
if not row:
return None
return {
'id': row['id'],
'identifier': row['identifier'],
'protocol': row['protocol'],
'name': row['name'],
'description': row['description'],
'location': row['location'],
'scope': row['scope'],
'added_at': row['added_at'],
'added_by': row['added_by'],
'last_verified': row['last_verified'],
'score_modifier': row['score_modifier'],
'metadata': json.loads(row['metadata']) if row['metadata'] else None
}
def get_all_known_devices(
location: str | None = None,
scope: str | None = None
) -> list[dict]:
"""Get all known devices, optionally filtered by location or scope."""
conditions = []
params = []
if location:
conditions.append('(location = ? OR scope = ?)')
params.extend([location, 'global'])
if scope:
conditions.append('scope = ?')
params.append(scope)
where_clause = f'WHERE {" AND ".join(conditions)}' if conditions else ''
with get_db() as conn:
cursor = conn.execute(f'''
SELECT * FROM tscm_known_devices
{where_clause}
ORDER BY added_at DESC
''', params)
return [
{
'id': row['id'],
'identifier': row['identifier'],
'protocol': row['protocol'],
'name': row['name'],
'description': row['description'],
'location': row['location'],
'scope': row['scope'],
'added_at': row['added_at'],
'added_by': row['added_by'],
'last_verified': row['last_verified'],
'score_modifier': row['score_modifier'],
'metadata': json.loads(row['metadata']) if row['metadata'] else None
}
for row in cursor
]
def delete_known_device(identifier: str) -> bool:
"""Remove a device from the known-good registry."""
with get_db() as conn:
cursor = conn.execute(
'DELETE FROM tscm_known_devices WHERE identifier = ?',
(identifier.upper(),)
)
return cursor.rowcount > 0
def is_known_good_device(identifier: str, location: str | None = None) -> dict | None:
"""Check if a device is in the known-good registry for a location."""
with get_db() as conn:
if location:
cursor = conn.execute('''
SELECT * FROM tscm_known_devices
WHERE identifier = ? AND (location = ? OR scope = 'global')
''', (identifier.upper(), location))
else:
cursor = conn.execute(
'SELECT * FROM tscm_known_devices WHERE identifier = ?',
(identifier.upper(),)
)
row = cursor.fetchone()
if not row:
return None
return {
'identifier': row['identifier'],
'name': row['name'],
'score_modifier': row['score_modifier'],
'scope': row['scope']
}
# =============================================================================
# TSCM Case Functions
# =============================================================================
def create_tscm_case(
name: str,
description: str | None = None,
location: str | None = None,
priority: str = 'normal',
created_by: str | None = None,
metadata: dict | None = None
) -> int:
"""Create a new TSCM case."""
with get_db() as conn:
cursor = conn.execute('''
INSERT INTO tscm_cases
(name, description, location, priority, created_by, metadata)
VALUES (?, ?, ?, ?, ?, ?)
''', (name, description, location, priority, created_by,
json.dumps(metadata) if metadata else None))
return cursor.lastrowid
def get_tscm_case(case_id: int) -> dict | None:
"""Get a TSCM case by ID."""
with get_db() as conn:
cursor = conn.execute('SELECT * FROM tscm_cases WHERE id = ?', (case_id,))
row = cursor.fetchone()
if not row:
return None
case = {
'id': row['id'],
'name': row['name'],
'description': row['description'],
'location': row['location'],
'status': row['status'],
'priority': row['priority'],
'created_at': row['created_at'],
'updated_at': row['updated_at'],
'closed_at': row['closed_at'],
'created_by': row['created_by'],
'assigned_to': row['assigned_to'],
'notes': row['notes'],
'metadata': json.loads(row['metadata']) if row['metadata'] else None,
'sweeps': [],
'threats': [],
'case_notes': []
}
# Get linked sweeps
cursor = conn.execute('''
SELECT s.* FROM tscm_sweeps s
JOIN tscm_case_sweeps cs ON s.id = cs.sweep_id
WHERE cs.case_id = ?
ORDER BY s.started_at DESC
''', (case_id,))
case['sweeps'] = [dict(row) for row in cursor]
# Get linked threats
cursor = conn.execute('''
SELECT t.* FROM tscm_threats t
JOIN tscm_case_threats ct ON t.id = ct.threat_id
WHERE ct.case_id = ?
ORDER BY t.detected_at DESC
''', (case_id,))
case['threats'] = [dict(row) for row in cursor]
# Get case notes
cursor = conn.execute('''
SELECT * FROM tscm_case_notes
WHERE case_id = ?
ORDER BY created_at DESC
''', (case_id,))
case['case_notes'] = [dict(row) for row in cursor]
return case
def get_all_tscm_cases(
status: str | None = None,
limit: int = 50
) -> list[dict]:
"""Get all TSCM cases."""
conditions = []
params = []
if status:
conditions.append('status = ?')
params.append(status)
where_clause = f'WHERE {" AND ".join(conditions)}' if conditions else ''
params.append(limit)
with get_db() as conn:
cursor = conn.execute(f'''
SELECT * FROM tscm_cases
{where_clause}
ORDER BY updated_at DESC
LIMIT ?
''', params)
return [dict(row) for row in cursor]
def update_tscm_case(
case_id: int,
status: str | None = None,
priority: str | None = None,
assigned_to: str | None = None,
notes: str | None = None
) -> bool:
"""Update a TSCM case."""
updates = ['updated_at = CURRENT_TIMESTAMP']
params = []
if status:
updates.append('status = ?')
params.append(status)
if status == 'closed':
updates.append('closed_at = CURRENT_TIMESTAMP')
if priority:
updates.append('priority = ?')
params.append(priority)
if assigned_to is not None:
updates.append('assigned_to = ?')
params.append(assigned_to)
if notes is not None:
updates.append('notes = ?')
params.append(notes)
params.append(case_id)
with get_db() as conn:
cursor = conn.execute(
f'UPDATE tscm_cases SET {", ".join(updates)} WHERE id = ?',
params
)
return cursor.rowcount > 0
def add_sweep_to_case(case_id: int, sweep_id: int) -> bool:
"""Link a sweep to a case."""
with get_db() as conn:
try:
conn.execute('''
INSERT INTO tscm_case_sweeps (case_id, sweep_id)
VALUES (?, ?)
''', (case_id, sweep_id))
conn.execute(
'UPDATE tscm_cases SET updated_at = CURRENT_TIMESTAMP WHERE id = ?',
(case_id,)
)
return True
except sqlite3.IntegrityError:
return False
def add_threat_to_case(case_id: int, threat_id: int) -> bool:
"""Link a threat to a case."""
with get_db() as conn:
try:
conn.execute('''
INSERT INTO tscm_case_threats (case_id, threat_id)
VALUES (?, ?)
''', (case_id, threat_id))
conn.execute(
'UPDATE tscm_cases SET updated_at = CURRENT_TIMESTAMP WHERE id = ?',
(case_id,)
)
return True
except sqlite3.IntegrityError:
return False
def add_case_note(
case_id: int,
content: str,
note_type: str = 'general',
created_by: str | None = None
) -> int:
"""Add a note to a case."""
with get_db() as conn:
cursor = conn.execute('''
INSERT INTO tscm_case_notes (case_id, content, note_type, created_by)
VALUES (?, ?, ?, ?)
''', (case_id, content, note_type, created_by))
conn.execute(
'UPDATE tscm_cases SET updated_at = CURRENT_TIMESTAMP WHERE id = ?',
(case_id,)
)
return cursor.lastrowid
# =============================================================================
# TSCM Meeting Window Functions
# =============================================================================
def start_meeting_window(
sweep_id: int | None = None,
name: str | None = None,
location: str | None = None,
notes: str | None = None
) -> int:
"""Start a meeting window."""
with get_db() as conn:
cursor = conn.execute('''
INSERT INTO tscm_meeting_windows (sweep_id, name, start_time, location, notes)
VALUES (?, ?, CURRENT_TIMESTAMP, ?, ?)
''', (sweep_id, name, location, notes))
return cursor.lastrowid
def end_meeting_window(meeting_id: int) -> bool:
"""End a meeting window."""
with get_db() as conn:
cursor = conn.execute('''
UPDATE tscm_meeting_windows
SET end_time = CURRENT_TIMESTAMP
WHERE id = ? AND end_time IS NULL
''', (meeting_id,))
return cursor.rowcount > 0
def get_active_meeting_window(sweep_id: int | None = None) -> dict | None:
"""Get currently active meeting window."""
with get_db() as conn:
if sweep_id:
cursor = conn.execute('''
SELECT * FROM tscm_meeting_windows
WHERE sweep_id = ? AND end_time IS NULL
ORDER BY start_time DESC LIMIT 1
''', (sweep_id,))
else:
cursor = conn.execute('''
SELECT * FROM tscm_meeting_windows
WHERE end_time IS NULL
ORDER BY start_time DESC LIMIT 1
''')
row = cursor.fetchone()
if row:
return dict(row)
return None
def get_meeting_windows(sweep_id: int) -> list[dict]:
"""Get all meeting windows for a sweep."""
with get_db() as conn:
cursor = conn.execute('''
SELECT * FROM tscm_meeting_windows
WHERE sweep_id = ?
ORDER BY start_time
''', (sweep_id,))
return [dict(row) for row in cursor]
# =============================================================================
# TSCM Sweep Capabilities Functions
# =============================================================================
def save_sweep_capabilities(
sweep_id: int,
capabilities: dict,
limitations: list[str] | None = None
) -> int:
"""Save sweep capabilities snapshot."""
with get_db() as conn:
cursor = conn.execute('''
INSERT INTO tscm_sweep_capabilities (sweep_id, capabilities, limitations)
VALUES (?, ?, ?)
ON CONFLICT(sweep_id) DO UPDATE SET
capabilities = excluded.capabilities,
limitations = excluded.limitations,
recorded_at = CURRENT_TIMESTAMP
''', (sweep_id, json.dumps(capabilities),
json.dumps(limitations) if limitations else None))
return cursor.lastrowid
def get_sweep_capabilities(sweep_id: int) -> dict | None:
"""Get capabilities for a sweep."""
with get_db() as conn:
cursor = conn.execute(
'SELECT * FROM tscm_sweep_capabilities WHERE sweep_id = ?',
(sweep_id,)
)
row = cursor.fetchone()
if not row:
return None
return {
'sweep_id': row['sweep_id'],
'capabilities': json.loads(row['capabilities']),
'limitations': json.loads(row['limitations']) if row['limitations'] else [],
'recorded_at': row['recorded_at']
}
utils/tscm/advanced.py
+2152
View File
File diff suppressed because it is too large Load Diff
utils/tscm/reports.py
+813
View File
@@ -0,0 +1,813 @@
"""
TSCM Report Generation Module
Generates:
1. Client-safe PDF reports with executive summary
2. Technical annex (JSON + CSV) with device timelines and indicators
DISCLAIMER: All reports include mandatory disclaimers.
No packet data. No claims of confirmed surveillance.
"""
from __future__ import annotations
import csv
import io
import json
import logging
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Optional
logger = logging.getLogger('intercept.tscm.reports')
# =============================================================================
# Report Data Structures
# =============================================================================
@dataclass
class ReportFinding:
"""A single finding for the report."""
identifier: str
protocol: str
name: Optional[str]
risk_level: str
risk_score: int
description: str
indicators: list[dict] = field(default_factory=list)
recommended_action: str = ''
playbook_reference: str = ''
@dataclass
class ReportMeetingSummary:
"""Meeting window summary for report."""
name: Optional[str]
start_time: str
end_time: Optional[str]
duration_minutes: float
devices_first_seen: int
behavior_changes: int
high_interest_devices: int
@dataclass
class TSCMReport:
"""
Complete TSCM sweep report.
Contains all data needed for both client-safe PDF and technical annex.
"""
# Report metadata
report_id: str
generated_at: datetime
sweep_id: int
sweep_type: str
# Location and context
location: Optional[str] = None
baseline_id: Optional[int] = None
baseline_name: Optional[str] = None
# Executive summary
executive_summary: str = ''
overall_risk_assessment: str = 'low' # low, moderate, elevated, high
key_findings_count: int = 0
# Capabilities used
capabilities: dict = field(default_factory=dict)
limitations: list[str] = field(default_factory=list)
# Findings by risk tier
high_interest_findings: list[ReportFinding] = field(default_factory=list)
needs_review_findings: list[ReportFinding] = field(default_factory=list)
informational_findings: list[ReportFinding] = field(default_factory=list)
# Meeting window summaries
meeting_summaries: list[ReportMeetingSummary] = field(default_factory=list)
# Statistics
total_devices_scanned: int = 0
wifi_devices: int = 0
bluetooth_devices: int = 0
rf_signals: int = 0
new_devices: int = 0
missing_devices: int = 0
# Sweep duration
sweep_start: Optional[datetime] = None
sweep_end: Optional[datetime] = None
duration_minutes: float = 0.0
# Technical data (for annex only)
device_timelines: list[dict] = field(default_factory=list)
all_indicators: list[dict] = field(default_factory=list)
baseline_diff: Optional[dict] = None
correlation_data: list[dict] = field(default_factory=list)
# =============================================================================
# Disclaimer Text
# =============================================================================
REPORT_DISCLAIMER = """
IMPORTANT DISCLAIMER
This report documents the findings of a Technical Surveillance Countermeasures
(TSCM) sweep conducted using electronic detection equipment. The following
limitations and considerations apply:
1. DETECTION LIMITATIONS: No TSCM sweep can guarantee detection of all
surveillance devices. Sophisticated devices may evade detection.
2. FINDINGS ARE INDICATORS: All findings represent patterns and indicators,
NOT confirmed surveillance devices. Each finding requires professional
interpretation and may have legitimate explanations.
3. ENVIRONMENTAL FACTORS: Wireless signals are affected by building
construction, interference, and other environmental factors that may
impact detection accuracy.
4. POINT-IN-TIME ASSESSMENT: This report reflects conditions at the time
of the sweep. Conditions may change after the assessment.
5. NOT LEGAL ADVICE: This report does not constitute legal advice. Consult
qualified legal counsel for guidance on surveillance-related matters.
6. PRIVACY CONSIDERATIONS: Some detected devices may be legitimate personal
devices of authorized individuals.
This report should be treated as confidential and distributed only to
authorized personnel on a need-to-know basis.
"""
ANNEX_DISCLAIMER = """
TECHNICAL ANNEX DISCLAIMER
This annex contains detailed technical data from the TSCM sweep. This data
is provided for documentation and audit purposes.
- No raw packet captures or intercepted communications are included
- Device identifiers (MAC addresses) are included for tracking purposes
- Signal strength values are approximate and environment-dependent
- Timeline data is time-bucketed to preserve privacy
- All interpretations require professional TSCM expertise
This data should be handled according to organizational data protection
policies and applicable privacy regulations.
"""
# =============================================================================
# Report Generation Functions
# =============================================================================
def generate_executive_summary(report: TSCMReport) -> str:
"""Generate executive summary text."""
lines = []
# Opening
lines.append(f"TSCM Sweep Report - {report.location or 'Location Not Specified'}")
lines.append(f"Conducted: {report.sweep_start.strftime('%Y-%m-%d %H:%M') if report.sweep_start else 'Unknown'}")
lines.append(f"Duration: {report.duration_minutes:.0f} minutes")
lines.append("")
# Overall assessment
assessment_text = {
'low': 'No significant indicators of surveillance activity were detected.',
'moderate': 'Some devices require review but no confirmed surveillance indicators.',
'elevated': 'Multiple indicators warrant further investigation.',
'high': 'Significant indicators detected requiring immediate attention.',
}
lines.append(f"OVERALL ASSESSMENT: {report.overall_risk_assessment.upper()}")
lines.append(assessment_text.get(report.overall_risk_assessment, ''))
lines.append("")
# Key statistics
lines.append("SCAN STATISTICS:")
lines.append(f" - Total devices scanned: {report.total_devices_scanned}")
lines.append(f" - WiFi access points: {report.wifi_devices}")
lines.append(f" - Bluetooth devices: {report.bluetooth_devices}")
lines.append(f" - RF signals: {report.rf_signals}")
lines.append("")
# Findings summary
lines.append("FINDINGS SUMMARY:")
lines.append(f" - High Interest (require investigation): {len(report.high_interest_findings)}")
lines.append(f" - Needs Review: {len(report.needs_review_findings)}")
lines.append(f" - Informational: {len(report.informational_findings)}")
lines.append("")
# Baseline comparison if available
if report.baseline_name:
lines.append(f"BASELINE COMPARISON (vs '{report.baseline_name}'):")
lines.append(f" - New devices: {report.new_devices}")
lines.append(f" - Missing devices: {report.missing_devices}")
lines.append("")
# Meeting window summary if available
if report.meeting_summaries:
lines.append("MEETING WINDOW ACTIVITY:")
for meeting in report.meeting_summaries:
lines.append(f" - {meeting.name or 'Unnamed meeting'}: "
f"{meeting.devices_first_seen} new devices, "
f"{meeting.high_interest_devices} high interest")
lines.append("")
# Limitations
if report.limitations:
lines.append("SWEEP LIMITATIONS:")
for limit in report.limitations[:3]: # Top 3 limitations
lines.append(f" - {limit}")
lines.append("")
return "\n".join(lines)
def generate_findings_section(findings: list[ReportFinding], title: str) -> str:
"""Generate a findings section for the report."""
if not findings:
return f"{title}\n\nNo findings in this category.\n"
lines = [title, "=" * len(title), ""]
for i, finding in enumerate(findings, 1):
lines.append(f"{i}. {finding.name or finding.identifier}")
lines.append(f" Protocol: {finding.protocol.upper()}")
lines.append(f" Identifier: {finding.identifier}")
lines.append(f" Risk Score: {finding.risk_score}")
lines.append(f" Description: {finding.description}")
if finding.indicators:
lines.append(" Indicators:")
for ind in finding.indicators[:5]: # Limit to 5 indicators
lines.append(f" - {ind.get('type', 'unknown')}: {ind.get('description', '')}")
lines.append(f" Recommended Action: {finding.recommended_action}")
if finding.playbook_reference:
lines.append(f" Reference: {finding.playbook_reference}")
lines.append("")
return "\n".join(lines)
def generate_meeting_section(summaries: list[ReportMeetingSummary]) -> str:
"""Generate meeting window summary section."""
if not summaries:
return "MEETING WINDOW SUMMARY\n\nNo meeting windows were marked during this sweep.\n"
lines = ["MEETING WINDOW SUMMARY", "=" * 22, ""]
for meeting in summaries:
lines.append(f"Meeting: {meeting.name or 'Unnamed'}")
lines.append(f" Time: {meeting.start_time} - {meeting.end_time or 'ongoing'}")
lines.append(f" Duration: {meeting.duration_minutes:.0f} minutes")
lines.append(f" Devices first seen during meeting: {meeting.devices_first_seen}")
lines.append(f" Behavior changes detected: {meeting.behavior_changes}")
lines.append(f" High interest devices active: {meeting.high_interest_devices}")
if meeting.devices_first_seen > 0 or meeting.high_interest_devices > 0:
lines.append(" NOTE: Meeting-correlated activity detected - see findings for details")
lines.append("")
lines.append("Meeting-correlated activity indicates temporal correlation only.")
lines.append("Devices appearing during meetings may have legitimate explanations.")
lines.append("")
return "\n".join(lines)
def generate_pdf_content(report: TSCMReport) -> str:
"""
Generate complete PDF report content.
Returns plain text that can be converted to PDF.
For actual PDF generation, use a library like reportlab or weasyprint.
"""
sections = []
# Header
sections.append("=" * 70)
sections.append("TECHNICAL SURVEILLANCE COUNTERMEASURES (TSCM) SWEEP REPORT")
sections.append("=" * 70)
sections.append("")
sections.append(f"Report ID: {report.report_id}")
sections.append(f"Generated: {report.generated_at.strftime('%Y-%m-%d %H:%M:%S')}")
sections.append(f"Sweep ID: {report.sweep_id}")
sections.append("")
# Executive Summary
sections.append("-" * 70)
sections.append("EXECUTIVE SUMMARY")
sections.append("-" * 70)
sections.append(report.executive_summary or generate_executive_summary(report))
sections.append("")
# High Interest Findings
if report.high_interest_findings:
sections.append("-" * 70)
sections.append(generate_findings_section(
report.high_interest_findings,
"HIGH INTEREST FINDINGS"
))
# Needs Review Findings
if report.needs_review_findings:
sections.append("-" * 70)
sections.append(generate_findings_section(
report.needs_review_findings,
"FINDINGS REQUIRING REVIEW"
))
# Meeting Window Summary
if report.meeting_summaries:
sections.append("-" * 70)
sections.append(generate_meeting_section(report.meeting_summaries))
# Capabilities & Limitations
sections.append("-" * 70)
sections.append("SWEEP CAPABILITIES & LIMITATIONS")
sections.append("=" * 33)
sections.append("")
if report.capabilities:
caps = report.capabilities
sections.append("Equipment Used:")
if caps.get('wifi', {}).get('mode') != 'unavailable':
sections.append(f" - WiFi: {caps.get('wifi', {}).get('mode', 'unknown')} mode")
if caps.get('bluetooth', {}).get('mode') != 'unavailable':
sections.append(f" - Bluetooth: {caps.get('bluetooth', {}).get('mode', 'unknown')}")
if caps.get('rf', {}).get('available'):
sections.append(f" - RF/SDR: {caps.get('rf', {}).get('device_type', 'unknown')}")
sections.append("")
if report.limitations:
sections.append("Limitations:")
for limit in report.limitations:
sections.append(f" - {limit}")
sections.append("")
# Disclaimer
sections.append("-" * 70)
sections.append(REPORT_DISCLAIMER)
# Footer
sections.append("")
sections.append("=" * 70)
sections.append("END OF REPORT")
sections.append("=" * 70)
return "\n".join(sections)
def generate_technical_annex_json(report: TSCMReport) -> dict:
"""
Generate technical annex as JSON.
Contains detailed device timelines, all indicators, and raw data
for audit and further analysis.
"""
return {
'annex_type': 'tscm_technical_annex',
'report_id': report.report_id,
'generated_at': report.generated_at.isoformat(),
'sweep_id': report.sweep_id,
'disclaimer': ANNEX_DISCLAIMER.strip(),
'sweep_details': {
'type': report.sweep_type,
'location': report.location,
'start_time': report.sweep_start.isoformat() if report.sweep_start else None,
'end_time': report.sweep_end.isoformat() if report.sweep_end else None,
'duration_minutes': report.duration_minutes,
'baseline_id': report.baseline_id,
'baseline_name': report.baseline_name,
},
'capabilities': report.capabilities,
'limitations': report.limitations,
'statistics': {
'total_devices': report.total_devices_scanned,
'wifi_devices': report.wifi_devices,
'bluetooth_devices': report.bluetooth_devices,
'rf_signals': report.rf_signals,
'new_devices': report.new_devices,
'missing_devices': report.missing_devices,
'high_interest_count': len(report.high_interest_findings),
'needs_review_count': len(report.needs_review_findings),
'informational_count': len(report.informational_findings),
},
'findings': {
'high_interest': [
{
'identifier': f.identifier,
'protocol': f.protocol,
'name': f.name,
'risk_score': f.risk_score,
'description': f.description,
'indicators': f.indicators,
'recommended_action': f.recommended_action,
}
for f in report.high_interest_findings
],
'needs_review': [
{
'identifier': f.identifier,
'protocol': f.protocol,
'name': f.name,
'risk_score': f.risk_score,
'description': f.description,
'indicators': f.indicators,
}
for f in report.needs_review_findings
],
},
'meeting_windows': [
{
'name': m.name,
'start_time': m.start_time,
'end_time': m.end_time,
'duration_minutes': m.duration_minutes,
'devices_first_seen': m.devices_first_seen,
'behavior_changes': m.behavior_changes,
'high_interest_devices': m.high_interest_devices,
}
for m in report.meeting_summaries
],
'device_timelines': report.device_timelines,
'all_indicators': report.all_indicators,
'baseline_diff': report.baseline_diff,
'correlations': report.correlation_data,
}
def generate_technical_annex_csv(report: TSCMReport) -> str:
"""
Generate device timeline data as CSV.
Provides spreadsheet-compatible format for further analysis.
"""
output = io.StringIO()
writer = csv.writer(output)
# Header
writer.writerow([
'identifier',
'protocol',
'name',
'risk_level',
'risk_score',
'first_seen',
'last_seen',
'observation_count',
'rssi_min',
'rssi_max',
'rssi_mean',
'rssi_stability',
'movement_pattern',
'meeting_correlated',
'indicators',
])
# Device data from timelines
for timeline in report.device_timelines:
indicators_str = '; '.join(
f"{i.get('type', '')}({i.get('score', 0)})"
for i in timeline.get('indicators', [])
)
signal = timeline.get('signal', {})
metrics = timeline.get('metrics', {})
movement = timeline.get('movement', {})
meeting = timeline.get('meeting_correlation', {})
writer.writerow([
timeline.get('identifier', ''),
timeline.get('protocol', ''),
timeline.get('name', ''),
timeline.get('risk_level', 'informational'),
timeline.get('risk_score', 0),
metrics.get('first_seen', ''),
metrics.get('last_seen', ''),
metrics.get('total_observations', 0),
signal.get('rssi_min', ''),
signal.get('rssi_max', ''),
signal.get('rssi_mean', ''),
signal.get('stability', ''),
movement.get('pattern', ''),
meeting.get('correlated', False),
indicators_str,
])
# Also add findings summary
writer.writerow([])
writer.writerow(['--- FINDINGS SUMMARY ---'])
writer.writerow(['identifier', 'protocol', 'risk_level', 'risk_score', 'description', 'recommended_action'])
all_findings = (
report.high_interest_findings +
report.needs_review_findings
)
for finding in all_findings:
writer.writerow([
finding.identifier,
finding.protocol,
finding.risk_level,
finding.risk_score,
finding.description,
finding.recommended_action,
])
return output.getvalue()
# =============================================================================
# Report Builder
# =============================================================================
class TSCMReportBuilder:
"""
Builder for constructing TSCM reports from sweep data.
Usage:
builder = TSCMReportBuilder(sweep_id=123)
builder.set_location("Conference Room A")
builder.add_capabilities(capabilities_dict)
builder.add_finding(finding)
report = builder.build()
"""
def __init__(self, sweep_id: int):
self.sweep_id = sweep_id
self.report = TSCMReport(
report_id=f"TSCM-{sweep_id}-{datetime.now().strftime('%Y%m%d%H%M%S')}",
generated_at=datetime.now(),
sweep_id=sweep_id,
sweep_type='standard',
)
def set_sweep_type(self, sweep_type: str) -> 'TSCMReportBuilder':
self.report.sweep_type = sweep_type
return self
def set_location(self, location: str) -> 'TSCMReportBuilder':
self.report.location = location
return self
def set_baseline(self, baseline_id: int, baseline_name: str) -> 'TSCMReportBuilder':
self.report.baseline_id = baseline_id
self.report.baseline_name = baseline_name
return self
def set_sweep_times(
self,
start: datetime,
end: Optional[datetime] = None
) -> 'TSCMReportBuilder':
self.report.sweep_start = start
self.report.sweep_end = end or datetime.now()
self.report.duration_minutes = (
(self.report.sweep_end - self.report.sweep_start).total_seconds() / 60
)
return self
def add_capabilities(self, capabilities: dict) -> 'TSCMReportBuilder':
self.report.capabilities = capabilities
self.report.limitations = capabilities.get('all_limitations', [])
return self
def add_finding(self, finding: ReportFinding) -> 'TSCMReportBuilder':
if finding.risk_level == 'high_interest':
self.report.high_interest_findings.append(finding)
elif finding.risk_level in ['review', 'needs_review']:
self.report.needs_review_findings.append(finding)
else:
self.report.informational_findings.append(finding)
return self
def add_findings_from_profiles(self, profiles: list[dict]) -> 'TSCMReportBuilder':
"""Add findings from correlation engine device profiles."""
for profile in profiles:
finding = ReportFinding(
identifier=profile.get('identifier', ''),
protocol=profile.get('protocol', ''),
name=profile.get('name'),
risk_level=profile.get('risk_level', 'informational'),
risk_score=profile.get('total_score', 0),
description=self._generate_finding_description(profile),
indicators=profile.get('indicators', []),
recommended_action=profile.get('recommended_action', 'monitor'),
playbook_reference=self._get_playbook_reference(profile),
)
self.add_finding(finding)
return self
def _generate_finding_description(self, profile: dict) -> str:
"""Generate description from profile indicators."""
indicators = profile.get('indicators', [])
if not indicators:
return f"{profile.get('protocol', 'Unknown').upper()} device detected"
# Use first indicator as primary description
primary = indicators[0]
desc = primary.get('description', 'Pattern detected')
if len(indicators) > 1:
desc += f" (+{len(indicators) - 1} additional indicators)"
return desc
def _get_playbook_reference(self, profile: dict) -> str:
"""Get playbook reference based on profile."""
risk_level = profile.get('risk_level', 'informational')
indicators = profile.get('indicators', [])
# Check for tracker
tracker_types = ['airtag_detected', 'tile_detected', 'smarttag_detected', 'known_tracker']
if any(i.get('type') in tracker_types for i in indicators):
if risk_level == 'high_interest':
return 'PB-001 (Tracker Detection)'
if risk_level == 'high_interest':
return 'PB-002 (Suspicious Device)'
elif risk_level in ['review', 'needs_review']:
return 'PB-003 (Unknown Device)'
return ''
def add_meeting_summary(self, summary: dict) -> 'TSCMReportBuilder':
"""Add meeting window summary."""
meeting = ReportMeetingSummary(
name=summary.get('name'),
start_time=summary.get('start_time', ''),
end_time=summary.get('end_time'),
duration_minutes=summary.get('duration_minutes', 0),
devices_first_seen=summary.get('devices_first_seen', 0),
behavior_changes=summary.get('behavior_changes', 0),
high_interest_devices=summary.get('high_interest_devices', 0),
)
self.report.meeting_summaries.append(meeting)
return self
def add_statistics(
self,
wifi: int = 0,
bluetooth: int = 0,
rf: int = 0,
new: int = 0,
missing: int = 0
) -> 'TSCMReportBuilder':
self.report.wifi_devices = wifi
self.report.bluetooth_devices = bluetooth
self.report.rf_signals = rf
self.report.total_devices_scanned = wifi + bluetooth + rf
self.report.new_devices = new
self.report.missing_devices = missing
return self
def add_device_timelines(self, timelines: list[dict]) -> 'TSCMReportBuilder':
self.report.device_timelines = timelines
return self
def add_all_indicators(self, indicators: list[dict]) -> 'TSCMReportBuilder':
self.report.all_indicators = indicators
return self
def add_baseline_diff(self, diff: dict) -> 'TSCMReportBuilder':
self.report.baseline_diff = diff
return self
def add_correlations(self, correlations: list[dict]) -> 'TSCMReportBuilder':
self.report.correlation_data = correlations
return self
def build(self) -> TSCMReport:
"""Build and return the complete report."""
# Calculate overall risk assessment
if self.report.high_interest_findings:
if len(self.report.high_interest_findings) >= 3:
self.report.overall_risk_assessment = 'high'
else:
self.report.overall_risk_assessment = 'elevated'
elif self.report.needs_review_findings:
self.report.overall_risk_assessment = 'moderate'
else:
self.report.overall_risk_assessment = 'low'
self.report.key_findings_count = (
len(self.report.high_interest_findings) +
len(self.report.needs_review_findings)
)
# Generate executive summary
self.report.executive_summary = generate_executive_summary(self.report)
return self.report
# =============================================================================
# Report Generation API Functions
# =============================================================================
def generate_report(
sweep_id: int,
sweep_data: dict,
device_profiles: list[dict],
capabilities: dict,
timelines: list[dict],
baseline_diff: Optional[dict] = None,
meeting_summaries: Optional[list[dict]] = None,
correlations: Optional[list[dict]] = None,
) -> TSCMReport:
"""
Generate a complete TSCM report from sweep data.
Args:
sweep_id: Sweep ID
sweep_data: Sweep dict from database
device_profiles: List of DeviceProfile dicts from correlation engine
capabilities: Capabilities dict
timelines: Device timeline dicts
baseline_diff: Optional baseline diff dict
meeting_summaries: Optional meeting summaries
correlations: Optional correlation data
Returns:
Complete TSCMReport
"""
builder = TSCMReportBuilder(sweep_id)
# Basic info
builder.set_sweep_type(sweep_data.get('sweep_type', 'standard'))
# Parse times
started_at = sweep_data.get('started_at')
completed_at = sweep_data.get('completed_at')
if started_at:
if isinstance(started_at, str):
started_at = datetime.fromisoformat(started_at.replace('Z', '+00:00')).replace(tzinfo=None)
if completed_at:
if isinstance(completed_at, str):
completed_at = datetime.fromisoformat(completed_at.replace('Z', '+00:00')).replace(tzinfo=None)
builder.set_sweep_times(started_at, completed_at)
# Capabilities
builder.add_capabilities(capabilities)
# Add findings from profiles
builder.add_findings_from_profiles(device_profiles)
# Statistics
results = sweep_data.get('results', {})
builder.add_statistics(
wifi=len(results.get('wifi', [])),
bluetooth=len(results.get('bluetooth', [])),
rf=len(results.get('rf', [])),
new=baseline_diff.get('summary', {}).get('new_devices', 0) if baseline_diff else 0,
missing=baseline_diff.get('summary', {}).get('missing_devices', 0) if baseline_diff else 0,
)
# Technical data
builder.add_device_timelines(timelines)
if baseline_diff:
builder.add_baseline_diff(baseline_diff)
if meeting_summaries:
for summary in meeting_summaries:
builder.add_meeting_summary(summary)
if correlations:
builder.add_correlations(correlations)
# Extract all indicators
all_indicators = []
for profile in device_profiles:
for ind in profile.get('indicators', []):
all_indicators.append({
'device': profile.get('identifier'),
'protocol': profile.get('protocol'),
**ind
})
builder.add_all_indicators(all_indicators)
return builder.build()
def get_pdf_report(report: TSCMReport) -> str:
"""Get PDF-ready report content."""
return generate_pdf_content(report)
def get_json_annex(report: TSCMReport) -> dict:
"""Get JSON technical annex."""
return generate_technical_annex_json(report)
def get_csv_annex(report: TSCMReport) -> str:
"""Get CSV technical annex."""
return generate_technical_annex_csv(report)