diff --git a/ROADMAP.md b/ROADMAP.md
index 40edaea..cc064c1 100644
--- a/ROADMAP.md
+++ b/ROADMAP.md
@@ -22,9 +22,11 @@ operational value and dependency, not by UI appeal.
Goal: make findings more accurate before adding more integrations.
-- [ ] Add baseline confidence based on sample count, profile age, and time-bucket coverage.
-- [ ] Add burst detection for event rate, authentication failures, DNS volume, and denied traffic.
-- [ ] Add rare-value detection with frequency thresholds and configurable field weights.
+- [x] Add baseline confidence based on sample count and time-bucket coverage.
+- [x] Add generic event-rate burst detection per stream/entity.
+- [x] Add rare-value detection with a minimum historical observation threshold.
+- [ ] Add detector-specific authentication failure, DNS volume, and denied-traffic burst thresholds.
+- [ ] Add configurable per-field detector weights.
- [ ] Add sequence detection, for example DNS lookup -> outbound connection -> authentication event.
- [ ] Add per-stream detector enablement and thresholds in the UI.
- [ ] Add a dry-run replay command to evaluate detector changes against a selected historic Graylog time range.
diff --git a/src/fgai/baseline.py b/src/fgai/baseline.py
index 97123c2..dc9bbfb 100644
--- a/src/fgai/baseline.py
+++ b/src/fgai/baseline.py
@@ -33,6 +33,14 @@ def _event_epoch(event: LogEvent, fallback: int) -> int:
return fallback
+def _baseline_confidence(samples: int, *, temporal: bool) -> str:
+ if samples >= 24 and temporal:
+ return "high"
+ if samples >= 18:
+ return "medium"
+ return "low"
+
+
class BaselineStore:
"""Persistent five-minute behavior baseline, implemented with stdlib SQLite."""
@@ -161,6 +169,7 @@ class BaselineStore:
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])
+ entity_events: dict[tuple[str, str], list[LogEvent]] = defaultdict(list)
for event in events:
profile = profiles.get(event.fields.get("fgai_stream_id", ""))
if not profile:
@@ -168,6 +177,7 @@ class BaselineStore:
entity = profile_entity(event, str(getattr(profile, "entity_field", "")))
if not entity:
continue
+ entity_events[(event.fields.get("fgai_stream_id", ""), entity)].append(event)
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())
@@ -181,25 +191,54 @@ class BaselineStore:
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
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"
+ 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:
- samples = sorted({event.fields.get(field, "") for event in matching})[:5]
+ sample_values = 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})
+ confidence = _baseline_confidence(len(rows), temporal=temporal)
+ score = 18 if confidence == "high" else 15 if confidence == "medium" else 10
+ output[entity].append({"detector": "numeric_baseline", "field": field, "stream_id": stream, "score": score, "confidence": confidence, "baseline_samples": len(rows), "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
+ 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)
+ score = min(30, (15 if confidence == "high" else 12 if confidence == "medium" else 8) + int(z_score))
+ samples = [{"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": "", "message": event.fields.get("message", event.fields.get("msg", ""))[:240]} for event in matching[:5]]
+ output[entity].append({"detector": "event_rate_burst", "field": "event_rate", "stream_id": stream, "score": score, "confidence": confidence, "baseline_samples": len(rows), "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})
# 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", ""))
@@ -215,10 +254,10 @@ class BaselineStore:
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:
+ 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:
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]]}
+ evidence = {"detector": "rare_value", "field": field, "stream_id": stream, "score": 12, "confidence": "medium", "baseline_samples": int(known_total), "baseline_scope": "known field values", "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
diff --git a/src/fgai/dashboard.py b/src/fgai/dashboard.py
index bfde00f..aa270c6 100644
--- a/src/fgai/dashboard.py
+++ b/src/fgai/dashboard.py
@@ -179,7 +179,7 @@ async function refresh() {
const streamTitles = Object.fromEntries((configuration.graylog_streams || []).map(item => [item.id, item.title || item.id]));
const fieldRows = Object.entries(data.field_deviations || {}).flatMap(([entity, deviations]) => (deviations || []).map(item => ({entity, stream_title: streamTitles[item.stream_id] || item.stream_id, ...item})));
document.getElementById('fieldDeviations').innerHTML = table(fieldRows, [
- {label:'Entity', key:'entity'}, {label:'Stream', key:'stream_title'}, {label:'Field', key:'field'}, {label:'Score', key:'score'}, {label:'Review', render:r => esc(r.feedback || 'unreviewed')}, {label:'Evidence', render:r => { const summary=esc(`${r.reason}; current ${r.current ?? '-'} vs baseline ${r.baseline ?? '-'}; values: ${(r.sample_values || []).join(', ') || '-'}`); const events=(r.sample_events || []).map(item => esc(`${item.timestamp} | ${item.source} -> ${item.destination} | ${item.action} ${item.service} | ${item.value} | ${item.message}`)).join(' ${events}
'); const id=`deviation:${r.entity}:${r.stream_id}:${r.field}:${r.value || ''}`; return events ? `${summary}
${events}