from __future__ import annotations import hashlib import sqlite3 import time from collections import Counter, defaultdict from pathlib import Path from statistics import mean, pstdev from .logs import THREAT_ACTIONS, is_utm_event from .models import LogEvent def _number(value: str | None) -> int: try: return int(float(value or 0)) except ValueError: return 0 class BaselineStore: """Persistent five-minute behavior baseline, implemented with stdlib SQLite.""" def __init__(self, path: str, *, bucket_seconds: int = 300) -> None: self.path = Path(path) self.path.parent.mkdir(parents=True, exist_ok=True) self.bucket_seconds = bucket_seconds with self._connect() as connection: connection.executescript( """ create table if not exists seen_events (fingerprint text primary key); create table if not exists source_buckets ( source_ip text not null, bucket_start integer not null, events integer not null, bytes integer not null, hitcount integer not null, denies integer not null, utm integer not null, primary key (source_ip, bucket_start) ); create table if not exists source_values ( source_ip text not null, kind text not null, value text not null, seen_count integer not null, primary key (source_ip, kind, value) ); create table if not exists profile_buckets ( stream_id text not null, entity text not null, field text not null, bucket_start integer not null, events integer not null, numeric_sum real not null, numeric_sum_squares real not null, primary key (stream_id, entity, field, bucket_start) ); create table if not exists profile_values ( stream_id text not null, entity text not null, field text not null, value text not null, seen_count integer not null, primary key (stream_id, entity, field, value) ); """ ) def _connect(self) -> sqlite3.Connection: return sqlite3.connect(self.path) def ingest(self, events: list[LogEvent], *, observed_at: int | None = None) -> int: observed_at = observed_at or int(time.time()) pending: dict[tuple[str, int], list[int]] = defaultdict(lambda: [0, 0, 0, 0, 0]) inserted = 0 with self._connect() as connection: for event in events: if not event.src_ip: continue fingerprint = hashlib.sha256(event.raw.encode("utf-8", errors="replace")).hexdigest() if connection.execute("insert or ignore into seen_events values (?)", (fingerprint,)).rowcount != 1: continue bucket = observed_at - (observed_at % self.bucket_seconds) values = pending[(event.src_ip, bucket)] values[0] += 1 values[1] += _number(event.fields.get("sentbyte")) + _number(event.fields.get("rcvdbyte")) values[2] += _number(event.fields.get("hitcount")) values[3] += int(event.action in THREAT_ACTIONS) values[4] += int(is_utm_event(event)) for kind, value in (("destination", event.dst_ip), ("destination_port", event.fields.get("dstport"))): if value: connection.execute( """insert into source_values values (?, ?, ?, 1) on conflict(source_ip, kind, value) do update set seen_count=seen_count+1""", (event.src_ip, kind, value), ) inserted += 1 for (source_ip, bucket), values in pending.items(): connection.execute( """insert into source_buckets values (?, ?, ?, ?, ?, ?, ?) on conflict(source_ip, bucket_start) do update set events=events+excluded.events, bytes=bytes+excluded.bytes, hitcount=hitcount+excluded.hitcount, denies=denies+excluded.denies, utm=utm+excluded.utm""", (source_ip, bucket, *values), ) return inserted def ingest_profile_fields(self, events: list[LogEvent], profiles: dict[str, object], *, observed_at: int | None = None) -> int: observed_at = observed_at or int(time.time()) bucket = observed_at - (observed_at % self.bucket_seconds) pending: dict[tuple[str, str, str], list[float]] = defaultdict(lambda: [0, 0.0, 0.0]) pending_values: Counter[tuple[str, str, str, str]] = Counter() for event in events: stream_id = event.fields.get("fgai_stream_id", "") profile = profiles.get(stream_id) if not profile: continue entity_field = str(getattr(profile, "entity_field", "")).lower() entity = event.fields.get(entity_field) if not entity: continue fields = [*getattr(profile, "categorical_fields", ()), *getattr(profile, "numeric_fields", ())] numeric = {str(field).lower() for field in getattr(profile, "numeric_fields", ())} for field in fields: key = (stream_id, entity, str(field).lower()) value = _number(event.fields.get(key[2])) if key[2] in numeric else 0 pending[key][0] += 1 pending[key][1] += value pending[key][2] += value * value if key[2] not in numeric: raw_value = event.fields.get(key[2]) if raw_value: pending_values[(*key, raw_value)] += 1 with self._connect() as connection: for (stream_id, entity, field), values in pending.items(): connection.execute("""insert into profile_buckets values (?, ?, ?, ?, ?, ?, ?) on conflict(stream_id, entity, field, bucket_start) do update set events=events+excluded.events,numeric_sum=numeric_sum+excluded.numeric_sum,numeric_sum_squares=numeric_sum_squares+excluded.numeric_sum_squares""", (stream_id, entity, field, bucket, *values)) for key, count in pending_values.items(): connection.execute("""insert into profile_values values (?, ?, ?, ?, ?) on conflict(stream_id, entity, field, value) do update set seen_count=seen_count+excluded.seen_count""", (*key, count)) return len(pending) def profile_deviations(self, events: list[LogEvent], profiles: dict[str, object]) -> dict[str, list[dict[str, object]]]: current: dict[tuple[str, str, str], list[float]] = defaultdict(lambda: [0, 0.0]) for event in events: profile = profiles.get(event.fields.get("fgai_stream_id", "")) if not profile: continue entity = event.fields.get(str(getattr(profile, "entity_field", "")).lower()) if not entity: continue numeric = {str(field).lower() for field in getattr(profile, "numeric_fields", ())} for field in [*getattr(profile, "categorical_fields", ()), *getattr(profile, "numeric_fields", ())]: key = (event.fields.get("fgai_stream_id", ""), entity, str(field).lower()) current[key][0] += 1 if key[2] in numeric: current[key][1] += _number(event.fields.get(key[2])) output: dict[str, list[dict[str, object]]] = defaultdict(list) with self._connect() as connection: for (stream, entity, field), values in current.items(): rows = connection.execute("select events, numeric_sum from profile_buckets where stream_id=? and entity=? and field=? order by bucket_start desc limit 25", (stream, entity, field)).fetchall() if len(rows) < 12: continue if values[1] == 0: history = [row[0] for row in rows] current_value = values[0] reason = f"{field} event rate deviates from its stream baseline" else: history = [row[1] / row[0] if row[0] else 0 for row in rows] current_value = values[1] / values[0] if values[0] else 0 reason = f"{field} value deviates from its stream baseline" deviation = abs(current_value - mean(history)) if deviation > (pstdev(history) or 1.0) * 3: output[entity].append({"field": field, "stream_id": stream, "score": 15, "reason": reason}) # Detect selected categorical values that have not appeared for this entity in prior data. for event in events: profile = profiles.get(event.fields.get("fgai_stream_id", "")) if not profile: continue entity = event.fields.get(str(getattr(profile, "entity_field", "")).lower()) if not entity: continue stream = event.fields.get("fgai_stream_id", "") for field in getattr(profile, "categorical_fields", ()): field = str(field).lower() value = event.fields.get(field) if not value: continue known = connection.execute("select seen_count from profile_values where stream_id=? and entity=? and field=? and value=?", (stream, entity, field, value)).fetchone() known_total = connection.execute("select count(*) from profile_values where stream_id=? and entity=? and field=?", (stream, entity, field)).fetchone()[0] if known is None and known_total >= 10: evidence = {"field": field, "stream_id": stream, "score": 12, "reason": f"new {field} value for this entity", "value": value} if evidence not in output[entity]: output[entity].append(evidence) return output def profiles(self, source_ips: set[str]) -> dict[str, dict[str, object]]: profiles: dict[str, dict[str, object]] = {} with self._connect() as connection: for source_ip in source_ips: rows = connection.execute( "select events, bytes, hitcount from source_buckets where source_ip=? order by bucket_start", (source_ip,) ).fetchall() # Do not compare the current (often incomplete) bucket to itself. rows = rows[:-1] if len(rows) < 12: continue rates = [row[0] * 60 / self.bucket_seconds for row in rows] hit_rates = [row[2] * 60 / self.bucket_seconds for row in rows] known = connection.execute( "select kind, value from source_values where source_ip=?", (source_ip,) ).fetchall() profiles[source_ip] = { "samples": len(rows), "event_rate_mean": mean(rates), "event_rate_stddev": pstdev(rates) or 1.0, "hitcount_rate_mean": mean(hit_rates), "hitcount_rate_stddev": pstdev(hit_rates) or 1.0, "known_destinations": [value for kind, value in known if kind == "destination"], "known_destination_ports": [value for kind, value in known if kind == "destination_port"], } return profiles