diff --git a/README.md b/README.md index b1489c0..0a05b7e 100644 --- a/README.md +++ b/README.md @@ -237,6 +237,29 @@ The continuous monitor also stores a local SQLite behavior baseline at five-minute windows. Historical rate and hitcount-rate deviations then contribute to its anomaly score. Set `FGAI_BASELINE_DB` to use another location. +SignalScope prunes old baseline buckets during each monitor cycle. The defaults +keep 14 days of buckets and dedupe history, prune stale one-off categorical +values after 7 days, and cap high-cardinality values per stream/entity/field. +Tune these in the dashboard or in `state/fgai-config.json`: + +```json +{ + "baseline_retention_days": 14, + "baseline_value_retention_days": 7, + "baseline_max_values_per_field": 2000 +} +``` + +If an existing baseline database has already grown large, stop the monitor and +run a manual prune plus SQLite compaction: + +```bash +signalscope baseline-maintenance --baseline-db state/fgai-baseline.sqlite3 --retention-days 14 --value-retention-days 7 --max-values-per-field 2000 --vacuum +``` + +`VACUUM` can take time on a large database and should not be run while the +monitor is actively writing. + Analyze local logs: ```bash diff --git a/ROADMAP.md b/ROADMAP.md index b5bf1c5..051e67f 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -33,6 +33,8 @@ Goal: make findings more accurate before adding more integrations. - [x] Add a dry-run replay command for historic JSONL or Graylog exports using temporary baselines. - [x] Add direct Graylog MCP time-range replay and result comparison against saved detector configurations. - [ ] Add dashboard controls for launching safe replay jobs and viewing detector deltas. +- [ ] Add baseline confidence tooling: per detector learning state, expected false-positive rate, and why a deviation crossed threshold. +- [ ] Add baseline maintenance tooling in the dashboard for retention, high-cardinality fields, and database compaction status. Acceptance: each finding shows its detector, confidence, baseline sample count, current value, expected value, and a bounded set of raw-event references. @@ -46,6 +48,7 @@ Goal: make one incident answer what happened, to whom, and across which sources. - [x] Persist incident state and analyst notes separately from transient detection output. - [x] Add direct Graylog query links or query details for each timeline event. - [x] Add investigation export as JSON and Markdown report. +- [ ] Add guided investigation tools that compare a selected incident against its baseline, related entities, and similar prior outcomes. Acceptance: an analyst can open an incident, see an ordered multi-stream timeline, review evidence, and record an outcome without losing it after the next monitor poll. @@ -79,7 +82,8 @@ Goal: run reliably in a monitored environment. - [ ] Add systemd unit files for monitor, dashboard, and optional local syslog listener. - [ ] Add health and readiness endpoints with last successful Graylog fetch time. -- [ ] Add structured application logs and configurable retention for status/history/baseline data. +- [x] Add configurable retention for baseline buckets and high-cardinality field values. +- [ ] Add structured application logs and configurable retention for status/history data. - [ ] Add backup and migration procedure for SQLite state. - [ ] Add Checkmk local-check output in addition to Prometheus metrics. - [ ] Add authentication/reverse-proxy guidance before exposing the dashboard beyond loopback. diff --git a/src/fgai/baseline.py b/src/fgai/baseline.py index 501139e..c212e93 100644 --- a/src/fgai/baseline.py +++ b/src/fgai/baseline.py @@ -15,6 +15,12 @@ from .detectors import DETECTOR_MINIMUMS, event_detector_categories from .normalization import canonical_value from .query_details import event_query_details +DEFAULT_RETENTION_DAYS = 14 +DEFAULT_VALUE_RETENTION_DAYS = 7 +DEFAULT_MAX_VALUES_PER_FIELD = 2000 +MIN_REPORTED_DEVIATION_SCORE = 15 +MAX_RARE_VALUES_PER_ENTITY = 5 + def _number(value: str | None) -> int: try: @@ -128,6 +134,27 @@ class BaselineStore: ); """ ) + 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) @@ -141,7 +168,7 @@ class BaselineStore: 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: + 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)] @@ -182,7 +209,7 @@ class BaselineStore: 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: + 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: @@ -217,8 +244,8 @@ class BaselineStore: 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)) + 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)) @@ -276,7 +303,8 @@ class BaselineStore: 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") - 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_scope": baseline_scope, "reason": reason, "current": round(current_value, 2), "baseline": round(mean(history), 2), "sample_values": sample_values, "sample_events": evidence_events}) + 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_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(): @@ -305,7 +333,8 @@ class BaselineStore: 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]] - 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_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}) + 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_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) @@ -336,8 +365,10 @@ class BaselineStore: 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]] - 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_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}) + 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_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 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: @@ -355,13 +386,69 @@ class BaselineStore: 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: + 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] - score, weight = _weighted_score(12, profile, field, "rare_value") - evidence = {"detector": "rare_value", "field": field, "stream_id": stream, "score": score, "base_score": 12, "weight": weight, "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": [_sample_event(item, value) for item in samples[:5]]} + 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_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, + ) -> 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") + return {"retention_days": retention_days, "value_retention_days": value_retention_days, "max_values_per_field": max_values_per_field, "vacuum": vacuum, "deleted": deleted, "size_bytes": self.path.stat().st_size if self.path.exists() else 0} + def profile_readiness(self, profiles: dict[str, object]) -> list[dict[str, object]]: rows: list[dict[str, object]] = [] with self._connect() as connection: diff --git a/src/fgai/cli.py b/src/fgai/cli.py index 1204666..e136368 100644 --- a/src/fgai/cli.py +++ b/src/fgai/cli.py @@ -16,6 +16,7 @@ from .syslog_server import listen_udp_syslog from .monitor import monitor_loop from .threat_intel import enrich_ips, is_public_ip from .config import ConfigStore +from .baseline import BaselineStore, DEFAULT_MAX_VALUES_PER_FIELD, DEFAULT_RETENTION_DAYS, DEFAULT_VALUE_RETENTION_DAYS from .exports import investigation_report, investigation_report_markdown from .graylog_mcp import GraylogMcpClient from .graylog_source import GraylogStreamSource @@ -237,6 +238,17 @@ def export_investigation(args: argparse.Namespace) -> int: return 0 +def baseline_maintenance(args: argparse.Namespace) -> int: + result = BaselineStore(args.baseline_db).maintenance( + retention_days=args.retention_days, + value_retention_days=args.value_retention_days, + max_values_per_field=args.max_values_per_field, + vacuum=args.vacuum, + ) + _print_json(result) + return 0 + + def _stream_name(config: dict[str, object], stream_id: str) -> str: return next((str(item.get("title", "")) for item in config.get("graylog_streams", []) if isinstance(item, dict) and item.get("id") == stream_id), stream_id) @@ -430,6 +442,14 @@ def build_parser() -> argparse.ArgumentParser: export.add_argument("--output", default="", help="Write the report to this file instead of stdout") export.set_defaults(func=export_investigation) + maintenance = subparsers.add_parser("baseline-maintenance", help="Prune and optionally compact the local SQLite baseline") + maintenance.add_argument("--baseline-db", default="state/fgai-baseline.sqlite3", help="SQLite baseline database") + maintenance.add_argument("--retention-days", type=int, default=DEFAULT_RETENTION_DAYS, help="Keep bucket and dedupe history for this many days") + maintenance.add_argument("--value-retention-days", type=int, default=DEFAULT_VALUE_RETENTION_DAYS, help="Prune stale one-off categorical values after this many days") + maintenance.add_argument("--max-values-per-field", type=int, default=DEFAULT_MAX_VALUES_PER_FIELD, help="Keep at most this many categorical values per stream/entity/field") + maintenance.add_argument("--vacuum", action="store_true", help="Run SQLite VACUUM after pruning to return disk space") + maintenance.set_defaults(func=baseline_maintenance) + replay = subparsers.add_parser("replay", help="Replay a historical log export against temporary baselines") replay.add_argument("--logs", required=True, help="Historic JSONL or key/value log export") replay.add_argument("--config-file", default="state/fgai-config.json", help="Stream profile configuration") diff --git a/src/fgai/config.py b/src/fgai/config.py index 29c147d..cfb36b8 100644 --- a/src/fgai/config.py +++ b/src/fgai/config.py @@ -13,6 +13,9 @@ DEFAULT_CONFIG: dict[str, object] = { "graylog_stream_profiles": [], "graylog_query": "*", "graylog_range_seconds": 3600, + "baseline_retention_days": 14, + "baseline_value_retention_days": 7, + "baseline_max_values_per_field": 2000, "graylog_field_mapping": "", "llm_enabled": False, "llm_model": "", @@ -49,9 +52,10 @@ class ConfigStore: current[key] = bool(value) elif key == "log_source" and value in {"local_syslog", "graylog_mcp"}: current[key] = value - elif key == "graylog_range_seconds": + elif key in {"graylog_range_seconds", "baseline_retention_days", "baseline_value_retention_days", "baseline_max_values_per_field"}: try: - current[key] = max(60, int(value)) + minimum = 60 if key == "graylog_range_seconds" else 1 + current[key] = max(minimum, int(value)) except (TypeError, ValueError): continue elif key == "graylog_streams" and isinstance(value, list): diff --git a/src/fgai/dashboard.py b/src/fgai/dashboard.py index 251a13e..b1ecc49 100644 --- a/src/fgai/dashboard.py +++ b/src/fgai/dashboard.py @@ -85,7 +85,7 @@ HTML = """

Events and Anomalies

Baseline and Stream Health

Correlation Map

AI Assessment

LLM assessment disabled.

Investigation Incidents

Anomalies

Recommendations

Field Baseline Deviations

Related Activity Across Sources

Block Candidates

Threat Intelligence

Policy Findings

Diagnostics

-

Runtime Configuration

+

Runtime Configuration