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

configurable per-field detector weights är klart.

Browse Source
This commit is contained in:
larssand
2026-06-25 19:36:08 +02:00
parent 899dbdfc84
commit 9da1aedf52
9 changed files with 106 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
tests/test_baseline.py
View File

@@ -61,3 +61,20 @@ class BaselineTests(unittest.TestCase):
deviations = store.profile_deviations(burst, profiles)["alice"]
auth = next(item for item in deviations if item["detector"] == "auth_failure_burst")
self.assertEqual(auth["current"], 5)
def test_profile_field_weights_raise_detector_score(self):
with tempfile.TemporaryDirectory() as directory:
store = BaselineStore(str(Path(directory) / "baseline.sqlite3"))
profiles = parse_profiles([{"stream_id": "windows", "entity_field": "username", "categorical_fields": ["action"], "field_weights": {"auth_failure_burst": 2}}])
for index in range(12):
event = parse_log_line(f"fgai_stream_id=windows username=alice action=failed baseline={index}")
store.ingest_profile_fields([event], profiles, observed_at=1_700_000_000 + index * 300)
burst = [
parse_log_line(f"fgai_stream_id=windows username=alice action=failed burst={index}")
for index in range(5)
]
deviations = store.profile_deviations(burst, profiles)["alice"]
auth = next(item for item in deviations if item["detector"] == "auth_failure_burst")
self.assertEqual(auth["weight"], 2.0)
self.assertGreater(auth["score"], auth["base_score"])

6
tests/test_stream_profiles.py
View File

@@ -13,3 +13,9 @@ class StreamProfileTests(unittest.TestCase):
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})
def test_parses_field_weights(self):
profiles = parse_profiles([{"stream_id": "dns", "entity_field": "IP", "field_weights": {"QH": 1.5, "query_domain": {"rare_value": 2}, "bad": "nope"}}])
self.assertEqual(profiles["dns"].field_weights["qh"], 1.5)
self.assertEqual(profiles["dns"].field_weights["query_domain"]["rare_value"], 2.0)
self.assertNotIn("bad", profiles["dns"].field_weights)