add dynamic

src/fgai/cli.py

@@ -6,7 +6,7 @@ import sys
 from pathlib import Path
 
 from .llm import ollama_summary
-from .logs import read_events, summarize_events
+from .logs import local_in_failures, read_events, summarize_events, top_field_values
 from .mitigation import FortiGateClient, parse_allowlist, suggest_block_candidates
 from .policies import audit_policies, read_policies
 from .syslog_server import listen_udp_syslog
@@ -27,6 +27,13 @@ def analyze_logs(args: argparse.Namespace) -> int:
     _print_json(
         {
             "summary": summarize_events(events),
+            "diagnostics": {
+                "top_source_ips": top_field_values(events, "srcip", limit=10),
+                "top_services": top_field_values(events, "service", limit=10),
+                "top_actions": top_field_values(events, "action", limit=10),
+                "top_subtypes": top_field_values(events, "subtype", limit=10),
+                "local_in_failures": local_in_failures(events, limit=10),
+            },
             "block_candidates": [
                 {"src_ip": candidate.src_ip, "score": candidate.score, "reasons": candidate.reasons}
                 for candidate in candidates

src/fgai/logs.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import json
 import shlex
+from collections import Counter
 from collections.abc import Iterable
 from pathlib import Path
 
@@ -117,3 +118,29 @@ def summarize_events(events: Iterable[LogEvent]) -> dict[str, int]:
         if event.severity in {"critical", "high", "alert", "emergency"}:
             summary["critical_or_high"] += 1
     return summary
+
+
+def top_field_values(events: Iterable[LogEvent], field: str, *, limit: int = 10) -> list[dict[str, int | str]]:
+    counter: Counter[str] = Counter()
+    for event in events:
+        value = event.fields.get(field)
+        if value:
+            counter[value] += 1
+    return [{"value": value, "count": count} for value, count in counter.most_common(limit)]
+
+
+def local_in_failures(events: Iterable[LogEvent], *, limit: int = 10) -> list[dict[str, int | str]]:
+    counter: Counter[tuple[str, str, str]] = Counter()
+    for event in events:
+        policy_type = event.fields.get("policytype", event.fields.get("type", "")).lower()
+        msg = event.fields.get("msg", "").lower()
+        if not (policy_type.startswith("local-in") or "local-in" in policy_type or msg == "connection failed"):
+            continue
+        src_ip = event.src_ip or "unknown"
+        service = event.fields.get("service", event.fields.get("app", "unknown"))
+        policy_id = event.fields.get("policyid", event.fields.get("poluuid", "unknown"))
+        counter[(src_ip, service, policy_id)] += 1
+    return [
+        {"src_ip": src_ip, "service": service, "policy": policy_id, "count": count}
+        for (src_ip, service, policy_id), count in counter.most_common(limit)
+    ]

tests/test_diagnostics.py

@@ -0,0 +1,35 @@
+import unittest
+
+from fgai.logs import local_in_failures, parse_log_line, top_field_values
+
+
+class DiagnosticTests(unittest.TestCase):
+    def test_local_in_failures_group_by_source_service_policy(self):
+        events = [
+            parse_log_line(
+                'type=traffic policytype="local-in-policy6" policyid=2 srcip=192.168.1.20 '
+                'service="udp/5353" msg="Connection Failed"'
+            ),
+            parse_log_line(
+                'type=traffic policytype="local-in-policy6" policyid=2 srcip=192.168.1.20 '
+                'service="udp/5353" msg="Connection Failed"'
+            ),
+        ]
+
+        self.assertEqual(
+            local_in_failures(events),
+            [{"src_ip": "192.168.1.20", "service": "udp/5353", "policy": "2", "count": 2}],
+        )
+
+    def test_top_field_values_counts_values(self):
+        events = [
+            parse_log_line("srcip=1.1.1.1 service=https"),
+            parse_log_line("srcip=1.1.1.1 service=http"),
+            parse_log_line("srcip=8.8.8.8 service=http"),
+        ]
+
+        self.assertEqual(top_field_values(events, "srcip", limit=1), [{"value": "1.1.1.1", "count": 2}])
+
+
+if __name__ == "__main__":
+    unittest.main()