diff --git a/correlation.py b/correlation.py new file mode 100644 index 0000000..33cfc48 --- /dev/null +++ b/correlation.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +from collections import defaultdict + +from .logs import THREAT_ACTIONS, is_utm_event +from .models import LogEvent + + +def correlate_source_ips(events: list[LogEvent], *, limit: int = 20) -> list[dict[str, object]]: + grouped: dict[str, list[LogEvent]] = defaultdict(list) + for event in events: + if event.src_ip: + grouped[event.src_ip].append(event) + correlations = [] + for source_ip, source_events in grouped.items(): + streams = sorted({event.fields.get("fgai_stream", "local_syslog") for event in source_events}) + if len(streams) < 2: + continue + threat_events = sum(event.action in THREAT_ACTIONS or is_utm_event(event) for event in source_events) + correlations.append({"source_ip": source_ip, "streams": streams, "events": len(source_events), "security_events": threat_events}) + return sorted(correlations, key=lambda item: (int(item["security_events"]), int(item["events"])), reverse=True)[:limit] diff --git a/src/fgai/dashboard.py b/src/fgai/dashboard.py index 0614ffb..76c036a 100644 --- a/src/fgai/dashboard.py +++ b/src/fgai/dashboard.py @@ -6,6 +6,7 @@ from pathlib import Path from .config import ConfigStore from .graylog_mcp import GraylogMcpClient +from .metrics import prometheus_metrics HTML = """ @@ -241,6 +242,13 @@ def serve_dashboard(host: str, port: int, status_file: str, *, image_dir: str | body = json.dumps({"summary": {}, "anomalies": [], "block_candidates": []}).encode("utf-8") self._send(200, "application/json", body) return + if self.path == "/metrics": + try: + status = json.loads(status_path.read_text(encoding="utf-8")) if status_path.exists() else {} + except json.JSONDecodeError: + status = {} + self._send(200, "text/plain; version=0.0.4; charset=utf-8", prometheus_metrics(status).encode("utf-8")) + return if self.path.startswith("/images/") and image_root: image_path = image_root / Path(self.path).name if image_path.exists() and image_path.is_file(): diff --git a/src/fgai/graylog_source.py b/src/fgai/graylog_source.py index fed2612..36ba153 100644 --- a/src/fgai/graylog_source.py +++ b/src/fgai/graylog_source.py @@ -41,10 +41,11 @@ def _records(value: object) -> Iterable[dict[str, object]]: class GraylogStreamSource: - def __init__(self, client: GraylogMcpClient, stream: str, query: str = "*", field_mapping: str = "") -> None: + def __init__(self, client: GraylogMcpClient, stream: str, query: str = "*", field_mapping: str = "", stream_label: str = "") -> None: self.client = client self.stream = stream self.query = query or "*" + self.stream_label = stream_label or stream try: self.mapping = json.loads(field_mapping) if field_mapping else {} except json.JSONDecodeError as exc: @@ -88,6 +89,7 @@ class GraylogStreamSource: def _event(self, record: dict[str, object]) -> LogEvent: fields = {str(key).lower(): str(value) for key, value in record.items() if value is not None} + fields["fgai_stream"] = self.stream_label for canonical, candidates in DEFAULT_FIELD_MAP.items(): mapped = self.mapping.get(canonical) candidates = (str(mapped),) if mapped else candidates diff --git a/src/fgai/llm.py b/src/fgai/llm.py index b258d45..e2972ed 100644 --- a/src/fgai/llm.py +++ b/src/fgai/llm.py @@ -59,6 +59,7 @@ def ollama_dashboard_assessment(analysis: dict[str, object], model: str | None = "event_context": analysis.get("event_context", {}), "diagnostics": analysis.get("diagnostics", {}), "capabilities": analysis.get("capabilities", {}), + "cross_source_correlations": analysis.get("cross_source_correlations", [])[:20], } return ollama_summary( [], diff --git a/src/fgai/metrics.py b/src/fgai/metrics.py new file mode 100644 index 0000000..4ac5b3a --- /dev/null +++ b/src/fgai/metrics.py @@ -0,0 +1,19 @@ +from __future__ import annotations + + +def prometheus_metrics(status: dict[str, object]) -> str: + summary = status.get("summary", {}) if isinstance(status.get("summary"), dict) else {} + anomalies = status.get("anomaly_summary", {}) if isinstance(status.get("anomaly_summary"), dict) else {} + baseline = status.get("baseline", {}) if isinstance(status.get("baseline"), dict) else {} + capabilities = status.get("capabilities", {}) if isinstance(status.get("capabilities"), dict) else {} + mcp = capabilities.get("graylog_mcp", {}) if isinstance(capabilities.get("graylog_mcp"), dict) else {} + lines = ["# HELP fgai_events_total Events in the latest analysis window.", "# TYPE fgai_events_total gauge"] + for key in ("total", "utm", "threat_actions", "critical_or_high"): + lines.append(f'fgai_events_total{{kind="{key}"}} {int(summary.get(key, 0) or 0)}') + lines += ["# HELP fgai_anomalies Anomalies by severity.", "# TYPE fgai_anomalies gauge"] + for severity in ("critical", "high", "medium", "low"): + lines.append(f'fgai_anomalies{{severity="{severity}"}} {int(anomalies.get(severity, 0) or 0)}') + lines += ["# HELP fgai_baseline_sources_ready Sources with sufficient baseline history.", "# TYPE fgai_baseline_sources_ready gauge", f'fgai_baseline_sources_ready {int(baseline.get("sources_ready", 0) or 0)}'] + lines += ["# HELP fgai_graylog_mcp_connected Graylog MCP connectivity state.", "# TYPE fgai_graylog_mcp_connected gauge", f'fgai_graylog_mcp_connected {1 if mcp.get("status") == "connected" else 0}'] + lines += ["# HELP fgai_graylog_events_fetched Events fetched from Graylog in the latest poll.", "# TYPE fgai_graylog_events_fetched gauge", f'fgai_graylog_events_fetched {int(mcp.get("events_fetched", 0) or 0)}'] + return "\n".join(lines) + "\n" diff --git a/src/fgai/monitor.py b/src/fgai/monitor.py index f3a88b5..5d7c47a 100644 --- a/src/fgai/monitor.py +++ b/src/fgai/monitor.py @@ -7,6 +7,7 @@ from pathlib import Path from .anomaly import anomaly_summary, detect_source_anomalies from .baseline import BaselineStore from .config import ConfigStore +from .correlation import correlate_source_ips from .event_context import build_event_context from .graylog_mcp import GraylogMcpClient from .graylog_source import GraylogStreamSource @@ -42,13 +43,15 @@ def build_status( else: try: configured_streams = runtime_values.get("graylog_streams", []) - stream_ids = [str(item.get("id")) for item in configured_streams if isinstance(item, dict) and item.get("enabled") and item.get("id")] + stream_configs = [item for item in configured_streams if isinstance(item, dict) and item.get("enabled") and item.get("id")] + stream_ids = [str(item.get("id")) for item in stream_configs] if not stream_ids: - stream_ids = [str(runtime_values.get("graylog_stream", ""))] + stream_configs = [{"id": str(runtime_values.get("graylog_stream", "")), "title": "Graylog"}] stream_statuses = [] events = [] - for stream_id in stream_ids: - stream_events, stream_status = GraylogStreamSource(GraylogMcpClient(url, token), stream_id, str(runtime_values.get("graylog_query", "*")), str(runtime_values.get("graylog_field_mapping", ""))).fetch() + for stream_config in stream_configs: + stream_id = str(stream_config["id"]) + stream_events, stream_status = GraylogStreamSource(GraylogMcpClient(url, token), stream_id, str(runtime_values.get("graylog_query", "*")), str(runtime_values.get("graylog_field_mapping", "")), str(stream_config.get("title", stream_id))).fetch() events.extend(stream_events) stream_statuses.append({"stream_id": stream_id, **stream_status}) mcp_status = {"status": "connected", "streams": stream_statuses, "events_fetched": len(events)} @@ -107,6 +110,7 @@ def build_status( "local_in_failures": local_in_failures(events, limit=10), }, "event_context": build_event_context(events), + "cross_source_correlations": correlate_source_ips(events), "anomalies": [ { "subject": finding.subject, diff --git a/tests/test_correlation.py b/tests/test_correlation.py new file mode 100644 index 0000000..9068ed6 --- /dev/null +++ b/tests/test_correlation.py @@ -0,0 +1,15 @@ +import unittest + +from fgai.correlation import correlate_source_ips +from fgai.logs import parse_log_line + + +class CorrelationTests(unittest.TestCase): + def test_correlates_same_source_across_streams(self): + events = [ + parse_log_line("srcip=10.0.0.5 fgai_stream=Fortigate action=blocked"), + parse_log_line("srcip=10.0.0.5 fgai_stream=DNS action=deny"), + ] + result = correlate_source_ips(events) + self.assertEqual(result[0]["source_ip"], "10.0.0.5") + self.assertEqual(result[0]["streams"], ["DNS", "Fortigate"]) diff --git a/tests/test_metrics.py b/tests/test_metrics.py new file mode 100644 index 0000000..2052fbc --- /dev/null +++ b/tests/test_metrics.py @@ -0,0 +1,10 @@ +import unittest + +from fgai.metrics import prometheus_metrics + + +class MetricsTests(unittest.TestCase): + def test_renders_low_cardinality_metrics(self): + output = prometheus_metrics({"summary": {"total": 10}, "anomaly_summary": {"high": 2}, "baseline": {"sources_ready": 3}, "capabilities": {"graylog_mcp": {"status": "connected", "events_fetched": 9}}}) + self.assertIn('fgai_events_total{kind="total"} 10', output) + self.assertIn("fgai_graylog_mcp_connected 1", output)