50 lines
1.9 KiB
Python
50 lines
1.9 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")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|