from __future__ import annotations import os import signal import subprocess import sys import tempfile import threading import re import time from dataclasses import replace from datetime import datetime, timezone from pathlib import Path from urllib.parse import urlparse from .analytics_cache import AnalyticsSnapshotCache from .backup import BackupManager from .config import Config from .eve import EVEWatcher from .forensics import ForensicPcapRing from .live import EventBus, LiveEventPipeline, TrafficHistory, TrafficNormalizer from .maintenance import clear_suricata_logs, storage_info from .metrics import PrometheusMetrics from .ndr import NDRAnalyzer, ThreatIntelManager from .notifier import WebhookNotifier from .policy import PolicyEngine from .redis_service import RedisSupervisor from .routeros import RouterOSClient from .rules import RuleManager from .state import RuntimeStats from .store import AlertStore from .tuning import AlertTuner 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 port = parsed.port or (443 if parsed.scheme == "https" else 80) return host, port def _ensure_suricata_state(cfg: Config) -> None: for path in (cfg.suricata_custom_rules, cfg.suricata_threshold_config): Path(path).parent.mkdir(parents=True, exist_ok=True) Path(path).touch(exist_ok=True) def _prepare_suricata_output_config(cfg: Config) -> Config: # The legacy "blocks" mode depended on Python seeing every TZSP frame to # maintain a pre-event RAM ring. The Rust data-plane intentionally removes # Python from that packet path. Keep forensic evidence without reintroducing # the bottleneck by falling back to Suricata's alert-associated PCAP output. if cfg.forensic_pcap_mode == "blocks": print( "[forensics] FORENSIC_PCAP_MODE=blocks is not used with the Rust data-plane; " "using Suricata alert PCAP capture instead", flush=True, ) cfg = replace(cfg, forensic_pcap_mode="alerts") source = Path(cfg.suricata_output_config) text = source.read_text(encoding="utf-8") match = re.search(r"(?ms)^ - pcap-log:\n.*?(?=^ - |\Z)", text) if match is None: raise RuntimeError("Suricata output profile has no pcap-log section") block = match.group(0) enabled = cfg.forensic_pcap_mode in {"alerts", "all"} conditional = "all" if cfg.forensic_pcap_mode == "all" else "alerts" block = re.sub(r"(?m)^ enabled: .*?$", f" enabled: {'yes' if enabled else 'no'}", block) block = re.sub(r"(?m)^ conditional: .*?$", f" conditional: {conditional}", block) rendered = text[:match.start()] + block + text[match.end():] runtime = Path("/run/suricata/ids-output.runtime.yaml") runtime.parent.mkdir(parents=True, exist_ok=True) runtime.write_text(rendered, encoding="utf-8") os.environ["SURICATA_OUTPUT_CONFIG"] = str(runtime) return replace(cfg, suricata_output_config=str(runtime)) def _suricata_common_args(cfg: Config, log_dir: str) -> list[str]: return [ "-c", cfg.suricata_config, "--include", cfg.suricata_output_config, "-l", log_dir, # Suricata exposes one additive -s signature path; use its supported # globbing so all persisted local/custom .rules files are loaded. "-s", cfg.suricata_extra_rules_glob, "--set", f"vars.address-groups.HOME_NET={cfg.suricata_home_net}", "--set", f"threshold-file={cfg.suricata_threshold_config}", "--set", f"default-rule-path={cfg.suricata_persist_lib_dir}/rules", # These fingerprints are useful IDS pivots but remain opt-in in the # upstream configuration unless a rule explicitly needs them. "--set", "app-layer.protocols.tls.ja3-fingerprints=yes", "--set", "app-layer.protocols.tls.ja4-fingerprints=yes", "--set", "app-layer.protocols.ssh.hassh=yes", ] def main() -> int: cfg = _prepare_suricata_output_config(Config.from_env()) stop_event = threading.Event() stats = RuntimeStats() started_at = datetime.now(timezone.utc) started_monotonic = time.monotonic() os.makedirs(os.path.dirname(cfg.eve_path) or ".", exist_ok=True) os.makedirs(os.path.dirname(cfg.db_path) or ".", exist_ok=True) _ensure_suricata_state(cfg) store = AlertStore(cfg.db_path) backup_manager = BackupManager(cfg.db_path, os.path.dirname(cfg.db_path) or ".") threat_intel = ThreatIntelManager(store, os.path.dirname(cfg.suricata_custom_rules)) purged_tests = store.purge_builtin_test_incidents() if purged_tests: print(f"[db] removed {purged_tests} legacy pipeline-test incidents", flush=True) purged = store.purge_older_than(cfg.alert_retention_days) if purged: print(f"[db] purged {purged} old alerts", flush=True) receiver = RustTZSPReceiver( binary=cfg.tzsp_receiver_bin, telemetry_socket=cfg.tzsp_telemetry_socket, stats=stats, stop_event=stop_event, ) try: receiver.start() if not receiver.wait_ready(timeout=8.0): raise RuntimeError("Rust receiver did not report ready state") except Exception as exc: print(f"[fatal] cannot start Rust TZSP data-plane: {exc}", file=sys.stderr, flush=True) print("[fatal] container needs /dev/net/tun and NET_ADMIN capability", file=sys.stderr, flush=True) receiver.close() store.close() return 2 print(f"[tap] {cfg.tap_name} is owned by Rust TZSP receiver, mtu={cfg.tap_mtu}", flush=True) log_dir = os.path.dirname(cfg.eve_path) or "/var/log/suricata" suricata_cmd = [ "suricata", *_suricata_common_args(cfg, log_dir), f"--af-packet={cfg.tap_name}", "--user", "suricata", "--group", "suricata", # Debian's default unix-command socket is directly under /var/run, # which is not writable after Suricata drops privileges. "--set", "unix-command.filename=suricata/suricata-command.socket", ] print("[suricata] validating configuration and managed rules", flush=True) with tempfile.TemporaryDirectory(prefix="suricata-config-test-") as test_log_dir: test_cmd = ["suricata", "-T", *_suricata_common_args(cfg, test_log_dir)] test = subprocess.run(test_cmd, check=False) if test.returncode != 0: print( f"[fatal] suricata configuration test failed with rc={test.returncode}", file=sys.stderr, flush=True, ) receiver.close() store.close() return test.returncode or 3 print("[suricata] starting IDS process", flush=True) suricata = subprocess.Popen(suricata_cmd) with open("/run/suricata.pid", "w", encoding="ascii") as pid_file: pid_file.write(str(suricata.pid)) tuner = AlertTuner( cfg.alert_max_severity, cfg.alert_ignore_sids, cfg.alert_ignore_categories, ) policy = PolicyEngine( cfg.auto_block, cfg.auto_block_max_severity, cfg.monitored_networks, cfg.never_block, ) routeros = RouterOSClient( cfg.routeros_url, cfg.routeros_user, cfg.routeros_password, cfg.routeros_verify_tls, cfg.routeros_address_list, cfg.routeros_http_timeout, ) notifier = WebhookNotifier(cfg.notify_webhook_url, cfg.notify_min_risk, cfg.notify_timeout_seconds) forensic_pcap = ForensicPcapRing( cfg.forensic_pcap_mode, log_dir, window_seconds=cfg.forensic_pcap_window_seconds, memory_mb=cfg.forensic_pcap_memory_mb, max_files=cfg.forensic_pcap_max_files, max_total_mb=cfg.forensic_pcap_max_total_mb, ) ndr_analyzer = NDRAnalyzer( store, threat_intel, routeros, cfg.monitored_networks, cfg.never_block, cfg.block_timeout, enabled=cfg.ndr_enabled, correlation_window_seconds=cfg.ndr_correlation_window_seconds, behavior_min_observations=cfg.behavior_min_observations, auto_block=cfg.ndr_auto_block, auto_block_risk=cfg.ndr_auto_block_risk, notifier=notifier, block_evidence_callback=lambda target, label: forensic_pcap.capture_target(target, label=label), ) redis_supervisor = RedisSupervisor( cfg.redis_managed, cfg.redis_data_dir, cfg.redis_port, cfg.redis_maxmemory_mb, cfg.redis_snapshot_seconds, cfg.redis_aof, ) if cfg.redis_managed and not redis_supervisor.start(wait_ready_seconds=120): raise RuntimeError( f"managed Redis failed to start: {redis_supervisor.status().get('last_error') or 'unknown error'}" ) event_bus = EventBus( history_size=0, subscriber_queue_size=cfg.websocket_queue_size, ) traffic_history = TrafficHistory( cfg.redis_url, cfg.traffic_retention_hours, cfg.traffic_max_events, cfg.traffic_memory_events, require_redis=True, allow_memory_fallback=False, archive_store=store, ) analytics_cache = AnalyticsSnapshotCache( store, traffic_history, stop_event, cfg.analytics_snapshot_interval_seconds, archive_interval_seconds=cfg.traffic_archive_interval_seconds, archive_lag_seconds=cfg.traffic_archive_lag_seconds, archive_batch_size=cfg.traffic_archive_batch_size, ) live_pipeline = LiveEventPipeline(event_bus, traffic_history) normalizer = TrafficNormalizer(cfg.monitored_networks) # The Rust data-plane emits one compact rate sample per second. Persisting it # is asynchronous and never sits in the packet receive/injection path. receiver.set_throughput_sink(live_pipeline.publish_throughput) watcher = EVEWatcher( cfg.eve_path, store, tuner, policy, routeros, cfg.block_timeout, cfg.alert_dedup_window_seconds, stats, stop_event, normalizer=normalizer, live_pipeline=live_pipeline, ndr_analyzer=ndr_analyzer, forensic_pcap=forensic_pcap, ) rule_manager = RuleManager( cfg, pid_provider=lambda: suricata.pid if suricata.poll() is None else None, suricata_available=True, ) 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() tap_up = os.path.exists(f"/sys/class/net/{cfg.tap_name}") eve_up = watcher.is_alive() routeros_status = "configured" if routeros.configured else "disabled" db = store.database_info() storage = storage_info(cfg.db_path, cfg.eve_path) rules = rule_manager.status() 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) alert_overflow = int(suri_stats.get("detect.alert_queue_overflow", 0) or 0) inject_errors = int(runtime.get("inject_errors", 0) or 0) tzsp_udp_drops = int(receiver_status.get("kernel_udp_drops", 0) or 0) tzsp_udp_drops_now = int(receiver_status.get("kernel_udp_drops_interval", 0) or 0) tzsp_queue_drops = int(receiver_status.get("queue_dropped_datagrams", 0) or 0) tzsp_queue_drops_now = int(receiver_status.get("queue_drops_interval", 0) or 0) tzsp_truncated_now = int(receiver_status.get("truncated_interval", 0) or 0) drop_pct = round((kernel_drops / kernel_packets) * 100.0, 3) if kernel_packets else 0.0 sensor_degraded = ( (kernel_packets >= 1000 and drop_pct >= 1.0) or alert_overflow > 0 or inject_errors > 0 or tzsp_udp_drops_now > 0 or tzsp_queue_drops_now > 0 or tzsp_truncated_now > 0 ) 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", "mode": "full", "dev_mode": False, "operational": operational, "started_at": started_at.isoformat(), "uptime_seconds": round(time.monotonic() - started_monotonic, 1), "suricata_running": suricata_up, "suricata_pid": suricata.pid, "auto_block": cfg.auto_block, "routeros_configured": routeros.configured, "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", "status": "up", "details": f"Listening on TCP {cfg.web_bind}:{cfg.web_port}", }, "tzsp": { "name": "TZSP receiver", "status": "up" if tzsp_up else "down", "details": ( f"Rust PID {receiver.pid or '—'} · UDP {cfg.tzsp_bind}:{cfg.tzsp_port} · " f"socket={int(receiver_status.get('rcvbuf_bytes', 0) or 0)} B · " f"queue={int(receiver_status.get('queue_depth_batches', 0) or 0)}/" f"{int(receiver_status.get('queue_capacity_batches', 0) or 0)} batches · " f"kernel drops={tzsp_udp_drops} · queue drops={tzsp_queue_drops}" ), }, "tap": { "name": "TAP interface", "status": "up" if tap_up else "down", "details": f"{cfg.tap_name}, MTU {cfg.tap_mtu}", }, "suricata": { "name": "Suricata IDS", "status": "up" if suricata_up else "down", "details": f"PID {suricata.pid}" if suricata_up else f"Process exited with code {suricata.poll()}", }, "sensor_quality": { "name": "Sensor quality / packet loss", "status": "degraded" if sensor_degraded else "up", "details": f"Suricata packets={kernel_packets}; Suricata kernel drops={kernel_drops} ({drop_pct}%); TZSP UDP drops={tzsp_udp_drops}; alert queue overflow={alert_overflow}; TAP inject errors={inject_errors}", }, "eve": { "name": "EVE JSON watcher", "status": "up" if eve_up else "down", "details": cfg.eve_path, }, "database": { "name": "SQLite database", "status": "up" if db["ok"] else "down", "details": f"{db['path']}; {db['rows']} incidents; WAL={db['journal_mode']}", }, "storage": { "name": "Persistent storage", "status": "up" if storage["free_bytes"] > 0 else "down", "details": f"{storage['path']}; {storage['used_percent']}% used", }, "live_flows": { "name": "Live session stream", "status": "up" if eve_up else "down", "details": "Suricata EVE sessions; packet capture is isolated in the Rust data-plane", }, "traffic_history": { "name": "Live traffic history", "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": { "name": "Persistent dashboard summaries", "status": "up", "details": f"SQLite snapshots for 15m/1h/5h/6h/24h; Redis is not scanned for charts", }, "ndr": { "name": "MikroSuricata NDR correlation", "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']}", }, "redis": { "name": "Managed Redis", "status": ( "up" if redis_status.get("ready") else "disabled" if not cfg.redis_managed else "degraded" ), "details": ( f"{cfg.redis_data_dir}; maxmemory={cfg.redis_maxmemory_mb} MB; persistence={redis_status.get('persistence')}" if cfg.redis_managed else "Managed Redis disabled; REDIS_URL may point to an external server" ), }, "rules": { "name": "Managed rules", "status": "up" if rules["available"] else "disabled", "details": f"{rules.get('builtin_rule_count', 0)} built-in; {rules['custom_rule_count']} custom; {rules['threshold_entry_count']} threshold/suppress entries", }, "routeros": { "name": "RouterOS REST integration", "status": routeros_status, "details": f"{cfg.routeros_url}; auto-block={'enabled' if cfg.auto_block else 'disabled'}", }, }, "ports": [ { "name": "Web UI / API", "direction": "listen", "protocol": "TCP", "address": cfg.web_bind, "port": cfg.web_port, "status": "up", }, { "name": "TZSP receiver", "direction": "listen", "protocol": "UDP", "address": cfg.tzsp_bind, "port": cfg.tzsp_port, "status": "up" if tzsp_up else "down", }, { "name": "RouterOS REST API", "direction": "outbound", "protocol": "TCP", "address": routeros_host, "port": routeros_port, "status": routeros_status, }, ], "runtime": runtime, } version_path = Path(__file__).resolve().parents[1] / "VERSION" try: app_version = version_path.read_text(encoding="ascii").strip() or "unknown" except OSError: app_version = "unknown" def metrics_state() -> dict: return { "components": { "web": True, "tzsp": receiver.is_alive(), "tap": os.path.exists(f"/sys/class/net/{cfg.tap_name}"), "suricata": suricata.poll() is None, "eve": watcher.is_alive(), }, "features": { "auto_block": cfg.auto_block, "ndr": cfg.ndr_enabled, "notifications": notifier.enabled, "routeros": routeros.configured, "redis_managed": cfg.redis_managed, "forensic_pcap": cfg.forensic_pcap_mode != "off", }, } prometheus_metrics = PrometheusMetrics( stats, version=app_version, mode="full", started_at=started_at, state_provider=metrics_state, flow_tracker=receiver, event_bus=event_bus, live_pipeline=live_pipeline, ndr_analyzer=ndr_analyzer, notifier=notifier, forensic_pcap=forensic_pcap, ) web = WebServer( cfg, store, health, stats=stats, rule_manager=rule_manager, traffic_history=traffic_history, event_bus=event_bus, live_pipeline=live_pipeline, routeros=routeros, analytics_cache=analytics_cache, threat_intel=threat_intel, ndr_analyzer=ndr_analyzer, backup_manager=backup_manager, forensic_pcap=forensic_pcap, traffic_source=receiver, metrics_provider=prometheus_metrics.render, healthcheck_provider=healthcheck, ) def housekeeping() -> None: interval_seconds = max(0, cfg.rule_update_interval_hours) * 3600 next_rule_update = time.monotonic() + interval_seconds if interval_seconds else None next_retention = time.monotonic() + 3600 next_routeros_inventory = time.monotonic() + 10 log_limit_bytes = max(0, cfg.suricata_log_max_mb) * 1024 * 1024 while not stop_event.wait(60): now = time.monotonic() if log_limit_bytes: try: current_storage = storage_info(cfg.db_path, cfg.eve_path) if int(current_storage.get("suricata_log_bytes", 0)) > log_limit_bytes: result = clear_suricata_logs(cfg.eve_path) stats.inc("log_auto_truncations") print( f"[housekeeping] Suricata logs exceeded {cfg.suricata_log_max_mb}MB; " f"freed {result['bytes_freed']} bytes", flush=True, ) except Exception as exc: print(f"[housekeeping] log cap failed: {exc}", file=sys.stderr, flush=True) if now >= next_retention: try: removed = store.purge_older_than(cfg.alert_retention_days) if removed: print(f"[db] purged {removed} expired incidents", flush=True) except Exception as exc: print(f"[housekeeping] alert retention failed: {exc}", file=sys.stderr, flush=True) next_retention = now + 3600 if now >= next_routeros_inventory: try: result = ndr_analyzer.sync_routeros_inventory() if result.get("assets"): print(f"[ndr] RouterOS inventory: {result['assets']} assets (DHCP={result['dhcp']}, ARP={result['arp']})", flush=True) except Exception as exc: print(f"[housekeeping] RouterOS inventory sync failed: {exc}", file=sys.stderr, flush=True) next_routeros_inventory = now + cfg.routeros_inventory_interval_seconds if next_rule_update is not None and now >= next_rule_update: result = rule_manager.update_vendor_rules() stream = sys.stdout if result.ok else sys.stderr print(f"[rules] scheduled update: {result.message}", file=stream, flush=True) next_rule_update = now + interval_seconds housekeeping_thread = threading.Thread(target=housekeeping, name="housekeeping", daemon=True) def request_stop(_signum=None, _frame=None): stop_event.set() signal.signal(signal.SIGTERM, request_stop) signal.signal(signal.SIGINT, request_stop) live_pipeline.start() analytics_cache.start() notifier.start() ndr_analyzer.start() watcher.start() housekeeping_thread.start() web.start() rc = 0 try: while not stop_event.is_set(): suricata_rc = suricata.poll() if suricata_rc is not None: print(f"[fatal] Suricata exited with rc={suricata_rc}", file=sys.stderr, flush=True) rc = suricata_rc or 4 break if not receiver.is_alive(): print("[fatal] Rust TZSP data-plane exited", file=sys.stderr, flush=True) rc = 5 break time.sleep(0.5) finally: stop_event.set() receiver.close() try: web.stop() except Exception: pass if suricata.poll() is None: suricata.terminate() try: suricata.wait(timeout=8) except subprocess.TimeoutExpired: suricata.kill() suricata.wait(timeout=3) try: os.remove("/run/suricata.pid") except FileNotFoundError: pass live_pipeline.stop() analytics_cache.stop() ndr_analyzer.stop() notifier.stop() redis_supervisor.stop() store.close() return rc if __name__ == "__main__": raise SystemExit(main())