Public/fgAI
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

MCP sample fetch now asks for common Winlogbeat/ECS fields before

Browse Source
This commit is contained in:
larssand
2026-06-30 15:10:47 +02:00
parent b8fe738aa1
commit 4a441bf443
8 changed files with 307 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
tests/test_graylog_source.py
View File

@@ -88,6 +88,17 @@ class GraylogSourceTests(unittest.TestCase):
self.assertIn("fw_action", client.arguments["fields"])
self.assertIn("full_message", client.arguments["fields"])
def test_requests_winlogbeat_discovery_fields_before_profile_exists(self):
client = _Client()
GraylogStreamSource(client, "windows").fetch()
self.assertIn("winlog.event_id", client.arguments["fields"])
self.assertIn("event.code", client.arguments["fields"])
self.assertIn("user.name", client.arguments["fields"])
self.assertIn("host.name", client.arguments["fields"])
self.assertIn("source.ip", client.arguments["fields"])
self.assertIn("winlog.channel", client.arguments["fields"])
self.assertIn("process.name", client.arguments["fields"])
if __name__ == "__main__":
unittest.main()

16
tests/test_normalization.py
View File

@@ -44,6 +44,22 @@ class NormalizationTests(unittest.TestCase):
self.assertEqual(event.fields["hostname"], "host01")
self.assertEqual(event.fields["eventid"], "4625")
def test_winlogbeat_ecs_aliases_populate_common_fields(self):
event = parse_log_line(
"@timestamp=2026-06-30T10:00:00Z event.code=4625 event.action=logon-failed "
"event.outcome=failure log.level=warning user.name=alice host.name=win01 "
"source.ip=10.0.0.5 winlog.channel=Security winlog.provider_name=Microsoft-Windows-Security-Auditing "
"winlog.event_data.LogonType=3 process.name=lsass.exe"
)
self.assertEqual(event.fields["eventtime"], "2026-06-30T10:00:00Z")
self.assertEqual(event.fields["eventid"], "4625")
self.assertEqual(event.fields["action"], "logon-failed")
self.assertEqual(event.fields["severity"], "warning")
self.assertEqual(event.fields["username"], "alice")
self.assertEqual(event.fields["hostname"], "win01")
self.assertEqual(event.fields["srcip"], "10.0.0.5")
if __name__ == "__main__":
unittest.main()

57
tests/test_profile_suggestions.py
View File

@@ -25,9 +25,10 @@ class ProfileSuggestionTests(unittest.TestCase):
def test_suggests_windows_event_fields(self):
events = [
parse_log_line(
f"fgai_stream_id=windows fgai_stream=Windows winlog_event_id=4625 "
f"fgai_stream_id=windows fgai_stream='Illuminate:Windows Security Event Log Messages' winlog_event_id=4625 "
f"winlog_event_data_targetusername=alice winlog_computer_name=host{index % 3} "
f"event_provider=Microsoft-Windows-Security-Auditing logon_type=3 "
f"event_provider=Microsoft-Windows-Security-Auditing channel=Security logon_type=3 "
f"winlog_event_data_ipaddress=10.0.0.{index % 5} "
f"eventtime=2026-06-29T10:00:{index:02d}Z action=failure"
)
for index in range(1, 30)
@@ -37,8 +38,60 @@ class ProfileSuggestionTests(unittest.TestCase):
self.assertIn("username", profile["entity_fields"])
self.assertIn("hostname", profile["entity_fields"])
self.assertIn("srcip", profile["entity_fields"])
self.assertNotIn("winlog_event_data_targetusername", profile["entity_fields"])
self.assertIn("eventid", profile["categorical_fields"])
self.assertIn("logon_type", profile["categorical_fields"])
self.assertIn("event_provider", profile["categorical_fields"])
self.assertNotIn("policyid", profile["categorical_fields"])
self.assertNotIn("dstport", profile["categorical_fields"])
self.assertNotIn("srcip", profile["categorical_fields"])
self.assertNotIn("eventid", profile["numeric_fields"])
self.assertIn("auth_failure", profile["detectors"])
def test_windows_named_stream_gets_auth_profile_even_before_failures(self):
events = [
parse_log_line(
f"fgai_stream_id=winsec fgai_stream='Illuminate:Windows Security Event Log Messages' "
f"eventid=4624 username=user{index % 4} hostname=host{index % 2} channel=Security "
f"eventtime=2026-06-29T10:00:{index:02d}Z action=success"
)
for index in range(1, 30)
]
suggestion = suggest_stream_profiles(events)[0]
profile = suggestion["profile"]
self.assertEqual(suggestion["stream_name"], "Illuminate:Windows Security Event Log Messages")
self.assertIn("username", profile["entity_fields"])
self.assertIn("hostname", profile["entity_fields"])
self.assertIn("eventid", profile["categorical_fields"])
self.assertIn("auth_failure", profile["detectors"])
def test_winlogbeat_ecs_fields_build_windows_profile(self):
events = [
parse_log_line(
f"fgai_stream_id=winbeat fgai_stream='Winlogbeat Security' "
f"@timestamp=2026-06-30T10:00:{index:02d}Z event.code=4625 event.action=logon-failed "
f"event.outcome=failure event.category=authentication event.type=start "
f"user.name=user{index % 4} host.name=win{index % 3} source.ip=10.0.0.{index % 5} "
f"winlog.channel=Security winlog.provider_name=Microsoft-Windows-Security-Auditing "
f"winlog.event_data.LogonType=3 process.name=proc{index % 3}.exe"
)
for index in range(1, 30)
]
profile = suggest_stream_profiles(events)[0]["profile"]
self.assertEqual(profile["timestamp_field"], "eventtime")
self.assertEqual(profile["entity_fields"][:3], ["username", "hostname", "srcip"])
self.assertIn("eventid", profile["categorical_fields"])
self.assertIn("action", profile["categorical_fields"])
self.assertIn("event.category", profile["categorical_fields"])
self.assertIn("winlog.channel", profile["categorical_fields"])
self.assertIn("process.name", profile["categorical_fields"])
self.assertNotIn("user.name", profile["categorical_fields"])
self.assertNotIn("source.ip", profile["categorical_fields"])
self.assertIn("auth_failure", profile["detectors"])
def test_discovers_non_priority_application_fields(self):