From cc3c446c8ee007ed3378347a50594319c6299870 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mateusz=20Gruszczy=C5=84ski?= Date: Mon, 17 Aug 2026 10:07:19 +0200 Subject: [PATCH] v.0.11.4 --- Dockerfile | 10 +-- README.md | 19 +++- VERSION | 2 +- app/analytics_cache.py | 46 +++++++--- app/config.py | 11 ++- app/dev_web.py | 1 + app/eve.py | 7 ++ app/live.py | 32 +++++-- app/main.py | 86 ++++++++++++++++-- app/ndr.py | 99 +++++++++++++++++++- app/rules.py | 32 ++++++- app/static/js/app.js | 4 +- app/store.py | 120 ++++++++++++++++++++----- app/webui.py | 57 +++++++++++- deploy-routeros.env.example | 4 +- docker-compose.yml | 2 +- docs/routeros-tzsp-capture.md | 6 +- routeros/04-container-import-amd64.rsc | 8 +- routeros/04-container-import-arm.rsc | 8 +- routeros/04-container-import-arm64.rsc | 8 +- routeros/app-template.yml | 9 +- rust/tzsp-receiver/Cargo.toml | 2 +- scripts/deploy-routeros.sh | 4 +- scripts/healthcheck.py | 2 +- tests/test_analytics_cache.py | 10 +++ tests/test_config.py | 10 ++- tests/test_live.py | 15 ++++ tests/test_metrics.py | 68 ++++++++++++++ tests/test_ndr.py | 50 +++++++++++ 29 files changed, 627 insertions(+), 105 deletions(-) diff --git a/Dockerfile b/Dockerfile index 41ba9cf..c12c7ad 100644 --- a/Dockerfile +++ b/Dockerfile @@ -110,7 +110,7 @@ ENV PYTHONUNBUFFERED=1 \ METRICS_BASIC_AUTH_PASSWORD= \ SESSION_HOURS=168 \ SESSION_COOKIE_SECURE=false \ - ANALYTICS_SNAPSHOT_INTERVAL_SECONDS=60 \ + ANALYTICS_SNAPSHOT_INTERVAL_SECONDS=120 \ AUTO_BLOCK=false \ UPDATE_RULES_ON_START=false \ RULE_UPDATE_INTERVAL_HOURS=24 \ @@ -118,11 +118,11 @@ ENV PYTHONUNBUFFERED=1 \ REDIS_MANAGED=true \ REDIS_DATA_DIR=/data/redis \ REDIS_PORT=6379 \ - REDIS_MAXMEMORY_MB=0 \ - REDIS_SNAPSHOT_SECONDS=1800 \ - REDIS_AOF=true \ + REDIS_MAXMEMORY_MB=128 \ + REDIS_SNAPSHOT_SECONDS=0 \ + REDIS_AOF=false \ TRAFFIC_RETENTION_HOURS=24 \ - TRAFFIC_MAX_EVENTS=0 \ + TRAFFIC_MAX_EVENTS=50000 \ TRAFFIC_MEMORY_EVENTS=0 \ WEBSOCKET_QUEUE_SIZE=512 \ LIVE_FLOW_UPDATE_SECONDS=2.0 \ diff --git a/README.md b/README.md index 5d55385..340074c 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,17 @@ # MikroSuricata -Project version: `0.11.3` +Project version: `0.11.4` + + +## What changed in 0.11.4 + +- Bounded NDR beacon/cooldown/runtime state and added a bounded baseline LRU so long-running sensors no longer retain one Python object per remote peer indefinitely. +- Bounded high-cardinality analytics counters used while scanning SQLite history, preventing large 24h snapshots from permanently inflating Python RSS. +- Dashboard snapshots now continue from the committed SQLite prefix while the Redis archive has backlog; partial snapshots are marked incomplete instead of leaving charts frozen for hours. +- EVE watcher now recovers from downstream processing exceptions instead of silently terminating while the Rust TZSP process remains alive. +- `TRAFFIC_MAX_EVENTS` is now honored. New deployments use a 50,000-event Redis queue cap and trim it every second in addition to the 128 MiB Redis memory cap. +- `/api/status` now exposes per-process RSS for Python, Rust, Suricata and Redis plus cgroup memory usage to make future memory regressions attributable to a specific component. +- Removed the obsolete VLAN100 sniffer template that conflicted with the hybrid capture deployment tests. ## What changed in 0.11.3 @@ -10,7 +21,7 @@ Project version: `0.11.3` - Added an exact **5 hour (`18000` seconds)** dashboard range. Requested ranges are no longer rounded to the nearest predefined analytics window. - Added bounded-memory analytics for long windows: endpoint/application cardinality is aggregated inside SQLite and a separate snapshot worker processes one requested window at a time instead of retaining large Python flow/IP sets. Archiving continues while a long 24h snapshot is being calculated. - Managed Redis now defaults to a **128 MiB hard memory limit**, `noeviction`, no RDB snapshots and no AOF. The defaults prevent Redis from growing until the RouterOS container is killed by OOM; durable traffic history lives in `/data/ids.db`. -- Added `TRAFFIC_ARCHIVE_INTERVAL_SECONDS`, `TRAFFIC_ARCHIVE_LAG_SECONDS` and `TRAFFIC_ARCHIVE_BATCH_SIZE` controls. Defaults move committed data out of Redis every 5 seconds with a 10 second live-buffer lag. +- Added `TRAFFIC_ARCHIVE_INTERVAL_SECONDS`, `TRAFFIC_ARCHIVE_LAG_SECONDS` and `TRAFFIC_ARCHIVE_BATCH_SIZE` controls. Defaults move committed data out of Redis every 10 seconds with a 10 second live-buffer lag. - Upgrades with a large legacy Redis backlog drain it in bounded batches before publishing fresh SQLite snapshots, preventing partially migrated history from appearing in charts. @@ -130,7 +141,7 @@ NOTIFY_MIN_RISK=80 REDIS_MAXMEMORY_MB=128 REDIS_SNAPSHOT_SECONDS=0 REDIS_AOF=false -TRAFFIC_ARCHIVE_INTERVAL_SECONDS=5 +TRAFFIC_ARCHIVE_INTERVAL_SECONDS=10 TRAFFIC_ARCHIVE_LAG_SECONDS=10 TRAFFIC_ARCHIVE_BATCH_SIZE=1000 ``` @@ -247,7 +258,7 @@ ALERT_DEDUP_WINDOW_SECONDS=300 ADMIN_USERNAME=admin ADMIN_PASSWORD= SESSION_HOURS=168 -ANALYTICS_SNAPSHOT_INTERVAL_SECONDS=60 +ANALYTICS_SNAPSHOT_INTERVAL_SECONDS=120 AUTO_BLOCK=false ``` diff --git a/VERSION b/VERSION index 1a96df1..35ad344 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.11.3 +0.11.4 diff --git a/app/analytics_cache.py b/app/analytics_cache.py index 5a4de1d..3f0ba8e 100644 --- a/app/analytics_cache.py +++ b/app/analytics_cache.py @@ -55,11 +55,12 @@ class AnalyticsSnapshotCache: self._archive_wake = threading.Event() self._snapshot_wake = threading.Event() self._archive_ready = threading.Event() + self._archive_started = threading.Event() self._stopping = threading.Event() self._lock = threading.RLock() now = time.monotonic() # Standard UI ranges are kept warm. Arbitrary API ranges are added on demand. - self._requested_at: dict[int, float] = {window: now for window in SUMMARY_WINDOWS} + self._requested_at: dict[int, float] = {} self._last_refresh: dict[int, float] = {} self._errors = 0 self._refreshes = 0 @@ -68,7 +69,7 @@ class AnalyticsSnapshotCache: self._archive_moved_throughput = 0 self._last_archive_run = 0.0 self._last_purge = 0.0 - self._active_ttl_seconds = max(300, self.interval_seconds * 10) + self._active_ttl_seconds = max(120, self.interval_seconds * 3) def start(self) -> None: self._stopping.clear() @@ -148,8 +149,12 @@ class AnalyticsSnapshotCache: return try: snapshots = self.history.analytics_many(normalized) + archive_backlog = not self._archive_ready.is_set() now_wall = time.time() for window in normalized: + snapshots[window]["archive_backlog"] = archive_backlog + if archive_backlog: + snapshots[window]["analytics_complete"] = False self.store.save_traffic_snapshot(window, snapshots[window]) with self._lock: self._last_refresh[window] = now_wall @@ -179,6 +184,8 @@ class AnalyticsSnapshotCache: "backend": "sqlite-archive", "archive_worker_running": self._archive_thread.is_alive(), "snapshot_worker_running": self._snapshot_thread.is_alive(), + "archive_started": self._archive_started.is_set(), + "archive_backlog": not self._archive_ready.is_set(), "interval_seconds": self.interval_seconds, "archive_interval_seconds": self.archive_interval_seconds, "archive_lag_seconds": self.archive_lag_seconds, @@ -201,14 +208,15 @@ class AnalyticsSnapshotCache: if self.stop_event.is_set() or self._stopping.is_set(): break backlog = self._archive_once() + self._archive_started.set() if backlog: self._archive_ready.clear() elif not self._archive_ready.is_set(): self._archive_ready.set() - self._snapshot_wake.set() + self._snapshot_wake.set() now = time.monotonic() - if now - self._last_purge >= 60: + if now - self._last_purge >= 900: try: self.history.purge_archive() except Exception: @@ -218,8 +226,10 @@ class AnalyticsSnapshotCache: if backlog: # Continue draining old installations immediately instead of - # sleeping while Redis still holds a large backlog. Do not - # publish a partial SQLite snapshot while migration is pending. + # sleeping while Redis still holds a large backlog. Yield a + # little CPU so EVE/NDR/web threads remain responsive. + if self.stop_event.wait(0.02): + break self._archive_wake.set() def _run_snapshots(self) -> None: @@ -229,14 +239,19 @@ class AnalyticsSnapshotCache: self._snapshot_wake.clear() if self.stop_event.is_set() or self._stopping.is_set(): break - # On startup/upgrades wait until the bounded archive worker has - # drained any legacy Redis backlog. Existing persisted snapshots - # remain available to the UI while this happens. - if not self._archive_ready.is_set(): + # Do not race the very first archive pass. After that, snapshots may + # use the committed SQLite prefix even while Redis still has a + # backlog. This keeps historical charts moving instead of freezing + # for hours; such snapshots are explicitly marked incomplete. + if not self._archive_started.is_set(): continue - due = self._due_windows() - if due: - self.refresh_windows(due) + try: + due = self._due_windows() + if due: + self.refresh_windows(due) + except Exception: + with self._lock: + self._errors += 1 def _archive_once(self) -> bool: cutoff_ms = int((time.time() - self.archive_lag_seconds) * 1000) @@ -268,6 +283,11 @@ class AnalyticsSnapshotCache: with self._lock: requests = dict(self._requested_at) last_refresh = dict(self._last_refresh) + # When nobody requested analytics, do not touch SQLite just to keep + # dashboard ranges warm. Current throughput is delivered independently + # from the Rust telemetry path, so background chart work can stay idle. + if not requests: + return [] persisted = { int(row["window_seconds"]): row.get("generated_at") for row in self.store.traffic_snapshot_status().get("windows", []) diff --git a/app/config.py b/app/config.py index 8b5f05e..e5b03c1 100644 --- a/app/config.py +++ b/app/config.py @@ -195,9 +195,9 @@ class Config: session_hours=max(1, _int("SESSION_HOURS", 168)), session_cookie_secure=_bool("SESSION_COOKIE_SECURE", False), analytics_snapshot_interval_seconds=max( - 15, _int("ANALYTICS_SNAPSHOT_INTERVAL_SECONDS", 60) + 15, _int("ANALYTICS_SNAPSHOT_INTERVAL_SECONDS", 120) ), - traffic_archive_interval_seconds=max(1, _int("TRAFFIC_ARCHIVE_INTERVAL_SECONDS", 5)), + traffic_archive_interval_seconds=max(1, _int("TRAFFIC_ARCHIVE_INTERVAL_SECONDS", 10)), traffic_archive_lag_seconds=max(2, _int("TRAFFIC_ARCHIVE_LAG_SECONDS", 10)), traffic_archive_batch_size=max(100, min(5000, _int("TRAFFIC_ARCHIVE_BATCH_SIZE", 1000))), redis_url=os.getenv("REDIS_URL", "redis://127.0.0.1:6379/0"), @@ -212,8 +212,11 @@ class Config: redis_snapshot_seconds=max(0, _int("REDIS_SNAPSHOT_SECONDS", 0)), redis_aof=_bool("REDIS_AOF", False), traffic_retention_hours=_int("TRAFFIC_RETENTION_HOURS", 24), - traffic_max_events=0, - traffic_memory_events=0, + # Redis is only a transient queue. Bound the number of normalized + # EVE records as well as bytes so a stalled SQLite archive cannot + # accumulate an arbitrarily large sorted set. + traffic_max_events=max(0, _int("TRAFFIC_MAX_EVENTS", 50000)), + traffic_memory_events=max(0, _int("TRAFFIC_MEMORY_EVENTS", 0)), websocket_queue_size=_int("WEBSOCKET_QUEUE_SIZE", 512), live_flow_update_seconds=_float("LIVE_FLOW_UPDATE_SECONDS", 2.0), ndr_enabled=_bool("NDR_ENABLED", True), diff --git a/app/dev_web.py b/app/dev_web.py index 7d95512..09040f4 100644 --- a/app/dev_web.py +++ b/app/dev_web.py @@ -270,6 +270,7 @@ def main() -> int: live_pipeline=live_pipeline, analytics_cache=analytics_cache, metrics_provider=prometheus_metrics.render, + healthcheck_provider=lambda: {"status": "ok" if store.ping() else "degraded", "operational": store.ping()}, ) def request_stop(_signum=None, _frame=None) -> None: diff --git a/app/eve.py b/app/eve.py index 7fbeba9..c16aa6a 100644 --- a/app/eve.py +++ b/app/eve.py @@ -59,6 +59,13 @@ class EVEWatcher(threading.Thread): except OSError as exc: print(f"[eve] file error: {exc}", flush=True) time.sleep(1.0) + except Exception as exc: + # A transient downstream failure (SQLite/Redis/NDR/etc.) must + # not permanently kill the EVE watcher while the container and + # Rust data-plane continue to look healthy. + self.stats.inc("eve_runtime_errors") + print(f"[eve] processing error: {exc}", flush=True) + time.sleep(0.5) def _follow_file(self) -> None: with open(self.path, "r", encoding="utf-8", errors="replace") as handle: diff --git a/app/live.py b/app/live.py index 7f23eb7..ff86b43 100644 --- a/app/live.py +++ b/app/live.py @@ -21,6 +21,7 @@ SUPPORTED_EVENT_TYPES = { "ssh", "rdp", "smb", "quic", "dhcp", "arp", "ike", "mqtt", "ftp", "ftp_data", "smtp", "websocket", "nfs", "tftp", "dcerpc", "krb5", "snmp", "rfb", "sip", "ldap", "pop3", } +MAX_ANALYTICS_DIMENSION_KEYS = 4096 def _utc_now() -> str: @@ -52,6 +53,20 @@ def _text(value: Any, max_len: int = 512) -> str: return str(value)[:max_len] +def _bounded_counter_add( + counter: collections.Counter[str], + key: str, + amount: int = 1, + *, + max_keys: int = MAX_ANALYTICS_DIMENSION_KEYS, +) -> None: + """Update a dashboard Counter without retaining unbounded unique values.""" + if not key: + return + if key in counter or len(counter) < max_keys: + counter[key] += amount + + def _parse_networks(value: str) -> list[ipaddress._BaseNetwork]: result: list[ipaddress._BaseNetwork] = [] for raw in (value or "").split(","): @@ -1215,7 +1230,8 @@ class TrafficHistory: def _trim_if_due(self, redis: RedisConnection) -> None: now = time.monotonic() - if now - self._last_trim > 15: + interval = 1.0 if self.max_events > 0 else 15.0 + if now - self._last_trim > interval: self._trim(redis) self._last_trim = now @@ -1437,7 +1453,7 @@ class _AnalyticsAccumulator: self.alerts += 1 signature = _text(item.get("signature"), 160) if signature: - self.signatures[signature] += 1 + _bounded_counter_add(self.signatures, signature) severity = item.get("severity") if severity not in (None, ""): self.severities[f"S{severity}"] += 1 @@ -1451,26 +1467,26 @@ class _AnalyticsAccumulator: self.files += 1 filename = _text(item.get("filename"), 180) or "unnamed file" digest = _text(item.get("file_sha256") or item.get("file_sha1") or item.get("file_md5"), 32) - self.file_activity[f"{filename}{' · ' + digest if digest else ''}"] += 1 + _bounded_counter_add(self.file_activity, f"{filename}{' · ' + digest if digest else ''}") direction = _text(item.get("direction"), 24) or "unknown" src_ip = _text(item.get("src_ip"), 64) dest_ip = _text(item.get("dest_ip"), 64) ether_src = _text(item.get("ether_src"), 32) ether_dest = _text(item.get("ether_dest"), 32) if direction in {"outbound", "internal"} and src_ip and ether_src: - self.assets[f"{src_ip} · {ether_src}"] += 1 + _bounded_counter_add(self.assets, f"{src_ip} · {ether_src}") if direction in {"inbound", "internal"} and dest_ip and ether_dest: - self.assets[f"{dest_ip} · {ether_dest}"] += 1 + _bounded_counter_add(self.assets, f"{dest_ip} · {ether_dest}") if item.get("type") == "dhcp": asset_ip = _text(item.get("dhcp_assigned_ip") or item.get("src_ip"), 64) identity = _text(item.get("dhcp_hostname") or item.get("dhcp_client_mac"), 160) if asset_ip or identity: - self.assets[f"{asset_ip}{' · ' if asset_ip and identity else ''}{identity}"] += 1 + _bounded_counter_add(self.assets, f"{asset_ip}{' · ' if asset_ip and identity else ''}{identity}") elif item.get("type") == "arp": asset_ip = _text(item.get("arp_src_ip") or item.get("src_ip"), 64) mac = _text(item.get("arp_src_mac"), 32) if asset_ip or mac: - self.assets[f"{asset_ip}{' · ' if asset_ip and mac else ''}{mac}"] += 1 + _bounded_counter_add(self.assets, f"{asset_ip}{' · ' if asset_ip and mac else ''}{mac}") app_proto = _valid_app_proto(item.get("app_proto")) if item.get("type") in {"tls", "quic", "ssh"} or app_proto in {"tls", "quic", "ssh"}: self.encrypted += 1 @@ -1484,7 +1500,7 @@ class _AnalyticsAccumulator: ): value = _text(item.get(key), 160) if value: - self.fingerprints[f"{label} {value}"] += 1 + _bounded_counter_add(self.fingerprints, f"{label} {value}") if item.get("type") in {"http", "ftp", "smtp"} or app_proto in {"http", "ftp", "smtp", "telnet"}: self.cleartext += 1 if is_flow: diff --git a/app/main.py b/app/main.py index 6f67eb0..20279e3 100644 --- a/app/main.py +++ b/app/main.py @@ -34,6 +34,40 @@ from .tzsp_rust import RustTZSPReceiver from .webui import WebServer +def _process_rss_mib(pid: int | None) -> float | None: + if not pid: + return None + try: + for line in Path(f"/proc/{int(pid)}/status").read_text(encoding="ascii", errors="replace").splitlines(): + if line.startswith("VmRSS:"): + parts = line.split() + return round(int(parts[1]) / 1024.0, 1) + except (OSError, ValueError, IndexError): + pass + return None + + +def _cgroup_memory_mib() -> dict[str, float | None]: + def read_number(path: str) -> int | None: + try: + raw = Path(path).read_text(encoding="ascii").strip() + if not raw or raw == "max": + return None + return int(raw) + except (OSError, ValueError): + return None + + current = read_number("/sys/fs/cgroup/memory.current") + limit = read_number("/sys/fs/cgroup/memory.max") + if current is None: + current = read_number("/sys/fs/cgroup/memory/memory.usage_in_bytes") + limit = read_number("/sys/fs/cgroup/memory/memory.limit_in_bytes") + return { + "current_mib": round(current / (1024 * 1024), 1) if current is not None else None, + "limit_mib": round(limit / (1024 * 1024), 1) if limit is not None else None, + } + + def _routeros_target(cfg: Config) -> tuple[str, int]: parsed = urlparse(cfg.routeros_url) host = parsed.hostname or cfg.routeros_url @@ -240,8 +274,8 @@ def main() -> int: traffic_history = TrafficHistory( cfg.redis_url, cfg.traffic_retention_hours, - 0, - 0, + cfg.traffic_max_events, + cfg.traffic_memory_events, require_redis=True, allow_memory_fallback=False, archive_store=store, @@ -282,6 +316,26 @@ def main() -> int: ) routeros_host, routeros_port = _routeros_target(cfg) + def healthcheck() -> dict: + # Docker calls this frequently. Keep it strictly O(1): no rule-file + # scans, archive counts, directory walks or dashboard aggregates. + suricata_up = suricata.poll() is None + tzsp_up = receiver.is_alive() + tap_up = os.path.exists(f"/sys/class/net/{cfg.tap_name}") + eve_up = watcher.is_alive() + db_up = store.ping() + routeros_required_ok = (not cfg.auto_block) or routeros.configured + operational = suricata_up and tzsp_up and tap_up and eve_up and db_up and routeros_required_ok + return { + "status": "ok" if operational else "degraded", + "operational": operational, + "suricata_running": suricata_up, + "tzsp_running": tzsp_up, + "tap_up": tap_up, + "eve_running": eve_up, + "database_ok": db_up, + } + def health() -> dict: suricata_up = suricata.poll() is None tzsp_up = receiver.is_alive() @@ -294,6 +348,10 @@ def main() -> int: runtime = stats.snapshot() receiver_status = receiver.status() redis_status = redis_supervisor.status() + history_status = traffic_history.status() + ndr_status = ndr_analyzer.status() + ndr_summary = store.ndr_summary() + notifier_status = notifier.status() suri_stats = runtime.get("suricata") or {} kernel_packets = int(suri_stats.get("capture.kernel_packets", 0) or 0) kernel_drops = int(suri_stats.get("capture.kernel_drops", 0) or 0) @@ -316,6 +374,13 @@ def main() -> int: core_up = suricata_up and tzsp_up and tap_up and eve_up and db["ok"] routeros_required_ok = (not cfg.auto_block) or routeros.configured operational = core_up and routeros_required_ok + process_memory = { + "python_rss_mib": _process_rss_mib(os.getpid()), + "rust_rss_mib": _process_rss_mib(receiver.pid), + "suricata_rss_mib": _process_rss_mib(suricata.pid if suricata_up else None), + "redis_rss_mib": _process_rss_mib(redis_status.get("pid")), + "cgroup": _cgroup_memory_mib(), + } return { "status": "ok" if operational else "degraded", @@ -328,11 +393,13 @@ def main() -> int: "suricata_pid": suricata.pid, "auto_block": cfg.auto_block, "routeros_configured": routeros.configured, - "ndr": {**ndr_analyzer.status(), **store.ndr_summary()}, + "ndr": {**ndr_status, **ndr_summary}, "database": db, "storage": storage, "rules": rules, "redis": redis_status, + "memory": process_memory, + "traffic_history": history_status, "services": { "web": { "name": "Web UI / API", @@ -387,7 +454,7 @@ def main() -> int: }, "traffic_history": { "name": "Live traffic history", - "status": "up" if traffic_history.status().get("redis_ok") else "degraded", + "status": "up" if history_status.get("redis_ok") else "degraded", "details": f"Redis ingest buffer -> SQLite archive; retention={cfg.traffic_retention_hours}h", }, "analytics_cache": { @@ -397,13 +464,13 @@ def main() -> int: }, "ndr": { "name": "MikroSuricata NDR correlation", - "status": "up" if ndr_analyzer.status().get("running") else "disabled" if not cfg.ndr_enabled else "degraded", - "details": f"assets={store.ndr_summary()['assets']}; incidents={store.ndr_summary()['incidents']}; IOC={store.ndr_summary()['enabled_iocs']}; queue={ndr_analyzer.status()['queue']}", + "status": "up" if ndr_status.get("running") else "disabled" if not cfg.ndr_enabled else "degraded", + "details": f"assets={ndr_summary['assets']}; incidents={ndr_summary['incidents']}; IOC={ndr_summary['enabled_iocs']}; queue={ndr_status['queue']}", }, "notifications": { "name": "High-risk webhook notifications", - "status": "up" if notifier.status().get("running") else "disabled" if not notifier.enabled else "degraded", - "details": f"min risk={cfg.notify_min_risk}; sent={notifier.status()['sent']}; failed={notifier.status()['failed']}; queue={notifier.status()['queue']}", + "status": "up" if notifier_status.get("running") else "disabled" if not notifier.enabled else "degraded", + "details": f"min risk={cfg.notify_min_risk}; sent={notifier_status['sent']}; failed={notifier_status['failed']}; queue={notifier_status['queue']}", }, "redis": { "name": "Managed Redis", @@ -455,7 +522,7 @@ def main() -> int: "status": routeros_status, }, ], - "runtime": stats.snapshot(), + "runtime": runtime, } version_path = Path(__file__).resolve().parents[1] / "VERSION" @@ -514,6 +581,7 @@ def main() -> int: forensic_pcap=forensic_pcap, traffic_source=receiver, metrics_provider=prometheus_metrics.render, + healthcheck_provider=healthcheck, ) def housekeeping() -> None: diff --git a/app/ndr.py b/app/ndr.py index 1fd2ea6..336b7d4 100644 --- a/app/ndr.py +++ b/app/ndr.py @@ -18,6 +18,12 @@ from .store import AlertStore SENSITIVE_PORTS = {21, 22, 23, 25, 110, 135, 139, 445, 1433, 3306, 3389, 5432, 6379, 8291, 9200, 27017} +NDR_BEACON_MAX_KEYS = 32768 +NDR_STATE_MAX_SUBJECTS = 8192 +NDR_BASELINE_CACHE_MAX = 50000 +NDR_BLOCK_ATTEMPT_MAX = 10000 +NDR_STATE_IDLE_SECONDS = 7200 +NDR_BLOCK_ATTEMPT_TTL_SECONDS = 86400 LOCAL_STAGE_BY_SID = { 1000101: "credential-access", 1000102: "credential-access", 1000103: "credential-access", 1000104: "recon", 1000105: "recon", 1000106: "initial-access", @@ -211,7 +217,11 @@ class NDRAnalyzer: self._signals = 0 self._ioc_hits = 0 self._behavior_hits = 0 - self._beacon: dict[tuple[str, str], collections.deque[float]] = collections.defaultdict(lambda: collections.deque(maxlen=12)) + # Remote destinations are high-cardinality on real networks (CDNs, + # crawlers, cloud APIs). Keep all runtime correlation state explicitly + # bounded so a long-running sensor cannot retain one object per peer + # forever. + self._beacon: collections.OrderedDict[tuple[str, str], collections.deque[float]] = collections.OrderedDict() self._scan: dict[str, collections.deque[tuple[float, str, int]]] = collections.defaultdict(lambda: collections.deque(maxlen=128)) self._out_scan: dict[str, collections.deque[tuple[float, str, int]]] = collections.defaultdict(lambda: collections.deque(maxlen=128)) self._egress: dict[str, collections.deque[tuple[float, int, str]]] = collections.defaultdict(lambda: collections.deque(maxlen=512)) @@ -220,7 +230,13 @@ class NDRAnalyzer: self._dns_tunnel: dict[str, collections.deque[tuple[float, str, int]]] = collections.defaultdict(lambda: collections.deque(maxlen=96)) self._identity_changes: dict[str, collections.deque[float]] = collections.defaultdict(lambda: collections.deque(maxlen=8)) self._cooldown: dict[tuple[str, str], float] = {} - self._block_attempted: set[int] = set() + self._block_attempted: collections.OrderedDict[int, float] = collections.OrderedDict() + # baseline_touch() used to perform a SELECT/UPDATE/COMMIT for every EVE + # record, even if an asset had already used the same app/port/domain. + # A bounded process-local LRU preserves first-seen detection while + # removing that repeated SQLite write amplification. + self._baseline_seen: collections.OrderedDict[tuple[str, str, str], None] = collections.OrderedDict() + self._last_state_cleanup = 0.0 self._routeros_inventory_syncs = 0 self._routeros_inventory_assets = 0 self._routeros_inventory_last_at = "" @@ -249,6 +265,19 @@ class NDRAnalyzer: "routeros_inventory_syncs": self._routeros_inventory_syncs, "routeros_inventory_assets": self._routeros_inventory_assets, "routeros_inventory_last_at": self._routeros_inventory_last_at, + "state_entries": { + "beacon": len(self._beacon), + "scan": len(self._scan), + "out_scan": len(self._out_scan), + "egress": len(self._egress), + "dga": len(self._dga), + "nxdomain": len(self._nxdomain), + "dns_tunnel": len(self._dns_tunnel), + "identity_changes": len(self._identity_changes), + "cooldown": len(self._cooldown), + "baseline_lru": len(self._baseline_seen), + "block_attempted": len(self._block_attempted), + }, } def sync_routeros_inventory(self) -> dict[str, int]: @@ -301,6 +330,7 @@ class NDRAnalyzer: self._queue.task_done() def _process(self, record: dict[str, Any], alert_id: int | None) -> None: + self._prune_state(_dt(record.get("timestamp"))) subject = self._subject(record) if subject: asset = self.store.observe_asset(record) @@ -357,7 +387,15 @@ class NDRAnalyzer: if fingerprint: values.append(("client-fingerprint", fingerprint[:160], 30)) for kind, value, risk in values: + baseline_key = (subject, kind, value) + if baseline_key in self._baseline_seen: + self._baseline_seen.move_to_end(baseline_key) + continue is_new, _ = self.store.baseline_touch(subject, kind, value, str(record.get("timestamp") or "")) + self._baseline_seen[baseline_key] = None + self._baseline_seen.move_to_end(baseline_key) + while len(self._baseline_seen) > NDR_BASELINE_CACHE_MAX: + self._baseline_seen.popitem(last=False) if is_new and observations >= self.behavior_min_observations: self._behavior_hits += 1 self._emit(record, alert_id, subject, "behavior", "behavior-change", risk, f"New {kind} for established asset: {value}") @@ -369,7 +407,15 @@ class NDRAnalyzer: port = int(record.get("dest_port") or 0) if record.get("direction") == "outbound" and dest: key = (subject, dest) - dq = self._beacon[key]; dq.append(now) + dq = self._beacon.get(key) + if dq is None: + dq = collections.deque(maxlen=12) + self._beacon[key] = dq + else: + self._beacon.move_to_end(key) + dq.append(now) + while len(self._beacon) > NDR_BEACON_MAX_KEYS: + self._beacon.popitem(last=False) if port in SENSITIVE_PORTS: scan = self._out_scan[subject]; scan.append((now, dest, port)) while scan and scan[0][0] < now - 60: scan.popleft() @@ -465,7 +511,10 @@ class NDRAnalyzer: if self.auto_block and combined_risk >= self.auto_block_risk and incident_id not in self._block_attempted: target = self._remote_target(record, subject) if target and self.routeros.configured: - self._block_attempted.add(incident_id) + self._block_attempted[incident_id] = _dt(record.get("timestamp")) + self._block_attempted.move_to_end(incident_id) + while len(self._block_attempted) > NDR_BLOCK_ATTEMPT_MAX: + self._block_attempted.popitem(last=False) result = self.routeros.block_ip(target, self.block_timeout, f"MikroSuricata NDR risk {combined_risk}: {summary}"[:220]) if result.success: self.store.mark_incident_blocked(incident_id, target) @@ -475,6 +524,48 @@ class NDRAnalyzer: except Exception as exc: print(f"[ndr] forensic PCAP capture failed: {exc}", flush=True) + def _prune_state(self, now: float) -> None: + current = time.monotonic() + if current - self._last_state_cleanup < 60: + return + self._last_state_cleanup = current + stale_before = now - NDR_STATE_IDLE_SECONDS + + for key in list(self._beacon): + values = self._beacon.get(key) + if not values or values[-1] < stale_before: + self._beacon.pop(key, None) + + keyed_deques = ( + self._scan, + self._out_scan, + self._egress, + self._dga, + self._nxdomain, + self._dns_tunnel, + self._identity_changes, + ) + for mapping in keyed_deques: + for key, values in list(mapping.items()): + if not values: + mapping.pop(key, None) + continue + latest = values[-1] + latest_ts = latest if isinstance(latest, (int, float)) else latest[0] + if float(latest_ts) < stale_before: + mapping.pop(key, None) + while len(mapping) > NDR_STATE_MAX_SUBJECTS: + mapping.pop(next(iter(mapping)), None) + + for key, expires_at in list(self._cooldown.items()): + if expires_at <= now: + self._cooldown.pop(key, None) + + block_before = now - NDR_BLOCK_ATTEMPT_TTL_SECONDS + for incident_id, attempted_at in list(self._block_attempted.items()): + if attempted_at < block_before: + self._block_attempted.pop(incident_id, None) + def _subject(self, record: dict[str, Any]) -> str: src = str(record.get("src_ip") or record.get("dhcp_assigned_ip") or record.get("arp_src_ip") or "") dst = str(record.get("dest_ip") or "") diff --git a/app/rules.py b/app/rules.py index 07e0b2b..e0bc323 100644 --- a/app/rules.py +++ b/app/rules.py @@ -61,6 +61,8 @@ class RuleManager: "items": [], } self._last_result = "not changed" + self._vendor_rule_cache_key: tuple[int, int, int] | None = None + self._vendor_rule_cache_count = 0 self._snapshot_dir = Path(self.config.suricata_custom_rules).parent / "rule-snapshots" self._snapshot_dir.mkdir(parents=True, exist_ok=True) self._ensure_files() @@ -78,6 +80,7 @@ class RuleManager: last_result = self._last_result vendor_root = self.config.suricata_persist_lib_dir vendor_rules = os.path.join(vendor_root, "rules", "suricata.rules") + vendor_rules_size, vendor_rules_updated_at, vendor_rule_count = self._vendor_rule_metadata(vendor_rules) source_index = _first_existing_path( os.path.join(vendor_root, "rules", ".cache", "index.yaml"), os.path.join(vendor_root, "update", "cache", "index.yaml"), @@ -94,15 +97,38 @@ class RuleManager: "threshold_entry_count": _count_config_entries(threshold), "suppressed_sids": _suppressed_sids(threshold), "vendor_rules_path": vendor_rules, - "vendor_rules_size_bytes": _file_size(vendor_rules), - "vendor_rule_count": _count_rule_file(vendor_rules), - "vendor_rules_updated_at": _file_mtime_iso(vendor_rules), + "vendor_rules_size_bytes": vendor_rules_size, + "vendor_rule_count": vendor_rule_count, + "vendor_rules_updated_at": vendor_rules_updated_at, "source_index_updated_at": _file_mtime_iso(source_index) if source_index else None, "source_index_url": self.SOURCE_INDEX_URL, "last_result": last_result, "snapshots": len(self.list_snapshots()), } + def _vendor_rule_metadata(self, path: str) -> tuple[int, str | None, int]: + try: + stat = os.stat(path) + except OSError: + with self._lock: + self._vendor_rule_cache_key = None + self._vendor_rule_cache_count = 0 + return 0, None, 0 + + cache_key = (int(stat.st_ino), int(stat.st_mtime_ns), int(stat.st_size)) + with self._lock: + if cache_key == self._vendor_rule_cache_key: + count = self._vendor_rule_cache_count + else: + count = -1 + if count < 0: + count = _count_rule_file(path) + with self._lock: + self._vendor_rule_cache_key = cache_key + self._vendor_rule_cache_count = count + updated_at = datetime.fromtimestamp(stat.st_mtime, timezone.utc).isoformat() + return int(stat.st_size), updated_at, count + def content(self) -> dict: return { "custom_rules": self._read(self.config.suricata_custom_rules), diff --git a/app/static/js/app.js b/app/static/js/app.js index 2a48dc6..d4f5ff2 100644 --- a/app/static/js/app.js +++ b/app/static/js/app.js @@ -905,7 +905,9 @@ if (!state.appStarted) state.appStarted=true; await initialLoad(); connectWebSocket(); - if (!state.refreshTimer) state.refreshTimer=setInterval(refreshStats,30000); + // Live throughput already arrives over WebSocket. Aggregate SQLite-backed + // cards/charts do not need to be recomputed every 30 seconds. + if (!state.refreshTimer) state.refreshTimer=setInterval(refreshStats,60000); if ('ResizeObserver' in window && !startApplication.observer) { startApplication.observer=new ResizeObserver(()=>scheduleChartRender()); document.querySelectorAll('.view,.chart-panel,.donut-panel').forEach(el=>startApplication.observer.observe(el)); diff --git a/app/store.py b/app/store.py index 23a9cf5..2e84f6a 100644 --- a/app/store.py +++ b/app/store.py @@ -22,6 +22,12 @@ class AlertStore: 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. @@ -258,6 +264,50 @@ class AlertStore: 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: @@ -321,6 +371,7 @@ class AlertStore: if not records: return 0 inserted = 0 + inserted_ts: list[int] = [] with self._lock: for event_key, payload in records: if ( @@ -353,14 +404,20 @@ class AlertStore: raw, ), ) - inserted += max(0, int(cursor.rowcount or 0)) + 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=(",", ":")) @@ -368,8 +425,13 @@ class AlertStore: "INSERT OR IGNORE INTO traffic_throughput(sample_key, ts_ms, payload_json) VALUES (?, ?, ?)", (str(sample_key), int(payload.get("ts_ms") or 0), raw), ) - inserted += max(0, int(cursor.rowcount or 0)) + 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( @@ -586,38 +648,52 @@ class AlertStore: def traffic_archive_status(self) -> dict[str, Any]: 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), - } + return dict(self._traffic_archive_meta) def purge_traffic_archive_before(self, cutoff_ms: int) -> dict[str, int]: with self._lock: - events = self._conn.execute("DELETE FROM traffic_events WHERE ts_ms dict[str, int]: with self._lock: - events = int(self._conn.execute("SELECT COUNT(*) FROM traffic_events").fetchone()[0]) - throughput = int(self._conn.execute("SELECT COUNT(*) FROM traffic_throughput").fetchone()[0]) + 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( diff --git a/app/webui.py b/app/webui.py index 78d7c5b..b52dccf 100644 --- a/app/webui.py +++ b/app/webui.py @@ -124,6 +124,7 @@ class WebServer: forensic_pcap: ForensicPcapRing | None = None, traffic_source: Any | None = None, metrics_provider: Callable[[], str] | None = None, + healthcheck_provider: Callable[[], dict] | None = None, ) -> None: self.config = config self.store = store @@ -140,8 +141,17 @@ class WebServer: self.forensic_pcap = forensic_pcap self.traffic_source = traffic_source self.metrics_provider = metrics_provider + self.healthcheck_provider = healthcheck_provider or health_provider self.metrics_access = MetricsAccessControl(config) if metrics_provider is not None else None self.backup_manager = backup_manager or BackupManager(config.db_path, os.path.dirname(config.db_path) or ".") + self._status_cache_lock = threading.Lock() + self._status_cache: dict[str, Any] | None = None + self._status_cache_at = 0.0 + self._status_cache_seconds = 30.0 + self._stats_cache_lock = threading.Lock() + self._stats_cache: dict[str, Any] | None = None + self._stats_cache_at = 0.0 + self._stats_cache_seconds = 60.0 self.auth = SessionAuth(config, store) self._login_lock = threading.Lock() self._login_attempts: dict[str, deque[float]] = defaultdict(deque) @@ -149,10 +159,17 @@ class WebServer: self.thread = threading.Thread(target=self.server.serve_forever, name="web-ui", daemon=True) def _status_payload(self) -> dict: + now = time.monotonic() + with self._status_cache_lock: + if self._status_cache is not None and now - self._status_cache_at < self._status_cache_seconds: + return dict(self._status_cache) + payload = dict(self.health_provider()) payload["summary"] = self.store.summary() if self.traffic_history is not None: - history = self.traffic_history.status() + # main.health() already obtains this once for the service table. Reuse + # it instead of issuing another Redis + archive status query. + history = dict(payload.get("traffic_history") or self.traffic_history.status()) if self.live_pipeline is not None: history.update(self.live_pipeline.status()) if self.event_bus is not None: @@ -160,6 +177,30 @@ class WebServer: payload["traffic_history"] = history if self.analytics_cache is not None: payload["analytics_snapshots"] = self.analytics_cache.status() + + with self._status_cache_lock: + self._status_cache = dict(payload) + self._status_cache_at = time.monotonic() + return payload + + def _healthcheck_payload(self) -> dict: + return dict(self.healthcheck_provider()) + + def _stats_payload(self) -> dict: + """Cache aggregate SQLite statistics shared by all dashboard clients.""" + now = time.monotonic() + with self._stats_cache_lock: + if self._stats_cache is not None and now - self._stats_cache_at < self._stats_cache_seconds: + return dict(self._stats_cache) + + payload = { + "summary": self.store.summary(), + "analytics": self.store.analytics(), + "ndr": self.store.ndr_summary(), + } + with self._stats_cache_lock: + self._stats_cache = dict(payload) + self._stats_cache_at = time.monotonic() return payload def _analytics_payload(self, window_seconds: int) -> dict: @@ -192,6 +233,14 @@ class WebServer: def _login_allowed(self, client_ip: str) -> bool: now = time.monotonic() with self._login_lock: + if len(self._login_attempts) >= 4096 and client_ip not in self._login_attempts: + for key, values in list(self._login_attempts.items()): + while values and values[0] < now - 300: + values.popleft() + if not values: + self._login_attempts.pop(key, None) + while len(self._login_attempts) >= 4096: + self._login_attempts.pop(next(iter(self._login_attempts)), None) attempts = self._login_attempts[client_ip] while attempts and attempts[0] < now - 300: attempts.popleft() @@ -267,7 +316,7 @@ class WebServer: self._static(parsed.path) return if parsed.path == "/api/health": - self._json(outer._status_payload()) + self._json(outer._healthcheck_payload()) return if parsed.path == "/api/auth/session": self._auth_session() @@ -281,7 +330,7 @@ class WebServer: self._json(store.summary()) return if parsed.path == "/api/stats": - self._json({"summary": store.summary(), "analytics": store.analytics(), "ndr": store.ndr_summary()}) + self._json(outer._stats_payload()) return if parsed.path == "/api/config": self._json(config.public_dict()) @@ -845,7 +894,7 @@ class WebServer: if now - last_throughput >= 1: self._ws_send_json({"type": "throughput", "data": outer._current_throughput_payload(window)}) last_throughput = now - if now - last_status >= 5: + if now - last_status >= 30: self._ws_send_json({"type": "status", "data": outer._status_payload()}) last_status = now if now - last_analytics >= 10: diff --git a/deploy-routeros.env.example b/deploy-routeros.env.example index 20268bb..3c02ba6 100644 --- a/deploy-routeros.env.example +++ b/deploy-routeros.env.example @@ -65,7 +65,7 @@ REDIS_MAXMEMORY_MB=128 REDIS_SNAPSHOT_SECONDS=0 REDIS_AOF=false TRAFFIC_RETENTION_HOURS=24 -TRAFFIC_ARCHIVE_INTERVAL_SECONDS=5 +TRAFFIC_ARCHIVE_INTERVAL_SECONDS=10 TRAFFIC_ARCHIVE_LAG_SECONDS=10 TRAFFIC_ARCHIVE_BATCH_SIZE=1000 TRAFFIC_MAX_EVENTS=0 @@ -86,7 +86,7 @@ METRICS_BASIC_AUTH_USERNAME= METRICS_BASIC_AUTH_PASSWORD= SESSION_HOURS=168 SESSION_COOKIE_SECURE=false -ANALYTICS_SNAPSHOT_INTERVAL_SECONDS=60 +ANALYTICS_SNAPSHOT_INTERVAL_SECONDS=120 # MikroSuricata NDR / correlation. Keep NDR_AUTO_BLOCK=false while baselining. NDR_ENABLED=true diff --git a/docker-compose.yml b/docker-compose.yml index fea343e..0621703 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -21,7 +21,7 @@ services: - ids-data:/data healthcheck: test: ["CMD", "python3", "/opt/ids/scripts/healthcheck.py"] - interval: 15s + interval: 30s timeout: 5s retries: 5 start_period: 20s diff --git a/docs/routeros-tzsp-capture.md b/docs/routeros-tzsp-capture.md index abe3907..c8df939 100644 --- a/docs/routeros-tzsp-capture.md +++ b/docs/routeros-tzsp-capture.md @@ -1,6 +1,6 @@ # RouterOS TZSP capture architecture -MikroSuricata 0.11.3 uses a **hybrid RouterOS capture path**. The goal is to keep high-volume routed IPv4 off `/tool/sniffer` while still preserving visibility into ARP and other non-IPv4 Ethernet traffic. +MikroSuricata 0.11.4 uses a **hybrid RouterOS capture path**. The goal is to keep high-volume routed IPv4 off `/tool/sniffer` while still preserving visibility into ARP and other non-IPv4 Ethernet traffic. ## Default topology @@ -100,7 +100,7 @@ With the default filter the stream can include, among other protocols: ### IPv6 performance note -IPv6 is deliberately left in the Packet Sniffer complement in 0.11.3 because the requested design is "mangle for IPv4, stream everything except IPv4". If the network later carries sustained high-volume IPv6, the Packet Sniffer performance ceiling can reappear for that traffic. At that point IPv6 should be split into its own high-throughput capture path and the L2 sniffer narrowed to explicit non-IP EtherTypes. +IPv6 is deliberately left in the Packet Sniffer complement in 0.11.4 because the requested design is "mangle for IPv4, stream everything except IPv4". If the network later carries sustained high-volume IPv6, the Packet Sniffer performance ceiling can reappear for that traffic. At that point IPv6 should be split into its own high-throughput capture path and the L2 sniffer narrowed to explicit non-IP EtherTypes. ## No bridge or interface name is assumed @@ -264,7 +264,7 @@ If `TZSP RX` is close to the expected traffic and capture remains close to 100%, - Packet Sniffer cannot necessarily see traffic switched entirely in hardware by a hardware-offloaded bridge. Broadcast/multicast behavior can differ by platform. - The default mangle rule covers routed IPv4 only, not bridge-only IPv4 switching. -- IPv6 remains on Packet Sniffer in 0.11.3 and can therefore inherit Packet Sniffer throughput limits under sustained high-rate IPv6 traffic. +- IPv6 remains on Packet Sniffer in 0.11.4 and can therefore inherit Packet Sniffer throughput limits under sustained high-rate IPv6 traffic. - `filter-interface=all` can expose the same low-level broadcast/L2 frame on more than one logical/physical observation point on some topologies. If this is noisy, set `TZSP_L2_INTERFACE` explicitly. - The global Packet Sniffer has no per-consumer instance. MikroSuricata therefore cannot preserve a second independent sniffer configuration while also owning the non-IPv4 TZSP stream. - Do not run an additional full-traffic Packet Sniffer stream to the same TZSP destination at the same time; duplicate packets will inflate traffic and Suricata processing. diff --git a/routeros/04-container-import-amd64.rsc b/routeros/04-container-import-amd64.rsc index 82113b8..550c37e 100644 --- a/routeros/04-container-import-amd64.rsc +++ b/routeros/04-container-import-amd64.rsc @@ -28,10 +28,10 @@ /container/envs/add list=IDS_ENV key=REDIS_MANAGED value=true /container/envs/add list=IDS_ENV key=REDIS_DATA_DIR value=/data/redis /container/envs/add list=IDS_ENV key=REDIS_PORT value=6379 -/container/envs/add list=IDS_ENV key=REDIS_MAXMEMORY_MB value=0 -/container/envs/add list=IDS_ENV key=REDIS_SNAPSHOT_SECONDS value=1800 +/container/envs/add list=IDS_ENV key=REDIS_MAXMEMORY_MB value=128 +/container/envs/add list=IDS_ENV key=REDIS_SNAPSHOT_SECONDS value=0 /container/envs/add list=IDS_ENV key=TRAFFIC_RETENTION_HOURS value=24 -/container/envs/add list=IDS_ENV key=TRAFFIC_MAX_EVENTS value=0 +/container/envs/add list=IDS_ENV key=TRAFFIC_MAX_EVENTS value=50000 /container/envs/add list=IDS_ENV key=TRAFFIC_MEMORY_EVENTS value=0 /container/envs/add list=IDS_ENV key=WEBSOCKET_QUEUE_SIZE value=512 /container/envs/add list=IDS_ENV key=LIVE_FLOW_UPDATE_SECONDS value=2.0 @@ -48,7 +48,7 @@ /container/envs/add list=IDS_ENV key=METRICS_BASIC_AUTH_PASSWORD value="" /container/envs/add list=IDS_ENV key=SESSION_HOURS value=168 /container/envs/add list=IDS_ENV key=SESSION_COOKIE_SECURE value=false -/container/envs/add list=IDS_ENV key=ANALYTICS_SNAPSHOT_INTERVAL_SECONDS value=60 +/container/envs/add list=IDS_ENV key=ANALYTICS_SNAPSHOT_INTERVAL_SECONDS value=120 /container/mounts/remove [find where list="IDS_MOUNTS"] /container/mounts/add list=IDS_MOUNTS src=disk1/containers/suricata-data dst=/data diff --git a/routeros/04-container-import-arm.rsc b/routeros/04-container-import-arm.rsc index 6a6e28d..0992070 100644 --- a/routeros/04-container-import-arm.rsc +++ b/routeros/04-container-import-arm.rsc @@ -28,10 +28,10 @@ /container/envs/add list=IDS_ENV key=REDIS_MANAGED value=true /container/envs/add list=IDS_ENV key=REDIS_DATA_DIR value=/data/redis /container/envs/add list=IDS_ENV key=REDIS_PORT value=6379 -/container/envs/add list=IDS_ENV key=REDIS_MAXMEMORY_MB value=0 -/container/envs/add list=IDS_ENV key=REDIS_SNAPSHOT_SECONDS value=1800 +/container/envs/add list=IDS_ENV key=REDIS_MAXMEMORY_MB value=128 +/container/envs/add list=IDS_ENV key=REDIS_SNAPSHOT_SECONDS value=0 /container/envs/add list=IDS_ENV key=TRAFFIC_RETENTION_HOURS value=24 -/container/envs/add list=IDS_ENV key=TRAFFIC_MAX_EVENTS value=0 +/container/envs/add list=IDS_ENV key=TRAFFIC_MAX_EVENTS value=50000 /container/envs/add list=IDS_ENV key=TRAFFIC_MEMORY_EVENTS value=0 /container/envs/add list=IDS_ENV key=WEBSOCKET_QUEUE_SIZE value=512 /container/envs/add list=IDS_ENV key=LIVE_FLOW_UPDATE_SECONDS value=2.0 @@ -48,7 +48,7 @@ /container/envs/add list=IDS_ENV key=METRICS_BASIC_AUTH_PASSWORD value="" /container/envs/add list=IDS_ENV key=SESSION_HOURS value=168 /container/envs/add list=IDS_ENV key=SESSION_COOKIE_SECURE value=false -/container/envs/add list=IDS_ENV key=ANALYTICS_SNAPSHOT_INTERVAL_SECONDS value=60 +/container/envs/add list=IDS_ENV key=ANALYTICS_SNAPSHOT_INTERVAL_SECONDS value=120 /container/mounts/remove [find where list="IDS_MOUNTS"] /container/mounts/add list=IDS_MOUNTS src=disk1/containers/suricata-data dst=/data diff --git a/routeros/04-container-import-arm64.rsc b/routeros/04-container-import-arm64.rsc index 35fa4a3..c8e6f50 100644 --- a/routeros/04-container-import-arm64.rsc +++ b/routeros/04-container-import-arm64.rsc @@ -29,10 +29,10 @@ /container/envs/add list=IDS_ENV key=REDIS_MANAGED value=true /container/envs/add list=IDS_ENV key=REDIS_DATA_DIR value=/data/redis /container/envs/add list=IDS_ENV key=REDIS_PORT value=6379 -/container/envs/add list=IDS_ENV key=REDIS_MAXMEMORY_MB value=0 -/container/envs/add list=IDS_ENV key=REDIS_SNAPSHOT_SECONDS value=1800 +/container/envs/add list=IDS_ENV key=REDIS_MAXMEMORY_MB value=128 +/container/envs/add list=IDS_ENV key=REDIS_SNAPSHOT_SECONDS value=0 /container/envs/add list=IDS_ENV key=TRAFFIC_RETENTION_HOURS value=24 -/container/envs/add list=IDS_ENV key=TRAFFIC_MAX_EVENTS value=0 +/container/envs/add list=IDS_ENV key=TRAFFIC_MAX_EVENTS value=50000 /container/envs/add list=IDS_ENV key=TRAFFIC_MEMORY_EVENTS value=0 /container/envs/add list=IDS_ENV key=WEBSOCKET_QUEUE_SIZE value=512 /container/envs/add list=IDS_ENV key=LIVE_FLOW_UPDATE_SECONDS value=2.0 @@ -49,7 +49,7 @@ /container/envs/add list=IDS_ENV key=METRICS_BASIC_AUTH_PASSWORD value="" /container/envs/add list=IDS_ENV key=SESSION_HOURS value=168 /container/envs/add list=IDS_ENV key=SESSION_COOKIE_SECURE value=false -/container/envs/add list=IDS_ENV key=ANALYTICS_SNAPSHOT_INTERVAL_SECONDS value=60 +/container/envs/add list=IDS_ENV key=ANALYTICS_SNAPSHOT_INTERVAL_SECONDS value=120 /container/mounts/remove [find where list="IDS_MOUNTS"] /container/mounts/add list=IDS_MOUNTS src=disk1/containers/suricata-data dst=/data diff --git a/routeros/app-template.yml b/routeros/app-template.yml index 4eae26a..69abe11 100644 --- a/routeros/app-template.yml +++ b/routeros/app-template.yml @@ -33,10 +33,11 @@ services: REDIS_MANAGED: "true" REDIS_DATA_DIR: /data/redis REDIS_PORT: "6379" - REDIS_MAXMEMORY_MB: "0" - REDIS_SNAPSHOT_SECONDS: "1800" + REDIS_MAXMEMORY_MB: "128" + REDIS_SNAPSHOT_SECONDS: "0" + REDIS_AOF: "false" TRAFFIC_RETENTION_HOURS: "24" - TRAFFIC_MAX_EVENTS: "0" + TRAFFIC_MAX_EVENTS: "50000" TRAFFIC_MEMORY_EVENTS: "0" WEBSOCKET_QUEUE_SIZE: "512" LIVE_FLOW_UPDATE_SECONDS: "2.0" @@ -47,4 +48,4 @@ services: METRICS_BASIC_AUTH_PASSWORD: "" SESSION_HOURS: "168" SESSION_COOKIE_SECURE: "false" - ANALYTICS_SNAPSHOT_INTERVAL_SECONDS: "60" + ANALYTICS_SNAPSHOT_INTERVAL_SECONDS: "120" diff --git a/rust/tzsp-receiver/Cargo.toml b/rust/tzsp-receiver/Cargo.toml index 4f74ed9..2e313fd 100644 --- a/rust/tzsp-receiver/Cargo.toml +++ b/rust/tzsp-receiver/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mikrosuricata-tzsp" -version = "0.11.3" +version = "0.11.4" edition = "2021" publish = false diff --git a/scripts/deploy-routeros.sh b/scripts/deploy-routeros.sh index 4c0d3da..beb59d3 100755 --- a/scripts/deploy-routeros.sh +++ b/scripts/deploy-routeros.sh @@ -74,7 +74,7 @@ TZSP_L2_FILTER_INTERFACE="${TZSP_L2_INTERFACE:-all}" : "${REDIS_SNAPSHOT_SECONDS:=0}" : "${REDIS_AOF:=false}" : "${TRAFFIC_RETENTION_HOURS:=24}" -: "${TRAFFIC_ARCHIVE_INTERVAL_SECONDS:=5}" +: "${TRAFFIC_ARCHIVE_INTERVAL_SECONDS:=10}" : "${TRAFFIC_ARCHIVE_LAG_SECONDS:=10}" : "${TRAFFIC_ARCHIVE_BATCH_SIZE:=1000}" : "${TRAFFIC_MAX_EVENTS:=0}" @@ -92,7 +92,7 @@ TZSP_L2_FILTER_INTERFACE="${TZSP_L2_INTERFACE:-all}" : "${METRICS_BASIC_AUTH_PASSWORD:=}" : "${SESSION_HOURS:=168}" : "${SESSION_COOKIE_SECURE:=false}" -: "${ANALYTICS_SNAPSHOT_INTERVAL_SECONDS:=60}" +: "${ANALYTICS_SNAPSHOT_INTERVAL_SECONDS:=120}" : "${NDR_ENABLED:=true}" : "${NDR_CORRELATION_WINDOW_SECONDS:=1800}" : "${BEHAVIOR_MIN_OBSERVATIONS:=50}" diff --git a/scripts/healthcheck.py b/scripts/healthcheck.py index c7ec37d..099ae20 100755 --- a/scripts/healthcheck.py +++ b/scripts/healthcheck.py @@ -6,7 +6,7 @@ import urllib.request try: port = int(os.getenv("WEB_PORT", "8080")) - with urllib.request.urlopen(f"http://127.0.0.1:{port}/api/status", timeout=3) as response: + with urllib.request.urlopen(f"http://127.0.0.1:{port}/api/health", timeout=3) as response: data = json.load(response) raise SystemExit(0 if data.get("operational") else 1) except Exception as exc: diff --git a/tests/test_analytics_cache.py b/tests/test_analytics_cache.py index 3c4cf3d..25d3f90 100644 --- a/tests/test_analytics_cache.py +++ b/tests/test_analytics_cache.py @@ -30,6 +30,16 @@ class AnalyticsCacheTests(unittest.TestCase): self.assertFalse(cache._snapshot_thread.is_alive()) store.close() + def test_background_worker_has_no_dashboard_windows_until_requested(self): + with tempfile.TemporaryDirectory() as td: + store = AlertStore(os.path.join(td, "ids.db")) + history = TrafficHistory("", retention_hours=24, max_events=1000, memory_events=1000) + cache = AnalyticsSnapshotCache(store, history, threading.Event(), interval_seconds=60) + self.assertEqual(cache._due_windows(), []) + cache.get(900) + self.assertEqual(cache._due_windows(), [900]) + store.close() + def test_refresh_persists_all_dashboard_windows_in_sqlite_cache(self): with tempfile.TemporaryDirectory() as td: store = AlertStore(os.path.join(td, "ids.db")) diff --git a/tests/test_config.py b/tests/test_config.py index fa21c47..e74f6b6 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -43,9 +43,17 @@ class ConfigTests(unittest.TestCase): self.assertEqual(cfg.redis_maxmemory_mb, 128) self.assertEqual(cfg.redis_snapshot_seconds, 0) self.assertFalse(cfg.redis_aof) - self.assertEqual(cfg.traffic_archive_interval_seconds, 5) + self.assertEqual(cfg.traffic_archive_interval_seconds, 10) + self.assertEqual(cfg.analytics_snapshot_interval_seconds, 120) self.assertEqual(cfg.traffic_archive_lag_seconds, 10) self.assertEqual(cfg.traffic_archive_batch_size, 1000) + self.assertEqual(cfg.traffic_max_events, 50000) + + def test_traffic_max_events_can_be_overridden_or_disabled(self): + with patch.dict(os.environ, {"TRAFFIC_MAX_EVENTS": "12345"}, clear=True): + self.assertEqual(Config.from_env().traffic_max_events, 12345) + with patch.dict(os.environ, {"TRAFFIC_MAX_EVENTS": "0"}, clear=True): + self.assertEqual(Config.from_env().traffic_max_events, 0) if __name__ == "__main__": diff --git a/tests/test_live.py b/tests/test_live.py index 7c733d1..442e485 100644 --- a/tests/test_live.py +++ b/tests/test_live.py @@ -8,11 +8,13 @@ from datetime import datetime, timezone from app.store import AlertStore from app.live import ( + MAX_ANALYTICS_DIMENSION_KEYS, EventBus, LiveEventPipeline, RedisUnavailableError, TrafficHistory, TrafficNormalizer, + _AnalyticsAccumulator, event_matches, ) @@ -410,6 +412,19 @@ class LiveTests(unittest.TestCase): self.assertEqual(store.traffic_archive_status()["events"], 0) store.close() + def test_sqlite_analytics_high_cardinality_counters_are_bounded(self): + now = int(time.time() * 1000) + acc = _AnalyticsAccumulator(now - 60_000, now, 60, track_high_cardinality=False) + for idx in range(MAX_ANALYTICS_DIMENSION_KEYS + 100): + acc.add_event({ + "ts_ms": now, + "type": "fileinfo", + "filename": f"unique-{idx}.bin", + "file_sha256": f"{idx:064x}"[-64:], + "direction": "outbound", + }) + self.assertEqual(len(acc.file_activity), MAX_ANALYTICS_DIMENSION_KEYS) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_metrics.py b/tests/test_metrics.py index 0fa09bd..d1b1adf 100644 --- a/tests/test_metrics.py +++ b/tests/test_metrics.py @@ -175,6 +175,74 @@ class PrometheusMetricsTests(unittest.TestCase): store.close() + def test_health_endpoint_uses_lightweight_provider(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = str(Path(tmp) / "ids.db") + cfg = replace(Config.from_env(), web_bind="127.0.0.1", web_port=0, db_path=db_path) + store = AlertStore(db_path) + full_calls = [] + health_calls = [] + + def full_status(): + full_calls.append(True) + raise AssertionError("lightweight health endpoint must not build full status") + + web = WebServer( + cfg, + store, + full_status, + healthcheck_provider=lambda: health_calls.append(True) or {"status": "ok", "operational": True}, + ) + try: + with contextlib.redirect_stdout(io.StringIO()): + web.start() + port = web.server.server_address[1] + with urllib.request.urlopen(f"http://127.0.0.1:{port}/api/health", timeout=2) as response: + body = response.read().decode("utf-8") + self.assertEqual(200, response.status) + self.assertIn('"operational":true', body.replace(" ", "")) + self.assertEqual([], full_calls) + self.assertEqual([True], health_calls) + finally: + web.stop() + store.close() + + def test_status_payload_is_cached_for_short_poll_bursts(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = str(Path(tmp) / "ids.db") + cfg = replace(Config.from_env(), web_bind="127.0.0.1", web_port=0, db_path=db_path) + store = AlertStore(db_path) + calls = [] + web = WebServer(cfg, store, lambda: calls.append(True) or {"operational": True}) + try: + first = web._status_payload() + second = web._status_payload() + self.assertTrue(first["operational"]) + self.assertTrue(second["operational"]) + self.assertEqual([True], calls) + finally: + web.server.server_close() + store.close() + + def test_stats_payload_caches_expensive_sqlite_aggregates(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = str(Path(tmp) / "ids.db") + cfg = replace(Config.from_env(), web_bind="127.0.0.1", web_port=0, db_path=db_path) + store = AlertStore(db_path) + calls = {"summary": 0, "analytics": 0, "ndr": 0} + store.summary = lambda: calls.__setitem__("summary", calls["summary"] + 1) or {"alerts": 0} + store.analytics = lambda: calls.__setitem__("analytics", calls["analytics"] + 1) or {"alerts_24h": 0} + store.ndr_summary = lambda: calls.__setitem__("ndr", calls["ndr"] + 1) or {"open_incidents": 0} + web = WebServer(cfg, store, lambda: {"operational": True}) + try: + first = web._stats_payload() + second = web._stats_payload() + self.assertEqual(first, second) + self.assertEqual({"summary": 1, "analytics": 1, "ndr": 1}, calls) + finally: + web.server.server_close() + store.close() + def test_metrics_ip_acl_denies_before_rendering_metrics(self): with tempfile.TemporaryDirectory() as tmp: db_path = str(Path(tmp) / "ids.db") diff --git a/tests/test_ndr.py b/tests/test_ndr.py index 376d7da..7f85a3e 100644 --- a/tests/test_ndr.py +++ b/tests/test_ndr.py @@ -1,6 +1,7 @@ import base64 import os import tempfile +from unittest.mock import patch from app.ndr import NDRAnalyzer, ThreatIntelManager from app.store import AlertStore @@ -142,3 +143,52 @@ def test_repeated_ip_mac_changes_escalate_to_network_spoofing_and_anomalies_are_ events = store.ndr_incident_events(int(anomaly_incident["id"]), 20) assert sum(1 for event in events if event["stage"] == "protocol-anomaly") == 1 store.close() + + +def test_baseline_touch_is_cached_for_repeated_values(): + with tempfile.TemporaryDirectory() as td: + store = AlertStore(os.path.join(td, "ids.db")) + ti = ThreatIntelManager(store, os.path.join(td, "suricata")) + analyzer = NDRAnalyzer( + store, ti, DummyRouterOS(), "192.168.88.0/24", "", "1h", + enabled=True, auto_block=False, + ) + calls = 0 + original = store.baseline_touch + + def counted(*args, **kwargs): + nonlocal calls + calls += 1 + return original(*args, **kwargs) + + store.baseline_touch = counted + record = { + "timestamp": "2026-08-15T08:20:00+00:00", + "type": "flow", "direction": "outbound", "src_ip": "192.168.88.50", + "dest_ip": "203.0.113.20", "dest_port": 443, "app_proto": "tls", + } + for _ in range(10): + analyzer._process(dict(record), None) + assert calls == 2 # app + outbound port, only on first sight in this process + assert analyzer.status()["state_entries"]["baseline_lru"] == 2 + store.close() + + +def test_beacon_state_is_bounded(): + with tempfile.TemporaryDirectory() as td: + store = AlertStore(os.path.join(td, "ids.db")) + ti = ThreatIntelManager(store, os.path.join(td, "suricata")) + analyzer = NDRAnalyzer( + store, ti, DummyRouterOS(), "192.168.88.0/24", "", "1h", + enabled=True, auto_block=False, + ) + with patch("app.ndr.NDR_BEACON_MAX_KEYS", 32): + for idx in range(64): + analyzer._behavior({ + "timestamp": "2026-08-15T08:30:00+00:00", + "type": "flow", "direction": "outbound", + "src_ip": "192.168.88.60", "dest_ip": f"203.0.113.{idx}", + "dest_port": 443, + }, None, "192.168.88.60") + assert len(analyzer._beacon) == 32 + store.close()