e35445da4f
Field roles are now inferred from observed values instead of names, with src/dst disambiguation, value-pattern detector hints, an Ollama advisor scoped to ambiguous fields only, confidence-gated auto-apply with manual-field protection, and drift flagging for review
69 lines
3.5 KiB
Python
69 lines
3.5 KiB
Python
import unittest
|
|
|
|
from fgai.stream_profiles import parse_profiles
|
|
|
|
|
|
class StreamProfileTests(unittest.TestCase):
|
|
def test_parses_field_level_profile(self):
|
|
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})
|
|
|
|
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)
|
|
|
|
def test_parses_relationship_fields(self):
|
|
profiles = parse_profiles([{"stream_id": "windows", "entity_field": "username", "relationship_fields": [{"left": "username", "right": "srcip", "name": "user source IP"}, {"left": "username", "right": "srcip"}]}])
|
|
|
|
relationships = profiles["windows"].relationship_fields
|
|
|
|
self.assertEqual(len(relationships), 1)
|
|
self.assertEqual(relationships[0].left, "username")
|
|
self.assertEqual(relationships[0].right, "srcip")
|
|
self.assertEqual(relationships[0].name, "user source IP")
|
|
|
|
def test_field_origin_defaults_missing_fields_to_manual(self):
|
|
profiles = parse_profiles([{
|
|
"stream_id": "windows",
|
|
"entity_field": "username",
|
|
"entity_fields": ["username", "hostname"],
|
|
"categorical_fields": ["action"],
|
|
"field_origin": {"hostname": "auto"},
|
|
}])
|
|
|
|
origin = profiles["windows"].field_origin
|
|
self.assertEqual(origin["username"], "manual")
|
|
self.assertEqual(origin["hostname"], "auto")
|
|
self.assertEqual(origin["action"], "manual")
|
|
|
|
def test_field_origin_round_trips_and_rejects_unknown_values(self):
|
|
profiles = parse_profiles([{
|
|
"stream_id": "windows",
|
|
"entity_field": "username",
|
|
"field_origin": {"username": "bogus"},
|
|
"updated_at": 1757000000,
|
|
}])
|
|
|
|
self.assertEqual(profiles["windows"].field_origin["username"], "manual")
|
|
self.assertEqual(profiles["windows"].updated_at, 1757000000)
|
|
|
|
def test_missing_field_origin_and_updated_at_default_safely(self):
|
|
profiles = parse_profiles([{"stream_id": "windows", "entity_field": "username"}])
|
|
|
|
self.assertEqual(profiles["windows"].field_origin, {"username": "manual"})
|
|
self.assertEqual(profiles["windows"].updated_at, 0)
|