${rows}
${esc(item.graylog_query)}` : ''}`).join('${rows}
${esc(r.graylog_query)}` : '-'}
], 'related-activity');
document.getElementById('reputation').innerHTML = table(reputationRows, [
{label:'IP', key:'ip'},
diff --git a/src/fgai/entities.py b/src/fgai/entities.py
index 024af20..f802376 100644
--- a/src/fgai/entities.py
+++ b/src/fgai/entities.py
@@ -5,6 +5,7 @@ from collections.abc import Iterable
from .models import LogEvent
from .normalization import canonical_value
+from .query_details import event_query_details
ENTITY_FIELDS: dict[str, tuple[str, ...]] = {
@@ -52,19 +53,23 @@ def profile_entities(event: LogEvent, profile: object) -> tuple[str, ...]:
return tuple(values)
-def sample_timeline(events: Iterable[LogEvent], *, limit: int = 20) -> list[dict[str, str]]:
- samples = [
- {
- "stream": event.fields.get("fgai_stream", "local_syslog"),
- "timestamp": event.fields.get("eventtime", event.fields.get("timestamp", "")),
- "type": canonical_value(event.fields, "type"),
- "subtype": event.subtype,
- "action": event.action,
- "severity": event.severity,
- "destination": event.dst_ip or canonical_value(event.fields, "context"),
- "service": canonical_value(event.fields, "service"),
- "context": canonical_value(event.fields, "context")[:240],
- }
- for event in events
- ]
+def _timeline_sample(event: LogEvent) -> dict[str, object]:
+ query_details = event_query_details(event)
+ return {
+ "stream": event.fields.get("fgai_stream", "local_syslog"),
+ "timestamp": event.fields.get("eventtime", event.fields.get("timestamp", "")),
+ "type": canonical_value(event.fields, "type"),
+ "subtype": event.subtype,
+ "action": event.action,
+ "severity": event.severity,
+ "destination": event.dst_ip or canonical_value(event.fields, "context"),
+ "service": canonical_value(event.fields, "service"),
+ "context": canonical_value(event.fields, "context")[:240],
+ "query_details": query_details,
+ "graylog_query": query_details["query"],
+ }
+
+
+def sample_timeline(events: Iterable[LogEvent], *, limit: int = 20) -> list[dict[str, object]]:
+ samples = [_timeline_sample(event) for event in events]
return sorted(samples, key=lambda item: item["timestamp"])[-limit:]
diff --git a/src/fgai/query_details.py b/src/fgai/query_details.py
new file mode 100644
index 0000000..efff7d2
--- /dev/null
+++ b/src/fgai/query_details.py
@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import json
+
+from .models import LogEvent
+from .normalization import FIELD_ALIASES, canonical_value
+
+
+def _quote(value: str) -> str:
+ escaped = value.replace("\\", "\\\\").replace('"', '\\"')
+ return f'"{escaped}"'
+
+
+def _field_group(canonical: str, value: str, *, limit: int = 6) -> str:
+ aliases = FIELD_ALIASES.get(canonical, (canonical,))[:limit]
+ return "(" + " OR ".join(f"{field}:{_quote(value)}" for field in aliases) + ")"
+
+
+def event_query_details(event: LogEvent) -> dict[str, str]:
+ """Build a copyable Graylog query for an event without assuming one vendor schema."""
+ terms: list[str] = []
+ for canonical in ("srcip", "dstip", "action", "dns_query"):
+ value = canonical_value(event.fields, canonical)
+ if value:
+ terms.append(_field_group(canonical, value))
+ timestamp = event.fields.get("eventtime", event.fields.get("timestamp", ""))
+ if not terms:
+ context = canonical_value(event.fields, "context")
+ if context:
+ terms.append(_field_group("context", context, limit=4))
+ return {
+ "stream": event.fields.get("fgai_stream", ""),
+ "stream_id": event.fields.get("fgai_stream_id", ""),
+ "timestamp": timestamp,
+ "query": " AND ".join(terms) if terms else "*",
+ "raw": event.raw if event.raw.startswith("{") else json.dumps(event.fields, sort_keys=True),
+ }
diff --git a/src/fgai/sequences.py b/src/fgai/sequences.py
index a4b32a3..c368d9f 100644
--- a/src/fgai/sequences.py
+++ b/src/fgai/sequences.py
@@ -8,6 +8,7 @@ from .entities import event_entities
from .logs import THREAT_ACTIONS
from .models import LogEvent
from .normalization import canonical_value
+from .query_details import event_query_details
DEFAULT_SEQUENCE_PATTERNS = {
@@ -46,7 +47,8 @@ def _is_network_event(event: LogEvent) -> bool:
return event.action in {"accept", "pass", "allowed", "allow", "close", "client-rst", "server-rst"} | THREAT_ACTIONS
-def _sample(event: LogEvent, value: str) -> dict[str, str]:
+def _sample(event: LogEvent, value: str) -> dict[str, object]:
+ query_details = event_query_details(event)
return {
"timestamp": event.fields.get("eventtime", event.fields.get("timestamp", "")),
"stream": event.fields.get("fgai_stream", event.fields.get("fgai_stream_id", "")),
@@ -57,6 +59,8 @@ def _sample(event: LogEvent, value: str) -> dict[str, str]:
"service": canonical_value(event.fields, "service"),
"value": value,
"message": canonical_value(event.fields, "context")[:240],
+ "query_details": query_details,
+ "graylog_query": query_details["query"],
}
diff --git a/tests/test_query_details.py b/tests/test_query_details.py
new file mode 100644
index 0000000..3e8c367
--- /dev/null
+++ b/tests/test_query_details.py
@@ -0,0 +1,28 @@
+import unittest
+
+from fgai.entities import sample_timeline
+from fgai.logs import parse_log_line
+from fgai.query_details import event_query_details
+
+
+class QueryDetailsTests(unittest.TestCase):
+ def test_builds_generic_graylog_query_from_aliases(self):
+ event = parse_log_line("src_addr=10.0.0.5 dst_addr=198.51.100.10 fw_action=allow")
+
+ details = event_query_details(event)
+
+ self.assertIn('srcip:"10.0.0.5"', details["query"])
+ self.assertIn('dstip:"198.51.100.10"', details["query"])
+ self.assertIn('action:"allow"', details["query"])
+
+ def test_timeline_contains_query_details(self):
+ event = parse_log_line('timestamp=2026-06-29T16:45:52Z fgai_stream=Firewall srcip=10.0.0.5 dstip=198.51.100.10 action=deny')
+
+ row = sample_timeline([event])[0]
+
+ self.assertEqual(row["query_details"]["stream"], "Firewall")
+ self.assertIn('srcip:"10.0.0.5"', row["graylog_query"])
+
+
+if __name__ == "__main__":
+ unittest.main()