66 lines
2.8 KiB
Python
66 lines
2.8 KiB
Python
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")
|
|
|
|
def test_identity_and_event_aliases_populate_common_fields(self):
|
|
event = parse_log_line(
|
|
"winlog_event_data_targetusername=alice winlog_computer_name=host01 "
|
|
"winlog_event_id=4625 actor_id=bob"
|
|
)
|
|
|
|
self.assertEqual(event.fields["username"], "alice")
|
|
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()
|