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

improve find fileds for default profiles

Browse Source
This commit is contained in:
larssand
2026-07-02 09:51:25 +02:00
parent 53437a3952
commit 7c0b482a22
6 changed files with 167 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
tests/test_profile_suggestions.py
View File

@@ -178,6 +178,30 @@ class ProfileSuggestionTests(unittest.TestCase):
self.assertIn("result_code", profile["categorical_fields"])
self.assertIn("latency_ms", profile["numeric_fields"])
def test_semantic_alias_fields_are_shared_across_streams(self):
events = []
events.extend(
parse_log_line(
f"fgai_stream_id=proxy fgai_stream=Proxy source.ip=10.0.0.{index % 7} url=/item/{index % 3} action=allow"
)
for index in range(1, 30)
)
events.extend(
parse_log_line(
f"fgai_stream_id=firewall fgai_stream=Firewall client_ip=10.0.0.{index % 7} dstport=443 action=accept"
)
for index in range(1, 30)
)
suggestions = {item["stream_id"]: item for item in suggest_stream_profiles(events)}
for suggestion in suggestions.values():
shared = {item["field"]: item for item in suggestion["shared_fields"]}
self.assertIn("srcip", shared)
self.assertEqual(shared["srcip"]["kind"], "entity")
self.assertIn("source.ip", shared["srcip"]["aliases"])
self.assertIn("client_ip", shared["srcip"]["aliases"])
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")
@@ -199,6 +223,16 @@ class ProfileSuggestionTests(unittest.TestCase):
self.assertNotIn("full_message", advised["profile"]["categorical_fields"])
self.assertEqual(set(advised["profile"]["detectors"]), {"auth_failure"})
def test_missing_llm_advice_keeps_heuristic_profile(self):
suggestion = suggest_stream_profiles([
parse_log_line("fgai_stream_id=app fgai_stream=App actor_id=user1 action=login")
])[0]
advised = apply_profile_advice([suggestion], [])[0]
self.assertEqual(advised["profile_advisor"]["status"], "heuristic")
self.assertEqual(advised["profile"], suggestion["profile"])
if __name__ == "__main__":
unittest.main()