feat: add category filter to TSCM report exports

Users can now choose which risk categories to include in generated
reports — High Interest, Needs Review, and Informational — via three
checkboxes that appear after a sweep completes. Applies to the HTML
report, PDF, JSON annex, and CSV annex. Defaults to High Interest +
Needs Review (Informational excluded) to keep reports concise.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
James Smith
2026-07-05 13:34:09 +01:00
parent 1c9bd34a28
commit 8b2879c71f
5 changed files with 153 additions and 76 deletions
+24 -2
View File
@@ -28,6 +28,22 @@ from utils.tscm.correlation import get_correlation_engine
logger = logging.getLogger('intercept.tscm')
_VALID_CATEGORIES = {'high_interest', 'needs_review', 'informational'}
def _parse_categories_param(raw: str) -> list[str] | None:
"""Parse a comma-separated `categories` query param into a validated list.
Returns None (meaning "all") when the param is absent or covers all three
categories, so callers can skip filtering in the common case.
"""
if not raw:
return None
cats = [c.strip() for c in raw.split(',') if c.strip() in _VALID_CATEGORIES]
if not cats or set(cats) == _VALID_CATEGORIES:
return None
return cats
# =============================================================================
# Threat Endpoints
@@ -264,6 +280,8 @@ def get_pdf_report():
if not sweep:
return jsonify({'status': 'error', 'message': 'Sweep not found'}), 404
categories = _parse_categories_param(request.args.get('categories', ''))
# Get data for report
correlation = get_correlation_engine()
profiles = [p.to_dict() for p in correlation.device_profiles.values()]
@@ -278,7 +296,8 @@ def get_pdf_report():
sweep_data=sweep,
device_profiles=profiles,
capabilities=caps,
timelines=timelines
timelines=timelines,
categories=categories,
)
pdf_content = get_pdf_report(report)
@@ -319,6 +338,8 @@ def get_technical_annex():
if not sweep:
return jsonify({'status': 'error', 'message': 'Sweep not found'}), 404
categories = _parse_categories_param(request.args.get('categories', ''))
# Get data for report
correlation = get_correlation_engine()
profiles = [p.to_dict() for p in correlation.device_profiles.values()]
@@ -333,7 +354,8 @@ def get_technical_annex():
sweep_data=sweep,
device_profiles=profiles,
capabilities=caps,
timelines=timelines
timelines=timelines,
categories=categories,
)
if format_type == 'csv':
+36 -11
View File
@@ -12260,6 +12260,7 @@
// Clear and reset the signal timeline for new sweep
SignalTimeline.clear();
document.getElementById('tscmReportBtn').style.display = 'none';
document.getElementById('tscmReportCategoryFilter').style.display = 'none';
// Show warnings if any devices unavailable
if (scanResult.warnings && scanResult.warnings.length > 0) {
@@ -12361,9 +12362,10 @@
document.getElementById('stopTscmBtn').style.display = 'none';
document.getElementById('tscmProgress').style.display = 'none';
// Show report button if we have any data
// Show report button and category filter if we have any data
const hasData = tscmWifiDevices.length > 0 || tscmBtDevices.length > 0 || tscmRfSignals.length > 0;
document.getElementById('tscmReportBtn').style.display = hasData ? 'block' : 'none';
document.getElementById('tscmReportCategoryFilter').style.display = hasData ? 'block' : 'none';
return postStopRequest(endpoint, timeoutMs);
}
@@ -12383,18 +12385,30 @@
...tscmRfSignals.map(d => ({ ...d, protocol: 'RF' }))
];
const highInterest = allDevices.filter(d => d.classification === 'high_interest' || d.score >= 6);
const needsReview = allDevices.filter(d => d.classification === 'review' || (d.score >= 3 && d.score < 6));
const informational = allDevices.filter(d => d.classification === 'informational' || d.score < 3);
const allHighInterest = allDevices.filter(d => d.classification === 'high_interest' || d.score >= 6);
const allNeedsReview = allDevices.filter(d => d.classification === 'review' || (d.score >= 3 && d.score < 6));
const allInformational = allDevices.filter(d => d.classification === 'informational' || d.score < 3);
// Determine overall assessment
// Apply category filter from sidebar checkboxes
const incHighInterest = document.getElementById('tscmCatHighInterest')?.checked ?? true;
const incNeedsReview = document.getElementById('tscmCatNeedsReview')?.checked ?? true;
const incInformational = document.getElementById('tscmCatInformational')?.checked ?? false;
const highInterest = incHighInterest ? allHighInterest : [];
const needsReview = incNeedsReview ? allNeedsReview : [];
const informational = incInformational ? allInformational : [];
const categoryNote = (!incHighInterest || !incNeedsReview || !incInformational)
? `<div style="font-size:11px; color:#6b7280; margin-top:8px;">Filtered: showing ${[incHighInterest && 'High Interest', incNeedsReview && 'Needs Review', incInformational && 'Informational'].filter(Boolean).join(', ') || 'none'} only</div>`
: '';
// Determine overall assessment (based on all data, not filter)
let assessment = 'LOW CONCERN';
let assessmentClass = 'informational';
if (highInterest.length > 0) {
assessment = `ELEVATED CONCERN: ${highInterest.length} high-interest item(s) detected requiring immediate attention`;
if (allHighInterest.length > 0) {
assessment = `ELEVATED CONCERN: ${allHighInterest.length} high-interest item(s) detected requiring immediate attention`;
assessmentClass = 'high-interest';
} else if (needsReview.length > 0) {
assessment = `MODERATE CONCERN: ${needsReview.length} item(s) requiring further review`;
} else if (allNeedsReview.length > 0) {
assessment = `MODERATE CONCERN: ${allNeedsReview.length} item(s) requiring further review`;
assessmentClass = 'needs-review';
} else {
assessment = 'LOW CONCERN: No anomalies flagged by automated scan. Manual inspection recommended for comprehensive assessment.';
@@ -12765,6 +12779,7 @@
<div class="assessment ${assessmentClass}">
<strong>Assessment:</strong> ${assessment}
</div>
${categoryNote}
</div>
${highInterest.length > 0 ? `
@@ -16216,9 +16231,18 @@
}
// Report Downloads
function _tscmCategoryParam() {
const cats = [];
if (document.getElementById('tscmCatHighInterest')?.checked) cats.push('high_interest');
if (document.getElementById('tscmCatNeedsReview')?.checked) cats.push('needs_review');
if (document.getElementById('tscmCatInformational')?.checked) cats.push('informational');
return cats.length < 3 ? `categories=${cats.join(',')}` : '';
}
async function tscmDownloadPdf() {
try {
const response = await fetch('/tscm/report/pdf');
const catParam = _tscmCategoryParam();
const response = await fetch(`/tscm/report/pdf${catParam ? '?' + catParam : ''}`);
if (response.ok) {
const blob = await response.blob();
const url = URL.createObjectURL(blob);
@@ -16241,7 +16265,8 @@
async function tscmDownloadAnnex(format) {
try {
const response = await fetch(`/tscm/report/annex?format=${format}`);
const catParam = _tscmCategoryParam();
const response = await fetch(`/tscm/report/annex?format=${format}${catParam ? '&' + catParam : ''}`);
if (response.ok) {
const blob = await response.blob();
const url = URL.createObjectURL(blob);
+17
View File
@@ -101,6 +101,23 @@
</div>
</div>
<!-- Report category filter (shown when sweep has data) -->
<div id="tscmReportCategoryFilter" style="display: none; margin-top: 8px; padding: 8px; background: rgba(255,255,255,0.04); border-radius: 4px; border: 1px solid var(--border-color);">
<label style="display: block; font-size: 10px; font-weight: 600; margin-bottom: 6px; color: var(--text-secondary); text-transform: uppercase; letter-spacing: 0.5px;">Include in Report</label>
<label style="display: flex; align-items: center; gap: 6px; font-size: 11px; margin-bottom: 4px; cursor: pointer;">
<input type="checkbox" id="tscmCatHighInterest" checked>
<span style="color: #ff3333; font-size: 9px;"></span> High Interest
</label>
<label style="display: flex; align-items: center; gap: 6px; font-size: 11px; margin-bottom: 4px; cursor: pointer;">
<input type="checkbox" id="tscmCatNeedsReview" checked>
<span style="color: #ffcc00; font-size: 9px;"></span> Needs Review
</label>
<label style="display: flex; align-items: center; gap: 6px; font-size: 11px; cursor: pointer;">
<input type="checkbox" id="tscmCatInformational">
<span style="color: #00cc00; font-size: 9px;"></span> Informational
</label>
</div>
<button class="preset-btn" id="tscmReportBtn" onclick="generateTscmReport()" style="display: none; width: 100%; margin-top: 8px; background: var(--accent-cyan); color: #000; font-weight: 600;">
Generate Report
</button>
+5 -4
View File
@@ -113,9 +113,9 @@ class TestMatchSignals:
def test_multi_range_signal_matched_by_any_range(self):
from utils.signal_db import match_signals
# POCSAG has ranges in 138-175 MHz and 450-470 MHz
# 162 MHz is in the first range (maritime VHF area, but also POCSAG territory)
results_vhf = match_signals(frequency_mhz=155.0)
results_uhf = match_signals(frequency_mhz=455.0)
# Use a high limit so the expanded DB doesn't push it out of the window
results_vhf = match_signals(frequency_mhz=155.0, limit=20)
results_uhf = match_signals(frequency_mhz=455.0, limit=20)
vhf_names = [r["name"] for r in results_vhf]
uhf_names = [r["name"] for r in results_uhf]
assert "POCSAG Pager" in vhf_names
@@ -124,7 +124,8 @@ class TestMatchSignals:
def test_region_mismatch_does_not_exclude_signal(self):
from utils.signal_db import match_signals
# PMR446 is EU/UK only; should still appear with US region but may score lower
results = match_signals(frequency_mhz=446.09375, region="US")
# Use a high limit since the expanded DB has more signals at 446 MHz
results = match_signals(frequency_mhz=446.09375, region="US", limit=20)
names = [r["name"] for r in results]
assert "PMR446 (Licence-Free UHF)" in names
+71 -59
View File
@@ -95,14 +95,14 @@ class TSCMReport:
# Meeting window summaries
meeting_summaries: list[ReportMeetingSummary] = field(default_factory=list)
# Statistics
total_devices_scanned: int = 0
wifi_devices: int = 0
wifi_clients: int = 0
bluetooth_devices: int = 0
rf_signals: int = 0
new_devices: int = 0
missing_devices: int = 0
# Statistics
total_devices_scanned: int = 0
wifi_devices: int = 0
wifi_clients: int = 0
bluetooth_devices: int = 0
rf_signals: int = 0
new_devices: int = 0
missing_devices: int = 0
# Sweep duration
sweep_start: datetime | None = None
@@ -194,13 +194,13 @@ def generate_executive_summary(report: TSCMReport) -> str:
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" - WiFi clients: {report.wifi_clients}")
lines.append(f" - Bluetooth devices: {report.bluetooth_devices}")
lines.append(f" - RF signals: {report.rf_signals}")
lines.append("")
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" - WiFi clients: {report.wifi_clients}")
lines.append(f" - Bluetooth devices: {report.bluetooth_devices}")
lines.append(f" - RF signals: {report.rf_signals}")
lines.append("")
# Findings summary
lines.append("FINDINGS SUMMARY:")
@@ -422,14 +422,14 @@ def generate_technical_annex_json(report: TSCMReport) -> dict:
'capabilities': report.capabilities,
'limitations': report.limitations,
'statistics': {
'total_devices': report.total_devices_scanned,
'wifi_devices': report.wifi_devices,
'wifi_clients': report.wifi_clients,
'bluetooth_devices': report.bluetooth_devices,
'rf_signals': report.rf_signals,
'new_devices': report.new_devices,
'missing_devices': report.missing_devices,
'statistics': {
'total_devices': report.total_devices_scanned,
'wifi_devices': report.wifi_devices,
'wifi_clients': report.wifi_clients,
'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),
@@ -776,23 +776,23 @@ class TSCMReportBuilder:
self.report.meeting_summaries.append(meeting)
return self
def add_statistics(
self,
wifi: int = 0,
wifi_clients: int = 0,
bluetooth: int = 0,
rf: int = 0,
new: int = 0,
missing: int = 0
) -> TSCMReportBuilder:
self.report.wifi_devices = wifi
self.report.wifi_clients = wifi_clients
self.report.bluetooth_devices = bluetooth
self.report.rf_signals = rf
self.report.total_devices_scanned = wifi + wifi_clients + bluetooth + rf
self.report.new_devices = new
self.report.missing_devices = missing
return self
def add_statistics(
self,
wifi: int = 0,
wifi_clients: int = 0,
bluetooth: int = 0,
rf: int = 0,
new: int = 0,
missing: int = 0
) -> TSCMReportBuilder:
self.report.wifi_devices = wifi
self.report.wifi_clients = wifi_clients
self.report.bluetooth_devices = bluetooth
self.report.rf_signals = rf
self.report.total_devices_scanned = wifi + wifi_clients + 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
@@ -847,6 +847,7 @@ def generate_report(
baseline_diff: dict | None = None,
meeting_summaries: list[dict] | None = None,
correlations: list[dict] | None = None,
categories: list[str] | None = None,
) -> TSCMReport:
"""
Generate a complete TSCM report from sweep data.
@@ -882,34 +883,45 @@ def generate_report(
# Capabilities
builder.add_capabilities(capabilities)
# Apply category filter before building findings
if categories:
_cat_map = {'needs_review': {'review', 'needs_review'}}
allowed = set()
for c in categories:
allowed |= _cat_map.get(c, {c})
device_profiles = [
p for p in device_profiles
if p.get('risk_level', 'informational') in allowed
]
# Add findings from profiles
builder.add_findings_from_profiles(device_profiles)
# Statistics
results = sweep_data.get('results', {})
wifi_count = results.get('wifi_count')
if wifi_count is None:
wifi_count = len(results.get('wifi_devices', results.get('wifi', [])))
wifi_client_count = results.get('wifi_client_count')
if wifi_client_count is None:
wifi_client_count = len(results.get('wifi_clients', []))
bt_count = results.get('bt_count')
if bt_count is None:
bt_count = len(results.get('bt_devices', results.get('bluetooth', [])))
results = sweep_data.get('results', {})
wifi_count = results.get('wifi_count')
if wifi_count is None:
wifi_count = len(results.get('wifi_devices', results.get('wifi', [])))
wifi_client_count = results.get('wifi_client_count')
if wifi_client_count is None:
wifi_client_count = len(results.get('wifi_clients', []))
bt_count = results.get('bt_count')
if bt_count is None:
bt_count = len(results.get('bt_devices', results.get('bluetooth', [])))
rf_count = results.get('rf_count')
if rf_count is None:
rf_count = len(results.get('rf_signals', results.get('rf', [])))
builder.add_statistics(
wifi=wifi_count,
wifi_clients=wifi_client_count,
bluetooth=bt_count,
rf=rf_count,
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,
builder.add_statistics(
wifi=wifi_count,
wifi_clients=wifi_client_count,
bluetooth=bt_count,
rf=rf_count,
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