671 lines
41 KiB
Python
671 lines
41 KiB
Python
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_entities
|
|
from .detectors import DETECTOR_MINIMUMS, event_detector_categories
|
|
from .normalization import canonical_value
|
|
from .query_details import event_query_details
|
|
|
|
DEFAULT_RETENTION_DAYS = 7
|
|
DEFAULT_VALUE_RETENTION_DAYS = 3
|
|
DEFAULT_MAX_VALUES_PER_FIELD = 500
|
|
DEFAULT_TRAINING_DAYS = 7
|
|
MIN_REPORTED_DEVIATION_SCORE = 15
|
|
MAX_RARE_VALUES_PER_ENTITY = 5
|
|
MAX_PROFILE_VALUE_LENGTH = 160
|
|
NOISY_VALUE_FIELD_PARTS = (
|
|
"answer",
|
|
"body",
|
|
"commandline",
|
|
"command_line",
|
|
"context",
|
|
"cookie",
|
|
"ephemeral",
|
|
"fingerprint",
|
|
"full_message",
|
|
"hash",
|
|
"message",
|
|
"payload",
|
|
"queryparameter",
|
|
"raw",
|
|
"request",
|
|
"response",
|
|
"session",
|
|
"stack",
|
|
"token",
|
|
"uri",
|
|
"url",
|
|
"user_agent",
|
|
)
|
|
|
|
|
|
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
|
|
|
|
|
|
def _baseline_confidence(samples: int, *, temporal: bool) -> str:
|
|
if samples >= 24 and temporal:
|
|
return "high"
|
|
if samples >= 18:
|
|
return "medium"
|
|
return "low"
|
|
|
|
|
|
def _age_days(oldest_bucket: int | None, newest_timestamp: int) -> float:
|
|
if not oldest_bucket:
|
|
return 0.0
|
|
return max(0.0, (newest_timestamp - int(oldest_bucket)) / 86400)
|
|
|
|
|
|
def _weighted_score(base: int, profile: object | None, field: str, detector: str) -> tuple[int, float]:
|
|
weights = getattr(profile, "field_weights", {}) if profile else {}
|
|
if not isinstance(weights, dict):
|
|
return base, 1.0
|
|
field_key = str(field).lower()
|
|
detector_key = str(detector).lower()
|
|
multiplier = 1.0
|
|
for key in (field_key, detector_key):
|
|
value = weights.get(key)
|
|
if isinstance(value, (int, float)):
|
|
multiplier *= float(value)
|
|
nested = weights.get(field_key)
|
|
if isinstance(nested, dict):
|
|
value = nested.get(detector_key)
|
|
if isinstance(value, (int, float)):
|
|
multiplier *= float(value)
|
|
multiplier = max(0.0, min(5.0, multiplier))
|
|
return min(100, max(0, int(round(base * multiplier)))), round(multiplier, 2)
|
|
|
|
|
|
def _sample_event(event: LogEvent, value: str = "") -> dict[str, object]:
|
|
query_details = event_query_details(event)
|
|
return {
|
|
"timestamp": event.fields.get("eventtime", event.fields.get("timestamp", "")),
|
|
"source": event.src_ip or event.fields.get("source", ""),
|
|
"destination": event.dst_ip or canonical_value(event.fields, "context"),
|
|
"action": event.action,
|
|
"severity": event.severity,
|
|
"service": canonical_value(event.fields, "service"),
|
|
"value": value,
|
|
"message": canonical_value(event.fields, "context")[:240],
|
|
"query_details": query_details,
|
|
"graylog_query": query_details["query"],
|
|
}
|
|
|
|
|
|
def _relationship_key(left: str, right: str) -> str:
|
|
return f"relationship:{left.lower()}->{right.lower()}"
|
|
|
|
|
|
def _should_store_profile_value(field: str, value: str) -> bool:
|
|
field = field.lower()
|
|
if field.startswith("relationship:"):
|
|
return True
|
|
if len(value) > MAX_PROFILE_VALUE_LENGTH:
|
|
return False
|
|
if any(part in field for part in NOISY_VALUE_FIELD_PARTS):
|
|
return False
|
|
return True
|
|
|
|
|
|
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)
|
|
);
|
|
create table if not exists profile_detector_buckets (
|
|
stream_id text not null, entity text not null, detector text not null, bucket_start integer not null,
|
|
events integer not null, primary key (stream_id, entity, detector, bucket_start)
|
|
);
|
|
create table if not exists profile_detector_temporal_buckets (
|
|
stream_id text not null, entity text not null, detector text not null,
|
|
weekday integer not null, hour integer not null, bucket_start integer not null,
|
|
events integer not null, primary key (stream_id, entity, detector, weekday, hour, bucket_start)
|
|
);
|
|
"""
|
|
)
|
|
self._migrate(connection)
|
|
|
|
def _migrate(self, connection: sqlite3.Connection) -> None:
|
|
columns = {row[1] for row in connection.execute("pragma table_info(seen_events)").fetchall()}
|
|
if "first_seen" not in columns:
|
|
connection.execute("alter table seen_events add column first_seen integer not null default 0")
|
|
columns = {row[1] for row in connection.execute("pragma table_info(profile_seen_events)").fetchall()}
|
|
if "first_seen" not in columns:
|
|
connection.execute("alter table profile_seen_events add column first_seen integer not null default 0")
|
|
columns = {row[1] for row in connection.execute("pragma table_info(profile_values)").fetchall()}
|
|
if "last_seen" not in columns:
|
|
connection.execute("alter table profile_values add column last_seen integer not null default 0")
|
|
connection.executescript(
|
|
"""
|
|
create index if not exists idx_profile_buckets_lookup on profile_buckets(stream_id, entity, field, bucket_start);
|
|
create index if not exists idx_profile_temporal_lookup on profile_temporal_buckets(stream_id, entity, field, weekday, hour, bucket_start);
|
|
create index if not exists idx_profile_detector_lookup on profile_detector_buckets(stream_id, entity, detector, bucket_start);
|
|
create index if not exists idx_profile_detector_temporal_lookup on profile_detector_temporal_buckets(stream_id, entity, detector, weekday, hour, bucket_start);
|
|
create index if not exists idx_profile_values_lookup on profile_values(stream_id, entity, field, seen_count);
|
|
"""
|
|
)
|
|
|
|
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(fingerprint, first_seen) values (?, ?)", (fingerprint, observed_at)).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,
|
|
max_values_per_field: int = DEFAULT_MAX_VALUES_PER_FIELD,
|
|
) -> 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])
|
|
detector_pending: Counter[tuple[str, str, str, int]] = Counter()
|
|
detector_temporal_pending: Counter[tuple[str, str, str, int, int, int]] = Counter()
|
|
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(fingerprint, first_seen) values (?, ?)", (fingerprint, observed_at)).rowcount != 1:
|
|
continue
|
|
entities = profile_entities(event, profile)
|
|
if not entities:
|
|
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 entity in entities:
|
|
for detector in event_detector_categories(event):
|
|
detector_pending[(stream_id, entity, detector, bucket)] += 1
|
|
detector_temporal_pending[(stream_id, entity, detector, moment.weekday(), moment.hour, bucket)] += 1
|
|
for relation in getattr(profile, "relationship_fields", ()):
|
|
left = str(getattr(relation, "left", "")).lower()
|
|
right = str(getattr(relation, "right", "")).lower()
|
|
left_value = event.fields.get(left)
|
|
right_value = event.fields.get(right)
|
|
if left_value and right_value:
|
|
pending_values[(stream_id, left_value, _relationship_key(left, right), right_value)] += 1
|
|
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 and _should_store_profile_value(key[2], 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))
|
|
value_counts: dict[tuple[str, str, str], int] = {}
|
|
for key, count in pending_values.items():
|
|
if max_values_per_field > 0:
|
|
group_key = key[:3]
|
|
known = connection.execute(
|
|
"select 1 from profile_values where stream_id=? and entity=? and field=? and value=?",
|
|
key,
|
|
).fetchone()
|
|
if known is None:
|
|
if group_key not in value_counts:
|
|
value_counts[group_key] = int(connection.execute(
|
|
"select count(*) from profile_values where stream_id=? and entity=? and field=?",
|
|
group_key,
|
|
).fetchone()[0])
|
|
if value_counts[group_key] >= max_values_per_field:
|
|
continue
|
|
value_counts[group_key] += 1
|
|
connection.execute("""insert into profile_values values (?, ?, ?, ?, ?, ?)
|
|
on conflict(stream_id, entity, field, value) do update set seen_count=seen_count+excluded.seen_count,last_seen=excluded.last_seen""", (*key, count, observed_at))
|
|
for (stream_id, entity, detector, bucket), count in detector_pending.items():
|
|
connection.execute("""insert into profile_detector_buckets values (?, ?, ?, ?, ?)
|
|
on conflict(stream_id, entity, detector, bucket_start) do update set events=events+excluded.events""", (stream_id, entity, detector, bucket, count))
|
|
for (stream_id, entity, detector, weekday, hour, bucket), count in detector_temporal_pending.items():
|
|
connection.execute("""insert into profile_detector_temporal_buckets values (?, ?, ?, ?, ?, ?, ?)
|
|
on conflict(stream_id, entity, detector, weekday, hour, bucket_start) do update set events=events+excluded.events""", (stream_id, entity, detector, weekday, hour, bucket, count))
|
|
return len(pending)
|
|
|
|
def profile_deviations(self, events: list[LogEvent], profiles: dict[str, object], *, min_training_days: int = 0) -> dict[str, list[dict[str, object]]]:
|
|
current: dict[tuple[str, str, str], list[float]] = defaultdict(lambda: [0, 0.0])
|
|
entity_events: dict[tuple[str, str], list[LogEvent]] = defaultdict(list)
|
|
detector_current: Counter[tuple[str, str, str]] = Counter()
|
|
for event in events:
|
|
profile = profiles.get(event.fields.get("fgai_stream_id", ""))
|
|
if not profile:
|
|
continue
|
|
entities = profile_entities(event, profile)
|
|
if not entities:
|
|
continue
|
|
numeric = {str(field).lower() for field in getattr(profile, "numeric_fields", ())}
|
|
for entity in entities:
|
|
entity_events[(event.fields.get("fgai_stream_id", ""), entity)].append(event)
|
|
for detector in event_detector_categories(event):
|
|
detector_current[(event.fields.get("fgai_stream_id", ""), entity, detector)] += 1
|
|
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():
|
|
profile = profiles.get(stream)
|
|
matching = [event for event in events if event.fields.get("fgai_stream_id") == stream and entity in profile_entities(event, profiles.get(stream)) 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()
|
|
temporal = True
|
|
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()
|
|
temporal = False
|
|
baseline_scope = "all observed periods"
|
|
if len(rows) < 12:
|
|
continue
|
|
oldest = connection.execute("select min(bucket_start) from profile_buckets where stream_id=? and entity=? and field=?", (stream, entity, field)).fetchone()[0]
|
|
baseline_age_days = _age_days(oldest, current_timestamp)
|
|
if baseline_age_days < min_training_days:
|
|
continue
|
|
if values[1] == 0:
|
|
continue
|
|
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:
|
|
sample_values = sorted({event.fields.get(field, "") for event in matching})[:5]
|
|
evidence_events = [_sample_event(event, event.fields.get(field, "")) for event in matching[:5]]
|
|
confidence = _baseline_confidence(len(rows), temporal=temporal)
|
|
base_score = 18 if confidence == "high" else 15 if confidence == "medium" else 10
|
|
score, weight = _weighted_score(base_score, profile, field, "numeric_baseline")
|
|
if score >= MIN_REPORTED_DEVIATION_SCORE:
|
|
output[entity].append({"detector": "numeric_baseline", "field": field, "stream_id": stream, "score": score, "base_score": base_score, "weight": weight, "confidence": confidence, "baseline_samples": len(rows), "baseline_age_days": round(baseline_age_days, 2), "baseline_scope": baseline_scope, "reason": reason, "current": round(current_value, 2), "baseline": round(mean(history), 2), "sample_values": sample_values, "sample_events": evidence_events})
|
|
|
|
# Event-rate burst is calculated once per stream/entity, rather than once per selected field.
|
|
for (stream, entity), matching in entity_events.items():
|
|
profile = profiles.get(stream)
|
|
fields = [*getattr(profile, "categorical_fields", ()), *getattr(profile, "numeric_fields", ())] if profile else []
|
|
if not fields:
|
|
continue
|
|
reference_field = str(fields[0]).lower()
|
|
current_timestamp = _event_epoch(matching[-1], int(time.time()))
|
|
moment = datetime.fromtimestamp(current_timestamp, tz=timezone.utc)
|
|
rows = connection.execute("select events from profile_temporal_buckets where stream_id=? and entity=? and field=? and weekday=? and hour=? order by bucket_start desc limit 25", (stream, entity, reference_field, moment.weekday(), moment.hour)).fetchall()
|
|
temporal = True
|
|
baseline_scope = "same weekday/hour"
|
|
if len(rows) < 12:
|
|
rows = connection.execute("select events from profile_buckets where stream_id=? and entity=? and field=? order by bucket_start desc limit 25", (stream, entity, reference_field)).fetchall()
|
|
temporal = False
|
|
baseline_scope = "all observed periods"
|
|
if len(rows) < 12:
|
|
continue
|
|
oldest = connection.execute("select min(bucket_start) from profile_buckets where stream_id=? and entity=? and field=?", (stream, entity, reference_field)).fetchone()[0]
|
|
baseline_age_days = _age_days(oldest, current_timestamp)
|
|
if baseline_age_days < min_training_days:
|
|
continue
|
|
history = [row[0] for row in rows]
|
|
current_value = len(matching)
|
|
z_score = (current_value - mean(history)) / (pstdev(history) or 1.0)
|
|
if z_score < 3:
|
|
continue
|
|
confidence = _baseline_confidence(len(rows), temporal=temporal)
|
|
base_score = min(30, (15 if confidence == "high" else 12 if confidence == "medium" else 8) + int(z_score))
|
|
score, weight = _weighted_score(base_score, profile, "event_rate", "event_rate_burst")
|
|
samples = [_sample_event(event) for event in matching[:5]]
|
|
if score >= MIN_REPORTED_DEVIATION_SCORE:
|
|
output[entity].append({"detector": "event_rate_burst", "field": "event_rate", "stream_id": stream, "score": score, "base_score": base_score, "weight": weight, "confidence": confidence, "baseline_samples": len(rows), "baseline_age_days": round(baseline_age_days, 2), "baseline_scope": baseline_scope, "reason": f"event rate burst above its {baseline_scope} baseline (z={z_score:.1f})", "current": current_value, "baseline": round(mean(history), 2), "sample_values": [], "sample_events": samples})
|
|
|
|
for (stream, entity, detector), current_value in detector_current.items():
|
|
profile = profiles.get(stream)
|
|
settings = getattr(profile, "detectors", {}).get(detector, {}) if profile else {}
|
|
if not settings.get("enabled", True):
|
|
continue
|
|
minimum = int(settings.get("minimum", DETECTOR_MINIMUMS[detector]))
|
|
z_threshold = float(settings.get("z_threshold", 3.0))
|
|
if current_value < minimum:
|
|
continue
|
|
matching = [event for event in entity_events[(stream, entity)] if detector in event_detector_categories(event)]
|
|
current_timestamp = _event_epoch(matching[-1], int(time.time()))
|
|
moment = datetime.fromtimestamp(current_timestamp, tz=timezone.utc)
|
|
rows = connection.execute("select events from profile_detector_temporal_buckets where stream_id=? and entity=? and detector=? and weekday=? and hour=? order by bucket_start desc limit 25", (stream, entity, detector, moment.weekday(), moment.hour)).fetchall()
|
|
temporal = True
|
|
baseline_scope = "same weekday/hour"
|
|
if len(rows) < 12:
|
|
rows = connection.execute("select events from profile_detector_buckets where stream_id=? and entity=? and detector=? order by bucket_start desc limit 25", (stream, entity, detector)).fetchall()
|
|
temporal = False
|
|
baseline_scope = "all observed periods"
|
|
if len(rows) < 12:
|
|
continue
|
|
oldest = connection.execute("select min(bucket_start) from profile_detector_buckets where stream_id=? and entity=? and detector=?", (stream, entity, detector)).fetchone()[0]
|
|
baseline_age_days = _age_days(oldest, current_timestamp)
|
|
if baseline_age_days < min_training_days:
|
|
continue
|
|
history = [row[0] for row in rows]
|
|
z_score = (current_value - mean(history)) / (pstdev(history) or 1.0)
|
|
if z_score < z_threshold:
|
|
continue
|
|
confidence = _baseline_confidence(len(rows), temporal=temporal)
|
|
base_score = min(35, (18 if detector == "auth_failure" else 15 if detector == "deny_action" else 12) + int(z_score))
|
|
score, weight = _weighted_score(base_score, profile, detector, f"{detector}_burst")
|
|
samples = [_sample_event(event, detector) for event in matching[:5]]
|
|
if score >= MIN_REPORTED_DEVIATION_SCORE:
|
|
output[entity].append({"detector": f"{detector}_burst", "field": detector, "stream_id": stream, "score": score, "base_score": base_score, "weight": weight, "confidence": confidence, "baseline_samples": len(rows), "baseline_age_days": round(baseline_age_days, 2), "baseline_scope": baseline_scope, "reason": f"{detector.replace('_', ' ')} burst above its {baseline_scope} baseline (z={z_score:.1f}, minimum={minimum})", "current": current_value, "baseline": round(mean(history), 2), "sample_values": [detector], "sample_events": samples})
|
|
# Detect custom behavior relationships, e.g. username -> srcip or host -> process.name.
|
|
relationship_counts: Counter[tuple[str, str, str, str, str]] = Counter()
|
|
relationship_samples: dict[tuple[str, str, str, str, str], list[LogEvent]] = defaultdict(list)
|
|
for event in events:
|
|
stream = event.fields.get("fgai_stream_id", "")
|
|
profile = profiles.get(stream)
|
|
if not profile:
|
|
continue
|
|
for relation in getattr(profile, "relationship_fields", ()):
|
|
left = str(getattr(relation, "left", "")).lower()
|
|
right = str(getattr(relation, "right", "")).lower()
|
|
left_value = event.fields.get(left)
|
|
right_value = event.fields.get(right)
|
|
if not left_value or not right_value:
|
|
continue
|
|
key = (stream, left_value, left, right, right_value)
|
|
relationship_counts[key] += 1
|
|
if len(relationship_samples[key]) < 5:
|
|
relationship_samples[key].append(event)
|
|
relationship_limit: Counter[tuple[str, str]] = Counter()
|
|
for (stream, left_value, left, right, right_value), count in relationship_counts.items():
|
|
field = _relationship_key(left, right)
|
|
known = connection.execute("select seen_count from profile_values where stream_id=? and entity=? and field=? and value=?", (stream, left_value, field, right_value)).fetchone()
|
|
known_total = connection.execute("select coalesce(sum(seen_count), 0) from profile_values where stream_id=? and entity=? and field=?", (stream, left_value, field)).fetchone()[0]
|
|
if known is not None or int(known_total or 0) < 12:
|
|
continue
|
|
if relationship_limit[(stream, left_value)] >= MAX_RARE_VALUES_PER_ENTITY:
|
|
continue
|
|
oldest = connection.execute("select min(bucket_start) from profile_buckets where stream_id=? and entity=?", (stream, left_value)).fetchone()[0]
|
|
if oldest is None:
|
|
oldest = connection.execute("select min(last_seen) from profile_values where stream_id=? and entity=? and field=?", (stream, left_value, field)).fetchone()[0]
|
|
baseline_age_days = _age_days(oldest, _event_epoch(relationship_samples[(stream, left_value, left, right, right_value)][0], int(time.time())))
|
|
if baseline_age_days < min_training_days:
|
|
continue
|
|
profile = profiles.get(stream)
|
|
base_score = 28
|
|
score, weight = _weighted_score(base_score, profile, field, "new_relationship")
|
|
if score < MIN_REPORTED_DEVIATION_SCORE:
|
|
continue
|
|
samples = relationship_samples[(stream, left_value, left, right, right_value)]
|
|
output[left_value].append({
|
|
"detector": "new_relationship",
|
|
"field": field,
|
|
"stream_id": stream,
|
|
"score": score,
|
|
"base_score": base_score,
|
|
"weight": weight,
|
|
"confidence": "medium",
|
|
"baseline_samples": int(known_total),
|
|
"baseline_age_days": round(baseline_age_days, 2),
|
|
"baseline_scope": "known field relationships",
|
|
"reason": f"new {right} value for {left}={left_value}",
|
|
"value": right_value,
|
|
"sample_values": [f"{left}={left_value}", f"{right}={right_value}"],
|
|
"sample_events": [_sample_event(item, f"{left}={left_value} {right}={right_value}") for item in samples],
|
|
})
|
|
relationship_limit[(stream, left_value)] += 1
|
|
# Detect selected categorical values that have not appeared for this entity in prior data.
|
|
rare_counts: Counter[tuple[str, str]] = Counter()
|
|
for event in events:
|
|
profile = profiles.get(event.fields.get("fgai_stream_id", ""))
|
|
if not profile:
|
|
continue
|
|
entities = profile_entities(event, profile)
|
|
if not entities:
|
|
continue
|
|
stream = event.fields.get("fgai_stream_id", "")
|
|
for entity in entities:
|
|
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 coalesce(sum(seen_count), 0) from profile_values where stream_id=? and entity=? and field=?", (stream, entity, field)).fetchone()[0]
|
|
if known is None and known_total >= 30:
|
|
oldest = connection.execute("select min(bucket_start) from profile_buckets where stream_id=? and entity=? and field=?", (stream, entity, field)).fetchone()[0]
|
|
baseline_age_days = _age_days(oldest, _event_epoch(event, int(time.time())))
|
|
if baseline_age_days < min_training_days:
|
|
continue
|
|
if rare_counts[(stream, entity)] >= MAX_RARE_VALUES_PER_ENTITY:
|
|
continue
|
|
samples = [item for item in events if item.fields.get("fgai_stream_id") == stream and entity in profile_entities(item, profile) and item.fields.get(field) == value]
|
|
base_score = 18
|
|
score, weight = _weighted_score(base_score, profile, field, "rare_value")
|
|
if score < MIN_REPORTED_DEVIATION_SCORE:
|
|
continue
|
|
evidence = {"detector": "rare_value", "field": field, "stream_id": stream, "score": score, "base_score": base_score, "weight": weight, "confidence": "medium", "baseline_samples": int(known_total), "baseline_age_days": round(baseline_age_days, 2), "baseline_scope": "known field values", "reason": f"new {field} value for this entity", "value": value, "sample_values": [value], "sample_events": [_sample_event(item, value) for item in samples[:5]]}
|
|
if evidence not in output[entity]:
|
|
output[entity].append(evidence)
|
|
rare_counts[(stream, entity)] += 1
|
|
return output
|
|
|
|
def maintenance(
|
|
self,
|
|
*,
|
|
retention_days: int = DEFAULT_RETENTION_DAYS,
|
|
value_retention_days: int = DEFAULT_VALUE_RETENTION_DAYS,
|
|
max_values_per_field: int = DEFAULT_MAX_VALUES_PER_FIELD,
|
|
vacuum: bool = False,
|
|
include_rows: bool = False,
|
|
) -> dict[str, object]:
|
|
now = int(time.time())
|
|
bucket_cutoff = now - max(1, retention_days) * 86400
|
|
value_cutoff = now - max(1, value_retention_days) * 86400
|
|
deleted: dict[str, int] = {}
|
|
with self._connect() as connection:
|
|
for table in (
|
|
"source_buckets",
|
|
"profile_buckets",
|
|
"profile_temporal_buckets",
|
|
"profile_detector_buckets",
|
|
"profile_detector_temporal_buckets",
|
|
):
|
|
deleted[table] = connection.execute(f"delete from {table} where bucket_start < ?", (bucket_cutoff,)).rowcount
|
|
deleted["seen_events"] = connection.execute("delete from seen_events where first_seen > 0 and first_seen < ?", (bucket_cutoff,)).rowcount
|
|
deleted["profile_seen_events"] = connection.execute("delete from profile_seen_events where first_seen > 0 and first_seen < ?", (bucket_cutoff,)).rowcount
|
|
deleted["profile_values_stale_low_count"] = connection.execute("delete from profile_values where last_seen > 0 and last_seen < ? and seen_count <= 1", (value_cutoff,)).rowcount
|
|
if max_values_per_field > 0:
|
|
rows = connection.execute("select stream_id, entity, field, count(*) from profile_values group by stream_id, entity, field having count(*) > ?", (max_values_per_field,)).fetchall()
|
|
trimmed = 0
|
|
for stream_id, entity, field, _count in rows:
|
|
keep = {
|
|
row[0]
|
|
for row in connection.execute(
|
|
"""select value from profile_values
|
|
where stream_id=? and entity=? and field=?
|
|
order by seen_count desc, last_seen desc limit ?""",
|
|
(stream_id, entity, field, max_values_per_field),
|
|
).fetchall()
|
|
}
|
|
placeholders = ",".join("?" for _ in keep)
|
|
if keep:
|
|
trimmed += connection.execute(
|
|
f"delete from profile_values where stream_id=? and entity=? and field=? and value not in ({placeholders})",
|
|
(stream_id, entity, field, *keep),
|
|
).rowcount
|
|
deleted["profile_values_capped"] = trimmed
|
|
connection.execute("pragma optimize")
|
|
if vacuum:
|
|
with self._connect() as connection:
|
|
connection.execute("vacuum")
|
|
stats = self.stats(include_rows=include_rows)
|
|
return {"retention_days": retention_days, "value_retention_days": value_retention_days, "max_values_per_field": max_values_per_field, "vacuum": vacuum, "deleted": deleted, **stats}
|
|
|
|
def stats(self, *, include_rows: bool = False) -> dict[str, object]:
|
|
with self._connect() as connection:
|
|
page_size = int(connection.execute("pragma page_size").fetchone()[0])
|
|
page_count = int(connection.execute("pragma page_count").fetchone()[0])
|
|
freelist_count = int(connection.execute("pragma freelist_count").fetchone()[0])
|
|
rows = {}
|
|
if include_rows:
|
|
tables = (
|
|
"seen_events",
|
|
"source_buckets",
|
|
"source_values",
|
|
"profile_seen_events",
|
|
"profile_buckets",
|
|
"profile_temporal_buckets",
|
|
"profile_detector_buckets",
|
|
"profile_detector_temporal_buckets",
|
|
"profile_values",
|
|
)
|
|
rows = {table: int(connection.execute(f"select count(*) from {table}").fetchone()[0]) for table in tables}
|
|
size_bytes = self.path.stat().st_size if self.path.exists() else 0
|
|
stats: dict[str, object] = {
|
|
"size_bytes": size_bytes,
|
|
"page_size": page_size,
|
|
"page_count": page_count,
|
|
"freelist_count": freelist_count,
|
|
"reclaimable_bytes": freelist_count * page_size,
|
|
}
|
|
if include_rows:
|
|
stats["rows"] = rows
|
|
return stats
|
|
|
|
def profile_readiness(self, profiles: dict[str, object], *, min_training_days: int = 0) -> 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]
|
|
oldest = connection.execute("select min(bucket_start) from profile_buckets where stream_id=? and field=?", (stream_id, str(field).lower())).fetchone()[0]
|
|
age_days = _age_days(oldest, int(time.time()))
|
|
rows.append({"stream_id": stream_id, "field": str(field), "buckets": count, "age_days": round(age_days, 2), "training_days": min_training_days, "ready": count >= 12 and age_days >= min_training_days})
|
|
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
|