181 lines
9.0 KiB
Python
181 lines
9.0 KiB
Python
import unittest
|
|
|
|
from fgai.logs import parse_log_line
|
|
from fgai.profile_suggestions import apply_profile_advice, suggest_stream_profiles
|
|
|
|
|
|
class ProfileSuggestionTests(unittest.TestCase):
|
|
def test_suggests_common_profile_fields_per_stream(self):
|
|
events = [
|
|
parse_log_line(f"fgai_stream_id=firewall fgai_stream=Firewall srcip=10.0.0.{index} dstip=8.8.8.8 dstport=443 action=deny timestamp=2026-06-29T10:00:{index:02d}Z hitcount={index}")
|
|
for index in range(1, 25)
|
|
]
|
|
|
|
suggestions = suggest_stream_profiles(events)
|
|
profile = suggestions[0]["profile"]
|
|
|
|
self.assertEqual(profile["stream_id"], "firewall")
|
|
self.assertIn("srcip", profile["entity_fields"])
|
|
self.assertEqual(profile["timestamp_field"], "eventtime")
|
|
self.assertIn("action", profile["categorical_fields"])
|
|
self.assertIn("dstport", profile["categorical_fields"])
|
|
self.assertIn("hitcount", profile["numeric_fields"])
|
|
self.assertIn("deny_action", profile["detectors"])
|
|
|
|
def test_placeholder_winlog_fields_do_not_make_firewall_windows(self):
|
|
events = [
|
|
parse_log_line(
|
|
f"fgai_stream_id=firewall fgai_stream='Firewall Logs' srcip=10.0.0.{index} dstip=8.8.8.8 "
|
|
f"dstport=443 action=deny policyid=12 eventid=- winlog.channel=- winlog.event_id=- "
|
|
f"winlog.event_data.targetusername=- hostname=- username=- timestamp=2026-06-29T10:00:{index:02d}Z"
|
|
)
|
|
for index in range(1, 25)
|
|
]
|
|
|
|
suggestion = suggest_stream_profiles(events)[0]
|
|
profile = suggestion["profile"]
|
|
|
|
self.assertEqual(suggestion["detected_log_type"], "firewall")
|
|
self.assertNotIn("username", profile["entity_fields"])
|
|
self.assertNotIn("hostname", profile["entity_fields"])
|
|
self.assertIn("srcip", profile["entity_fields"])
|
|
self.assertIn("policyid", profile["categorical_fields"])
|
|
self.assertIn("deny_action", profile["detectors"])
|
|
|
|
def test_suggests_windows_event_fields(self):
|
|
events = [
|
|
parse_log_line(
|
|
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 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)
|
|
]
|
|
|
|
profile = suggest_stream_profiles(events)[0]["profile"]
|
|
|
|
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):
|
|
events = [
|
|
parse_log_line(
|
|
f"fgai_stream_id=app fgai_stream=BillingApp tenant_id=t{index % 4} actor_id=user{index % 7} "
|
|
f"workflow_state={'approved' if index % 2 else 'rejected'} payment_provider=stripe risk_points={index % 9} "
|
|
f"created_at=2026-06-29T10:00:{index:02d}Z"
|
|
)
|
|
for index in range(1, 40)
|
|
]
|
|
|
|
profile = suggest_stream_profiles(events)[0]["profile"]
|
|
|
|
self.assertIn("username", profile["entity_fields"])
|
|
self.assertIn("tenant_id", profile["categorical_fields"])
|
|
self.assertIn("workflow_state", profile["categorical_fields"])
|
|
self.assertIn("risk_points", profile["numeric_fields"])
|
|
|
|
def test_profile_discovery_explains_selected_and_rejected_fields(self):
|
|
events = [
|
|
parse_log_line(
|
|
f"fgai_stream_id=app fgai_stream=BillingApp tenant_id=t{index % 4} actor_id=user{index % 7} "
|
|
f"workflow_state={'approved' if index % 2 else 'rejected'} request_id=req-{index} "
|
|
f"constant_field=same raw='{{json}}' risk_points={index % 9} created_at=2026-06-29T10:00:{index:02d}Z"
|
|
)
|
|
for index in range(1, 40)
|
|
]
|
|
|
|
suggestion = suggest_stream_profiles(events)[0]
|
|
discovery = suggestion["discovery"]
|
|
|
|
self.assertEqual(suggestion["detected_log_type"], "application")
|
|
self.assertEqual(discovery["log_type"], "application")
|
|
self.assertIn("actor_id", discovery["selected_fields"]["entity"])
|
|
self.assertIn("workflow_state", discovery["selected_fields"]["categorical"])
|
|
self.assertIn("risk_points", discovery["selected_fields"]["numeric"])
|
|
rejected = {item["field"]: item["reason"] for item in discovery["rejected_fields"]}
|
|
self.assertEqual(rejected["raw"], "raw/internal field")
|
|
self.assertEqual(rejected["constant_field"], "constant value")
|
|
self.assertEqual(rejected["request_id"], "too high cardinality")
|
|
|
|
def test_applies_valid_llm_advice_and_rejects_unknown_fields(self):
|
|
suggestion = suggest_stream_profiles([
|
|
parse_log_line("fgai_stream_id=windows fgai_stream=Windows username=alice hostname=host01 eventid=4625 action=failure")
|
|
])[0]
|
|
|
|
advised = apply_profile_advice([suggestion], [{
|
|
"stream_id": "windows",
|
|
"entity_fields": ["username", "not_a_field"],
|
|
"timestamp_field": "eventtime",
|
|
"categorical_fields": ["eventid", "full_message"],
|
|
"numeric_fields": ["missing_number"],
|
|
"detectors": {"auth_failure": {"enabled": True, "minimum": 3, "z_threshold": 2.5}, "made_up": {"enabled": True}},
|
|
"reason": "Windows auth fields",
|
|
}])[0]
|
|
|
|
self.assertEqual(advised["profile_advisor"]["status"], "ok")
|
|
self.assertEqual(advised["profile"]["entity_fields"], ["username"])
|
|
self.assertIn("eventid", advised["profile"]["categorical_fields"])
|
|
self.assertNotIn("full_message", advised["profile"]["categorical_fields"])
|
|
self.assertEqual(set(advised["profile"]["detectors"]), {"auth_failure"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|