64 lines
3.5 KiB
Python
64 lines
3.5 KiB
Python
import tempfile
|
|
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_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)
|