from __future__ import annotations import hashlib import sqlite3 import time from collections import Counter, defaultdict from datetime import datetime, timezone from pathlib import Path from statistics import mean, pstdev from .logs import THREAT_ACTIONS, is_utm_event from .models import LogEvent from .entities import profile_entity def _number(value: str | None) -> int: try: return int(float(value or 0)) except ValueError: return 0 def _event_epoch(event: LogEvent, fallback: int) -> int: value = event.fields.get("eventtime", event.fields.get("timestamp", "")) if not value: return fallback try: return int(float(value)) except ValueError: try: return int(datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()) except ValueError: return fallback 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) ); create table if not exists profile_seen_events (fingerprint text primary key); create table if not exists profile_temporal_buckets ( stream_id text not null, entity text not null, field text not null, weekday integer not null, hour integer 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, weekday, hour, bucket_start) ); """ ) 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()) pending: dict[tuple[str, str, str, int], list[float]] = defaultdict(lambda: [0, 0.0, 0.0]) temporal_pending: dict[tuple[str, str, str, int, int, int], list[float]] = defaultdict(lambda: [0, 0.0, 0.0]) pending_values: Counter[tuple[str, str, str, str]] = Counter() with self._connect() as connection: for event in events: stream_id = event.fields.get("fgai_stream_id", "") profile = profiles.get(stream_id) if not profile: continue fingerprint = hashlib.sha256(f"{stream_id}|{event.raw}".encode("utf-8", errors="replace")).hexdigest() if connection.execute("insert or ignore into profile_seen_events values (?)", (fingerprint,)).rowcount != 1: continue entity = profile_entity(event, str(getattr(profile, "entity_field", ""))) if not entity: continue timestamp = _event_epoch(event, observed_at) bucket = timestamp - (timestamp % self.bucket_seconds) moment = datetime.fromtimestamp(timestamp, tz=timezone.utc) 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(), bucket) 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 temporal = (*key[:3], moment.weekday(), moment.hour, bucket) temporal_pending[temporal][0] += 1 temporal_pending[temporal][1] += value temporal_pending[temporal][2] += value * value if key[2] not in numeric: raw_value = event.fields.get(key[2]) if raw_value: pending_values[(*key[:3], raw_value)] += 1 for (stream_id, entity, field, bucket), 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 (stream_id, entity, field, weekday, hour, bucket), values in temporal_pending.items(): connection.execute("""insert into profile_temporal_buckets values (?, ?, ?, ?, ?, ?, ?, ?, ?) on conflict(stream_id, entity, field, weekday, hour, 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, weekday, hour, 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 = profile_entity(event, str(getattr(profile, "entity_field", ""))) 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(): matching = [event for event in events if event.fields.get("fgai_stream_id") == stream and profile_entity(event, str(getattr(profiles.get(stream), "entity_field", ""))) == entity and event.fields.get(field)] current_timestamp = _event_epoch(matching[-1], int(time.time())) if matching else int(time.time()) moment = datetime.fromtimestamp(current_timestamp, tz=timezone.utc) rows = connection.execute("select events, numeric_sum from profile_temporal_buckets where stream_id=? and entity=? and field=? and weekday=? and hour=? order by bucket_start desc limit 25", (stream, entity, field, moment.weekday(), moment.hour)).fetchall() baseline_scope = "same weekday/hour" if len(rows) < 12: 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() baseline_scope = "all observed periods" 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 {baseline_scope} 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 {baseline_scope} baseline" deviation = abs(current_value - mean(history)) if deviation > (pstdev(history) or 1.0) * 3: samples = sorted({event.fields.get(field, "") for event in matching})[:5] evidence_events = [{"timestamp": event.fields.get("eventtime", event.fields.get("timestamp", "")), "source": event.src_ip or event.fields.get("source", ""), "destination": event.dst_ip or "", "action": event.action, "severity": event.severity, "service": event.fields.get("service", ""), "value": event.fields.get(field, ""), "message": event.fields.get("message", event.fields.get("msg", ""))[:240]} for event in matching[:5]] output[entity].append({"field": field, "stream_id": stream, "score": 15, "reason": reason, "current": round(current_value, 2), "baseline": round(mean(history), 2), "sample_values": samples, "sample_events": evidence_events}) # 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 = profile_entity(event, str(getattr(profile, "entity_field", ""))) 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: samples = [item for item in events if item.fields.get("fgai_stream_id") == stream and profile_entity(item, str(getattr(profile, "entity_field", ""))) == entity and item.fields.get(field) == value] evidence = {"field": field, "stream_id": stream, "score": 12, "reason": f"new {field} value for this entity", "value": value, "sample_values": [value], "sample_events": [{"timestamp": item.fields.get("eventtime", item.fields.get("timestamp", "")), "source": item.src_ip or item.fields.get("source", ""), "destination": item.dst_ip or "", "action": item.action, "severity": item.severity, "service": item.fields.get("service", ""), "value": value, "message": item.fields.get("message", item.fields.get("msg", ""))[:240]} for item in samples[:5]]} if evidence not in output[entity]: output[entity].append(evidence) return output def profile_readiness(self, profiles: dict[str, object]) -> list[dict[str, object]]: rows: list[dict[str, object]] = [] with self._connect() as connection: for stream_id, profile in profiles.items(): fields = [*getattr(profile, "categorical_fields", ()), *getattr(profile, "numeric_fields", ())] for field in fields: count = connection.execute("select count(distinct bucket_start) from profile_buckets where stream_id=? and field=?", (stream_id, str(field).lower())).fetchone()[0] rows.append({"stream_id": stream_id, "field": str(field), "buckets": count, "ready": count >= 12}) return rows 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