Add Incident / Correlation Engine

.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

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)

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"),

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: