Ny central normalisering

README.md

@@ -79,6 +79,13 @@ The dashboard and Ollama then correlate behavior across sources, for example a
 client IP appearing in FortiGate, AdGuard/DNS, Windows Security, Nginx, Squid,
 VPN, or Proxmox.
 
+SignalScope keeps a common alias map for fields such as source IP, destination
+IP, ports, action, severity, service/protocol, DNS query, URL, message, and event
+type. This lets Related Activity and correlations work with firewall/proxy/DNS
+streams that use names like `src_addr`, `destination.ip`, `dest_port`,
+`fw_action`, `priority`, `proto`, or `full_message` without adding a new parser
+for every product.
+
 Correlation is entity-aware rather than FortiGate-specific. SignalScope recognizes
 common IP fields such as `srcip`, `source_ip`, `remote_addr`, and Windows event
 IP fields; account fields such as `username`, `user`, and `TargetUserName`; and

ROADMAP.md

@@ -11,6 +11,7 @@ operational value and dependency, not by UI appeal.
 - [x] Local five-minute field baselines with duplicate-event protection.
 - [x] Time-aware baseline comparison using matching UTC weekday/hour when available.
 - [x] Generic entity correlation for IP addresses, users, and hostnames.
+- [x] Centralized common field alias normalization for firewall, DNS, proxy, endpoint, and web logs.
 - [x] Cross-stream correlation timelines and investigation incident grouping.
 - [x] Field-deviation review: expected, false positive, confirmed, note, and expiry.
 - [x] Local Ollama analyst assessment with incident and feedback context.
@@ -106,6 +107,7 @@ Goal: add log sources and outputs without adding source-specific logic everywher
 
 - [ ] Define versioned stream-profile templates for FortiGate, Windows, DNS/AdGuard, Nginx, Squid, VPN, and Proxmox.
 - [x] Add inventory-style stream coverage to guide which streams need profiles before templates are added.
+- [x] Add common alias normalizer so new firewall/proxy/DNS streams can populate source, destination, action, severity, service, and context without source-specific code.
 - [ ] Add import/export for profile templates and detector settings.
 - [ ] Separate source adapters, normalizers, detectors, enrichers, and output adapters into explicit extension interfaces.
 - [ ] Add optional webhook/SIEM ticket output for confirmed high-severity incidents.

src/fgai/baseline.py

@@ -12,6 +12,7 @@ from .logs import THREAT_ACTIONS, is_utm_event
 from .models import LogEvent
 from .entities import profile_entity
 from .detectors import DETECTOR_MINIMUMS, event_detector_categories
+from .normalization import canonical_value
 
 
 def _number(value: str | None) -> int:
@@ -62,6 +63,19 @@ def _weighted_score(base: int, profile: object | None, field: str, detector: str
     return min(100, max(0, int(round(base * multiplier)))), round(multiplier, 2)
 
 
+def _sample_event(event: LogEvent, value: str = "") -> dict[str, str]:
+    return {
+        "timestamp": event.fields.get("eventtime", event.fields.get("timestamp", "")),
+        "source": event.src_ip or event.fields.get("source", ""),
+        "destination": event.dst_ip or canonical_value(event.fields, "context"),
+        "action": event.action,
+        "severity": event.severity,
+        "service": canonical_value(event.fields, "service"),
+        "value": value,
+        "message": canonical_value(event.fields, "context")[:240],
+    }
+
+
 class BaselineStore:
     """Persistent five-minute behavior baseline, implemented with stdlib SQLite."""
 
@@ -252,7 +266,7 @@ class BaselineStore:
                 deviation = abs(current_value - mean(history))
                 if deviation > (pstdev(history) or 1.0) * 3:
                     sample_values = sorted({event.fields.get(field, "") for event in matching})[:5]
-                    evidence_events = [{"timestamp": event.fields.get("eventtime", event.fields.get("timestamp", "")), "source": event.src_ip or event.fields.get("source", ""), "destination": event.dst_ip or "", "action": event.action, "severity": event.severity, "service": event.fields.get("service", ""), "value": event.fields.get(field, ""), "message": event.fields.get("message", event.fields.get("msg", ""))[:240]} for event in matching[:5]]
+                    evidence_events = [_sample_event(event, event.fields.get(field, "")) for event in matching[:5]]
                     confidence = _baseline_confidence(len(rows), temporal=temporal)
                     base_score = 18 if confidence == "high" else 15 if confidence == "medium" else 10
                     score, weight = _weighted_score(base_score, profile, field, "numeric_baseline")
@@ -284,7 +298,7 @@ class BaselineStore:
                 confidence = _baseline_confidence(len(rows), temporal=temporal)
                 base_score = min(30, (15 if confidence == "high" else 12 if confidence == "medium" else 8) + int(z_score))
                 score, weight = _weighted_score(base_score, profile, "event_rate", "event_rate_burst")
-                samples = [{"timestamp": event.fields.get("eventtime", event.fields.get("timestamp", "")), "source": event.src_ip or event.fields.get("source", ""), "destination": event.dst_ip or "", "action": event.action, "severity": event.severity, "service": event.fields.get("service", ""), "value": "", "message": event.fields.get("message", event.fields.get("msg", ""))[:240]} for event in matching[:5]]
+                samples = [_sample_event(event) for event in matching[:5]]
                 output[entity].append({"detector": "event_rate_burst", "field": "event_rate", "stream_id": stream, "score": score, "base_score": base_score, "weight": weight, "confidence": confidence, "baseline_samples": len(rows), "baseline_scope": baseline_scope, "reason": f"event rate burst above its {baseline_scope} baseline (z={z_score:.1f})", "current": current_value, "baseline": round(mean(history), 2), "sample_values": [], "sample_events": samples})
 
             for (stream, entity, detector), current_value in detector_current.items():
@@ -315,7 +329,7 @@ class BaselineStore:
                 confidence = _baseline_confidence(len(rows), temporal=temporal)
                 base_score = min(35, (18 if detector == "auth_failure" else 15 if detector == "deny_action" else 12) + int(z_score))
                 score, weight = _weighted_score(base_score, profile, detector, f"{detector}_burst")
-                samples = [{"timestamp": event.fields.get("eventtime", event.fields.get("timestamp", "")), "source": event.src_ip or event.fields.get("source", ""), "destination": event.dst_ip or event.fields.get("query_domain", ""), "action": event.action, "severity": event.severity, "service": event.fields.get("service", event.fields.get("query_type", "")), "value": detector, "message": event.fields.get("message", event.fields.get("msg", ""))[:240]} for event in matching[:5]]
+                samples = [_sample_event(event, detector) for event in matching[:5]]
                 output[entity].append({"detector": f"{detector}_burst", "field": detector, "stream_id": stream, "score": score, "base_score": base_score, "weight": weight, "confidence": confidence, "baseline_samples": len(rows), "baseline_scope": baseline_scope, "reason": f"{detector.replace('_', ' ')} burst above its {baseline_scope} baseline (z={z_score:.1f}, minimum={minimum})", "current": current_value, "baseline": round(mean(history), 2), "sample_values": [detector], "sample_events": samples})
             # Detect selected categorical values that have not appeared for this entity in prior data.
             for event in events:
@@ -336,7 +350,7 @@ class BaselineStore:
                     if known is None and known_total >= 30:
                         samples = [item for item in events if item.fields.get("fgai_stream_id") == stream and profile_entity(item, str(getattr(profile, "entity_field", ""))) == entity and item.fields.get(field) == value]
                         score, weight = _weighted_score(12, profile, field, "rare_value")
-                        evidence = {"detector": "rare_value", "field": field, "stream_id": stream, "score": score, "base_score": 12, "weight": weight, "confidence": "medium", "baseline_samples": int(known_total), "baseline_scope": "known field values", "reason": f"new {field} value for this entity", "value": value, "sample_values": [value], "sample_events": [{"timestamp": item.fields.get("eventtime", item.fields.get("timestamp", "")), "source": item.src_ip or item.fields.get("source", ""), "destination": item.dst_ip or "", "action": item.action, "severity": item.severity, "service": item.fields.get("service", ""), "value": value, "message": item.fields.get("message", item.fields.get("msg", ""))[:240]} for item in samples[:5]]}
+                        evidence = {"detector": "rare_value", "field": field, "stream_id": stream, "score": score, "base_score": 12, "weight": weight, "confidence": "medium", "baseline_samples": int(known_total), "baseline_scope": "known field values", "reason": f"new {field} value for this entity", "value": value, "sample_values": [value], "sample_events": [_sample_event(item, value) for item in samples[:5]]}
                         if evidence not in output[entity]:
                             output[entity].append(evidence)
         return output

src/fgai/detectors.py

@@ -18,7 +18,7 @@ def event_detector_categories(event: LogEvent) -> tuple[str, ...]:
     categories: list[str] = []
     if action in {"fail", "failed", "failure", "login_failed", "logon_failed", "authentication_failed"} or event_id == "4625":
         categories.append("auth_failure")
-    if fields.get("query_domain") or fields.get("qh") or fields.get("dns_query"):
+    if fields.get("dns_query"):
         categories.append("dns_query")
     if action in THREAT_ACTIONS:
         categories.append("deny_action")

src/fgai/entities.py

@@ -4,6 +4,7 @@ import ipaddress
 from collections.abc import Iterable
 
 from .models import LogEvent
+from .normalization import canonical_value
 
 
 ENTITY_FIELDS: dict[str, tuple[str, ...]] = {
@@ -46,13 +47,13 @@ def sample_timeline(events: Iterable[LogEvent], *, limit: int = 20) -> list[dict
         {
             "stream": event.fields.get("fgai_stream", "local_syslog"),
             "timestamp": event.fields.get("eventtime", event.fields.get("timestamp", "")),
-            "type": event.fields.get("type", ""),
+            "type": canonical_value(event.fields, "type"),
             "subtype": event.subtype,
             "action": event.action,
             "severity": event.severity,
-            "destination": event.dst_ip or event.fields.get("query_domain", event.fields.get("url", "")),
+            "destination": event.dst_ip or canonical_value(event.fields, "context"),
-            "service": event.fields.get("service", event.fields.get("query_type", "")),
+            "service": canonical_value(event.fields, "service"),
-            "context": event.fields.get("query_domain", event.fields.get("qh", event.fields.get("url", event.fields.get("message", event.fields.get("msg", "")))))[:240],
+            "context": canonical_value(event.fields, "context")[:240],
         }
         for event in events
     ]

src/fgai/event_context.py

@@ -4,6 +4,7 @@ from collections import Counter, defaultdict
 
 from .logs import THREAT_ACTIONS, is_utm_event
 from .models import LogEvent
+from .normalization import canonical_value
 
 
 def _entity(event: LogEvent) -> str:
@@ -30,8 +31,8 @@ def build_event_context(events: list[LogEvent], *, source_limit: int = 30, sampl
 
     suspicious = [event for event in events if is_utm_event(event) or event.action in THREAT_ACTIONS or event.severity in {"critical", "high", "alert", "emergency"}]
     samples = [{
-        "entity": _entity(event), "type": event.fields.get("type", ""), "subtype": event.subtype,
+        "entity": _entity(event), "type": canonical_value(event.fields, "type"), "subtype": event.subtype,
-        "action": event.action, "severity": event.severity, "dst": event.dst_ip or "", "service": event.fields.get("service", ""),
+        "action": event.action, "severity": event.severity, "dst": event.dst_ip or canonical_value(event.fields, "context"), "service": canonical_value(event.fields, "service"),
         "policyid": event.fields.get("policyid", ""), "timestamp": event.fields.get("eventtime", event.fields.get("timestamp", "")),
     } for event in suspicious[:sample_limit]]
     return {

src/fgai/graylog_source.py

@@ -5,19 +5,10 @@ from collections.abc import Iterable
 
 from .graylog_mcp import GraylogMcpClient
 from .models import LogEvent
+from .normalization import DEFAULT_SEARCH_FIELDS, normalize_fields
 
 
-DEFAULT_FIELD_MAP = {
+DEFAULT_FIELDS = list(dict.fromkeys([*DEFAULT_SEARCH_FIELDS, "source", "eventid", "event_id", "winlog_event_id", "policyid", "policy_id", "policy", "rule", "rule_name", "sentbyte", "rcvdbyte", "bytes_in", "bytes_out", "hitcount", "elapsed"]))
-    "srcip": ("srcip", "src_ip", "source_ip", "client_ip", "ip", "client"),
-    "dstip": ("dstip", "dst_ip", "destination_ip", "server_ip", "upstream"),
-    "dstport": ("dstport", "dst_port", "destination_port"),
-    "srcport": ("srcport", "src_port", "source_port"),
-    "eventtime": ("eventtime", "timestamp", "time"),
-    "severity": ("severity", "level"),
-    "action": ("action", "event_action", "disposition"),
-}
-
-DEFAULT_FIELDS = ["timestamp", "source", "srcip", "src_ip", "source_ip", "client_ip", "ip", "dstip", "dst_ip", "destination_ip", "upstream", "query_domain", "qh", "qt", "query_type", "dns_query", "srcport", "dstport", "service", "action", "status", "eventid", "event_id", "winlog_event_id", "severity", "policyid", "subtype", "type", "sentbyte", "rcvdbyte", "hitcount", "elapsed", "message"]
 
 
 def _records(value: object) -> Iterable[dict[str, object]]:
@@ -92,11 +83,4 @@ class GraylogStreamSource:
         fields = {str(key).lower(): str(value) for key, value in record.items() if value is not None}
         fields["fgai_stream"] = self.stream_label
         fields["fgai_stream_id"] = self.stream
-        for canonical, candidates in DEFAULT_FIELD_MAP.items():
+        return LogEvent(raw=json.dumps(record, sort_keys=True), fields=normalize_fields(fields, self.mapping))
-            mapped = self.mapping.get(canonical)
-            candidates = (str(mapped),) if mapped else candidates
-            for candidate in candidates:
-                if candidate.lower() in fields:
-                    fields[canonical] = fields[candidate.lower()]
-                    break
-        return LogEvent(raw=json.dumps(record, sort_keys=True), fields=fields)

src/fgai/logs.py

@@ -7,6 +7,7 @@ from collections.abc import Iterable
 from pathlib import Path
 
 from .models import LogEvent
+from .normalization import canonical_value, normalize_fields
 
 UTM_SUBTYPES = {
     "ips",
@@ -60,7 +61,7 @@ def parse_log_line(line: str) -> LogEvent:
         except json.JSONDecodeError:
             data = {}
         else:
-            return LogEvent(raw=line, fields={str(k).lower(): str(v) for k, v in data.items()})
+            return LogEvent(raw=line, fields=normalize_fields({str(k).lower(): str(v) for k, v in data.items()}))
 
     fields: dict[str, str] = {}
     try:
@@ -74,7 +75,7 @@ def parse_log_line(line: str) -> LogEvent:
         key, value = part.split("=", 1)
         fields[key.strip().lower()] = value.strip().strip('"')
 
-    return LogEvent(raw=line, fields=fields)
+    return LogEvent(raw=line, fields=normalize_fields(fields))
 
 
 def read_events(path: str | Path) -> list[LogEvent]:
@@ -137,7 +138,7 @@ def local_in_failures(events: Iterable[LogEvent], *, limit: int = 10) -> list[di
         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"))
+        service = canonical_value(event.fields, "service") or "unknown"
         policy_id = event.fields.get("policyid", event.fields.get("poluuid", "unknown"))
         counter[(src_ip, service, policy_id)] += 1
     return [

src/fgai/models.py

@@ -2,6 +2,8 @@ from __future__ import annotations
 
 from dataclasses import dataclass, field
 
+from .normalization import canonical_value
+
 
 @dataclass(frozen=True)
 class LogEvent:
@@ -10,23 +12,23 @@ class LogEvent:
 
     @property
     def src_ip(self) -> str | None:
-        return self.fields.get("srcip") or self.fields.get("src_ip") or self.fields.get("source_ip")
+        return canonical_value(self.fields, "srcip") or None
 
     @property
     def dst_ip(self) -> str | None:
-        return self.fields.get("dstip") or self.fields.get("dst_ip") or self.fields.get("destination_ip")
+        return canonical_value(self.fields, "dstip") or None
 
     @property
     def subtype(self) -> str:
-        return self.fields.get("subtype", "").lower()
+        return canonical_value(self.fields, "subtype").lower()
 
     @property
     def action(self) -> str:
-        return self.fields.get("action", "").lower()
+        return canonical_value(self.fields, "action").lower()
 
     @property
     def severity(self) -> str:
-        return self.fields.get("severity", self.fields.get("level", "")).lower()
+        return canonical_value(self.fields, "severity").lower()
 
 
 @dataclass(frozen=True)

src/fgai/normalization.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+
+FIELD_ALIASES: dict[str, tuple[str, ...]] = {
+    "srcip": ("srcip", "src_ip", "source_ip", "source.ip", "src", "src_addr", "srcaddr", "sourceaddress", "client_ip", "client.ip", "clientip", "client", "remote_addr", "remote_ip", "ip", "ipaddress", "gl2_remote_ip", "winlog_event_data_ipaddress", "event_data_ipaddress"),
+    "dstip": ("dstip", "dst_ip", "destination_ip", "destination.ip", "dst", "dst_addr", "dstaddr", "destinationaddress", "dest_ip", "dest", "server_ip", "server.ip", "server", "upstream"),
+    "srcport": ("srcport", "src_port", "source_port", "source.port", "sport", "client_port", "client.port", "clientport"),
+    "dstport": ("dstport", "dst_port", "destination_port", "destination.port", "dest_port", "dport", "server_port", "server.port", "serverport"),
+    "eventtime": ("eventtime", "timestamp", "time", "event_time", "eventtime_ms", "created_at"),
+    "severity": ("severity", "level", "priority", "sev", "loglevel", "log_level", "event_severity"),
+    "action": ("action", "act", "fw_action", "rule_action", "policy_action", "event_action", "disposition", "outcome", "result", "status", "event_outcome"),
+    "type": ("type", "event_type", "log_type", "category", "event_category", "facility"),
+    "subtype": ("subtype", "sub_type", "event_subtype", "subcategory"),
+    "service": ("service", "service_name", "dst_service", "app", "appname", "application", "application_name", "proto", "protocol", "transport", "network.transport", "query_type", "qt"),
+    "dns_query": ("dns_query", "query_domain", "qh", "dns_question", "question", "queried_domain"),
+    "context": ("query_domain", "qh", "dns_query", "url", "uri", "request", "request_uri", "path", "domain", "hostname", "message", "msg", "full_message", "event_message"),
+}
+
+DEFAULT_SEARCH_FIELDS = tuple(dict.fromkeys(field for aliases in FIELD_ALIASES.values() for field in aliases))
+
+
+def first_field(fields: dict[str, str], *names: str) -> str:
+    for name in names:
+        value = str(fields.get(name.lower(), "")).strip()
+        if value and value not in {"-", "unknown", "n/a", "none", "null"}:
+            return value
+    return ""
+
+
+def canonical_value(fields: dict[str, str], canonical: str) -> str:
+    return first_field(fields, *FIELD_ALIASES.get(canonical, (canonical,)))
+
+
+def normalize_fields(fields: dict[str, str], mapping: dict[str, object] | None = None) -> dict[str, str]:
+    normalized = dict(fields)
+    mapping = mapping or {}
+    for canonical, aliases in FIELD_ALIASES.items():
+        mapped = mapping.get(canonical)
+        candidates = (str(mapped).lower(),) if isinstance(mapped, str) and mapped else aliases
+        value = first_field(normalized, *candidates)
+        if value:
+            normalized[canonical] = value
+    return normalized

src/fgai/sequences.py

@@ -7,6 +7,7 @@ from .detectors import event_detector_categories
 from .entities import event_entities
 from .logs import THREAT_ACTIONS
 from .models import LogEvent
+from .normalization import canonical_value
 
 
 DEFAULT_SEQUENCE_PATTERNS = {
@@ -38,9 +39,9 @@ def _timestamp(event: LogEvent, fallback: int) -> int:
 
 
 def _is_network_event(event: LogEvent) -> bool:
-    if event.fields.get("dstip") or event.fields.get("destination_ip") or event.fields.get("dst_ip"):
+    if event.dst_ip:
         return True
-    if event.fields.get("dstport") or event.fields.get("destination_port") or event.fields.get("service"):
+    if canonical_value(event.fields, "dstport") or canonical_value(event.fields, "service"):
         return True
     return event.action in {"accept", "pass", "allowed", "allow", "close", "client-rst", "server-rst"} | THREAT_ACTIONS
 
@@ -50,12 +51,12 @@ def _sample(event: LogEvent, value: str) -> dict[str, str]:
         "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", "")),
+        "destination": event.dst_ip or canonical_value(event.fields, "context"),
         "action": event.action,
         "severity": event.severity,
-        "service": event.fields.get("service", event.fields.get("query_type", "")),
+        "service": canonical_value(event.fields, "service"),
         "value": value,
-        "message": event.fields.get("message", event.fields.get("msg", ""))[:240],
+        "message": canonical_value(event.fields, "context")[:240],
     }
 
 

tests/test_graylog_source.py

@@ -41,6 +41,14 @@ class GraylogSourceTests(unittest.TestCase):
         self.assertIn("targetusername", client.arguments["fields"])
         self.assertIn("eventid", client.arguments["fields"])
 
+    def test_requests_common_firewall_alias_fields(self):
+        client = _Client()
+        GraylogStreamSource(client, "firewall").fetch()
+        self.assertIn("src_addr", client.arguments["fields"])
+        self.assertIn("dest_port", client.arguments["fields"])
+        self.assertIn("fw_action", client.arguments["fields"])
+        self.assertIn("full_message", client.arguments["fields"])
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_normalization.py

@@ -0,0 +1,39 @@
+import unittest
+
+from fgai.entities import sample_timeline
+from fgai.logs import parse_log_line
+
+
+class NormalizationTests(unittest.TestCase):
+    def test_firewall_aliases_populate_common_event_fields(self):
+        event = parse_log_line(
+            'timestamp=2026-06-29T16:45:52.000Z fgai_stream="Illuminate:Fortigate Messages" '
+            "src_addr=10.251.62.26 dst_addr=198.51.100.10 dest_port=443 fw_action=allow priority=high proto=tcp "
+            'full_message="allowed outbound session"'
+        )
+
+        self.assertEqual(event.src_ip, "10.251.62.26")
+        self.assertEqual(event.dst_ip, "198.51.100.10")
+        self.assertEqual(event.action, "allow")
+        self.assertEqual(event.severity, "high")
+        self.assertEqual(event.fields["dstport"], "443")
+        self.assertEqual(event.fields["service"], "tcp")
+
+    def test_related_activity_timeline_uses_aliases(self):
+        event = parse_log_line(
+            'timestamp=2026-06-29T16:45:52.000Z fgai_stream="Illuminate:Fortigate Messages" '
+            "src_addr=10.251.62.26 dst_addr=198.51.100.10 dest_port=443 fw_action=allow priority=high proto=tcp "
+            'full_message="allowed outbound session"'
+        )
+
+        row = sample_timeline([event])[0]
+
+        self.assertEqual(row["action"], "allow")
+        self.assertEqual(row["severity"], "high")
+        self.assertEqual(row["destination"], "198.51.100.10")
+        self.assertEqual(row["service"], "tcp")
+        self.assertEqual(row["context"], "allowed outbound session")
+
+
+if __name__ == "__main__":
+    unittest.main()