import tempfile import time import unittest from pathlib import Path from fgai.baseline import BaselineStore from fgai.logs import parse_log_line from fgai.stream_profiles import parse_profiles class BaselineTests(unittest.TestCase): def test_creates_profile_after_twelve_completed_windows(self): with tempfile.TemporaryDirectory() as directory: store = BaselineStore(str(Path(directory) / "baseline.sqlite3")) for index in range(13): store.ingest([parse_log_line(f"srcip=10.0.0.1 dstport={1000 + index} hitcount=2 sentbyte=5")], observed_at=1_700_000_000 + index * 300) profiles = store.profiles({"10.0.0.1"}) self.assertEqual(profiles["10.0.0.1"]["samples"], 12) self.assertIn("1000", profiles["10.0.0.1"]["known_destination_ports"]) def test_profile_events_are_deduplicated_between_polls(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"]}]) events = [ parse_log_line(f"fgai_stream_id=windows username=alice action=login event={index}") for index in range(12) ] 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")) profiles = parse_profiles([{"stream_id": "windows", "entity_field": "username", "categorical_fields": ["action"]}]) for index in range(12): event = parse_log_line(f"fgai_stream_id=windows username=alice action=login 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=login burst={index}") for index in range(20) ] deviations = store.profile_deviations(burst, profiles)["alice"] rate = next(item for item in deviations if item["detector"] == "event_rate_burst") self.assertEqual(rate["field"], "event_rate") self.assertEqual(rate["baseline_samples"], 12) def test_profile_authentication_failure_burst(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"]}]) 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["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"]) def test_maintenance_prunes_old_baseline_rows(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"]}]) old = int(time.time()) - 20 * 86400 recent = int(time.time()) store.ingest([parse_log_line("srcip=10.0.0.1 dstport=443 old=1")], observed_at=old) store.ingest_profile_fields([parse_log_line("fgai_stream_id=windows username=alice action=login old=1")], profiles, observed_at=old) store.ingest([parse_log_line("srcip=10.0.0.2 dstport=443 recent=1")], observed_at=recent) store.ingest_profile_fields([parse_log_line("fgai_stream_id=windows username=bob action=login recent=1")], profiles, observed_at=recent) result = store.maintenance(retention_days=7, value_retention_days=7, max_values_per_field=2000, include_rows=True) self.assertGreaterEqual(result["deleted"]["source_buckets"], 1) self.assertGreaterEqual(result["deleted"]["profile_buckets"], 1) self.assertGreaterEqual(result["deleted"]["seen_events"], 1) self.assertGreaterEqual(result["deleted"]["profile_seen_events"], 1) self.assertIn("rows", result) self.assertIn("reclaimable_bytes", result) def test_profile_value_storage_skips_noisy_high_cardinality_fields(self): with tempfile.TemporaryDirectory() as directory: store = BaselineStore(str(Path(directory) / "baseline.sqlite3")) profiles = parse_profiles([{"stream_id": "web", "entity_field": "srcip", "categorical_fields": ["url", "action"]}]) events = [ parse_log_line(f"fgai_stream_id=web srcip=10.0.0.5 action=allow url=/download/{index}/very/noisy/path event={index}") for index in range(10) ] store.ingest_profile_fields(events, profiles, observed_at=1_700_000_000) stats = store.stats(include_rows=True) self.assertEqual(stats["rows"]["profile_buckets"], 2) with store._connect() as connection: fields = {row[0] for row in connection.execute("select distinct field from profile_values").fetchall()} self.assertIn("action", fields) self.assertNotIn("url", fields) def test_profile_value_storage_caps_new_values_during_ingest(self): with tempfile.TemporaryDirectory() as directory: store = BaselineStore(str(Path(directory) / "baseline.sqlite3")) profiles = parse_profiles([{"stream_id": "dns", "entity_field": "srcip", "categorical_fields": ["query_domain"]}]) events = [ parse_log_line(f"fgai_stream_id=dns srcip=10.0.0.5 query_domain=value-{index}.example event={index}") for index in range(5) ] store.ingest_profile_fields(events, profiles, observed_at=1_700_000_000, max_values_per_field=2) with store._connect() as connection: stored = connection.execute("select count(*) from profile_values").fetchone()[0] self.assertEqual(stored, 2) def test_rare_values_are_limited_per_entity(self): with tempfile.TemporaryDirectory() as directory: store = BaselineStore(str(Path(directory) / "baseline.sqlite3")) profiles = parse_profiles([{"stream_id": "dns", "entity_field": "srcip", "categorical_fields": ["query_domain"]}]) now = int(time.time()) - 86400 baseline = [ parse_log_line(f"fgai_stream_id=dns srcip=10.0.0.5 query_domain=known-{index}.example baseline={index}") for index in range(30) ] store.ingest_profile_fields(baseline, profiles, observed_at=now) burst = [ parse_log_line(f"fgai_stream_id=dns srcip=10.0.0.5 query_domain=new-{index}.example burst={index}") for index in range(12) ] deviations = store.profile_deviations(burst, profiles)["10.0.0.5"] rare = [item for item in deviations if item["detector"] == "rare_value"] self.assertLessEqual(len(rare), 5) self.assertTrue(all(item["score"] >= 15 for item in rare)) def test_training_days_gate_profile_deviations(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"]}]) now = int(time.time()) 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=now - 86400 + index * 300) burst = [ parse_log_line(f"fgai_stream_id=windows username=alice action=failed burst={index}") for index in range(5) ] self.assertNotIn("alice", store.profile_deviations(burst, profiles, min_training_days=7)) self.assertIn("alice", store.profile_deviations(burst, profiles, min_training_days=0)) def test_custom_relationship_flags_new_user_source_ip(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"], "relationship_fields": [{"left": "username", "right": "srcip", "name": "user source IP"}], }]) for index in range(12): event = parse_log_line(f"fgai_stream_id=windows username=peter srcip=10.0.0.10 action=success baseline={index}") store.ingest_profile_fields([event], profiles, observed_at=1_700_000_000 + index * 300) current = [parse_log_line("fgai_stream_id=windows username=peter srcip=10.0.0.99 action=success current=1")] deviations = store.profile_deviations(current, profiles)["peter"] relation = next(item for item in deviations if item["detector"] == "new_relationship") self.assertEqual(relation["field"], "relationship:username->srcip") self.assertEqual(relation["value"], "10.0.0.99") self.assertIn("new srcip value for username=peter", relation["reason"])