from __future__ import annotations import json import os import sqlite3 import threading from datetime import datetime, timedelta, timezone from typing import Any from .adaptive import score_rule from .mitre import classify as classify_mitre, merge as merge_mitre class AlertStore: SCHEMA_VERSION = 12 def __init__(self, path: str) -> None: self.path = path os.makedirs(os.path.dirname(path) or ".", exist_ok=True) self._lock = threading.RLock() self._analytics_lock = threading.RLock() self._conn = sqlite3.connect(path, check_same_thread=False) self._conn.row_factory = sqlite3.Row self._init_schema() # Keep exact archive counters/bounds in memory. A full COUNT(*) over a # multi-million-row traffic archive on every /api/status request was a # major source of periodic CPU spikes. The archive is only mutated by # this process, so incremental metadata stays exact after one startup # scan. self._traffic_archive_meta = self._load_traffic_archive_meta() # WAL permits a dedicated read connection to run long dashboard # aggregates while the archive worker keeps committing new batches. # This avoids coupling Redis draining to a 24h GROUP BY query. self._analytics_conn: sqlite3.Connection | None = None if path != ":memory:": self._analytics_conn = sqlite3.connect(path, check_same_thread=False) self._analytics_conn.row_factory = sqlite3.Row self._analytics_conn.executescript( """ PRAGMA query_only=ON; PRAGMA temp_store=FILE; PRAGMA busy_timeout=5000; """ ) def _init_schema(self) -> None: with self._lock: previous_version = int(self._conn.execute("PRAGMA user_version").fetchone()[0]) self._conn.executescript( """ PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL; PRAGMA foreign_keys=ON; PRAGMA temp_store=FILE; PRAGMA busy_timeout=5000; CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, first_seen TEXT, last_seen TEXT, hit_count INTEGER NOT NULL DEFAULT 1, flow_id TEXT, src_ip TEXT, src_port INTEGER, dest_ip TEXT, dest_port INTEGER, proto TEXT, signature_id INTEGER, signature TEXT, category TEXT, severity INTEGER, action TEXT, blocked INTEGER NOT NULL DEFAULT 0, incident_id INTEGER, risk_score INTEGER NOT NULL DEFAULT 0, block_target TEXT, block_reason TEXT, raw_json TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS traffic_snapshots ( window_seconds INTEGER PRIMARY KEY, generated_at TEXT NOT NULL, payload_json TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS traffic_events ( event_key TEXT PRIMARY KEY, ts_ms INTEGER NOT NULL, event_type TEXT NOT NULL DEFAULT '', proto TEXT NOT NULL DEFAULT '', app_proto TEXT NOT NULL DEFAULT '', direction TEXT NOT NULL DEFAULT '', flow_id TEXT NOT NULL DEFAULT '', src_ip TEXT NOT NULL DEFAULT '', dest_ip TEXT NOT NULL DEFAULT '', flow_bytes INTEGER NOT NULL DEFAULT 0, payload_json TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS traffic_throughput ( sample_key TEXT PRIMARY KEY, ts_ms INTEGER NOT NULL, payload_json TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS web_sessions ( token_hash TEXT PRIMARY KEY, username TEXT NOT NULL, csrf_token TEXT NOT NULL, created_at TEXT NOT NULL, expires_at TEXT NOT NULL, last_seen_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS assets ( ip TEXT PRIMARY KEY, mac TEXT, hostname TEXT, first_seen TEXT NOT NULL, last_seen TEXT NOT NULL, observations INTEGER NOT NULL DEFAULT 0, bytes_total INTEGER NOT NULL DEFAULT 0, alert_count INTEGER NOT NULL DEFAULT 0, incident_count INTEGER NOT NULL DEFAULT 0, risk_score INTEGER NOT NULL DEFAULT 0, last_event_type TEXT, last_app_proto TEXT, identity_source TEXT, protocols_json TEXT NOT NULL DEFAULT '[]', ports_json TEXT NOT NULL DEFAULT '[]', domains_json TEXT NOT NULL DEFAULT '[]', fingerprints_json TEXT NOT NULL DEFAULT '[]' ); CREATE TABLE IF NOT EXISTS asset_baseline ( asset_ip TEXT NOT NULL, kind TEXT NOT NULL, value TEXT NOT NULL, first_seen TEXT NOT NULL, last_seen TEXT NOT NULL, seen_count INTEGER NOT NULL DEFAULT 1, PRIMARY KEY(asset_ip, kind, value) ); CREATE TABLE IF NOT EXISTS threat_iocs ( id INTEGER PRIMARY KEY AUTOINCREMENT, indicator TEXT NOT NULL, indicator_type TEXT NOT NULL, source TEXT NOT NULL DEFAULT 'manual', confidence INTEGER NOT NULL DEFAULT 80, severity INTEGER NOT NULL DEFAULT 1, enabled INTEGER NOT NULL DEFAULT 1, note TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL, expires_at TEXT, last_hit_at TEXT, hit_count INTEGER NOT NULL DEFAULT 0, UNIQUE(indicator_type, indicator) ); CREATE TABLE IF NOT EXISTS ndr_incidents ( id INTEGER PRIMARY KEY AUTOINCREMENT, subject_ip TEXT NOT NULL, opened_at TEXT NOT NULL, last_seen TEXT NOT NULL, title TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'open', risk_score INTEGER NOT NULL DEFAULT 0, severity INTEGER NOT NULL DEFAULT 3, event_count INTEGER NOT NULL DEFAULT 0, alert_count INTEGER NOT NULL DEFAULT 0, ioc_hits INTEGER NOT NULL DEFAULT 0, behavior_hits INTEGER NOT NULL DEFAULT 0, blocked INTEGER NOT NULL DEFAULT 0, block_target TEXT, summary TEXT NOT NULL DEFAULT '', stages_json TEXT NOT NULL DEFAULT '[]', signals_json TEXT NOT NULL DEFAULT '[]', flow_ids_json TEXT NOT NULL DEFAULT '[]', community_ids_json TEXT NOT NULL DEFAULT '[]', destinations_json TEXT NOT NULL DEFAULT '[]', mitre_json TEXT NOT NULL DEFAULT '[]' ); CREATE TABLE IF NOT EXISTS ndr_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, incident_id INTEGER NOT NULL REFERENCES ndr_incidents(id) ON DELETE CASCADE, timestamp TEXT NOT NULL, kind TEXT NOT NULL, stage TEXT NOT NULL DEFAULT '', risk INTEGER NOT NULL DEFAULT 0, summary TEXT NOT NULL, src_ip TEXT, dest_ip TEXT, signature_id INTEGER, flow_id TEXT, community_id TEXT, details_json TEXT NOT NULL DEFAULT '{}', mitre_json TEXT NOT NULL DEFAULT '[]' ); CREATE TABLE IF NOT EXISTS audit_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, username TEXT NOT NULL DEFAULT '', action TEXT NOT NULL, target TEXT NOT NULL DEFAULT '', result TEXT NOT NULL DEFAULT 'ok', remote_ip TEXT NOT NULL DEFAULT '', details_json TEXT NOT NULL DEFAULT '{}' ); """ ) # Existing 0.3.x databases do not have last_seen/hit_count. Add # columns before creating indexes that reference the new schema. self._migrate_columns() self._migrate_traffic_archive_columns() self._conn.executescript( """ CREATE INDEX IF NOT EXISTS idx_alerts_timestamp ON alerts(timestamp DESC); CREATE INDEX IF NOT EXISTS idx_traffic_events_ts ON traffic_events(ts_ms DESC); CREATE INDEX IF NOT EXISTS idx_traffic_events_type_ts ON traffic_events(event_type, ts_ms DESC); CREATE INDEX IF NOT EXISTS idx_traffic_events_proto_ts ON traffic_events(proto, ts_ms DESC); CREATE INDEX IF NOT EXISTS idx_traffic_events_app_ts ON traffic_events(app_proto, ts_ms DESC); CREATE INDEX IF NOT EXISTS idx_traffic_events_direction_ts ON traffic_events(direction, ts_ms DESC); CREATE INDEX IF NOT EXISTS idx_traffic_events_src_ts ON traffic_events(src_ip, ts_ms DESC); CREATE INDEX IF NOT EXISTS idx_traffic_events_dest_ts ON traffic_events(dest_ip, ts_ms DESC); CREATE INDEX IF NOT EXISTS idx_traffic_events_flow_ts ON traffic_events(flow_id, ts_ms DESC); CREATE INDEX IF NOT EXISTS idx_traffic_throughput_ts ON traffic_throughput(ts_ms DESC); CREATE INDEX IF NOT EXISTS idx_alerts_last_seen ON alerts(last_seen DESC); CREATE INDEX IF NOT EXISTS idx_alerts_signature_id ON alerts(signature_id); CREATE INDEX IF NOT EXISTS idx_alerts_blocked ON alerts(blocked); CREATE INDEX IF NOT EXISTS idx_alerts_src_ip ON alerts(src_ip); CREATE INDEX IF NOT EXISTS idx_alerts_dest_ip ON alerts(dest_ip); CREATE INDEX IF NOT EXISTS idx_web_sessions_expires ON web_sessions(expires_at); CREATE INDEX IF NOT EXISTS idx_assets_last_seen ON assets(last_seen DESC); CREATE INDEX IF NOT EXISTS idx_assets_risk ON assets(risk_score DESC); CREATE INDEX IF NOT EXISTS idx_asset_baseline_asset ON asset_baseline(asset_ip, kind); CREATE INDEX IF NOT EXISTS idx_iocs_enabled ON threat_iocs(enabled, indicator_type); CREATE INDEX IF NOT EXISTS idx_iocs_expires ON threat_iocs(expires_at); CREATE INDEX IF NOT EXISTS idx_ndr_incidents_last_seen ON ndr_incidents(last_seen DESC); CREATE INDEX IF NOT EXISTS idx_ndr_incidents_subject ON ndr_incidents(subject_ip, status, last_seen DESC); CREATE INDEX IF NOT EXISTS idx_ndr_events_incident ON ndr_events(incident_id, timestamp DESC); CREATE INDEX IF NOT EXISTS idx_audit_log_timestamp ON audit_log(timestamp DESC); CREATE INDEX IF NOT EXISTS idx_audit_log_action ON audit_log(action, timestamp DESC); """ ) self._conn.execute( "UPDATE alerts SET first_seen=COALESCE(first_seen,timestamp), " "last_seen=COALESCE(last_seen,timestamp), hit_count=COALESCE(hit_count,1)" ) self._normalise_existing_timestamps() self._purge_expired_sessions_locked() if previous_version < self.SCHEMA_VERSION: self._compact_existing_incidents(300) if previous_version < 11: self._backfill_mitre_locked() self._conn.execute(f"PRAGMA user_version={self.SCHEMA_VERSION}") self._conn.commit() def _migrate_traffic_archive_columns(self) -> None: columns = { str(row["name"]) for row in self._conn.execute("PRAGMA table_info(traffic_events)").fetchall() } additions = { "flow_id": "TEXT NOT NULL DEFAULT ''", "src_ip": "TEXT NOT NULL DEFAULT ''", "dest_ip": "TEXT NOT NULL DEFAULT ''", "flow_bytes": "INTEGER NOT NULL DEFAULT 0", } for name, ddl in additions.items(): if name not in columns: self._conn.execute(f"ALTER TABLE traffic_events ADD COLUMN {name} {ddl}") def _load_traffic_archive_meta(self) -> dict[str, int]: with self._lock: events = self._conn.execute( "SELECT COUNT(*) AS count, MIN(ts_ms) AS oldest, MAX(ts_ms) AS newest FROM traffic_events" ).fetchone() throughput = self._conn.execute( "SELECT COUNT(*) AS count, MIN(ts_ms) AS oldest, MAX(ts_ms) AS newest FROM traffic_throughput" ).fetchone() return { "events": int(events["count"] or 0), "throughput_samples": int(throughput["count"] or 0), "oldest_event_ms": int(events["oldest"] or 0), "newest_event_ms": int(events["newest"] or 0), "oldest_throughput_ms": int(throughput["oldest"] or 0), "newest_throughput_ms": int(throughput["newest"] or 0), } def _traffic_meta_insert_locked(self, kind: str, ts_ms: int) -> None: ts_ms = int(ts_ms) if kind == "event": count_key, oldest_key, newest_key = "events", "oldest_event_ms", "newest_event_ms" else: count_key, oldest_key, newest_key = "throughput_samples", "oldest_throughput_ms", "newest_throughput_ms" previous_count = int(self._traffic_archive_meta[count_key]) self._traffic_archive_meta[count_key] = previous_count + 1 if previous_count == 0: self._traffic_archive_meta[oldest_key] = ts_ms self._traffic_archive_meta[newest_key] = ts_ms return self._traffic_archive_meta[oldest_key] = min(int(self._traffic_archive_meta[oldest_key]), ts_ms) self._traffic_archive_meta[newest_key] = max(int(self._traffic_archive_meta[newest_key]), ts_ms) def _oldest_ts_locked(self, table: str) -> int: row = self._conn.execute(f"SELECT ts_ms FROM {table} ORDER BY ts_ms ASC LIMIT 1").fetchone() return int(row[0] or 0) if row is not None else 0 def ping(self) -> bool: try: with self._lock: self._conn.execute("SELECT 1").fetchone() return True except sqlite3.Error: return False def save_traffic_snapshot(self, window_seconds: int, payload: dict[str, Any]) -> None: window_seconds = int(window_seconds) if window_seconds <= 0: raise ValueError("window_seconds must be positive") generated_at = datetime.now(timezone.utc).isoformat() stored = dict(payload) stored["window_seconds"] = window_seconds stored["generated_at"] = generated_at raw = json.dumps(stored, ensure_ascii=False, separators=(",", ":")) with self._lock: self._conn.execute( """ INSERT INTO traffic_snapshots(window_seconds, generated_at, payload_json) VALUES (?, ?, ?) ON CONFLICT(window_seconds) DO UPDATE SET generated_at=excluded.generated_at, payload_json=excluded.payload_json """, (window_seconds, generated_at, raw), ) self._conn.commit() def traffic_snapshot(self, window_seconds: int) -> dict[str, Any] | None: with self._lock: row = self._conn.execute( "SELECT generated_at, payload_json FROM traffic_snapshots WHERE window_seconds=?", (int(window_seconds),), ).fetchone() if row is None: return None try: payload = json.loads(str(row["payload_json"])) except (TypeError, ValueError, json.JSONDecodeError): return None if not isinstance(payload, dict): return None payload.setdefault("generated_at", row["generated_at"]) payload["persisted_snapshot"] = True return payload def traffic_snapshot_status(self) -> dict[str, Any]: with self._lock: rows = self._conn.execute( "SELECT window_seconds, generated_at FROM traffic_snapshots ORDER BY window_seconds" ).fetchall() return { "windows": [ {"window_seconds": int(row["window_seconds"]), "generated_at": row["generated_at"]} for row in rows ] } def clear_traffic_snapshots(self) -> int: with self._lock: count = int(self._conn.execute("SELECT COUNT(*) FROM traffic_snapshots").fetchone()[0]) self._conn.execute("DELETE FROM traffic_snapshots") self._conn.commit() return count def archive_traffic_events(self, records: list[tuple[str, dict[str, Any]]]) -> int: if not records: return 0 inserted = 0 inserted_ts: list[int] = [] with self._lock: for event_key, payload in records: if ( str(payload.get("type") or "").strip().lower() == "alert" and str(payload.get("signature") or "").strip().casefold() in {"suricata ipv4 truncated packet", "suricata ipv6 truncated packet"} ): # These decoder alerts have always been excluded from the # dashboard. Do not spend disk or analytics work on them. continue raw = json.dumps(payload, ensure_ascii=False, separators=(",", ":")) cursor = self._conn.execute( """ INSERT OR IGNORE INTO traffic_events( event_key, ts_ms, event_type, proto, app_proto, direction, flow_id, src_ip, dest_ip, flow_bytes, payload_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( str(event_key), int(payload.get("ts_ms") or 0), str(payload.get("type") or "")[:32].lower(), str(payload.get("proto") or "")[:24].lower(), str(payload.get("app_proto") or "")[:48].lower(), str(payload.get("direction") or "")[:24].lower(), str(payload.get("flow_id") or payload.get("community_id") or "")[:128], str(payload.get("src_ip") or "")[:64], str(payload.get("dest_ip") or "")[:64], max(0, int(payload.get("bytes") or 0)) if str(payload.get("type") or "").lower() == "flow" else 0, raw, ), ) added = max(0, int(cursor.rowcount or 0)) inserted += added if added: inserted_ts.append(int(payload.get("ts_ms") or 0)) self._conn.commit() for ts_ms in inserted_ts: self._traffic_meta_insert_locked("event", ts_ms) return inserted def archive_traffic_throughput(self, records: list[tuple[str, dict[str, Any]]]) -> int: if not records: return 0 inserted = 0 inserted_ts: list[int] = [] with self._lock: for sample_key, payload in records: raw = json.dumps(payload, ensure_ascii=False, separators=(",", ":")) cursor = self._conn.execute( "INSERT OR IGNORE INTO traffic_throughput(sample_key, ts_ms, payload_json) VALUES (?, ?, ?)", (str(sample_key), int(payload.get("ts_ms") or 0), raw), ) added = max(0, int(cursor.rowcount or 0)) inserted += added if added: inserted_ts.append(int(payload.get("ts_ms") or 0)) self._conn.commit() for ts_ms in inserted_ts: self._traffic_meta_insert_locked("throughput", ts_ms) return inserted def traffic_event_page( self, since_ms: int, until_ms: int, *, offset: int = 0, limit: int = 2000, event_type: str = "", proto: str = "", app_proto: str = "", direction: str = "", text: str = "", ) -> list[dict[str, Any]]: clauses = ["ts_ms>=?", "ts_ms<=?"] params: list[Any] = [int(since_ms), int(until_ms)] for column, value in ( ("event_type", event_type), ("proto", proto), ("app_proto", app_proto), ("direction", direction), ): value = str(value or "").strip().lower() if value: clauses.append(f"{column}=?") params.append(value) text = str(text or "").strip().lower() if text: clauses.append("LOWER(payload_json) LIKE ?") params.append(f"%{text}%") params.extend((max(1, min(int(limit), 5000)), max(0, int(offset)))) sql = ( "SELECT event_key, payload_json FROM traffic_events WHERE " + " AND ".join(clauses) + " ORDER BY ts_ms DESC, event_key DESC LIMIT ? OFFSET ?" ) conn = self._analytics_conn or self._conn lock = self._analytics_lock if self._analytics_conn is not None else self._lock with lock: rows = conn.execute(sql, params).fetchall() result: list[dict[str, Any]] = [] for row in rows: try: payload = json.loads(str(row["payload_json"])) except (TypeError, ValueError, json.JSONDecodeError): continue if isinstance(payload, dict): payload["_archive_key"] = str(row["event_key"]) result.append(payload) return result def traffic_throughput_page( self, since_ms: int, until_ms: int, *, offset: int = 0, limit: int = 5000, ) -> list[dict[str, Any]]: conn = self._analytics_conn or self._conn lock = self._analytics_lock if self._analytics_conn is not None else self._lock with lock: rows = conn.execute( """ SELECT sample_key, payload_json FROM traffic_throughput WHERE ts_ms>=? AND ts_ms<=? ORDER BY ts_ms DESC, sample_key DESC LIMIT ? OFFSET ? """, (int(since_ms), int(until_ms), max(1, min(int(limit), 10000)), max(0, int(offset))), ).fetchall() result: list[dict[str, Any]] = [] for row in rows: try: payload = json.loads(str(row["payload_json"])) except (TypeError, ValueError, json.JSONDecodeError): continue if isinstance(payload, dict): payload["_archive_key"] = str(row["sample_key"]) result.append(payload) return result def traffic_dimension_summary(self, since_ms: int, until_ms: int, limit: int = 10) -> dict[str, Any]: """High-cardinality dashboard dimensions calculated inside SQLite. Keeping endpoint/flow DISTINCT work in SQLite prevents the Python worker from retaining millions of flow IDs or remote peers in RAM for 24h views. """ since_ms = int(since_ms) until_ms = int(until_ms) limit = max(1, min(int(limit), 25)) base = (since_ms, until_ms) ignored_apps = ("", "failed", "unknown", "none", "null", "notset") conn = self._analytics_conn or self._conn lock = self._analytics_lock if self._analytics_conn is not None else self._lock def rows(sql: str, params: tuple[Any, ...]) -> list[dict[str, Any]]: with lock: return [dict(row) for row in conn.execute(sql, params).fetchall()] def scalar(sql: str, params: tuple[Any, ...]) -> int: with lock: row = conn.execute(sql, params).fetchone() return int(row[0] or 0) if row is not None else 0 local_cte = """ WITH endpoints AS ( SELECT src_ip AS name, flow_bytes AS bytes FROM traffic_events WHERE ts_ms>=? AND ts_ms<=? AND direction IN ('outbound','internal') AND src_ip<>'' UNION ALL SELECT dest_ip AS name, flow_bytes AS bytes FROM traffic_events WHERE ts_ms>=? AND ts_ms<=? AND direction IN ('inbound','internal') AND dest_ip<>'' AND (direction<>'internal' OR dest_ip<>src_ip) ) """ remote_cte = """ WITH endpoints AS ( SELECT dest_ip AS name, flow_bytes AS bytes FROM traffic_events WHERE ts_ms>=? AND ts_ms<=? AND direction='outbound' AND dest_ip<>'' UNION ALL SELECT src_ip AS name, flow_bytes AS bytes FROM traffic_events WHERE ts_ms>=? AND ts_ms<=? AND direction='inbound' AND src_ip<>'' UNION ALL SELECT src_ip AS name, flow_bytes AS bytes FROM traffic_events WHERE ts_ms>=? AND ts_ms<=? AND direction NOT IN ('outbound','inbound','internal') AND src_ip<>'' UNION ALL SELECT dest_ip AS name, flow_bytes AS bytes FROM traffic_events WHERE ts_ms>=? AND ts_ms<=? AND direction NOT IN ('outbound','inbound','internal') AND dest_ip<>'' AND dest_ip<>src_ip ) """ top_sources = rows( """ SELECT src_ip AS name, COUNT(*) AS count FROM traffic_events WHERE ts_ms>=? AND ts_ms<=? AND src_ip<>'' GROUP BY src_ip ORDER BY count DESC, name LIMIT ? """, base + (limit,), ) top_destinations = rows( """ SELECT dest_ip AS name, COUNT(*) AS count FROM traffic_events WHERE ts_ms>=? AND ts_ms<=? AND dest_ip<>'' GROUP BY dest_ip ORDER BY count DESC, name LIMIT ? """, base + (limit,), ) app_params = base + ignored_apps + (limit,) top_apps = rows( """ SELECT app_proto AS name, COUNT(DISTINCT CASE WHEN flow_id<>'' THEN flow_id ELSE event_key END) AS count FROM traffic_events WHERE ts_ms>=? AND ts_ms<=? AND app_proto NOT IN (?, ?, ?, ?, ?, ?) GROUP BY app_proto ORDER BY count DESC, name LIMIT ? """, app_params, ) top_apps_by_bytes = rows( """ SELECT app_proto AS name, COALESCE(SUM(flow_bytes),0) AS bytes FROM traffic_events WHERE ts_ms>=? AND ts_ms<=? AND app_proto NOT IN (?, ?, ?, ?, ?, ?) GROUP BY app_proto ORDER BY bytes DESC, name LIMIT ? """, app_params, ) local_params = base + base top_local = rows( local_cte + " SELECT name, COUNT(*) AS count FROM endpoints GROUP BY name ORDER BY count DESC, name LIMIT ?", local_params + (limit,), ) local_bytes = rows( local_cte + " SELECT name, COALESCE(SUM(bytes),0) AS bytes FROM endpoints GROUP BY name ORDER BY bytes DESC, name LIMIT ?", local_params + (limit,), ) unique_local = scalar( local_cte + " SELECT COUNT(DISTINCT name) FROM endpoints", local_params, ) remote_params = base + base + base + base top_remote = rows( remote_cte + " SELECT name, COUNT(*) AS count FROM endpoints GROUP BY name ORDER BY count DESC, name LIMIT ?", remote_params + (limit,), ) remote_bytes = rows( remote_cte + " SELECT name, COALESCE(SUM(bytes),0) AS bytes FROM endpoints GROUP BY name ORDER BY bytes DESC, name LIMIT ?", remote_params + (limit,), ) unique_remote = scalar( remote_cte + " SELECT COUNT(DISTINCT name) FROM endpoints", remote_params, ) return { "top_apps": top_apps, "top_apps_by_bytes": top_apps_by_bytes, "top_sources": top_sources, "top_destinations": top_destinations, "top_local_clients": top_local, "top_remote_peers": top_remote, "top_local_clients_by_bytes": local_bytes, "top_remote_peers_by_bytes": remote_bytes, "unique_local_clients": unique_local, "unique_remote_peers": unique_remote, } def traffic_archive_status(self) -> dict[str, Any]: with self._lock: return dict(self._traffic_archive_meta) def purge_traffic_archive_before(self, cutoff_ms: int) -> dict[str, int]: with self._lock: events_cursor = self._conn.execute("DELETE FROM traffic_events WHERE ts_ms dict[str, int]: with self._lock: events = int(self._traffic_archive_meta["events"]) throughput = int(self._traffic_archive_meta["throughput_samples"]) self._conn.execute("DELETE FROM traffic_events") self._conn.execute("DELETE FROM traffic_throughput") self._conn.commit() self._traffic_archive_meta.update({ "events": 0, "throughput_samples": 0, "oldest_event_ms": 0, "newest_event_ms": 0, "oldest_throughput_ms": 0, "newest_throughput_ms": 0, }) return {"events": events, "throughput_samples": throughput} def create_web_session( self, token_hash: str, username: str, csrf_token: str, expires_at: datetime, ) -> None: now = datetime.now(timezone.utc).isoformat() expiry = expires_at.astimezone(timezone.utc).isoformat() with self._lock: self._purge_expired_sessions_locked() self._conn.execute( """ INSERT OR REPLACE INTO web_sessions( token_hash, username, csrf_token, created_at, expires_at, last_seen_at ) VALUES (?, ?, ?, ?, ?, ?) """, (token_hash, username, csrf_token, now, expiry, now), ) self._conn.commit() def get_web_session(self, token_hash: str, *, touch: bool = True) -> dict[str, Any] | None: now = datetime.now(timezone.utc) with self._lock: row = self._conn.execute( """ SELECT token_hash, username, csrf_token, created_at, expires_at, last_seen_at FROM web_sessions WHERE token_hash=? """, (token_hash,), ).fetchone() if row is None: return None expires_at = _parse_timestamp(row["expires_at"]) if expires_at <= now: self._conn.execute("DELETE FROM web_sessions WHERE token_hash=?", (token_hash,)) self._conn.commit() return None if touch: last_seen_at = now.isoformat() self._conn.execute( "UPDATE web_sessions SET last_seen_at=? WHERE token_hash=?", (last_seen_at, token_hash), ) self._conn.commit() else: last_seen_at = str(row["last_seen_at"]) return { "username": str(row["username"]), "csrf_token": str(row["csrf_token"]), "created_at": str(row["created_at"]), "expires_at": expires_at.isoformat(), "last_seen_at": last_seen_at, } def delete_web_session(self, token_hash: str) -> None: with self._lock: self._conn.execute("DELETE FROM web_sessions WHERE token_hash=?", (token_hash,)) self._conn.commit() def purge_expired_sessions(self) -> int: with self._lock: count = self._purge_expired_sessions_locked() self._conn.commit() return count def _purge_expired_sessions_locked(self) -> int: cutoff = datetime.now(timezone.utc).isoformat() cursor = self._conn.execute("DELETE FROM web_sessions WHERE expires_at<=?", (cutoff,)) return int(cursor.rowcount) def _migrate_columns(self) -> None: def add_missing(table: str, additions: dict[str, str]) -> None: columns = { str(row["name"]) for row in self._conn.execute(f"PRAGMA table_info({table})").fetchall() } for name, definition in additions.items(): if name not in columns: self._conn.execute(f"ALTER TABLE {table} ADD COLUMN {name} {definition}") add_missing("alerts", { "first_seen": "TEXT", "last_seen": "TEXT", "hit_count": "INTEGER NOT NULL DEFAULT 1", "incident_id": "INTEGER", "risk_score": "INTEGER NOT NULL DEFAULT 0", }) add_missing("ndr_incidents", { "mitre_json": "TEXT NOT NULL DEFAULT '[]'", }) add_missing("ndr_events", { "mitre_json": "TEXT NOT NULL DEFAULT '[]'", }) def _backfill_mitre_locked(self) -> None: rows = self._conn.execute( "SELECT id,incident_id,stage,summary,src_ip,dest_ip,details_json,mitre_json FROM ndr_events" ).fetchall() incident_map: dict[int, list[dict[str, str]]] = {} for row in rows: current = _json_objects(row["mitre_json"] or "[]") try: details = json.loads(row["details_json"] or "{}") except (TypeError, ValueError, json.JSONDecodeError): details = {} if not isinstance(details, dict): details = {} record = dict(details) record.setdefault("src_ip", row["src_ip"]) record.setdefault("dest_ip", row["dest_ip"]) mapped = merge_mitre(current, classify_mitre(str(row["stage"] or ""), str(row["summary"] or ""), record)) if mapped != current: self._conn.execute( "UPDATE ndr_events SET mitre_json=? WHERE id=?", (json.dumps(mapped, ensure_ascii=False, separators=(",", ":")), int(row["id"])), ) incident_id = int(row["incident_id"]) incident_map[incident_id] = merge_mitre(incident_map.get(incident_id, []), mapped) for incident_id, mapped in incident_map.items(): row = self._conn.execute("SELECT mitre_json FROM ndr_incidents WHERE id=?", (incident_id,)).fetchone() if row is None: continue merged = merge_mitre(_json_objects(row["mitre_json"] or "[]"), mapped) self._conn.execute( "UPDATE ndr_incidents SET mitre_json=? WHERE id=?", (json.dumps(merged, ensure_ascii=False, separators=(",", ":")), incident_id), ) def _normalise_existing_timestamps(self) -> None: rows = self._conn.execute( "SELECT id, timestamp, first_seen, last_seen FROM alerts" ).fetchall() for row in rows: timestamp = _normalise_timestamp(row["timestamp"]) first_seen = _normalise_timestamp(row["first_seen"] or row["timestamp"]) last_seen = _normalise_timestamp(row["last_seen"] or row["timestamp"]) if ( timestamp != row["timestamp"] or first_seen != row["first_seen"] or last_seen != row["last_seen"] ): self._conn.execute( "UPDATE alerts SET timestamp=?, first_seen=?, last_seen=? WHERE id=?", (timestamp, first_seen, last_seen, int(row["id"])), ) def _compact_existing_incidents(self, window_seconds: int) -> int: """Merge legacy duplicate rows created before incident aggregation existed.""" rows = self._conn.execute( """ SELECT id, timestamp, first_seen, last_seen, hit_count, src_ip, dest_ip, dest_port, proto, signature_id, blocked, block_target, block_reason, raw_json FROM alerts ORDER BY signature_id, src_ip, dest_ip, dest_port, proto, COALESCE(first_seen,timestamp), id """ ).fetchall() groups: dict[tuple[Any, ...], list[sqlite3.Row]] = {} for row in rows: key = ( row["signature_id"], row["src_ip"], row["dest_ip"], row["dest_port"], row["proto"], ) groups.setdefault(key, []).append(row) merged_rows = 0 for group_rows in groups.values(): current: list[sqlite3.Row] = [] current_start: datetime | None = None for row in group_rows: row_first = _parse_timestamp(row["first_seen"] or row["timestamp"]) if ( current and current_start is not None and (row_first - current_start).total_seconds() > window_seconds ): merged_rows += self._merge_row_group(current) current = [] current_start = None if current_start is None: current_start = row_first current.append(row) if current: merged_rows += self._merge_row_group(current) return merged_rows def _merge_row_group(self, rows: list[sqlite3.Row]) -> int: if len(rows) < 2: return 0 keep = rows[0] latest = max(rows, key=lambda row: _parse_timestamp(row["last_seen"] or row["timestamp"])) first_seen = min(_parse_timestamp(row["first_seen"] or row["timestamp"]) for row in rows).isoformat() last_seen = max(_parse_timestamp(row["last_seen"] or row["timestamp"]) for row in rows).isoformat() hit_count = sum(max(1, int(row["hit_count"] or 1)) for row in rows) blocked_rows = [row for row in rows if int(row["blocked"] or 0)] block_row = blocked_rows[-1] if blocked_rows else latest self._conn.execute( """ UPDATE alerts SET timestamp=?, first_seen=?, last_seen=?, hit_count=?, blocked=?, block_target=?, block_reason=?, raw_json=? WHERE id=? """, ( last_seen, first_seen, last_seen, hit_count, 1 if blocked_rows else 0, block_row["block_target"], block_row["block_reason"], latest["raw_json"], int(keep["id"]), ), ) ids = [int(row["id"]) for row in rows[1:]] placeholders = ",".join("?" for _ in ids) self._conn.execute(f"DELETE FROM alerts WHERE id IN ({placeholders})", ids) return len(ids) def purge_builtin_test_incidents(self) -> int: with self._lock: # SID 1000001 is reserved by this project for the deterministic # TZSP self-test and should never become a production incident. cursor = self._conn.execute( "DELETE FROM alerts WHERE signature_id=1000001" ) self._conn.commit() return int(cursor.rowcount) def insert_alert( self, event: dict[str, Any], blocked: bool, block_target: str | None, block_reason: str, ) -> int: alert = event.get("alert") or {} timestamp = _normalise_timestamp(event.get("timestamp")) values = ( timestamp, timestamp, timestamp, 1, str(event.get("flow_id") or ""), event.get("src_ip"), event.get("src_port"), event.get("dest_ip"), event.get("dest_port"), event.get("proto"), _as_int(alert.get("signature_id")), alert.get("signature"), alert.get("category"), _as_int(alert.get("severity")), alert.get("action"), 1 if blocked else 0, block_target, block_reason, json.dumps(event, ensure_ascii=False, separators=(",", ":")), ) with self._lock: cursor = self._conn.execute( """ INSERT INTO alerts ( timestamp, first_seen, last_seen, hit_count, flow_id, src_ip, src_port, dest_ip, dest_port, proto, signature_id, signature, category, severity, action, blocked, block_target, block_reason, raw_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, values, ) self._conn.commit() return int(cursor.lastrowid) def find_recent_duplicate(self, event: dict[str, Any], window_seconds: int) -> int | None: if window_seconds <= 0: return None alert = event.get("alert") or {} sid = _as_int(alert.get("signature_id")) if sid is None: return None # Use the event timestamp instead of wall-clock time. EVE timestamps can # arrive with different UTC offsets and may be delayed slightly by log # rotation. Comparing normalized event time keeps aggregation stable. event_time = datetime.fromisoformat( _normalise_timestamp(event.get("timestamp")).replace("Z", "+00:00") ) cutoff = (event_time - timedelta(seconds=window_seconds)).isoformat() upper = (event_time + timedelta(seconds=window_seconds)).isoformat() values = ( sid, event.get("src_ip"), event.get("dest_ip"), event.get("dest_port"), event.get("proto"), cutoff, upper, ) with self._lock: row = self._conn.execute( """ SELECT id FROM alerts WHERE signature_id=? AND src_ip IS ? AND dest_ip IS ? AND dest_port IS ? AND proto IS ? AND COALESCE(first_seen,timestamp) BETWEEN ? AND ? ORDER BY COALESCE(first_seen,timestamp) DESC, id DESC LIMIT 1 """, values, ).fetchone() return int(row["id"]) if row else None def bump_duplicate(self, alert_id: int, event: dict[str, Any]) -> None: timestamp = _normalise_timestamp(event.get("timestamp")) raw = json.dumps(event, ensure_ascii=False, separators=(",", ":")) with self._lock: self._conn.execute( """ UPDATE alerts SET timestamp=MAX(timestamp, ?), first_seen=MIN(COALESCE(first_seen,timestamp), ?), last_seen=MAX(COALESCE(last_seen,timestamp), ?), hit_count=COALESCE(hit_count,1)+1, raw_json=? WHERE id=? """, (timestamp, timestamp, timestamp, raw, int(alert_id)), ) self._conn.commit() def recent(self, limit: int = 100) -> list[dict[str, Any]]: limit = min(max(int(limit), 1), 500) with self._lock: rows = self._conn.execute( """ SELECT id, timestamp, first_seen, last_seen, hit_count, src_ip, src_port, dest_ip, dest_port, proto, signature_id, signature, category, severity, action, blocked, block_target, block_reason, incident_id, risk_score FROM alerts ORDER BY COALESCE(last_seen,timestamp) DESC, id DESC LIMIT ? """, (limit,), ).fetchall() result = [] for row in rows: item = dict(row) item["blocked"] = bool(item["blocked"]) result.append(item) return result def summary(self) -> dict[str, Any]: with self._lock: row = self._conn.execute( """ SELECT COUNT(*) AS incidents, COALESCE(SUM(hit_count),0) AS total_alerts, COALESCE(SUM(CASE WHEN blocked=1 THEN 1 ELSE 0 END),0) AS blocked_alerts, COUNT(DISTINCT signature_id) AS unique_signatures FROM alerts """ ).fetchone() sev = self._conn.execute( """ SELECT severity, COALESCE(SUM(hit_count),0) AS count FROM alerts GROUP BY severity ORDER BY severity """ ).fetchall() return { "total_alerts": int(row["total_alerts"]), "incidents": int(row["incidents"]), "blocked_alerts": int(row["blocked_alerts"]), "unique_signatures": int(row["unique_signatures"]), "by_severity": {str(item["severity"]): int(item["count"]) for item in sev}, } def analytics(self, top_limit: int = 8) -> dict[str, Any]: top_limit = min(max(int(top_limit), 1), 25) now = datetime.now(timezone.utc) cutoff_1h = (now - timedelta(hours=1)).isoformat() cutoff_24h = (now - timedelta(hours=24)).isoformat() with self._lock: windows = self._conn.execute( """ SELECT COALESCE(SUM(CASE WHEN COALESCE(last_seen,timestamp)>=? THEN hit_count ELSE 0 END),0) AS alerts_1h, COALESCE(SUM(CASE WHEN COALESCE(last_seen,timestamp)>=? THEN hit_count ELSE 0 END),0) AS alerts_24h, COUNT(DISTINCT CASE WHEN COALESCE(last_seen,timestamp)>=? THEN src_ip END) AS sources_24h, COUNT(DISTINCT CASE WHEN COALESCE(last_seen,timestamp)>=? THEN signature_id END) AS signatures_24h FROM alerts """, (cutoff_1h, cutoff_24h, cutoff_24h, cutoff_24h), ).fetchone() top_signatures = self._conn.execute( """ SELECT signature_id, signature, severity, COALESCE(SUM(hit_count),0) AS count, MAX(COALESCE(last_seen,timestamp)) AS last_seen FROM alerts WHERE COALESCE(last_seen,timestamp)>=? GROUP BY signature_id, signature, severity ORDER BY count DESC, last_seen DESC LIMIT ? """, (cutoff_24h, top_limit), ).fetchall() top_sources = self._conn.execute( """ SELECT src_ip, COALESCE(SUM(hit_count),0) AS count, MAX(COALESCE(last_seen,timestamp)) AS last_seen FROM alerts WHERE COALESCE(last_seen,timestamp)>=? AND src_ip IS NOT NULL GROUP BY src_ip ORDER BY count DESC, last_seen DESC LIMIT ? """, (cutoff_24h, top_limit), ).fetchall() top_destinations = self._conn.execute( """ SELECT dest_ip, COALESCE(SUM(hit_count),0) AS count, MAX(COALESCE(last_seen,timestamp)) AS last_seen FROM alerts WHERE COALESCE(last_seen,timestamp)>=? AND dest_ip IS NOT NULL GROUP BY dest_ip ORDER BY count DESC, last_seen DESC LIMIT ? """, (cutoff_24h, top_limit), ).fetchall() return { "alerts_1h": int(windows["alerts_1h"]), "alerts_24h": int(windows["alerts_24h"]), "sources_24h": int(windows["sources_24h"]), "signatures_24h": int(windows["signatures_24h"]), "top_signatures": [dict(row) for row in top_signatures], "top_sources": [dict(row) for row in top_sources], "top_destinations": [dict(row) for row in top_destinations], } def database_info(self) -> dict[str, Any]: with self._lock: self._conn.execute("SELECT 1").fetchone() journal_mode = str(self._conn.execute("PRAGMA journal_mode").fetchone()[0]) user_version = int(self._conn.execute("PRAGMA user_version").fetchone()[0]) row_count = int(self._conn.execute("SELECT COUNT(*) FROM alerts").fetchone()[0]) size = _file_size(self.path) wal_size = _file_size(self.path + "-wal") return { "ok": True, "path": self.path, "exists": os.path.exists(self.path), "size_bytes": size, "wal_size_bytes": wal_size, "rows": row_count, "journal_mode": journal_mode, "schema_version": user_version, } def purge_older_than(self, days: int) -> int: if days <= 0: return 0 cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat() with self._lock: cursor = self._conn.execute( "DELETE FROM alerts WHERE COALESCE(last_seen,timestamp) < ?", (cutoff,) ) self._conn.commit() return int(cursor.rowcount) def clear_alerts(self) -> int: with self._lock: count = int(self._conn.execute("SELECT COUNT(*) FROM alerts").fetchone()[0]) self._conn.execute("DELETE FROM alerts") self._conn.commit() return count def observe_asset(self, record: dict[str, Any], *, risk_score: int = 0, incident: bool = False) -> dict[str, Any] | None: ip = _local_subject(record) if not ip: return None now = _normalise_timestamp(record.get("timestamp")) mac = _asset_mac(record, ip) hostname = str(record.get("dhcp_hostname") or "")[:255] app_proto = str(record.get("app_proto") or "")[:48].lower() event_type = str(record.get("type") or "")[:32] bytes_count = max(_as_int(record.get("bytes")) or 0, 0) port = _asset_dest_port(record, ip) domain = str(record.get("dns_query") or record.get("tls_sni") or record.get("quic_sni") or record.get("http_host") or "")[:255].lower().rstrip(".") fingerprints = [str(record.get(k) or "")[:160] for k in ("tls_ja4", "tls_ja3", "quic_ja4", "quic_ja3", "ssh_hassh_client")] fingerprints = [x for x in fingerprints if x] is_alert = 1 if event_type == "alert" else 0 with self._lock: old = self._conn.execute("SELECT * FROM assets WHERE ip=?", (ip,)).fetchone() protocols = _json_set(old["protocols_json"] if old else "[]") ports = _json_set(old["ports_json"] if old else "[]") domains = _json_set(old["domains_json"] if old else "[]") fps = _json_set(old["fingerprints_json"] if old else "[]") if app_proto: protocols.add(app_proto) if port: ports.add(str(port)) if domain: domains.add(domain) fps.update(fingerprints) # Keep bounded identity metadata. Baseline details live in asset_baseline. protocols = set(sorted(protocols)[:64]) ports = set(sorted(ports, key=lambda x: int(x) if x.isdigit() else 65536)[:128]) domains = set(sorted(domains)[-128:]) fps = set(sorted(fps)[-128:]) previous_mac = str(old["mac"] or "") if old else "" new_mac = mac or previous_mac new_hostname = hostname or (str(old["hostname"] or "") if old else "") new_risk = max(int(old["risk_score"] or 0) if old else 0, max(0, min(100, int(risk_score)))) self._conn.execute( """ INSERT INTO assets(ip,mac,hostname,first_seen,last_seen,observations,bytes_total,alert_count,incident_count,risk_score,last_event_type,last_app_proto,identity_source,protocols_json,ports_json,domains_json,fingerprints_json) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(ip) DO UPDATE SET mac=excluded.mac, hostname=excluded.hostname, last_seen=excluded.last_seen, observations=assets.observations+1, bytes_total=assets.bytes_total+excluded.bytes_total, alert_count=assets.alert_count+excluded.alert_count, incident_count=assets.incident_count+excluded.incident_count, risk_score=MAX(assets.risk_score, excluded.risk_score), last_event_type=excluded.last_event_type, last_app_proto=excluded.last_app_proto, identity_source=CASE WHEN excluded.identity_source<>'' THEN excluded.identity_source ELSE assets.identity_source END, protocols_json=excluded.protocols_json, ports_json=excluded.ports_json, domains_json=excluded.domains_json, fingerprints_json=excluded.fingerprints_json """, ( ip,new_mac,new_hostname,now,now,1,bytes_count,is_alert,1 if incident else 0,new_risk,event_type,app_proto, "dhcp" if hostname or record.get("dhcp_client_mac") else "arp" if record.get("arp_src_mac") else "ethernet" if mac else "eve", _json_dump_set(protocols),_json_dump_set(ports),_json_dump_set(domains),_json_dump_set(fps), ), ) self._conn.commit() row = self._conn.execute("SELECT * FROM assets WHERE ip=?", (ip,)).fetchone() result = dict(row) if row else None if result: result["mac_changed"] = bool(previous_mac and mac and previous_mac.lower() != mac.lower()) result["previous_mac"] = previous_mac return result def baseline_touch(self, asset_ip: str, kind: str, value: str, timestamp: str) -> tuple[bool, int]: if not asset_ip or not kind or not value: return False, 0 timestamp = _normalise_timestamp(timestamp) with self._lock: row = self._conn.execute( "SELECT seen_count FROM asset_baseline WHERE asset_ip=? AND kind=? AND value=?", (asset_ip, kind, value), ).fetchone() is_new = row is None if row is None: self._conn.execute( "INSERT INTO asset_baseline(asset_ip,kind,value,first_seen,last_seen,seen_count) VALUES(?,?,?,?,?,1)", (asset_ip, kind, value, timestamp, timestamp), ) count = 1 else: count = int(row["seen_count"] or 0) + 1 self._conn.execute( "UPDATE asset_baseline SET last_seen=?, seen_count=? WHERE asset_ip=? AND kind=? AND value=?", (timestamp, count, asset_ip, kind, value), ) self._conn.commit() return is_new, count def asset_observation_count(self, asset_ip: str) -> int: with self._lock: row = self._conn.execute("SELECT observations FROM assets WHERE ip=?", (asset_ip,)).fetchone() return int(row["observations"] or 0) if row else 0 def assets(self, limit: int = 250) -> list[dict[str, Any]]: limit = min(max(int(limit), 1), 1000) with self._lock: rows = self._conn.execute("SELECT * FROM assets ORDER BY risk_score DESC,last_seen DESC LIMIT ?", (limit,)).fetchall() result = [] for row in rows: item = dict(row) for key in ("protocols_json","ports_json","domains_json","fingerprints_json"): item[key.removesuffix("_json")] = sorted(_json_set(item.pop(key, "[]"))) result.append(item) return result def raise_asset_risk(self, asset_ip: str, risk_score: int) -> None: with self._lock: self._conn.execute( "UPDATE assets SET risk_score=MAX(risk_score,?) WHERE ip=?", (max(0, min(100, int(risk_score))), str(asset_ip)[:64]), ) self._conn.commit() def add_ioc(self, indicator: str, indicator_type: str, *, source: str = "manual", confidence: int = 80, severity: int = 1, note: str = "", expires_at: str | None = None) -> int: indicator_type = str(indicator_type).strip().lower() indicator = _normalise_ioc(indicator, indicator_type) if indicator_type not in {"ip","domain","sha256","ja3","ja4","hassh"}: raise ValueError("unsupported IOC type") if not indicator: raise ValueError("indicator is required") now = datetime.now(timezone.utc).isoformat() with self._lock: self._conn.execute( """ INSERT INTO threat_iocs(indicator,indicator_type,source,confidence,severity,enabled,note,created_at,expires_at) VALUES(?,?,?,?,?,1,?,?,?) ON CONFLICT(indicator_type,indicator) DO UPDATE SET source=excluded.source,confidence=excluded.confidence,severity=excluded.severity,enabled=1,note=excluded.note,expires_at=excluded.expires_at """, (indicator,indicator_type,str(source)[:120],max(0,min(100,int(confidence))),max(1,min(4,int(severity))),str(note)[:500],now,expires_at), ) self._conn.commit() row = self._conn.execute("SELECT id FROM threat_iocs WHERE indicator_type=? AND indicator=?", (indicator_type,indicator)).fetchone() return int(row["id"]) def list_iocs(self, limit: int = 1000, *, enabled_only: bool = False) -> list[dict[str, Any]]: limit = min(max(int(limit), 1), 5000) now = datetime.now(timezone.utc).isoformat() where = "WHERE enabled=1 AND (expires_at IS NULL OR expires_at>?)" if enabled_only else "" params: tuple[Any, ...] = (now, limit) if enabled_only else (limit,) sql = f"SELECT * FROM threat_iocs {where} ORDER BY enabled DESC,severity ASC,confidence DESC,id DESC LIMIT ?" with self._lock: rows = self._conn.execute(sql, params).fetchall() result=[] for row in rows: item=dict(row); item["enabled"]=bool(item["enabled"]); result.append(item) return result def remove_ioc(self, ioc_id: int) -> bool: with self._lock: cur=self._conn.execute("DELETE FROM threat_iocs WHERE id=?", (int(ioc_id),)); self._conn.commit() return bool(cur.rowcount) def mark_ioc_hit(self, ioc_id: int, timestamp: str) -> None: with self._lock: self._conn.execute("UPDATE threat_iocs SET hit_count=hit_count+1,last_hit_at=? WHERE id=?", (_normalise_timestamp(timestamp),int(ioc_id))) self._conn.commit() def correlate_signal(self, signal: dict[str, Any], window_seconds: int = 1800) -> int: subject_ip = str(signal.get("subject_ip") or "")[:64] if not subject_ip: raise ValueError("subject_ip is required") ts = _normalise_timestamp(signal.get("timestamp")) cutoff = (_parse_timestamp(ts) - timedelta(seconds=max(60,int(window_seconds)))).isoformat() risk = max(0,min(100,int(signal.get("risk") or 0))) stage = str(signal.get("stage") or "")[:64] kind = str(signal.get("kind") or "signal")[:64] summary = str(signal.get("summary") or kind)[:500] flow_id = str(signal.get("flow_id") or "")[:64] community_id = str(signal.get("community_id") or "")[:128] dest_ip = str(signal.get("dest_ip") or "")[:64] mitre = [dict(item) for item in (signal.get("mitre") or []) if isinstance(item, dict)] with self._lock: row = self._conn.execute( "SELECT * FROM ndr_incidents WHERE subject_ip=? AND status='open' AND last_seen>=? ORDER BY last_seen DESC,id DESC LIMIT 1", (subject_ip, cutoff), ).fetchone() if row is None: stages=set(); signals=[]; flows=set(); communities=set(); destinations=set(); if stage: stages.add(stage) signals.append(summary) if flow_id: flows.add(flow_id) if community_id: communities.add(community_id) if dest_ip and dest_ip != subject_ip: destinations.add(dest_ip) cursor=self._conn.execute( """INSERT INTO ndr_incidents(subject_ip,opened_at,last_seen,title,risk_score,severity,event_count,alert_count,ioc_hits,behavior_hits,summary,stages_json,signals_json,flow_ids_json,community_ids_json,destinations_json,mitre_json) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (subject_ip,ts,ts,str(signal.get("title") or summary)[:220],risk,_risk_severity(risk),1,1 if kind=="alert" else 0,1 if kind=="ioc" else 0,1 if kind=="behavior" else 0,summary,_json_dump_set(stages),json.dumps(signals,ensure_ascii=False),_json_dump_set(flows),_json_dump_set(communities),_json_dump_set(destinations),json.dumps(mitre,ensure_ascii=False,separators=(",",":"))), ) incident_id=int(cursor.lastrowid) else: incident_id=int(row["id"]) stages=_json_set(row["stages_json"]); signals=_json_list(row["signals_json"]); flows=_json_set(row["flow_ids_json"]); communities=_json_set(row["community_ids_json"]); destinations=_json_set(row["destinations_json"]); mitre=merge_mitre(_json_objects(row["mitre_json"]), mitre) if stage: stages.add(stage) if summary and summary not in signals: signals=(signals+[summary])[-20:] if flow_id: flows.add(flow_id) if community_id: communities.add(community_id) if dest_ip and dest_ip != subject_ip: destinations.add(dest_ip) stage_bonus=10 if len(stages)>=2 else 0 stage_bonus+=10 if len(stages)>=3 else 0 combined=max(int(row["risk_score"] or 0), min(100,risk+stage_bonus)) title=str(row["title"] or signal.get("title") or summary)[:220] if risk >= int(row["risk_score"] or 0): title=str(signal.get("title") or summary)[:220] self._conn.execute( """UPDATE ndr_incidents SET last_seen=?,title=?,risk_score=?,severity=?,event_count=event_count+1,alert_count=alert_count+?,ioc_hits=ioc_hits+?,behavior_hits=behavior_hits+?,summary=?,stages_json=?,signals_json=?,flow_ids_json=?,community_ids_json=?,destinations_json=?,mitre_json=? WHERE id=?""", (ts,title,combined,_risk_severity(combined),1 if kind=="alert" else 0,1 if kind=="ioc" else 0,1 if kind=="behavior" else 0,summary,_json_dump_set(stages),json.dumps(signals,ensure_ascii=False),_json_dump_set(flows),_json_dump_set(communities),_json_dump_set(destinations),json.dumps(mitre,ensure_ascii=False,separators=(",",":")),incident_id), ) self._conn.execute( """INSERT INTO ndr_events(incident_id,timestamp,kind,stage,risk,summary,src_ip,dest_ip,signature_id,flow_id,community_id,details_json,mitre_json) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)""", (incident_id,ts,kind,stage,risk,summary,str(signal.get("src_ip") or "")[:64],dest_ip,_as_int(signal.get("signature_id")),flow_id,community_id,json.dumps(signal.get("details") or {},ensure_ascii=False,separators=(",",":")),json.dumps(mitre,ensure_ascii=False,separators=(",",":"))), ) # Bound evidence rows per incident without losing the incident summary. self._conn.execute("DELETE FROM ndr_events WHERE incident_id=? AND id NOT IN (SELECT id FROM ndr_events WHERE incident_id=? ORDER BY id DESC LIMIT 200)", (incident_id,incident_id)) self._conn.commit() return incident_id def link_alert_incident(self, alert_id: int, incident_id: int, risk_score: int) -> None: with self._lock: self._conn.execute("UPDATE alerts SET incident_id=?,risk_score=MAX(COALESCE(risk_score,0),?) WHERE id=?", (int(incident_id),max(0,min(100,int(risk_score))),int(alert_id))) self._conn.commit() def mark_incident_blocked(self, incident_id: int, target: str) -> None: with self._lock: self._conn.execute("UPDATE ndr_incidents SET blocked=1,block_target=? WHERE id=?", (str(target)[:64],int(incident_id))); self._conn.commit() def set_ndr_incident_status(self, incident_id: int, status: str) -> bool: status = str(status or "").strip().lower() if status not in {"open", "acknowledged", "closed"}: raise ValueError("status must be open, acknowledged or closed") with self._lock: cursor = self._conn.execute( "UPDATE ndr_incidents SET status=? WHERE id=?", (status, int(incident_id)), ) self._conn.commit() return int(cursor.rowcount) > 0 def ndr_incident(self, incident_id: int) -> dict[str, Any] | None: with self._lock: row = self._conn.execute("SELECT * FROM ndr_incidents WHERE id=?", (int(incident_id),)).fetchone() if row is None: return None item = dict(row); item["blocked"] = bool(item["blocked"]) for key in ("stages_json", "signals_json", "flow_ids_json", "community_ids_json", "destinations_json"): out = key.removesuffix("_json"); item[out] = _json_list(item.pop(key, "[]")) item["mitre"] = _json_objects(item.pop("mitre_json", "[]")) return item def recent_ndr_incidents(self, limit: int = 100) -> list[dict[str, Any]]: limit=min(max(int(limit),1),500) with self._lock: rows=self._conn.execute("SELECT * FROM ndr_incidents ORDER BY last_seen DESC,id DESC LIMIT ?", (limit,)).fetchall() result=[] for row in rows: item=dict(row); item["blocked"]=bool(item["blocked"]) for key in ("stages_json","signals_json","flow_ids_json","community_ids_json","destinations_json"): out=key.removesuffix("_json"); item[out]=_json_list(item.pop(key,"[]")) item["mitre"]=_json_objects(item.pop("mitre_json","[]")) result.append(item) return result def ndr_incident_events(self, incident_id: int, limit: int = 100) -> list[dict[str, Any]]: with self._lock: rows=self._conn.execute("SELECT * FROM ndr_events WHERE incident_id=? ORDER BY timestamp DESC,id DESC LIMIT ?", (int(incident_id),min(max(int(limit),1),200))).fetchall() result=[] for row in rows: item=dict(row) try: item["details"]=json.loads(item.pop("details_json") or "{}") except (ValueError,TypeError,json.JSONDecodeError): item["details"]={} item["mitre"]=_json_objects(item.pop("mitre_json","[]")) result.append(item) return result def ndr_summary(self) -> dict[str, Any]: with self._lock: row=self._conn.execute("SELECT COUNT(*) total,SUM(CASE WHEN status='open' THEN 1 ELSE 0 END) open_count,SUM(CASE WHEN risk_score>=80 THEN 1 ELSE 0 END) criticalish,MAX(risk_score) max_risk FROM ndr_incidents").fetchone() assets=self._conn.execute("SELECT COUNT(*) total,SUM(CASE WHEN risk_score>=60 THEN 1 ELSE 0 END) risky FROM assets").fetchone() iocs=self._conn.execute("SELECT COUNT(*) total,SUM(CASE WHEN enabled=1 THEN 1 ELSE 0 END) enabled,SUM(hit_count) hits FROM threat_iocs").fetchone() return {"incidents":int(row["total"] or 0),"open_incidents":int(row["open_count"] or 0),"high_risk_incidents":int(row["criticalish"] or 0),"max_risk":int(row["max_risk"] or 0),"assets":int(assets["total"] or 0),"risky_assets":int(assets["risky"] or 0),"iocs":int(iocs["total"] or 0),"enabled_iocs":int(iocs["enabled"] or 0),"ioc_hits":int(iocs["hits"] or 0)} def rule_intelligence(self, hours: int = 24, limit: int = 100) -> dict[str, Any]: hours = min(max(int(hours), 1), 24 * 30) limit = min(max(int(limit), 1), 500) cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat() with self._lock: rows = self._conn.execute( """ SELECT signature_id, MAX(signature) signature, MAX(category) category, MIN(COALESCE(severity,4)) severity, COUNT(*) rows, SUM(COALESCE(hit_count,1)) hits, COUNT(DISTINCT NULLIF(src_ip,'')) unique_src, COUNT(DISTINCT NULLIF(dest_ip,'')) unique_dst, COUNT(DISTINCT incident_id) incidents, SUM(CASE WHEN blocked=1 THEN 1 ELSE 0 END) blocked, MIN(COALESCE(first_seen,timestamp)) first_seen, MAX(COALESCE(last_seen,timestamp)) last_seen FROM alerts WHERE signature_id IS NOT NULL AND COALESCE(last_seen,timestamp)>=? GROUP BY signature_id ORDER BY hits DESC, last_seen DESC LIMIT ? """, (cutoff, limit), ).fetchall() scored = [score_rule(dict(row)) for row in rows] return { "window_hours": hours, "rules": scored, "noisy": sum(1 for row in scored if row["recommendation"] == "limit"), "review": sum(1 for row in scored if row["recommendation"] == "review"), } def audit(self, username: str, action: str, *, target: str = "", result: str = "ok", remote_ip: str = "", details: dict[str, Any] | None = None) -> int: now = datetime.now(timezone.utc).isoformat() payload = json.dumps(details or {}, ensure_ascii=False, separators=(",", ":")) with self._lock: cur = self._conn.execute( "INSERT INTO audit_log(timestamp,username,action,target,result,remote_ip,details_json) VALUES(?,?,?,?,?,?,?)", (now, str(username or "")[:120], str(action or "")[:160], str(target or "")[:300], str(result or "")[:32], str(remote_ip or "")[:64], payload[:12000]), ) self._conn.execute( "DELETE FROM audit_log WHERE id NOT IN (SELECT id FROM audit_log ORDER BY id DESC LIMIT 10000)" ) self._conn.commit() return int(cur.lastrowid) def audit_events(self, limit: int = 200) -> list[dict[str, Any]]: limit = min(max(int(limit), 1), 1000) with self._lock: rows = self._conn.execute( "SELECT * FROM audit_log ORDER BY timestamp DESC,id DESC LIMIT ?", (limit,) ).fetchall() out = [] for row in rows: item = dict(row) try: item["details"] = json.loads(item.pop("details_json") or "{}") except (TypeError, ValueError, json.JSONDecodeError): item["details"] = {} out.append(item) return out def vacuum(self) -> None: with self._lock: self._conn.execute("VACUUM") def close(self) -> None: if self._analytics_conn is not None: with self._analytics_lock: self._analytics_conn.close() self._analytics_conn = None with self._lock: self._conn.close() def _parse_timestamp(value: Any) -> datetime: text = _normalise_timestamp(value) return datetime.fromisoformat(text.replace("Z", "+00:00")) def _normalise_timestamp(value: Any) -> str: if value not in (None, ""): text = str(value).strip() try: parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return parsed.astimezone(timezone.utc).isoformat() except ValueError: pass return datetime.now(timezone.utc).isoformat() def _json_set(raw: Any) -> set[str]: try: value = json.loads(str(raw or "[]")) except (ValueError, TypeError, json.JSONDecodeError): return set() return {str(x) for x in value if str(x)} if isinstance(value, list) else set() def _json_list(raw: Any) -> list[str]: try: value = json.loads(str(raw or "[]")) except (ValueError, TypeError, json.JSONDecodeError): return [] return [str(x) for x in value if str(x)] if isinstance(value, list) else [] def _json_dump_set(values: set[str]) -> str: return json.dumps(sorted(values), ensure_ascii=False, separators=(",", ":")) def _json_objects(raw: Any) -> list[dict[str, str]]: try: value = json.loads(str(raw or "[]")) except (ValueError, TypeError, json.JSONDecodeError): return [] if not isinstance(value, list): return [] out: list[dict[str, str]] = [] for item in value: if isinstance(item, dict): out.append({str(k): str(v) for k, v in item.items() if v not in (None, "")}) return out def _local_subject(record: dict[str, Any]) -> str: direction = str(record.get("direction") or "") if direction in {"outbound", "internal"}: return str(record.get("src_ip") or record.get("dhcp_assigned_ip") or record.get("arp_src_ip") or "")[:64] if direction == "inbound": return str(record.get("dest_ip") or "")[:64] return str(record.get("dhcp_assigned_ip") or record.get("arp_src_ip") or "")[:64] def _asset_mac(record: dict[str, Any], ip: str) -> str: if str(record.get("dhcp_assigned_ip") or "") == ip: return str(record.get("dhcp_client_mac") or "")[:32] if str(record.get("arp_src_ip") or "") == ip: return str(record.get("arp_src_mac") or "")[:32] if str(record.get("src_ip") or "") == ip: return str(record.get("ether_src") or "")[:32] if str(record.get("dest_ip") or "") == ip: return str(record.get("ether_dest") or "")[:32] return "" def _asset_dest_port(record: dict[str, Any], ip: str) -> int | None: if str(record.get("src_ip") or "") == ip: return _as_int(record.get("dest_port")) return None def _risk_severity(risk: int) -> int: if risk >= 80: return 1 if risk >= 55: return 2 if risk >= 30: return 3 return 4 def _normalise_ioc(indicator: str, indicator_type: str) -> str: value = str(indicator or "").strip() if indicator_type == "ip": try: return str(__import__("ipaddress").ip_address(value)) except ValueError: raise ValueError("invalid IP IOC") if indicator_type == "domain": value = value.lower().rstrip(".") if value.startswith("*."): value = value[2:] if not value or "." not in value or any(ch.isspace() for ch in value): raise ValueError("invalid domain IOC") return value if indicator_type == "sha256": value=value.lower() if len(value)!=64 or any(c not in "0123456789abcdef" for c in value): raise ValueError("invalid SHA256 IOC") return value if indicator_type in {"ja3", "hassh"}: value = value.lower() if len(value) != 32 or any(c not in "0123456789abcdef" for c in value): raise ValueError(f"invalid {indicator_type.upper()} IOC") return value if indicator_type == "ja4": value = value.lower() if len(value) < 20 or len(value) > 96 or any(c.isspace() for c in value): raise ValueError("invalid JA4 IOC") return value return value.lower() def _file_size(path: str) -> int: try: return int(os.path.getsize(path)) except OSError: return 0 def _as_int(value: Any) -> int | None: if value is None or value == "": return None try: return int(value) except (TypeError, ValueError): return None