sequence detection

src/fgai/monitor.py

@@ -20,6 +20,7 @@ from .logs import local_in_failures, read_events, summarize_events, top_field_va
 from .mitigation import parse_allowlist, suggest_block_candidates
 from .policies import audit_policies, read_policies
 from .recommendations import build_recommendations
+from .sequences import detect_sequences
 from .threat_intel import ThreatIntelClient, enrich_ips, is_public_ip
 from .stream_profiles import parse_profiles
 
@@ -104,6 +105,9 @@ def build_status(
     baseline = BaselineStore(baseline_path) if baseline_path else None
     profiles = baseline.profiles({event.src_ip for event in events if event.src_ip}) if baseline else {}
     field_deviations = baseline.profile_deviations(events, stream_profiles) if baseline else {}
+    sequence_findings = detect_sequences(events)
+    for entity, findings in sequence_findings.items():
+        field_deviations.setdefault(entity, []).extend(findings)
     for deviations in field_deviations.values():
         for deviation in deviations:
             stream_id = str(deviation.get("stream_id", ""))
@@ -194,6 +198,7 @@ def build_status(
         },
         "event_context": build_event_context(events),
         "field_deviations": field_deviations,
+        "sequence_findings": sequence_findings,
         "feedback": feedback,
         "cross_source_correlations": correlations,
         "incidents": build_incidents(anomalies, field_deviations, correlations),

src/fgai/sequences.py

@@ -0,0 +1,123 @@
+from __future__ import annotations
+
+from collections import defaultdict
+from datetime import datetime
+
+from .detectors import event_detector_categories
+from .entities import event_entities
+from .logs import THREAT_ACTIONS
+from .models import LogEvent
+
+
+DEFAULT_SEQUENCE_PATTERNS = {
+    "dns_network_auth_sequence": ("dns_query", "network_connection", "auth_failure"),
+}
+
+
+def _timestamp(event: LogEvent, fallback: int) -> int:
+    value = event.fields.get("eventtime", event.fields.get("timestamp", ""))
+    if value:
+        try:
+            parsed = float(value)
+            while parsed > 10_000_000_000:
+                parsed /= 1_000
+            return int(parsed)
+        except ValueError:
+            try:
+                return int(datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp())
+            except ValueError:
+                pass
+    date = event.fields.get("date")
+    clock = event.fields.get("time")
+    if date and clock:
+        try:
+            return int(datetime.fromisoformat(f"{date}T{clock}").timestamp())
+        except ValueError:
+            pass
+    return fallback
+
+
+def _is_network_event(event: LogEvent) -> bool:
+    if event.fields.get("dstip") or event.fields.get("destination_ip") or event.fields.get("dst_ip"):
+        return True
+    if event.fields.get("dstport") or event.fields.get("destination_port") or event.fields.get("service"):
+        return True
+    return event.action in {"accept", "pass", "allowed", "allow", "close", "client-rst", "server-rst"} | THREAT_ACTIONS
+
+
+def _sample(event: LogEvent, value: str) -> dict[str, str]:
+    return {
+        "timestamp": event.fields.get("eventtime", event.fields.get("timestamp", "")),
+        "stream": event.fields.get("fgai_stream", event.fields.get("fgai_stream_id", "")),
+        "source": event.src_ip or event.fields.get("source", ""),
+        "destination": event.dst_ip or event.fields.get("query_domain", event.fields.get("qh", "")),
+        "action": event.action,
+        "severity": event.severity,
+        "service": event.fields.get("service", event.fields.get("query_type", "")),
+        "value": value,
+        "message": event.fields.get("message", event.fields.get("msg", ""))[:240],
+    }
+
+
+def detect_sequences(events: list[LogEvent], *, window_seconds: int = 900, limit: int = 50, patterns: dict[str, tuple[str, ...]] | None = None) -> dict[str, list[dict[str, object]]]:
+    """Detect ordered multi-event behavior for any entity shared across streams."""
+    patterns = patterns or DEFAULT_SEQUENCE_PATTERNS
+    by_entity: dict[str, list[tuple[int, LogEvent, set[str]]]] = defaultdict(list)
+    for index, event in enumerate(events):
+        categories = set(event_detector_categories(event))
+        if _is_network_event(event):
+            categories.add("network_connection")
+        if not categories:
+            continue
+        timestamp = _timestamp(event, index)
+        for identity in event_entities(event):
+            by_entity[identity["entity"]].append((timestamp, event, categories))
+
+    findings: dict[str, list[dict[str, object]]] = defaultdict(list)
+    for entity, items in by_entity.items():
+        ordered = sorted(items, key=lambda item: item[0])
+        for detector, sequence in patterns.items():
+            match: list[tuple[int, LogEvent, str]] = []
+            start_at = 0
+            for category in sequence:
+                found = next(
+                    (
+                        (timestamp, event, category)
+                        for timestamp, event, categories in ordered
+                        if timestamp >= start_at and (not match or timestamp <= match[0][0] + window_seconds) and category in categories
+                    ),
+                    None,
+                )
+                if not found:
+                    match = []
+                    break
+                match.append(found)
+                start_at = found[0]
+            if len(match) != len(sequence):
+                continue
+            duration = max(0, match[-1][0] - match[0][0])
+            streams = sorted({event.fields.get("fgai_stream", "") for _, event, _ in match} - {""})
+            score = 28 if len(streams) >= 2 else 20
+            findings[entity].append({
+                "detector": detector,
+                "field": "sequence",
+                "stream_id": "multi_stream",
+                "stream_name": "Multi-stream",
+                "stream_title": "Multi-stream",
+                "score": score,
+                "base_score": score,
+                "weight": 1.0,
+                "confidence": "medium" if len(streams) >= 2 else "low",
+                "baseline_samples": 0,
+                "baseline_scope": f"{window_seconds}s sequence window",
+                "reason": f"ordered {' -> '.join(sequence)} sequence observed in {duration}s",
+                "current": len(sequence),
+                "baseline": 0,
+                "value": detector,
+                "sample_values": streams,
+                "sample_events": [_sample(event, category) for _, event, category in match],
+            })
+            if sum(len(values) for values in findings.values()) >= limit:
+                return findings
+            break
+    return findings