45 lines
2.1 KiB
Python
45 lines
2.1 KiB
Python
import unittest
|
|
|
|
from fgai.logs import parse_log_line
|
|
from fgai.sequences import detect_sequences
|
|
|
|
|
|
class SequenceTests(unittest.TestCase):
|
|
def test_detects_generic_dns_network_auth_sequence(self):
|
|
events = [
|
|
parse_log_line("timestamp=2026-06-25T10:00:00Z fgai_stream=Resolver srcip=10.0.0.5 query_domain=example.test"),
|
|
parse_log_line("timestamp=2026-06-25T10:02:00Z fgai_stream=Proxy srcip=10.0.0.5 dstip=203.0.113.10 dstport=443 action=accept"),
|
|
parse_log_line("timestamp=2026-06-25T10:05:00Z fgai_stream=Identity srcip=10.0.0.5 action=failed eventid=4625"),
|
|
]
|
|
|
|
result = detect_sequences(events)
|
|
|
|
self.assertIn("10.0.0.5", result)
|
|
finding = result["10.0.0.5"][0]
|
|
self.assertEqual(finding["detector"], "dns_network_auth_sequence")
|
|
self.assertEqual(finding["sample_values"], ["Identity", "Proxy", "Resolver"])
|
|
self.assertEqual([item["value"] for item in finding["sample_events"]], ["dns_query", "network_connection", "auth_failure"])
|
|
|
|
def test_does_not_match_sequence_outside_window(self):
|
|
events = [
|
|
parse_log_line("timestamp=2026-06-25T10:00:00Z fgai_stream=Resolver srcip=10.0.0.5 query_domain=example.test"),
|
|
parse_log_line("timestamp=2026-06-25T10:02:00Z fgai_stream=Proxy srcip=10.0.0.5 dstip=203.0.113.10 dstport=443 action=accept"),
|
|
parse_log_line("timestamp=2026-06-25T11:00:00Z fgai_stream=Identity srcip=10.0.0.5 action=failed eventid=4625"),
|
|
]
|
|
|
|
self.assertEqual(detect_sequences(events, window_seconds=900), {})
|
|
|
|
def test_supports_custom_generic_patterns(self):
|
|
events = [
|
|
parse_log_line("timestamp=2026-06-25T10:00:00Z fgai_stream=Proxy srcip=10.0.0.5 dstip=203.0.113.10 action=accept"),
|
|
parse_log_line("timestamp=2026-06-25T10:01:00Z fgai_stream=Firewall srcip=10.0.0.5 action=deny"),
|
|
]
|
|
|
|
result = detect_sequences(events, patterns={"connection_then_deny": ("network_connection", "deny_action")})
|
|
|
|
self.assertEqual(result["10.0.0.5"][0]["detector"], "connection_then_deny")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|