From b8ef8ef9298f84841c7426b70d3f9c2ab12aa6fe Mon Sep 17 00:00:00 2001 From: maximus Date: Wed, 15 Jul 2026 22:01:06 +0330 Subject: [PATCH] Add bounded incident correlation engine --- .env.example | 11 + README.md | 2 + backend/app/main.py | 38 + backend/app/services/event_aggregator.py | 20 +- backend/app/services/incident_correlation.py | 680 ++++++++++++++++++ backend/app/services/live_ring_buffer.py | 3 + backend/app/services/monitoring_service.py | 43 ++ .../app/services/sniffer_event_publisher.py | 4 + backend/app/services/sniffer_service.py | 49 +- docs/ARCHITECTURE.md | 15 + docs/INCIDENT_CORRELATION.md | 81 +++ docs/PERFORMANCE_PIPELINE.md | 9 +- frontend/src/App.jsx | 11 + frontend/src/components/AppNav.jsx | 1 + frontend/src/components/IncidentsPanel.jsx | 96 +++ .../src/components/IncidentsPanel.test.jsx | 47 ++ frontend/src/components/OpsPanel.jsx | 20 + frontend/src/components/OpsPanel.test.jsx | 21 + frontend/src/hooks/useApiClient.js | 5 + frontend/src/hooks/useDashboardController.js | 2 +- frontend/src/lib/opsHealth.js | 34 +- frontend/src/styles.css | 113 +++ tests/test_event_aggregator.py | 12 + tests/test_incident_correlation.py | 136 ++++ tests/test_live_ring_buffer.py | 15 +- tests/test_release_readiness.py | 21 + 26 files changed, 1467 insertions(+), 22 deletions(-) create mode 100644 backend/app/services/incident_correlation.py create mode 100644 docs/INCIDENT_CORRELATION.md create mode 100644 frontend/src/components/IncidentsPanel.jsx create mode 100644 frontend/src/components/IncidentsPanel.test.jsx create mode 100644 tests/test_incident_correlation.py diff --git a/.env.example b/.env.example index b0738b4..b7d93e6 100644 --- a/.env.example +++ b/.env.example @@ -46,6 +46,7 @@ NETBOT_LIVE_RING_ENABLED=true NETBOT_LIVE_RING_PACKET_MAX=5000 NETBOT_LIVE_RING_FLOW_MAX=2000 NETBOT_LIVE_RING_ALERT_MAX=1000 +NETBOT_LIVE_RING_INCIDENT_MAX=1000 NETBOT_LIVE_RING_EXPERT_MAX=1000 NETBOT_LIVE_RING_PROTOCOL_MAX=2000 NETBOT_LIVE_RING_AGENT_MAX=1000 @@ -61,6 +62,16 @@ NETBOT_SERVICE_ATTRIBUTION_DNS_WINDOW_SEC=300 NETBOT_SERVICE_ATTRIBUTION_MAX_REASONS=8 NETBOT_SERVICE_ATTRIBUTION_UNKNOWN_RATE_WARN=0.75 +# Bounded, read-only correlation of redacted analysis signals. +NETBOT_INCIDENTS_ENABLED=true +NETBOT_INCIDENT_CORRELATION_WINDOW_SEC=600 +NETBOT_INCIDENT_MAX_OPEN=1000 +NETBOT_INCIDENT_MAX_SIGNALS_PER_INCIDENT=500 +NETBOT_INCIDENT_RETENTION_HOURS=24 +NETBOT_INCIDENT_MIN_SEVERITY=low +NETBOT_INCIDENT_HIGH_SIGNAL_THRESHOLD=5 +NETBOT_INCIDENT_CRITICAL_SIGNAL_THRESHOLD=10 + # Bounded batch persistence. Counters and health are exposed in Ops Snapshot. NETBOT_PERSISTENCE_BATCH_ENABLED=true NETBOT_PERSISTENCE_PACKET_BATCH_SIZE=500 diff --git a/README.md b/README.md index bcde07a..6da04d4 100644 --- a/README.md +++ b/README.md @@ -67,6 +67,7 @@ or PCAP artifacts. | Flow Risk Scoring | Active | Explainable `0..100` scoring covers alerts, volume, DNS failures, unusual protocols/ports, and destinations. | Risk is an investigation aid, not an automated verdict. | | Offline PCAP Flow Summary | Active | Offline analysis returns flows, conversations, protocol summaries, timelines, and risk distribution. | Existing API response fields remain compatible; output stays redacted. | | Service Attribution / Destination Intelligence | Active MVP | DNS, visible TLS SNI, HTTP Host, recent DNS-answer correlation, ASN metadata, and a local fingerprint registry produce confidence-scored destination labels with evidence. Browser/container identity and CDN infrastructure are handled conservatively; encrypted traffic without evidence remains Unknown. | Local metadata-only inference; no external lookup, TLS decryption, browser history, cookies, credentials, or payload inspection. | +| Incident Correlation Engine | Active MVP | Related redacted alerts, flows, service attribution, and expert signals are grouped into bounded incidents with timeline, severity, confidence, and manual investigation guidance. | Read-only and in-memory in this release; no response actions, blocking, AI decisions, or agent commands. | | Deep Packet Inspection | Active MVP | Inspect renders a searchable layer tree, safe bytes view, streams, and Expert Info. | No TLS decryption; visible metadata and ASCII previews are centrally redacted. | | Display Filters | Active MVP | Safe packet and flow filter parser covers text, equality, range, and boolean operators. | Filters run on redacted metadata and never use Python `eval`. | | Offline PCAP Deep Analysis | Active MVP | Offline results include packet details, Expert Info, and stream summaries. | Previous API fields remain compatible; raw secrets are not exposed. | @@ -258,6 +259,7 @@ and surfaced in Ops Snapshot metrics. - [Performance Validation](docs/PERFORMANCE_VALIDATION.md) - [Flow Analysis And Protocol Intelligence](docs/FLOW_ANALYSIS.md) - [Service Attribution And Destination Intelligence](docs/SERVICE_ATTRIBUTION.md) +- [Incident Correlation](docs/INCIDENT_CORRELATION.md) - [Deep Packet Inspection](docs/DEEP_PACKET_INSPECTION.md) - [Agent Operational QA Checklist](docs/AGENT_QA_CHECKLIST.md) - [Deployment Overview](docs/DEPLOYMENT_OVERVIEW.md) diff --git a/backend/app/main.py b/backend/app/main.py index f0ad486..d15f808 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -120,6 +120,7 @@ def _observability_snapshot() -> dict[str, Any]: "flow_worker_pool": sniffer_service.flow_worker_pool_stats(), "live_ring_buffer": sniffer_service.live_ring_buffer_stats(), "service_attribution": sniffer_service.service_attribution_stats(), + "incidents": sniffer_service.incident_correlation_stats(), "persistence": sniffer_service.persistence_stats(), "auto_block": sniffer_service.auto_block_stats(), } @@ -250,6 +251,43 @@ def api_live_ring_metrics( return sniffer_service.live_ring_buffer_stats() +@app.get("/api/incidents") +def api_incidents( + status: str = "open", + severity: str = "", + limit: int = 100, + since: str | None = None, + _: None = Depends(require_trusted_client), + __: None = Depends(require_local_token), +) -> dict[str, Any]: + if status not in {"open", "resolved", "investigating", "suppressed", "all"}: + raise HTTPException(status_code=400, detail="Invalid incident status") + if severity and severity not in {"info", "low", "medium", "high", "critical"}: + raise HTTPException(status_code=400, detail="Invalid incident severity") + if not 1 <= limit <= 1000: + raise HTTPException( + status_code=400, detail="Incident limit must be between 1 and 1000" + ) + return sniffer_service.list_incidents( + status=status, + severity=severity, + limit=limit, + since=since, + ) + + +@app.get("/api/incidents/{incident_id}") +def api_incident_detail( + incident_id: str, + _: None = Depends(require_trusted_client), + __: None = Depends(require_local_token), +) -> dict[str, Any]: + incident = sniffer_service.get_incident(incident_id) + if incident is None: + raise HTTPException(status_code=404, detail="Incident not found") + return {"incident": incident} + + def _agent_headers( agent_id: str = Header("", alias="X-NetBot-Agent-Id"), agent_token: str = Header("", alias="X-NetBot-Agent-Token"), diff --git a/backend/app/services/event_aggregator.py b/backend/app/services/event_aggregator.py index bd67b01..8ca4715 100644 --- a/backend/app/services/event_aggregator.py +++ b/backend/app/services/event_aggregator.py @@ -8,7 +8,6 @@ from backend.app.services.redaction import redact_sensitive_data - VALID_SLOW_CLIENT_POLICIES = {"coalesce", "drop_oldest", "drop_newest"} @@ -71,6 +70,7 @@ def __init__( self._packet_events: list[dict[str, Any]] = [] self._alert_events: list[dict[str, Any]] = [] self._flow_events: list[dict[str, Any]] = [] + self._incident_events: list[dict[str, Any]] = [] self._agent_events: list[dict[str, Any]] = [] self._dashboard_summary: dict[str, Any] | None = None self._ops_health: dict[str, Any] | None = None @@ -95,6 +95,9 @@ def publish(self, event_type: str, payload: dict[str, Any]) -> None: if event_type.startswith("flow:"): self._append_buffer("flow", message, self.flow_batch_max) return + if event_type.startswith("incident:"): + self._append_buffer("incident", message, self.alert_batch_max) + return if event_type.startswith("agent:"): self._append_buffer("agent", message, self.flow_batch_max) return @@ -111,6 +114,7 @@ def flush_all(self) -> None: self._flush("packet") self._flush("alert") self._flush("flow") + self._flush("incident") self._flush("agent") self._flush("dashboard") self._flush("ops") @@ -130,6 +134,7 @@ def stats(self) -> dict[str, Any]: pending_packet = len(self._packet_events) pending_alert = len(self._alert_events) pending_flow = len(self._flow_events) + pending_incident = len(self._incident_events) batch_avg = round(mean(self._batch_sizes), 2) if self._batch_sizes else 0.0 pressure_reasons: list[str] = [] if self._events_dropped_total: @@ -154,6 +159,7 @@ def stats(self) -> dict[str, Any]: "pending_packet_events": pending_packet, "pending_alert_events": pending_alert, "pending_flow_events": pending_flow, + "pending_incident_events": pending_incident, "batches_sent_total": self._batches_sent_total, "events_received_total": self._events_received_total, "events_sent_total": self._events_sent_total, @@ -175,7 +181,9 @@ def record_coalesced(self, count: int = 1) -> None: with self._lock: self._events_coalesced_total += max(0, int(count)) - def _append_buffer(self, category: str, message: dict[str, Any], max_size: int) -> None: + def _append_buffer( + self, category: str, message: dict[str, Any], max_size: int + ) -> None: self._record_received() should_flush = False with self._lock: @@ -217,6 +225,10 @@ def _flush(self, category: str) -> None: events = self._flow_events self._flow_events = [] batch = self._batch("flow_delta", "updates", events) + elif category == "incident": + events = self._incident_events + self._incident_events = [] + batch = self._batch("incident_batch", "incidents", events) elif category == "agent": events = self._agent_events self._agent_events = [] @@ -257,6 +269,8 @@ def _delay_for(self, category: str) -> int: return self.alert_batch_ms if category == "flow": return self.flow_batch_ms + if category == "incident": + return self.alert_batch_ms if category == "agent": return self.agent_batch_ms return self.summary_batch_ms @@ -268,6 +282,8 @@ def _buffer_for(self, category: str) -> list[dict[str, Any]]: return self._alert_events if category == "flow": return self._flow_events + if category == "incident": + return self._incident_events return self._agent_events def _record_received(self) -> None: diff --git a/backend/app/services/incident_correlation.py b/backend/app/services/incident_correlation.py new file mode 100644 index 0000000..1293987 --- /dev/null +++ b/backend/app/services/incident_correlation.py @@ -0,0 +1,680 @@ +from __future__ import annotations + +import os +import threading +import time +import uuid +from collections import OrderedDict, deque +from copy import deepcopy +from datetime import datetime, timedelta, timezone +from statistics import mean +from typing import Any + +from backend.app.services.redaction import redact_sensitive_data, redact_sensitive_text + +SEVERITIES = ("info", "low", "medium", "high", "critical") +INCIDENT_TYPES = { + "possible_beaconing": { + "title": "Possible Beaconing", + "steps": [ + "Review the destination, process, and connection timing.", + "Compare related DNS and encrypted-flow evidence.", + ], + "false_positive": "Background updates and long-lived application sessions can look periodic.", + }, + "possible_port_scan": { + "title": "Possible Port Scan / Connection Sweep", + "steps": [ + "Review the source host and targeted ports or destinations.", + "Confirm whether an authorized discovery tool was running.", + ], + "false_positive": "Inventory, monitoring, and vulnerability scanners may be authorized.", + }, + "suspicious_dns": { + "title": "Suspicious DNS Activity", + "steps": [ + "Review related domains and DNS response patterns.", + "Compare the requesting process with expected host activity.", + ], + "false_positive": "CDNs and security products may generate many unique or high-entropy names.", + }, + "unusual_external_service": { + "title": "Unusual External Service", + "steps": [ + "Validate the external destination and attributed service.", + "Confirm that the source application is expected to make this connection.", + ], + "false_positive": "New SaaS endpoints and shared CDN infrastructure can reduce attribution confidence.", + }, + "data_exfiltration_indicator": { + "title": "Data Exfiltration Indicator", + "steps": [ + "Review outbound byte volume, duration, and destination ownership.", + "Validate the source process and related alerts without exposing payload data.", + ], + "false_positive": "Authorized backups, uploads, and software distribution can produce high outbound volume.", + }, +} + + +def _env_bool(name: str, default: bool) -> bool: + value = str(os.environ.get(name, str(default))).strip().lower() + if value in {"1", "true", "yes", "on"}: + return True + if value in {"0", "false", "no", "off"}: + return False + return default + + +def _env_int(name: str, default: int, minimum: int, maximum: int) -> int: + try: + value = int(os.environ.get(name, str(default))) + except (TypeError, ValueError): + return default + return value if minimum <= value <= maximum else default + + +def _parse_time(value: Any) -> datetime: + if isinstance(value, datetime): + parsed = value + else: + try: + parsed = datetime.fromisoformat(str(value or "").replace("Z", "+00:00")) + except (TypeError, ValueError): + parsed = datetime.now(timezone.utc) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def _string(value: Any, limit: int = 300) -> str: + return redact_sensitive_text(str(value or "").strip())[:limit] + + +def _list(value: Any, limit: int = 20) -> list[str]: + rows = value if isinstance(value, (list, tuple, set)) else [value] + return list(dict.fromkeys(_string(item) for item in rows if _string(item)))[:limit] + + +class IncidentCorrelationEngine: + """Bounded, deterministic correlation over already-derived metadata signals.""" + + def __init__( + self, + *, + enabled: bool = True, + correlation_window_sec: int = 600, + max_open: int = 1000, + max_signals_per_incident: int = 500, + retention_hours: int = 24, + min_severity: str = "low", + high_signal_threshold: int = 5, + critical_signal_threshold: int = 10, + ) -> None: + self.enabled = bool(enabled) + self.correlation_window_sec = max(10, min(int(correlation_window_sec), 86400)) + self.max_open = max(1, min(int(max_open), 100_000)) + self.max_signals_per_incident = max(2, min(int(max_signals_per_incident), 5000)) + self.retention_hours = max(1, min(int(retention_hours), 8760)) + self.min_severity = min_severity if min_severity in SEVERITIES else "low" + self.high_signal_threshold = max(2, int(high_signal_threshold)) + self.critical_signal_threshold = max( + self.high_signal_threshold + 1, int(critical_signal_threshold) + ) + self._lock = threading.RLock() + self._incidents: OrderedDict[str, dict[str, Any]] = OrderedDict() + self._pending: OrderedDict[str, deque[dict[str, Any]]] = OrderedDict() + self._latencies: deque[float] = deque(maxlen=512) + self._created_total = 0 + self._updated_total = 0 + self._signals_received = 0 + self._signals_correlated = 0 + self._signals_ignored = 0 + self._signals_dropped = 0 + self._errors = 0 + self._last_created_at = "" + self._last_updated_at = "" + self._last_error = "" + + @classmethod + def from_env(cls) -> "IncidentCorrelationEngine": + return cls( + enabled=_env_bool("NETBOT_INCIDENTS_ENABLED", True), + correlation_window_sec=_env_int( + "NETBOT_INCIDENT_CORRELATION_WINDOW_SEC", 600, 10, 86400 + ), + max_open=_env_int("NETBOT_INCIDENT_MAX_OPEN", 1000, 1, 100_000), + max_signals_per_incident=_env_int( + "NETBOT_INCIDENT_MAX_SIGNALS_PER_INCIDENT", 500, 2, 5000 + ), + retention_hours=_env_int("NETBOT_INCIDENT_RETENTION_HOURS", 24, 1, 8760), + min_severity=os.environ.get("NETBOT_INCIDENT_MIN_SEVERITY", "low").lower(), + high_signal_threshold=_env_int( + "NETBOT_INCIDENT_HIGH_SIGNAL_THRESHOLD", 5, 2, 1000 + ), + critical_signal_threshold=_env_int( + "NETBOT_INCIDENT_CRITICAL_SIGNAL_THRESHOLD", 10, 3, 5000 + ), + ) + + def correlate( + self, + *, + packet: dict[str, Any], + flow: dict[str, Any] | None, + alerts: list[dict[str, Any]], + expert_items: list[dict[str, Any]], + ) -> list[dict[str, Any]]: + if not self.enabled: + return [] + signals = self._derive_signals(packet, flow or {}, alerts, expert_items) + updated: OrderedDict[str, dict[str, Any]] = OrderedDict() + for signal in signals: + incident = self.ingest_signal(signal) + if incident: + updated[incident["incident_id"]] = incident + return list(updated.values()) + + def ingest_signal(self, signal: dict[str, Any]) -> dict[str, Any] | None: + started = time.perf_counter() + with self._lock: + self._signals_received += 1 + try: + normalized = self._normalize_signal(signal) + if not self.enabled or not normalized: + self._signals_ignored += 1 + return None + self._cleanup_locked(normalized["timestamp_dt"]) + incident = self._matching_incident_locked(normalized) + if incident is None: + pending = self._pending_for_locked(normalized) + pending.append(normalized) + while len(pending) > self.max_signals_per_incident: + pending.popleft() + self._signals_dropped += 1 + if not self._threshold_met(pending): + self._signals_ignored += 1 + return None + incident = self._create_locked(list(pending)) + pending.clear() + else: + self._update_locked(incident, normalized) + self._signals_correlated += 1 + return self._public(incident) + except Exception as exc: # pragma: no cover - defensive hot path + self._errors += 1 + self._last_error = type(exc).__name__ + self._signals_ignored += 1 + return None + finally: + self._latencies.append((time.perf_counter() - started) * 1000.0) + + def list_incidents( + self, + *, + status: str = "open", + severity: str = "", + limit: int = 100, + since: str | None = None, + ) -> dict[str, Any]: + with self._lock: + self._cleanup_locked(datetime.now(timezone.utc)) + since_dt = _parse_time(since) if since else None + items = [] + for incident in reversed(self._incidents.values()): + if status != "all" and incident["status"] != status: + continue + if severity and incident["severity"] != severity: + continue + if since_dt and _parse_time(incident["last_seen"]) < since_dt: + continue + items.append(self._public(incident)) + if len(items) >= max(1, min(int(limit), 1000)): + break + return { + "items": items, + "count": len(items), + "generated_at": datetime.now(timezone.utc).isoformat(), + } + + def get_incident(self, incident_id: str) -> dict[str, Any] | None: + with self._lock: + incident = self._incidents.get(str(incident_id)) + return self._public(incident) if incident else None + + def reset(self) -> None: + with self._lock: + self._incidents.clear() + self._pending.clear() + + def metrics(self) -> dict[str, Any]: + with self._lock: + incidents = list(self._incidents.values()) + open_total = sum(row["status"] == "open" for row in incidents) + high_total = sum(row["severity"] == "high" for row in incidents) + critical_total = sum(row["severity"] == "critical" for row in incidents) + reasons: list[str] = [] + avg_latency = round(mean(self._latencies), 3) if self._latencies else 0.0 + ordered = sorted(self._latencies) + p95 = ( + round(ordered[min(len(ordered) - 1, int((len(ordered) - 1) * 0.95))], 3) + if ordered + else 0.0 + ) + if open_total >= max(1, int(self.max_open * 0.8)): + reasons.append("incident_high_open_count") + if self._signals_dropped: + reasons.append("incident_dropped_signals") + if p95 >= 50.0: + reasons.append("incident_correlation_latency") + if self._errors: + reasons.append("incident_engine_errors") + health = "degraded" if reasons else "healthy" + if self._errors >= 3 or open_total >= self.max_open: + health = "critical" + return redact_sensitive_data( + { + "enabled": self.enabled, + "health": health, + "open_total": open_total, + "created_total": self._created_total, + "updated_total": self._updated_total, + "resolved_total": sum( + row["status"] == "resolved" for row in incidents + ), + "suppressed_total": sum( + row["status"] == "suppressed" for row in incidents + ), + "signals_received_total": self._signals_received, + "signals_correlated_total": self._signals_correlated, + "signals_ignored_total": self._signals_ignored, + "signals_dropped_total": self._signals_dropped, + "high_severity_total": high_total, + "critical_severity_total": critical_total, + "avg_correlation_latency_ms": avg_latency, + "p95_correlation_latency_ms": p95, + "max_open_incidents": self.max_open, + "max_signals_per_incident": self.max_signals_per_incident, + "last_created_at": self._last_created_at, + "last_updated_at": self._last_updated_at, + "last_error": self._last_error, + "pressure_reasons": reasons, + } + ) + + def _derive_signals( + self, + packet: dict[str, Any], + flow: dict[str, Any], + alerts: list[dict[str, Any]], + experts: list[dict[str, Any]], + ) -> list[dict[str, Any]]: + base = { + "timestamp": packet.get("ts") or packet.get("timestamp"), + "source_host": packet.get("src") or flow.get("src_ip"), + "destination_host": packet.get("dst") or flow.get("dst_ip"), + "application": packet.get("process_name") or flow.get("process_name"), + "service": packet.get("service_name") or flow.get("service_name"), + "domain": packet.get("service_domain") or flow.get("service_domain"), + "category": packet.get("service_category") or flow.get("service_category"), + "flow_key": flow.get("flow_id") or packet.get("flow_id"), + "bytes_out": flow.get("bytes_sent") or packet.get("length") or 0, + } + signals: list[dict[str, Any]] = [] + for alert in alerts: + text = " ".join( + str(alert.get(key) or "") + for key in ("attack_type", "detail", "risk_reasons") + ) + incident_type = self._classify(text, packet, flow) + if incident_type: + signals.append( + { + **base, + "incident_type": incident_type, + "source": "alert", + "summary": alert.get("attack_type") or alert.get("detail"), + "severity": alert.get("severity") or "medium", + "alert_id": alert.get("id"), + "risk_reasons": alert.get("risk_reasons") + or [alert.get("detail")], + } + ) + combined = " ".join( + str(value or "") + for value in [ + flow.get("risk_reasons"), + packet.get("service_risk_hint"), + packet.get("service_name"), + ] + ) + flow_type = self._classify(combined, packet, flow) + if flow_type: + signals.append( + { + **base, + "incident_type": flow_type, + "source": "flow", + "summary": self._summary_for(flow_type, packet), + "severity": flow.get("risk_level") or "low", + "risk_reasons": flow.get("risk_reasons") + or packet.get("service_reasons"), + } + ) + for expert in experts[:5]: + text = " ".join( + str(expert.get(key) or "") for key in ("summary", "message", "category") + ) + incident_type = self._classify(text, packet, flow) + if incident_type: + signals.append( + { + **base, + "incident_type": incident_type, + "source": "expert_info", + "summary": expert.get("summary") or expert.get("message"), + "severity": expert.get("severity") or "low", + "risk_reasons": [text], + } + ) + return signals + + @staticmethod + def _classify(text: str, packet: dict[str, Any], flow: dict[str, Any]) -> str: + lowered = text.lower() + if ( + any(word in lowered for word in ("exfil", "large outbound", "upload spike")) + or int(flow.get("bytes_sent") or 0) >= 50_000_000 + ): + return "data_exfiltration_indicator" + if any( + word in lowered + for word in ( + "port scan", + "connection sweep", + "scan burst", + "many destination ports", + ) + ): + return "possible_port_scan" + if any( + word in lowered + for word in ( + "dns entropy", + "nxdomain", + "dga", + "dns tunnel", + "suspicious dns", + ) + ): + return "suspicious_dns" + encrypted = bool(packet.get("service_encrypted")) or str( + flow.get("app_protocol") or "" + ).upper() in {"TLS", "QUIC"} + unknown = bool(packet.get("service_unknown")) or "unknown encrypted" in lowered + if encrypted and unknown: + return "possible_beaconing" + if unknown or any( + word in lowered + for word in ( + "unusual destination", + "rare external", + "low confidence attribution", + ) + ): + return "unusual_external_service" + return "" + + @staticmethod + def _summary_for(incident_type: str, packet: dict[str, Any]) -> str: + service = _string(packet.get("service_name") or "unknown service", 100) + return f"Observed metadata associated with {service} for {INCIDENT_TYPES[incident_type]['title'].lower()}." + + def _normalize_signal(self, value: dict[str, Any]) -> dict[str, Any] | None: + if ( + not isinstance(value, dict) + or value.get("incident_type") not in INCIDENT_TYPES + ): + return None + safe = redact_sensitive_data(deepcopy(value)) + timestamp_dt = _parse_time(safe.get("timestamp")) + severity = str(safe.get("severity") or "low").lower() + if severity not in SEVERITIES: + severity = "low" + return { + "signal_id": _string(safe.get("signal_id") or uuid.uuid4().hex, 80), + "incident_type": safe["incident_type"], + "timestamp": timestamp_dt.isoformat(), + "timestamp_dt": timestamp_dt, + "source_host": _string(safe.get("source_host"), 100), + "destination_host": _string(safe.get("destination_host"), 100), + "application": _string(safe.get("application"), 100), + "service": _string(safe.get("service"), 100), + "domain": _string(safe.get("domain"), 253), + "category": _string(safe.get("category"), 100), + "flow_key": _string(safe.get("flow_key"), 100), + "alert_id": _string(safe.get("alert_id"), 100), + "agent_id": _string(safe.get("agent_id"), 100), + "source": _string(safe.get("source") or "analysis", 80), + "summary": _string(safe.get("summary") or "Related analysis signal", 500), + "severity": severity, + "risk_reasons": _list(safe.get("risk_reasons"), 20), + "bytes_out": max(0, int(safe.get("bytes_out") or 0)), + } + + def _pending_for_locked(self, signal: dict[str, Any]) -> deque[dict[str, Any]]: + key = f"{signal['incident_type']}|{signal['source_host'] or signal['flow_key'] or signal['destination_host']}" + pending = self._pending.setdefault(key, deque()) + cutoff = signal["timestamp_dt"] - timedelta(seconds=self.correlation_window_sec) + while pending and pending[0]["timestamp_dt"] < cutoff: + pending.popleft() + self._pending.move_to_end(key) + while len(self._pending) > self.max_open * 2: + _, removed = self._pending.popitem(last=False) + self._signals_dropped += len(removed) + return pending + + @staticmethod + def _threshold_met(signals: deque[dict[str, Any]]) -> bool: + if len(signals) < 2: + return False + dimensions = {signal.get("source") for signal in signals} + severe = any( + SEVERITIES.index(signal["severity"]) >= SEVERITIES.index("high") + for signal in signals + ) + return len(dimensions) >= 2 or severe or len(signals) >= 3 + + def _matching_incident_locked( + self, signal: dict[str, Any] + ) -> dict[str, Any] | None: + for incident in reversed(self._incidents.values()): + if ( + incident["status"] != "open" + or incident["type"] != signal["incident_type"] + ): + continue + if ( + abs( + ( + signal["timestamp_dt"] - _parse_time(incident["last_seen"]) + ).total_seconds() + ) + > self.correlation_window_sec + ): + continue + source_match = ( + signal["source_host"] + and signal["source_host"] in incident["source_hosts"] + ) + context_match = any( + [ + signal["flow_key"] + and signal["flow_key"] in incident["related_flows"], + signal["domain"] and signal["domain"] in incident["domains"], + signal["service"] and signal["service"] in incident["services"], + signal["destination_host"] + and signal["destination_host"] in incident["destination_hosts"], + ] + ) + if source_match and (context_match or incident["signal_count"] >= 2): + return incident + return None + + def _create_locked(self, signals: list[dict[str, Any]]) -> dict[str, Any]: + while len(self._incidents) >= self.max_open: + self._incidents.popitem(last=False) + self._signals_dropped += 1 + now = datetime.now(timezone.utc).isoformat() + first = signals[0] + definition = INCIDENT_TYPES[first["incident_type"]] + incident = { + "incident_id": f"inc-{uuid.uuid4().hex[:16]}", + "title": definition["title"], + "type": first["incident_type"], + "severity": "low", + "confidence": "low", + "status": "open", + "first_seen": min(row["timestamp"] for row in signals), + "last_seen": max(row["timestamp"] for row in signals), + "source_hosts": [], + "destination_hosts": [], + "applications": [], + "services": [], + "domains": [], + "categories": [], + "related_flows": [], + "related_alerts": [], + "related_agents": [], + "risk_reasons": [], + "evidence": [], + "timeline": [], + "recommended_investigation_steps": definition["steps"], + "false_positive_notes": [definition["false_positive"]], + "correlation_reasons": [], + "signal_count": 0, + "created_at": now, + "updated_at": now, + } + for signal in signals: + self._apply_signal(incident, signal) + self._incidents[incident["incident_id"]] = incident + self._created_total += 1 + self._last_created_at = now + return incident + + def _update_locked(self, incident: dict[str, Any], signal: dict[str, Any]) -> None: + self._apply_signal(incident, signal) + incident["updated_at"] = datetime.now(timezone.utc).isoformat() + self._incidents.move_to_end(incident["incident_id"]) + self._updated_total += 1 + self._last_updated_at = incident["updated_at"] + + def _apply_signal(self, incident: dict[str, Any], signal: dict[str, Any]) -> None: + incident["signal_count"] += 1 + incident["last_seen"] = max(incident["last_seen"], signal["timestamp"]) + mappings = ( + ("source_hosts", "source_host"), + ("destination_hosts", "destination_host"), + ("applications", "application"), + ("services", "service"), + ("domains", "domain"), + ("categories", "category"), + ("related_flows", "flow_key"), + ("related_alerts", "alert_id"), + ("related_agents", "agent_id"), + ) + for target, source in mappings: + value = signal[source] + if value and value not in incident[target]: + incident[target].append(value) + incident[target] = incident[target][-100:] + incident["risk_reasons"] = list( + dict.fromkeys(incident["risk_reasons"] + signal["risk_reasons"]) + )[-100:] + if signal["summary"] not in incident["evidence"]: + incident["evidence"].append(signal["summary"]) + incident["evidence"] = incident["evidence"][ + -self.max_signals_per_incident : + ] + incident["timeline"].append( + { + "timestamp": signal["timestamp"], + "event_type": "signal_correlated", + "summary": signal["summary"], + "source": signal["source"], + "flow_key": signal["flow_key"], + "alert_id": signal["alert_id"], + "service_name": signal["service"], + "domain": signal["domain"], + "severity": signal["severity"], + } + ) + incident["timeline"] = sorted( + incident["timeline"], key=lambda row: row["timestamp"] + )[-self.max_signals_per_incident :] + self._score(incident) + + def _score(self, incident: dict[str, Any]) -> None: + count = incident["signal_count"] + max_signal = max( + (SEVERITIES.index(row["severity"]) for row in incident["timeline"]), + default=1, + ) + if count >= self.critical_signal_threshold or max_signal >= 4: + severity = "critical" + elif count >= self.high_signal_threshold or max_signal >= 3: + severity = "high" + elif count >= 3 or max_signal >= 2: + severity = "medium" + else: + severity = "low" + if SEVERITIES.index(severity) < SEVERITIES.index(self.min_severity): + severity = self.min_severity + incident["severity"] = severity + dimensions = sum( + bool(incident[key]) + for key in ( + "source_hosts", + "related_flows", + "services", + "domains", + "applications", + "related_alerts", + ) + ) + incident["confidence"] = ( + "high" + if dimensions >= 5 and count >= 3 + else "medium" if dimensions >= 3 else "low" + ) + reasons = [] + if incident["source_hosts"]: + reasons.append("Same source host within correlation window") + if incident["domains"]: + reasons.append("Same service attribution domain") + if len(incident["evidence"]) >= 2: + reasons.append("Multiple related analysis signals") + if incident["related_flows"]: + reasons.append("Related flow metadata") + incident["correlation_reasons"] = reasons + + def _cleanup_locked(self, now: datetime) -> None: + cutoff = now - timedelta(hours=self.retention_hours) + for incident_id in list(self._incidents): + if _parse_time(self._incidents[incident_id]["last_seen"]) < cutoff: + del self._incidents[incident_id] + pending_cutoff = now - timedelta(seconds=self.correlation_window_sec) + for key in list(self._pending): + pending = self._pending[key] + while pending and pending[0]["timestamp_dt"] < pending_cutoff: + pending.popleft() + if not pending: + del self._pending[key] + + @staticmethod + def _public(incident: dict[str, Any]) -> dict[str, Any]: + return redact_sensitive_data(deepcopy(incident)) + + +__all__ = ["IncidentCorrelationEngine", "INCIDENT_TYPES", "SEVERITIES"] diff --git a/backend/app/services/live_ring_buffer.py b/backend/app/services/live_ring_buffer.py index 89efb46..fb102a7 100644 --- a/backend/app/services/live_ring_buffer.py +++ b/backend/app/services/live_ring_buffer.py @@ -15,6 +15,7 @@ "packet", "flow", "alert", + "incident", "expert_info", "protocol_metadata", "agent_status", @@ -25,6 +26,7 @@ "packet": 5000, "flow": 2000, "alert": 1000, + "incident": 1000, "expert_info": 1000, "protocol_metadata": 2000, "agent_status": 1000, @@ -35,6 +37,7 @@ "packet": "NETBOT_LIVE_RING_PACKET_MAX", "flow": "NETBOT_LIVE_RING_FLOW_MAX", "alert": "NETBOT_LIVE_RING_ALERT_MAX", + "incident": "NETBOT_LIVE_RING_INCIDENT_MAX", "expert_info": "NETBOT_LIVE_RING_EXPERT_MAX", "protocol_metadata": "NETBOT_LIVE_RING_PROTOCOL_MAX", "agent_status": "NETBOT_LIVE_RING_AGENT_MAX", diff --git a/backend/app/services/monitoring_service.py b/backend/app/services/monitoring_service.py index b8e9411..3cce28b 100644 --- a/backend/app/services/monitoring_service.py +++ b/backend/app/services/monitoring_service.py @@ -188,6 +188,7 @@ def build_monitoring_metrics( flow_worker_pool = dict(observability.get("flow_worker_pool") or {}) live_ring_buffer = dict(observability.get("live_ring_buffer") or {}) service_attribution = dict(observability.get("service_attribution") or {}) + incidents = dict(observability.get("incidents") or {}) event_aggregator = dict(observability.get("event_aggregator") or {}) websocket = dict(observability.get("websocket") or {}) persistence = dict(observability.get("persistence") or {}) @@ -283,6 +284,16 @@ def build_monitoring_metrics( pressure_reasons.extend( reason for reason in flow_worker_reasons if reason not in pressure_reasons ) + incident_reasons = [ + reason + for value in incidents.get("pressure_reasons") or [] + if (reason := str(value)).startswith("incident_") + and len(reason) <= 80 + and reason.replace("_", "").isalnum() + ] + pressure_reasons.extend( + reason for reason in incident_reasons if reason not in pressure_reasons + ) live_ring_reasons = _safe_live_ring_reasons( live_ring_buffer.get("pressure_reasons") or [] ) @@ -357,6 +368,7 @@ def build_monitoring_metrics( or flow_worker_pool.get("health") == "critical" or live_ring_buffer.get("health") == "critical" or service_attribution.get("health") == "critical" + or incidents.get("health") == "critical" ): health = "critical" @@ -395,6 +407,9 @@ def build_monitoring_metrics( ), "pending_alert_events": _int(event_aggregator.get("pending_alert_events")), "pending_flow_events": _int(event_aggregator.get("pending_flow_events")), + "pending_incident_events": _int( + event_aggregator.get("pending_incident_events") + ), "batches_sent_total": _int(event_aggregator.get("batches_sent_total")), "events_received_total": _int( event_aggregator.get("events_received_total") @@ -593,6 +608,7 @@ def build_monitoring_metrics( "packet", "flow", "alert", + "incident", "expert_info", "protocol_metadata", "agent_status", @@ -637,6 +653,33 @@ def build_monitoring_metrics( ), "pressure_reasons": service_attribution_reasons, }, + "incidents": { + "enabled": bool(incidents.get("enabled", True)), + "health": str(incidents.get("health") or "healthy"), + "open_total": _int(incidents.get("open_total")), + "created_total": _int(incidents.get("created_total")), + "updated_total": _int(incidents.get("updated_total")), + "resolved_total": _int(incidents.get("resolved_total")), + "suppressed_total": _int(incidents.get("suppressed_total")), + "signals_received_total": _int(incidents.get("signals_received_total")), + "signals_correlated_total": _int(incidents.get("signals_correlated_total")), + "signals_ignored_total": _int(incidents.get("signals_ignored_total")), + "signals_dropped_total": _int(incidents.get("signals_dropped_total")), + "high_severity_total": _int(incidents.get("high_severity_total")), + "critical_severity_total": _int(incidents.get("critical_severity_total")), + "avg_correlation_latency_ms": _number( + incidents.get("avg_correlation_latency_ms") + ), + "p95_correlation_latency_ms": _number( + incidents.get("p95_correlation_latency_ms") + ), + "max_open_incidents": _int(incidents.get("max_open_incidents")), + "max_signals_per_incident": _int(incidents.get("max_signals_per_incident")), + "last_created_at": str(incidents.get("last_created_at") or ""), + "last_updated_at": str(incidents.get("last_updated_at") or ""), + "last_error": _safe_persistence_error(incidents.get("last_error")), + "pressure_reasons": incident_reasons, + }, "persistence": { "enabled": bool( persistence.get("persistence_enabled", persistence.get("enabled")) diff --git a/backend/app/services/sniffer_event_publisher.py b/backend/app/services/sniffer_event_publisher.py index fdbe34c..d1e7a2f 100644 --- a/backend/app/services/sniffer_event_publisher.py +++ b/backend/app/services/sniffer_event_publisher.py @@ -16,5 +16,9 @@ def publish_alerts(self, alerts: list[dict[str, Any]]) -> None: for alert in alerts: self._event_bus.publish("alert:new", alert) + def publish_incidents(self, incidents: list[dict[str, Any]]) -> None: + for incident in incidents: + self._event_bus.publish("incident:update", incident) + def publish_state(self, event_type: str, state: dict[str, Any]) -> None: self._event_bus.publish(event_type, state) diff --git a/backend/app/services/sniffer_service.py b/backend/app/services/sniffer_service.py index 7a59c30..6cfb5da 100644 --- a/backend/app/services/sniffer_service.py +++ b/backend/app/services/sniffer_service.py @@ -12,6 +12,7 @@ from backend.app.services.event_bus import EventBus from backend.app.services.flow_service import FlowService from backend.app.services.flow_worker_pool import FlowWorkerPool +from backend.app.services.incident_correlation import IncidentCorrelationEngine from backend.app.services.live_ring_buffer import LiveRingBuffer from backend.app.services.packet_queue import BoundedPacketQueue from backend.app.services.service_attribution import ServiceAttributionEngine @@ -55,6 +56,7 @@ def __init__( flow_service: FlowService | None = None, live_ring_buffer: LiveRingBuffer | None = None, service_attribution: ServiceAttributionEngine | None = None, + incident_correlation: IncidentCorrelationEngine | None = None, ) -> None: self._lock = threading.Lock() self._event_bus = event_bus @@ -78,6 +80,9 @@ def __init__( self._service_attribution = ( service_attribution or ServiceAttributionEngine.from_env() ) + self._incident_correlation = ( + incident_correlation or IncidentCorrelationEngine.from_env() + ) self._packet_queue = BoundedPacketQueue( max_size=PACKET_QUEUE_MAX_SIZE, overflow_policy=PACKET_QUEUE_OVERFLOW_POLICY, @@ -199,9 +204,23 @@ def _process_packet(self, meta: dict[str, Any]) -> None: flow = self._flow_service.ingest(packet, alerts, persist=False) else: flow = self._flow_service.ingest(packet, alerts) + flow_key = str((flow or {}).get("flow_id") or flow_id_for(packet)) + try: + expert_items = packet_expert_items(packet, flow_key) + except Exception as exc: # pragma: no cover - defensive hot-path guard + logger.warning( + "live ring expert generation failed error_type=%s", + type(exc).__name__, + ) + expert_items = [] + incidents = self._incident_correlation.correlate( + packet=packet, + flow=flow, + alerts=alerts, + expert_items=expert_items, + ) self._state.add_packet(packet) self._state.add_alerts(alerts) - flow_key = str((flow or {}).get("flow_id") or flow_id_for(packet)) self._live_ring_buffer.append( "packet", packet, @@ -225,14 +244,6 @@ def _process_packet(self, meta: dict[str, Any]) -> None: timestamp=str(alert.get("ts") or alert.get("timestamp") or ""), source="detection_engine", ) - try: - expert_items = packet_expert_items(packet, flow_key) - except Exception as exc: # pragma: no cover - defensive hot-path guard - logger.warning( - "live ring expert generation failed error_type=%s", - type(exc).__name__, - ) - expert_items = [] for expert_item in expert_items: self._live_ring_buffer.append( "expert_info", @@ -241,11 +252,20 @@ def _process_packet(self, meta: dict[str, Any]) -> None: timestamp=str(expert_item.get("last_seen") or ""), source="expert_info", ) + for incident in incidents: + self._live_ring_buffer.append( + "incident", + incident, + flow_key=flow_key, + timestamp=str(incident.get("last_seen") or ""), + source="incident_correlation", + ) self._persistence.persist(packet, alerts) if self._central_flow_persistence and isinstance(flow, dict): self._persistence.persist_flow(flow) self._publisher.publish_packet(packet) self._publisher.publish_alerts(alerts) + self._publisher.publish_incidents(incidents) def _packet_worker_loop(self) -> None: while not self._packet_worker_stop.is_set() or not self._packet_queue.empty(): @@ -349,6 +369,7 @@ def reset_session(self) -> dict[str, Any]: self._flow_service.reset() self._live_ring_buffer.clear() self._service_attribution.reset_runtime() + self._incident_correlation.reset() state = self._state.state(running=running, iface=iface) state["observability"] = self.observability() self._publisher.publish_state("sniffer:reset", state) @@ -374,6 +395,15 @@ def live_ring_buffer_stats(self) -> dict[str, Any]: def service_attribution_stats(self) -> dict[str, Any]: return self._service_attribution.metrics() + def incident_correlation_stats(self) -> dict[str, Any]: + return self._incident_correlation.metrics() + + def list_incidents(self, **filters: Any) -> dict[str, Any]: + return self._incident_correlation.list_incidents(**filters) + + def get_incident(self, incident_id: str) -> dict[str, Any] | None: + return self._incident_correlation.get_incident(incident_id) + def recent_live_records( self, category: str = "all", @@ -399,6 +429,7 @@ def observability(self) -> dict[str, Any]: "flow_worker_pool": self.flow_worker_pool_stats(), "live_ring_buffer": self.live_ring_buffer_stats(), "service_attribution": self.service_attribution_stats(), + "incidents": self.incident_correlation_stats(), "persistence": self.persistence_stats(), "auto_block": self.auto_block_stats(), } diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index f6e5260..4cfb67d 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -214,6 +214,21 @@ proxies, shared CDNs, and connection reuse can legitimately leave a destination Unknown. No TLS decryption, MITM, browser history inspection, cookie/session inspection, or credential collection is part of this layer. +## Incident Correlation Layer + +After protocol analysis, alert generation, expert information, and Service +Attribution, `IncidentCorrelationEngine` consumes bounded, centrally redacted +metadata signals. It requires repeated or multi-source evidence before creating an +incident and matches later signals by source, flow, destination, application, +service/domain, and a bounded time window. + +Incident summaries and sorted bounded timelines are exposed through protected +read-only APIs, the Incident workspace, the Live Ring Buffer, and batched WebSocket +updates. Ops Monitoring reports engine count, latency, drops, and pressure. Incident +persistence is not enabled in this step; the in-memory store has explicit count and +retention limits. A future read-only analyst may consume this evidence, but no AI or +response action is part of the current architecture. + ## Performance Pipeline Foundation ### Batch Persistence Foundation diff --git a/docs/INCIDENT_CORRELATION.md b/docs/INCIDENT_CORRELATION.md new file mode 100644 index 0000000..c6ad03f --- /dev/null +++ b/docs/INCIDENT_CORRELATION.md @@ -0,0 +1,81 @@ +# Incident Correlation + +NetBotPro groups related, already-derived analysis signals into a smaller set of +explainable security incidents. The engine is local, deterministic, read-only, +bounded, and designed to reduce alert fatigue without claiming that a correlated +pattern is automatically malicious. + +## Signal Flow + +```text +Packet / flow metadata + -> protocol and detection analysis + -> service attribution and expert information + -> bounded incident correlation + -> Live Ring Buffer and WebSocket incident batches + -> Incident API and UI +``` + +The engine currently runs in memory. Incident persistence is intentionally left +for a later schema-reviewed step. Incidents age out after the configured retention +period and both incident count and timeline length have hard limits. + +## Supported Incident Types + +- **Possible Beaconing:** repeated encrypted traffic to an unknown destination, + optionally strengthened by related alerts or expert evidence. +- **Possible Port Scan / Connection Sweep:** existing scan or sweep signals from + the same source in the correlation window. +- **Suspicious DNS Activity:** existing entropy, NXDOMAIN, DGA, or DNS-tunnel + signals. +- **Unusual External Service:** repeated unknown, rare, or low-confidence external + service attribution. +- **Data Exfiltration Indicator:** existing exfiltration/upload signals or very + high outbound flow volume combined with related context. + +Agent/host-health incidents are not generated in this step because agent health +snapshots are not yet connected to the live correlation stream. No unsupported +detection is simulated. + +## Correlation And Scoring + +Signals are compared by source host, flow key, destination, application, service, +domain, alert ID, and time. A single weak signal does not create an incident. +Multiple related sources of evidence or repeated strong signals are required. + +Severity describes accumulated impact evidence: `info`, `low`, `medium`, `high`, +or `critical`. Confidence is `low`, `medium`, or `high` and describes how strongly +the signals appear to belong together, not whether activity is malicious. +Every incident includes correlation reasons, bounded evidence, recommended manual +investigation steps, and false-positive notes. + +## Timeline + +Timeline entries contain only a timestamp, event type, redacted summary, source, +and related safe identifiers. Entries are timestamp-sorted and capped by +`NETBOT_INCIDENT_MAX_SIGNALS_PER_INCIDENT`. + +## Configuration + +| Variable | Default | Purpose | +| --- | ---: | --- | +| `NETBOT_INCIDENTS_ENABLED` | `true` | Enables read-only correlation. | +| `NETBOT_INCIDENT_CORRELATION_WINDOW_SEC` | `600` | Maximum time between related signals. | +| `NETBOT_INCIDENT_MAX_OPEN` | `1000` | Hard cap for in-memory incidents. | +| `NETBOT_INCIDENT_MAX_SIGNALS_PER_INCIDENT` | `500` | Evidence and timeline cap per incident. | +| `NETBOT_INCIDENT_RETENTION_HOURS` | `24` | In-memory retention. | +| `NETBOT_INCIDENT_MIN_SEVERITY` | `low` | Minimum displayed severity. | +| `NETBOT_INCIDENT_HIGH_SIGNAL_THRESHOLD` | `5` | Signal count that promotes severity to high. | +| `NETBOT_INCIDENT_CRITICAL_SIGNAL_THRESHOLD` | `10` | Signal count that promotes severity to critical. | + +Invalid values fall back to bounded defaults. + +## Security Boundaries + +All signal data passes through central redaction. Raw payloads, credentials, +authorization headers, cookies, tokens, and sessions are not part of an incident. +The engine does not block hosts, execute response actions, send commands to agents, +decrypt TLS, inspect browser data, or change Remote Sensor and Agent boundaries. + +False positives remain possible. Analysts should validate context and use the +recommended steps before drawing conclusions. diff --git a/docs/PERFORMANCE_PIPELINE.md b/docs/PERFORMANCE_PIPELINE.md index f410c33..bb074da 100644 --- a/docs/PERFORMANCE_PIPELINE.md +++ b/docs/PERFORMANCE_PIPELINE.md @@ -584,7 +584,7 @@ behavior, or AI autonomous actions. ## Next Planned Steps -1. Incident / Correlation Engine +1. Validate Incident / Correlation Engine quality and false-positive rates 2. Read-only AI Analyst ### Recorded Product Direction @@ -596,7 +596,10 @@ registry. The attribution stage adds bounded local matching work; its average and p95 latency, unknown rate, errors, and registry health are visible in Ops Snapshot. Missing or weak evidence remains `Unknown / Encrypted`. -Incident correlation follows attribution, and a read-only AI Analyst follows -incident quality validation. None of these roadmap items authorize TLS +The Incident Correlation Engine now follows attribution. It consumes only bounded, +redacted alerts, flows, attribution results, and expert signals, and exposes its +own latency, count, and pressure metrics. Its in-memory incident and timeline +limits preserve the performance pipeline's bounded-data guarantees. A read-only +AI Analyst may follow incident quality validation. None of these roadmap items authorize TLS decryption, MITM, credential collection, command/control, autonomous actions, or Agent raw packet, payload, or PCAP forwarding. diff --git a/frontend/src/App.jsx b/frontend/src/App.jsx index 422dc95..16b8452 100644 --- a/frontend/src/App.jsx +++ b/frontend/src/App.jsx @@ -7,6 +7,7 @@ import { DeepPacketPanel } from "./components/DeepPacketPanel"; import { ExportPanel } from "./components/ExportPanel"; import { FocusedIpPanel } from "./components/FocusedIpPanel"; import { FlowsPanel } from "./components/FlowsPanel"; +import { IncidentsPanel } from "./components/IncidentsPanel"; import { HeroPanel } from "./components/HeroPanel"; import { LiveGraphPanel } from "./components/LiveGraphPanel"; import { MiniList } from "./components/MiniList"; @@ -46,6 +47,7 @@ const WORKSPACE_META = { monitor: { eyebrow: "Analyze", title: "Live Monitor", subtitle: "Capture status, packets, alerts, and operational health.", links: [["inspect", "Inspect packet"], ["flows", "Open flows"]] }, inspect: { eyebrow: "Analyze", title: "Packet Inspection", subtitle: "Investigate selected packets, alerts, streams, and protocol layers.", links: [["monitor", "Select packets"], ["flows", "Related flows"]] }, flows: { eyebrow: "Analyze", title: "Flows & Conversations", subtitle: "Review protocol-aware sessions, timelines, and explainable risk.", links: [["monitor", "Live packets"], ["inspect", "Inspect packet"], ["reports", "Reports"]] }, + incidents: { eyebrow: "Analyze", title: "Security Incidents", subtitle: "Review explainable groups of related alerts, flows, and service evidence.", links: [["monitor", "Live packets"], ["flows", "Related flows"], ["inspect", "Inspect evidence"]] }, agents: { eyebrow: "Operations", title: "Agent Fleet", subtitle: "Read-only server health, capture status, alerts, and risk history.", links: [["reports", "Fleet reports"], ["settings", "Settings"]] }, traceroute: { eyebrow: "Operations", title: "Traceroute", subtitle: "Probe and review a network route from the local backend.", links: [["monitor", "Monitor"], ["inspect", "Inspect"]] }, offline: { eyebrow: "Operations", title: "Offline PCAP Analysis", subtitle: "Analyze an authorized capture without starting live monitoring.", links: [["inspect", "Inspect"], ["flows", "Flows"]] }, @@ -608,6 +610,14 @@ function App() { ); + const incidentsPage = ( +
+ + + +
+ ); + const traceroutePage = (
@@ -661,6 +671,7 @@ function App() { monitor: monitorPage, inspect: inspectPage, flows: flowsPage, + incidents: incidentsPage, agents: agentsPage, settings: settingsPage, traceroute: traceroutePage, diff --git a/frontend/src/components/AppNav.jsx b/frontend/src/components/AppNav.jsx index 6205eaf..96c6ad2 100644 --- a/frontend/src/components/AppNav.jsx +++ b/frontend/src/components/AppNav.jsx @@ -5,6 +5,7 @@ const NAV_GROUPS = [ { id: "monitor", label: "Monitor" }, { id: "inspect", label: "Inspect" }, { id: "flows", label: "Flows" }, + { id: "incidents", label: "Incidents" }, ], }, { diff --git a/frontend/src/components/IncidentsPanel.jsx b/frontend/src/components/IncidentsPanel.jsx new file mode 100644 index 0000000..7f7740d --- /dev/null +++ b/frontend/src/components/IncidentsPanel.jsx @@ -0,0 +1,96 @@ +import { useEffect, useState } from "react"; + +function safeText(value) { + return String(value || "") + .replace(/\b(?:bearer|basic)\s+[A-Za-z0-9._~+/=-]+/gi, "[REDACTED]") + .replace(/\b(password|token|api_key|secret|session)\s*[:=]\s*[^\s,;]+/gi, "$1=[REDACTED]"); +} + +function TextList({ title, items = [] }) { + return ( +
+

{title}

+ {items.length ?
    {items.map((item, index) =>
  • {safeText(item)}
    • )}
:

No evidence available.

} +
+ ); +} + +export function IncidentsPanel({ api }) { + const [items, setItems] = useState([]); + const [selected, setSelected] = useState(null); + const [status, setStatus] = useState("open"); + const [severity, setSeverity] = useState(""); + const [loading, setLoading] = useState(false); + const [error, setError] = useState(""); + + async function load(nextStatus = status, nextSeverity = severity) { + if (!api) return; + setLoading(true); + setError(""); + try { + const result = await api.getIncidents({ status: nextStatus, severity: nextSeverity, limit: 200 }); + const nextItems = result.items || []; + setItems(nextItems); + if (!selected && nextItems[0]) await selectIncident(nextItems[0].incident_id); + } catch (err) { + setError(err?.message || "Unable to load incidents"); + } finally { + setLoading(false); + } + } + + async function selectIncident(id) { + if (!api || !id) return; + try { + const result = await api.getIncident(id); + setSelected(result.incident || null); + } catch (err) { + setError(err?.message || "Unable to load incident details"); + } + } + + useEffect(() => { load(); }, [api]); + + return ( +
+
+ + + +
+ {error ?

{safeText(error)}

: null} + {!loading && !items.length ?

No correlated incidents

The engine waits for multiple related signals before creating an incident.

: null} + {items.length ?
+
+ {items.map((incident) => ( + + ))} +
+ +
: null} +
+ ); +} diff --git a/frontend/src/components/IncidentsPanel.test.jsx b/frontend/src/components/IncidentsPanel.test.jsx new file mode 100644 index 0000000..1a8adcc --- /dev/null +++ b/frontend/src/components/IncidentsPanel.test.jsx @@ -0,0 +1,47 @@ +import { fireEvent, render, screen, waitFor } from "@testing-library/react"; +import { describe, expect, it, vi } from "vitest"; + +import { IncidentsPanel } from "./IncidentsPanel"; + +const incident = { + incident_id: "inc-1", + title: "Possible Beaconing", + type: "possible_beaconing", + severity: "high", + confidence: "high", + status: "open", + signal_count: 4, + source_hosts: ["10.0.0.5"], + applications: ["browser.exe"], + services: ["Unknown encrypted destination"], + domains: ["unknown.example"], + evidence: ["Repeated outbound TLS", "token=raw-secret"], + correlation_reasons: ["Same source host within correlation window"], + recommended_investigation_steps: ["Review the destination."], + false_positive_notes: ["Background updates may be periodic."], + related_flows: ["flow-1"], related_alerts: ["alert-1"], related_agents: [], + timeline: [{ timestamp: "2026-07-15T12:00:00Z", source: "alert", severity: "high", summary: "Bearer another-secret" }], +}; + +describe("IncidentsPanel", () => { + it("renders an empty state", async () => { + const api = { getIncidents: vi.fn().mockResolvedValue({ items: [] }), getIncident: vi.fn() }; + render(); + expect(await screen.findByText("No correlated incidents")).toBeTruthy(); + }); + + it("renders incident details, evidence, timeline, and masks sensitive mock data", async () => { + const api = { getIncidents: vi.fn().mockResolvedValue({ items: [incident] }), getIncident: vi.fn().mockResolvedValue({ incident }) }; + render(); + await screen.findAllByText("Possible Beaconing"); + await waitFor(() => expect(screen.getByText("Evidence")).toBeTruthy()); + expect(screen.getByText("Correlation reasons")).toBeTruthy(); + expect(screen.getByText("Timeline")).toBeTruthy(); + expect(screen.getByText("1 flows")).toBeTruthy(); + expect(document.body.textContent).not.toContain("raw-secret"); + expect(document.body.textContent).not.toContain("another-secret"); + expect(screen.getAllByText("high").length).toBeGreaterThan(0); + fireEvent.click(screen.getAllByRole("button", { name: "Refresh incidents" }).at(-1)); + await waitFor(() => expect(api.getIncidents).toHaveBeenCalledTimes(2)); + }); +}); diff --git a/frontend/src/components/OpsPanel.jsx b/frontend/src/components/OpsPanel.jsx index 2dc011e..02a18c1 100644 --- a/frontend/src/components/OpsPanel.jsx +++ b/frontend/src/components/OpsPanel.jsx @@ -39,6 +39,7 @@ export function OpsPanel({ observability, operationalMetrics = null, isRefreshin const flowWorkerLastDropReason = snapshot.safeFlowWorkerDropReason || "No drops recorded"; const liveRingBuffer = snapshot.liveRingBuffer; const serviceAttribution = snapshot.serviceAttribution; + const incidents = snapshot.incidents; const persistence = snapshot.persistence; const eventBus = snapshot.eventBus; const eventAggregator = snapshot.eventAggregator; @@ -142,6 +143,25 @@ export function OpsPanel({ observability, operationalMetrics = null, isRefreshin + +
+ + = incidents.max_open_incidents && incidents.max_open_incidents ? "degraded" : "healthy"} /> + + + + + + + +
+ + { expect(screen.getAllByText(`Health ${health}`).length).toBeGreaterThan(0); } ); + + it("renders incident engine pressure without exposing diagnostics", () => { + render( + + ); + expect(screen.getByText("Incident Correlation Engine")).toBeTruthy(); + expect(screen.getByText("Many incidents are open. Review high-severity incidents and suppress known benign patterns.")).toBeTruthy(); + expect(screen.getByText("Incident signals were dropped due to correlation pressure. Review incident limits and event volume.")).toBeTruthy(); + expect(document.body.textContent).not.toMatch(/ops-secret|authorization/i); + }); }); diff --git a/frontend/src/hooks/useApiClient.js b/frontend/src/hooks/useApiClient.js index cd15707..b4160ec 100644 --- a/frontend/src/hooks/useApiClient.js +++ b/frontend/src/hooks/useApiClient.js @@ -80,6 +80,11 @@ export function useApiClient(localToken) { getProtocolsSummary: () => request("/protocols/summary"), getProtocolIntelligence: () => request("/protocols/intelligence"), getFlowSummaryReport: () => request("/reports/flows/summary"), + getIncidents: (params = {}) => { + const query = buildQueryString(params); + return request(`/incidents${query ? `?${query}` : ""}`); + }, + getIncident: (id) => request(`/incidents/${encodeURIComponent(id)}`), getAgentsOverview: () => request("/agents/overview"), getAgentAlertsSummary: () => request("/agents/alerts/summary"), getAgentRiskSummary: () => request("/agents/risk/summary"), diff --git a/frontend/src/hooks/useDashboardController.js b/frontend/src/hooks/useDashboardController.js index 714a038..19ed84a 100644 --- a/frontend/src/hooks/useDashboardController.js +++ b/frontend/src/hooks/useDashboardController.js @@ -5,7 +5,7 @@ import { useLiveEvents } from "./useLiveEvents"; import { getPeerInfo, isPrivateIp } from "../lib/networkView"; export const PAGE_SIZE = 25; -const VALID_PAGES = new Set(["monitor", "inspect", "flows", "agents", "settings", "traceroute", "exports", "reports", "offline"]); +const VALID_PAGES = new Set(["monitor", "inspect", "flows", "incidents", "agents", "settings", "traceroute", "exports", "reports", "offline"]); const TIMELINE_BUCKETS = 30; const TIMELINE_BUCKET_MS = 2_000; const HISTORY_CACHE_TTL_MS = 5_000; diff --git a/frontend/src/lib/opsHealth.js b/frontend/src/lib/opsHealth.js index 74f25f2..707bb1c 100644 --- a/frontend/src/lib/opsHealth.js +++ b/frontend/src/lib/opsHealth.js @@ -114,7 +114,7 @@ export function buildOpsSnapshot(observability, operationalMetrics = null) { const liveRingBuffer = operationalMetrics?.live_ring_buffer || observability?.live_ring_buffer || {}; const safeLiveRingCategories = Object.fromEntries( Object.entries(liveRingBuffer.categories || {}) - .filter(([category]) => ["packet", "flow", "alert", "expert_info", "protocol_metadata", "agent_status", "ops_event"].includes(category)) + .filter(([category]) => ["packet", "flow", "alert", "incident", "expert_info", "protocol_metadata", "agent_status", "ops_event"].includes(category)) .map(([category, values]) => [category, { records: toNumber(values?.records), capacity: toNumber(values?.capacity), @@ -166,6 +166,23 @@ export function buildOpsSnapshot(observability, operationalMetrics = null) { pressure_reasons: safeServiceAttributionReasons, }; const persistence = operationalMetrics?.persistence || observability?.persistence || {}; + const rawIncidents = operationalMetrics?.incidents || observability?.incidents || {}; + const incidentReasons = Array.isArray(rawIncidents.pressure_reasons) + ? rawIncidents.pressure_reasons.filter((reason) => ["incident_high_open_count", "incident_signal_pressure", "incident_correlation_latency", "incident_dropped_signals", "incident_engine_errors"].includes(reason)) + : []; + const incidents = { + enabled: rawIncidents.enabled !== false, + health: ["healthy", "degraded", "critical"].includes(rawIncidents.health) ? rawIncidents.health : "healthy", + open_total: toNumber(rawIncidents.open_total), created_total: toNumber(rawIncidents.created_total), updated_total: toNumber(rawIncidents.updated_total), + resolved_total: toNumber(rawIncidents.resolved_total), suppressed_total: toNumber(rawIncidents.suppressed_total), + signals_received_total: toNumber(rawIncidents.signals_received_total), signals_correlated_total: toNumber(rawIncidents.signals_correlated_total), + signals_ignored_total: toNumber(rawIncidents.signals_ignored_total), signals_dropped_total: toNumber(rawIncidents.signals_dropped_total), + high_severity_total: toNumber(rawIncidents.high_severity_total), critical_severity_total: toNumber(rawIncidents.critical_severity_total), + avg_correlation_latency_ms: toNumber(rawIncidents.avg_correlation_latency_ms), p95_correlation_latency_ms: toNumber(rawIncidents.p95_correlation_latency_ms), + max_open_incidents: toNumber(rawIncidents.max_open_incidents), last_created_at: String(rawIncidents.last_created_at || ""), + last_updated_at: String(rawIncidents.last_updated_at || ""), last_error: /^[A-Za-z0-9_]{1,80}$/.test(String(rawIncidents.last_error || "")) ? String(rawIncidents.last_error) : "", + pressure_reasons: incidentReasons, + }; const history = observability?.history || {}; const autoBlock = observability?.auto_block || {}; const capture = operationalMetrics?.capture || {}; @@ -283,10 +300,11 @@ export function buildOpsSnapshot(observability, operationalMetrics = null) { const serviceAttributionLevel = safeServiceAttribution.enabled ? normalizeLevel(safeServiceAttribution.health) : "healthy"; + const incidentLevel = incidents.enabled ? normalizeLevel(incidents.health) : "healthy"; const freshnessLevel = ageSeconds == null || ageSeconds <= 120 ? "healthy" : ageSeconds <= 300 ? "warning" : "degraded"; const criticalFlows = toNumber(flows.risk_distribution?.critical); const highFlows = toNumber(flows.risk_distribution?.high); - const overall = worstLevel(backendLevel, freshnessLevel, packetQueueLevel, flowWorkerLevel, liveRingLevel, serviceAttributionLevel, persistenceLevel, streamLevel, queryLevel, autoBlockLevel); + const overall = worstLevel(backendLevel, freshnessLevel, packetQueueLevel, flowWorkerLevel, liveRingLevel, serviceAttributionLevel, incidentLevel, persistenceLevel, streamLevel, queryLevel, autoBlockLevel); const recommendedActions = []; if (backendLevel !== "healthy") { @@ -346,6 +364,10 @@ export function buildOpsSnapshot(observability, operationalMetrics = null) { if (safeServiceAttribution.attribution_errors_total > 0) { recommendedActions.push("Service attribution errors were observed. Inspect backend logs and recent flow metadata."); } + if (incidentReasons.includes("incident_high_open_count")) recommendedActions.push("Many incidents are open. Review high-severity incidents and suppress known benign patterns."); + if (incidents.signals_dropped_total > 0) recommendedActions.push("Incident signals were dropped due to correlation pressure. Review incident limits and event volume."); + if (incidentReasons.includes("incident_correlation_latency")) recommendedActions.push("Incident correlation latency is elevated. Review correlation window and signal volume."); + if (incidents.last_error) recommendedActions.push("Incident correlation errors occurred. Inspect backend logs and recent signal payloads."); if (criticalFlows > 0) { recommendedActions.push("Review critical flows and related alerts first."); } else if (highFlows > 0) { @@ -446,6 +468,12 @@ export function buildOpsSnapshot(observability, operationalMetrics = null) { hint: `Unknown ${safeServiceAttribution.unknown_flows_total} | High ${safeServiceAttribution.high_confidence_total}`, level: serviceAttributionLevel, }, + { + label: "Incidents", + value: String(incidents.open_total), + hint: `High ${incidents.high_severity_total} | Critical ${incidents.critical_severity_total}`, + level: incidentLevel, + }, { label: "Write Queue", value: String(queueSize), @@ -513,6 +541,8 @@ export function buildOpsSnapshot(observability, operationalMetrics = null) { liveRingLevel, serviceAttribution: safeServiceAttribution, serviceAttributionLevel, + incidents, + incidentLevel, persistence, history, autoBlock, diff --git a/frontend/src/styles.css b/frontend/src/styles.css index a7b89ac..e2894c5 100644 --- a/frontend/src/styles.css +++ b/frontend/src/styles.css @@ -1832,6 +1832,117 @@ a { gap: 4px; } +.incidents-workspace { + display: grid; + gap: 16px; + min-width: 0; +} + +.incident-toolbar { + display: grid; + grid-template-columns: minmax(150px, 200px) minmax(150px, 200px) max-content; + gap: 8px; + align-items: center; +} + +.incident-toolbar select { + width: 100%; + min-width: 150px; +} + +.incidents-layout { + display: grid; + grid-template-columns: minmax(280px, 0.8fr) minmax(0, 1.4fr); + gap: 12px; + min-width: 0; +} + +.incident-list, +.incident-detail { + min-width: 0; + border: 1px solid var(--border); + background: var(--panel-muted); +} + +.incident-list { + display: grid; + align-content: start; + max-height: 720px; + overflow-y: auto; +} + +.incident-row { + display: grid; + grid-template-columns: auto minmax(0, 1fr); + gap: 8px 10px; + width: 100%; + padding: 12px; + border: 0; + border-bottom: 1px solid var(--border); + border-radius: 0; + background: transparent; + color: inherit; + text-align: left; +} + +.incident-row:hover, +.incident-row-selected { + background: rgba(48, 198, 166, 0.09); +} + +.incident-row span:nth-child(2) { + display: grid; + gap: 4px; + min-width: 0; +} + +.incident-row small, +.incident-row em { + overflow-wrap: anywhere; + color: var(--muted); + font-style: normal; +} + +.incident-row em { + grid-column: 2; + font-size: 12px; +} + +.incident-detail { + display: grid; + align-content: start; + gap: 16px; + padding: 16px; + overflow-wrap: anywhere; +} + +.incident-detail-head { + display: flex; + justify-content: space-between; + gap: 12px; + align-items: flex-start; +} + +.incident-detail h3, +.incident-detail h4 { + margin: 0; +} + +.incident-detail section { + display: grid; + gap: 8px; +} + +.incident-related { + grid-template-columns: repeat(3, minmax(0, 1fr)); +} + +.incident-related span { + padding: 10px; + border: 1px solid var(--border); + text-align: center; +} + .flow-attribution-summary { min-width: 0; } @@ -2151,6 +2262,8 @@ button:disabled { } .flows-layout, + .incidents-layout, + .incident-toolbar, .flow-summary-grid, .flow-toolbar { grid-template-columns: 1fr; diff --git a/tests/test_event_aggregator.py b/tests/test_event_aggregator.py index dbebdce..6f3c919 100644 --- a/tests/test_event_aggregator.py +++ b/tests/test_event_aggregator.py @@ -70,6 +70,18 @@ def test_batches_flow_events_as_delta_updates(self): self.assertEqual(emitted[0]["count"], 2) self.assertEqual(emitted[0]["updates"][0]["payload"]["flow_id"], "flow-1") + def test_batches_and_redacts_incident_updates(self): + emitted = [] + aggregator = EventAggregator( + emitted.append, alert_batch_ms=10000, alert_batch_max=1 + ) + aggregator.publish( + "incident:update", {"incident_id": "inc-1", "evidence": "token=raw-secret"} + ) + self.assertEqual(emitted[0]["type"], "incident_batch") + self.assertNotIn("raw-secret", str(emitted[0])) + self.assertEqual(aggregator.stats()["pending_incident_events"], 0) + def test_batch_interval_triggers_flush(self): emitted = [] aggregator = EventAggregator( diff --git a/tests/test_incident_correlation.py b/tests/test_incident_correlation.py new file mode 100644 index 0000000..9488d82 --- /dev/null +++ b/tests/test_incident_correlation.py @@ -0,0 +1,136 @@ +from __future__ import annotations + +import unittest +from datetime import datetime, timedelta, timezone +from unittest.mock import patch + +from fastapi.testclient import TestClient + +from backend.app.main import app, sniffer_service +from backend.app.security import require_local_token, require_trusted_client +from backend.app.services.incident_correlation import IncidentCorrelationEngine + + +def signal(index: int = 1, **overrides): + value = { + "incident_type": "possible_beaconing", + "timestamp": ( + datetime(2026, 7, 15, tzinfo=timezone.utc) + timedelta(seconds=index) + ).isoformat(), + "source_host": "10.0.0.5", + "destination_host": "203.0.113.20", + "application": "browser.exe", + "service": "Unknown encrypted destination", + "domain": "unknown.example", + "flow_key": "flow-1", + "source": "alert" if index % 2 else "flow", + "summary": f"Repeated encrypted signal {index}", + "severity": "medium", + "risk_reasons": ["Unknown encrypted service"], + } + value.update(overrides) + return value + + +class IncidentCorrelationTests(unittest.TestCase): + def test_single_weak_signal_does_not_create_incident(self): + engine = IncidentCorrelationEngine() + self.assertIsNone(engine.ingest_signal(signal())) + self.assertEqual(engine.list_incidents()["count"], 0) + + def test_related_signals_create_and_update_incident(self): + engine = IncidentCorrelationEngine( + high_signal_threshold=3, critical_signal_threshold=5 + ) + engine.ingest_signal(signal(1)) + created = engine.ingest_signal(signal(2)) + updated = engine.ingest_signal(signal(3)) + + self.assertIsNotNone(created) + self.assertEqual(created["incident_id"], updated["incident_id"]) + self.assertEqual(updated["signal_count"], 3) + self.assertEqual(updated["severity"], "high") + self.assertEqual(updated["confidence"], "high") + self.assertTrue(updated["correlation_reasons"]) + self.assertTrue(updated["recommended_investigation_steps"]) + self.assertTrue(updated["false_positive_notes"]) + + def test_unrelated_sources_are_separated(self): + engine = IncidentCorrelationEngine() + engine.ingest_signal(signal(1)) + engine.ingest_signal(signal(2)) + engine.ingest_signal(signal(3, source_host="10.0.0.8", flow_key="flow-2")) + engine.ingest_signal(signal(4, source_host="10.0.0.8", flow_key="flow-2")) + self.assertEqual(engine.list_incidents()["count"], 2) + + def test_timeline_is_sorted_bounded_and_redacted(self): + engine = IncidentCorrelationEngine(max_signals_per_incident=3) + engine.ingest_signal(signal(2, summary="Authorization: Bearer raw-secret")) + incident = engine.ingest_signal(signal(1)) + for index in range(3, 7): + incident = engine.ingest_signal(signal(index)) + rendered = str(incident) + timestamps = [row["timestamp"] for row in incident["timeline"]] + self.assertNotIn("raw-secret", rendered) + self.assertLessEqual(len(timestamps), 3) + self.assertEqual(timestamps, sorted(timestamps)) + + def test_max_open_and_retention_are_bounded(self): + engine = IncidentCorrelationEngine(max_open=1, retention_hours=24) + engine.ingest_signal(signal(1)) + engine.ingest_signal(signal(2)) + engine.ingest_signal( + signal( + 3, + incident_type="suspicious_dns", + source_host="10.0.0.8", + flow_key="flow-2", + ) + ) + engine.ingest_signal( + signal( + 4, + incident_type="suspicious_dns", + source_host="10.0.0.8", + flow_key="flow-2", + ) + ) + self.assertEqual(engine.list_incidents(status="all")["count"], 1) + self.assertGreater(engine.metrics()["signals_dropped_total"], 0) + + def test_malformed_signal_is_ignored_and_metrics_are_recorded(self): + engine = IncidentCorrelationEngine() + self.assertIsNone(engine.ingest_signal({"password": "secret"})) + metrics = engine.metrics() + self.assertEqual(metrics["signals_received_total"], 1) + self.assertEqual(metrics["signals_ignored_total"], 1) + self.assertNotIn("secret", str(metrics)) + + +class IncidentApiTests(unittest.TestCase): + def setUp(self): + app.dependency_overrides[require_trusted_client] = lambda: None + app.dependency_overrides[require_local_token] = lambda: None + self.client = TestClient(app) + self.engine = IncidentCorrelationEngine() + self.engine.ingest_signal(signal(1)) + self.incident = self.engine.ingest_signal( + signal(2, summary="Cookie: token=api-secret") + ) + + def tearDown(self): + app.dependency_overrides.clear() + + def test_list_detail_and_monitoring_are_redacted(self): + with patch.object(sniffer_service, "_incident_correlation", self.engine): + listing = self.client.get("/api/incidents") + detail = self.client.get(f"/api/incidents/{self.incident['incident_id']}") + monitoring = self.client.get("/api/monitoring/metrics") + self.assertEqual(listing.status_code, 200) + self.assertEqual(detail.status_code, 200) + self.assertIn("incidents", monitoring.json()) + self.assertNotIn("api-secret", listing.text + detail.text + monitoring.text) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_live_ring_buffer.py b/tests/test_live_ring_buffer.py index 1df421d..749003e 100644 --- a/tests/test_live_ring_buffer.py +++ b/tests/test_live_ring_buffer.py @@ -10,10 +10,7 @@ from backend.app.main import app, sniffer_service from backend.app.security import require_local_token, require_trusted_client -from backend.app.services.live_ring_buffer import ( - DEFAULT_CAPACITIES, - LiveRingBuffer, -) +from backend.app.services.live_ring_buffer import DEFAULT_CAPACITIES, LiveRingBuffer from backend.app.services.monitoring_service import build_monitoring_metrics @@ -30,7 +27,15 @@ def test_creates_all_category_buffers_with_configured_capacity(self): self.assertEqual(metrics["categories"]["packet"]["capacity"], 2) self.assertEqual(metrics["categories"]["flow"]["capacity"], 4) - self.assertEqual(len(metrics["categories"]), 7) + self.assertEqual(len(metrics["categories"]), 8) + + def test_incident_records_are_supported_and_redacted(self): + ring = self.make_buffer() + ring.append( + "incident", {"id": "inc-1", "evidence": "Authorization: Bearer ring-secret"} + ) + rendered = str(ring.query("incident")) + self.assertNotIn("ring-secret", rendered) def test_appends_supported_live_records(self): ring = self.make_buffer() diff --git a/tests/test_release_readiness.py b/tests/test_release_readiness.py index d7d4e0a..3f5ab39 100644 --- a/tests/test_release_readiness.py +++ b/tests/test_release_readiness.py @@ -126,6 +126,27 @@ def test_service_attribution_docs_and_boundaries_are_explicit(self): ]: self.assertIn(boundary, service_lower) + def test_incident_correlation_docs_and_boundaries_are_explicit(self): + incident_docs = self._read("docs/INCIDENT_CORRELATION.md") + incident_lower = incident_docs.lower() + readme = self._read("README.md") + architecture = self._read("docs/ARCHITECTURE.md") + + self.assertIn("docs/INCIDENT_CORRELATION.md", readme) + self.assertIn("Incident Correlation Engine", readme) + self.assertIn("IncidentCorrelationEngine", architecture) + self.assertIn("NETBOT_INCIDENT_MAX_OPEN", incident_docs) + self.assertIn("single weak signal does not create", incident_lower) + for boundary in [ + "raw payloads", + "credentials", + "does not block hosts", + "execute response actions", + "send commands to agents", + "decrypt tls", + ]: + self.assertIn(boundary, incident_lower) + def test_performance_pipeline_docs_are_linked_and_scoped(self): readme = self._read("README.md") architecture = self._read("docs/ARCHITECTURE.md")