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

Fortsatte roadmapen med multi-entity stream profiles

Browse Source
This commit is contained in:
larssand
2026-06-29 19:08:23 +02:00
parent 20b0e0a92e
commit 68370217da
10 changed files with 102 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
tests/test_baseline.py
View File

@@ -28,6 +28,17 @@ class BaselineTests(unittest.TestCase):
self.assertEqual(store.ingest_profile_fields(events, profiles, observed_at=1_700_000_000), 1)
self.assertEqual(store.ingest_profile_fields(events, profiles, observed_at=1_700_000_300), 0)
def test_profile_can_baseline_multiple_entities_from_same_event(self):
with tempfile.TemporaryDirectory() as directory:
store = BaselineStore(str(Path(directory) / "baseline.sqlite3"))
profiles = parse_profiles([{"stream_id": "vpn", "entity_fields": ["username", "srcip"], "categorical_fields": ["action"]}])
events = [parse_log_line("fgai_stream_id=vpn username=alice srcip=10.0.0.5 action=login")]
self.assertEqual(store.ingest_profile_fields(events, profiles, observed_at=1_700_000_000), 2)
readiness = store.profile_readiness(profiles)
self.assertEqual(readiness[0]["buckets"], 1)
def test_profile_rate_burst_is_one_confident_detector(self):
with tempfile.TemporaryDirectory() as directory:
store = BaselineStore(str(Path(directory) / "baseline.sqlite3"))

6
tests/test_stream_profiles.py
View File

@@ -8,8 +8,14 @@ class StreamProfileTests(unittest.TestCase):
profiles = parse_profiles([{"stream_id": "dns", "name": "DNS client behavior", "entity_field": "IP", "categorical_fields": ["QH"], "numeric_fields": ["Elapsed"]}])
self.assertEqual(profiles["dns"].name, "DNS client behavior")
self.assertEqual(profiles["dns"].entity_field, "IP")
self.assertEqual(profiles["dns"].entity_fields, ("IP",))
self.assertEqual(profiles["dns"].numeric_fields, ("Elapsed",))
def test_parses_multiple_entity_fields(self):
profiles = parse_profiles([{"stream_id": "vpn", "entity_fields": ["username", "srcip", "hostname"], "categorical_fields": ["action"]}])
self.assertEqual(profiles["vpn"].entity_field, "username")
self.assertEqual(profiles["vpn"].entity_fields, ("username", "srcip", "hostname"))
def test_parses_detector_thresholds(self):
profiles = parse_profiles([{"stream_id": "windows", "entity_field": "username", "detectors": {"auth_failure": {"enabled": False, "minimum": 7, "z_threshold": 4.5}}}])
self.assertEqual(profiles["windows"].detectors["auth_failure"], {"enabled": False, "minimum": 7, "z_threshold": 4.5})