219 lines
13 KiB
Python
219 lines
13 KiB
Python
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:
|
|
matching = [event for event in events if event.fields.get("fgai_stream_id") == stream and event.fields.get(str(getattr(profiles.get(stream), "entity_field", "")).lower()) == entity and event.fields.get(field)]
|
|
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 = 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 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
|