from __future__ import annotations import base64 import errno import hashlib import hmac import ipaddress import json import mimetypes import os import queue import re import select import sys import socket import struct import threading import time import urllib.parse from collections import defaultdict, deque from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any, Callable from .config import Config from .auth import SessionAuth from .analytics_cache import AnalyticsSnapshotCache from .backup import BackupManager from .forensics import ForensicPcapRing from .live import EventBus, LiveEventPipeline, RedisUnavailableError, TrafficHistory, event_matches from .maintenance import clear_suricata_logs from .ndr import NDRAnalyzer, ThreatIntelManager from .routeros import RouterOSClient from .rules import RuleManager from .state import RuntimeStats from .store import AlertStore BASE_DIR = Path(__file__).resolve().parent TEMPLATE_PATH = BASE_DIR / "templates" / "index.html" STATIC_DIR = BASE_DIR / "static" DASHBOARD = TEMPLATE_PATH.read_text(encoding="utf-8") _TIMEOUT_RE = re.compile(r"^[1-9][0-9]{0,5}[smhdw]$") class _WebHTTPServer(ThreadingHTTPServer): allow_reuse_address = True daemon_threads = True def handle_error(self, request, client_address) -> None: exc = sys.exc_info()[1] if isinstance(exc, (BrokenPipeError, ConnectionResetError, ConnectionAbortedError)): return if isinstance(exc, OSError) and exc.errno in {errno.EPIPE, errno.ECONNRESET, errno.ECONNABORTED}: return super().handle_error(request, client_address) class MetricsAccessControl: """Pre-parsed ACL for the lightweight Prometheus endpoint.""" def __init__(self, config: Config) -> None: networks = [] for item in config.metrics_allowed_ips.split(","): item = item.strip() if not item: continue try: networks.append(ipaddress.ip_network(item, strict=False)) except ValueError as exc: raise ValueError(f"invalid METRICS_ALLOWED_IPS entry: {item}") from exc self.networks = tuple(networks) self.username = config.metrics_basic_auth_username self.password = config.metrics_basic_auth_password if bool(self.username) != bool(self.password): raise ValueError( "METRICS_BASIC_AUTH_USERNAME and METRICS_BASIC_AUTH_PASSWORD " "must either both be set or both be empty" ) @property def basic_auth_enabled(self) -> bool: return bool(self.username and self.password) def ip_allowed(self, client_ip: str) -> bool: try: address = ipaddress.ip_address(client_ip) except ValueError: return False if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped is not None: address = address.ipv4_mapped return any(address in network for network in self.networks if network.version == address.version) def basic_auth_allowed(self, authorization: str) -> bool: if not self.basic_auth_enabled: return True scheme, separator, encoded = authorization.partition(" ") if not separator or scheme.lower() != "basic" or not encoded.strip(): return False try: supplied = base64.b64decode(encoded.strip().encode("ascii"), validate=True) except (ValueError, UnicodeEncodeError): return False expected = f"{self.username}:{self.password}".encode("utf-8") return hmac.compare_digest(supplied, expected) class WebServer: def __init__( self, config: Config, store: AlertStore, health_provider: Callable[[], dict], stats: RuntimeStats | None = None, rule_manager: RuleManager | None = None, traffic_history: TrafficHistory | None = None, event_bus: EventBus | None = None, live_pipeline: LiveEventPipeline | None = None, routeros: RouterOSClient | None = None, analytics_cache: AnalyticsSnapshotCache | None = None, threat_intel: ThreatIntelManager | None = None, ndr_analyzer: NDRAnalyzer | None = None, backup_manager: BackupManager | None = None, 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 self.health_provider = health_provider self.stats = stats self.rule_manager = rule_manager self.traffic_history = traffic_history self.event_bus = event_bus self.live_pipeline = live_pipeline self.routeros = routeros self.analytics_cache = analytics_cache self.threat_intel = threat_intel self.ndr_analyzer = ndr_analyzer 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) self.server = _WebHTTPServer((config.web_bind, config.web_port), self._handler()) 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: # 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: history.update(self.event_bus.status()) 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: if self.analytics_cache is not None: payload = self.analytics_cache.get(window_seconds) elif self.traffic_history is not None: payload = self.traffic_history.analytics(window_seconds) else: payload = {"window_seconds": window_seconds, "events": 0, "timeline": []} return self._overlay_current_throughput(payload, window_seconds) def _overlay_current_throughput(self, payload: dict, window_seconds: int) -> dict: source = self.traffic_source if source is None or not hasattr(source, "overlay_current"): return payload try: return source.overlay_current(payload, window_seconds) except Exception: return payload def _current_throughput_payload(self, window_seconds: int) -> dict: source = self.traffic_source if source is None or not hasattr(source, "current_throughput"): return {"window_seconds": window_seconds, "current_bps": 0} try: return source.current_throughput(window_seconds) except Exception: return {"window_seconds": window_seconds, "current_bps": 0} 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() if len(attempts) >= 10: return False attempts.append(now) return True def _clear_login_attempts(self, client_ip: str) -> None: with self._login_lock: self._login_attempts.pop(client_ip, None) def _handler(self): outer = self store = self.store config = self.config stats = self.stats rule_manager = self.rule_manager traffic_history = self.traffic_history event_bus = self.event_bus routeros = self.routeros threat_intel = self.threat_intel ndr_analyzer = self.ndr_analyzer backup_manager = self.backup_manager forensic_pcap = self.forensic_pcap metrics_provider = self.metrics_provider metrics_access = self.metrics_access auth = self.auth class Handler(BaseHTTPRequestHandler): MAX_BODY = 1024 * 1024 server_version = "MikroSuricata/1.0" def do_GET(self): parsed = urllib.parse.urlparse(self.path) view_path = parsed.path.rstrip("/") or "/" if parsed.path == "/ws/live": self._websocket(parsed) return if parsed.path == "/metrics": if metrics_provider is None or metrics_access is None: self._send(404, b"metrics unavailable\n", "text/plain; charset=utf-8") return client_ip = self.client_address[0] if self.client_address else "" if not metrics_access.ip_allowed(client_ip): self._send(403, b"metrics access denied\n", "text/plain; charset=utf-8") return if not metrics_access.basic_auth_allowed(self.headers.get("Authorization", "")): self._send( 401, b"metrics authentication required\n", "text/plain; charset=utf-8", extra_headers={ "WWW-Authenticate": 'Basic realm="metrics", charset="UTF-8"' }, ) return try: payload = metrics_provider().encode("utf-8") except Exception: self._send(500, b"metrics scrape failed\n", "text/plain; charset=utf-8") return self._send( 200, payload, "text/plain; version=0.0.4; charset=utf-8", ) return if view_path in {"/", "/live", "/security", "/intelligence", "/blocks", "/reports", "/feeds", "/rules", "/system"}: self._send(200, DASHBOARD.encode("utf-8"), "text/html; charset=utf-8") return if parsed.path.startswith("/static/"): self._static(parsed.path) return if parsed.path == "/api/health": self._json(outer._healthcheck_payload()) return if parsed.path == "/api/auth/session": self._auth_session() return if parsed.path.startswith("/api/") and not self._require_session(): return if parsed.path == "/api/status": self._json(outer._status_payload()) return if parsed.path == "/api/summary": self._json(store.summary()) return if parsed.path == "/api/stats": self._json(outer._stats_payload()) return if parsed.path == "/api/config": self._json(config.public_dict()) return if parsed.path == "/api/alerts": query = urllib.parse.parse_qs(parsed.query) self._json({"alerts": store.recent(self._query_int(query, "limit", 100, 1, 500))}) return if parsed.path == "/api/traffic": if traffic_history is None: self._json({"events": [], "history": {"backend": "disabled"}}) return query = urllib.parse.parse_qs(parsed.query) window = self._query_int(query, "window", 3600, 60, config.traffic_retention_hours * 3600) now_ms = int(time.time() * 1000) try: events = traffic_history.search( limit=self._query_int(query, "limit", 250, 1, 2000), since_ms=now_ms - window * 1000, text=self._query_text(query, "q", 256), event_type=self._query_text(query, "type", 32), proto=self._query_text(query, "proto", 24), app_proto=self._query_text(query, "app", 48), direction=self._query_text(query, "direction", 24), ) except RedisUnavailableError as exc: self._json({"error": f"Traffic history unavailable: {exc}"}, status=503) return self._json({"events": events, "history": traffic_history.status()}) return if parsed.path == "/api/traffic/throughput": if traffic_history is None: self._json({"window_seconds": 3600, "bytes": 0, "timeline": []}) return query = urllib.parse.parse_qs(parsed.query) window = self._query_int(query, "window", 3600, 60, config.traffic_retention_hours * 3600) try: payload = traffic_history.throughput_analytics(window) self._json(outer._overlay_current_throughput(payload, window)) except RedisUnavailableError as exc: self._json({"error": f"Traffic throughput unavailable: {exc}"}, status=503) return if parsed.path == "/api/traffic/analytics": if traffic_history is None: self._json({"window_seconds": 3600, "events": 0, "timeline": []}) return query = urllib.parse.parse_qs(parsed.query) window = self._query_int(query, "window", 3600, 60, config.traffic_retention_hours * 3600) try: payload = outer._analytics_payload(window) except RedisUnavailableError as exc: self._json({"error": f"Traffic analytics unavailable: {exc}"}, status=503) return self._json(payload) return if parsed.path == "/api/ndr/summary": self._json({"summary": store.ndr_summary(), "analyzer": ndr_analyzer.status() if ndr_analyzer else {"enabled": False}}) return if parsed.path == "/api/ndr/incidents": query = urllib.parse.parse_qs(parsed.query) self._json({"incidents": store.recent_ndr_incidents(self._query_int(query, "limit", 100, 1, 500))}) return if parsed.path.startswith("/api/ndr/incidents/"): try: incident_id = int(parsed.path.rsplit("/", 1)[-1]) except ValueError: self._json({"error": "invalid incident id"}, status=400); return rows = [x for x in store.recent_ndr_incidents(500) if int(x.get("id", 0)) == incident_id] if not rows: self._json({"error": "incident not found"}, status=404); return self._json({"incident": rows[0], "events": store.ndr_incident_events(incident_id, 200)}) return if parsed.path == "/api/assets": query = urllib.parse.parse_qs(parsed.query) self._json({"assets": store.assets(self._query_int(query, "limit", 250, 1, 1000))}) return if parsed.path == "/api/threat-intel": query = urllib.parse.parse_qs(parsed.query) self._json({"iocs": store.list_iocs(self._query_int(query, "limit", 1000, 1, 5000)), "summary": store.ndr_summary()}) return if parsed.path == "/api/forensics/pcaps": files = self._pcap_files() max_bytes = config.forensic_pcap_max_total_mb * 1024 * 1024 self._json({"files": [ {"name": path.name, "size_bytes": path.stat().st_size, "modified_at": path.stat().st_mtime} for path in files ], "mode": config.forensic_pcap_mode, "max_bytes": max_bytes}) return if parsed.path == "/api/forensics/pcap": query = urllib.parse.parse_qs(parsed.query) name = self._query_text(query, "name", 255) allowed = {path.name: path for path in self._pcap_files()} path = allowed.get(name) if path is None: self._json({"error": "PCAP not found"}, status=404); return self._send_file(path, "application/vnd.tcpdump.pcap") return if parsed.path == "/api/rules/intelligence": query = urllib.parse.parse_qs(parsed.query) self._json(store.rule_intelligence(self._query_int(query, "hours", 24, 1, 720), self._query_int(query, "limit", 100, 1, 500))) return if parsed.path == "/api/audit": query = urllib.parse.parse_qs(parsed.query) self._json({"events": store.audit_events(self._query_int(query, "limit", 200, 1, 1000))}) return if parsed.path == "/api/system/backups": self._json({"backups": backup_manager.list()}) return if parsed.path == "/api/system/backup": query = urllib.parse.parse_qs(parsed.query) name = self._query_text(query, "name", 255) path = backup_manager.path(name) if path is None: self._json({"error": "backup not found"}, status=404); return self._send_file(path, "application/gzip") return if parsed.path == "/api/admin/rules/snapshots": if not self._require_admin(): return self._json({"snapshots": rule_manager.list_snapshots() if rule_manager else []}) return if parsed.path == "/api/blocks": blocks = routeros.list_blocks() if routeros is not None and routeros.configured else [] self._json({ "configured": bool(routeros and routeros.configured), "address_list": config.routeros_address_list, "blocks": blocks, }) return if parsed.path == "/api/rules/sources": if rule_manager is None: self._json({"error": "rule manager unavailable"}, status=503) else: payload = rule_manager.source_catalog() self._json(payload, status=200 if payload.get("ok") else 503) return if parsed.path == "/api/rules/merged": if rule_manager is None: self._json({"error": "rule manager unavailable"}, status=503) else: query = urllib.parse.parse_qs(parsed.query) payload = rule_manager.merged_rules( self._query_text(query, "q", 300), self._query_int(query, "offset", 0, 0, 100000000), self._query_int(query, "limit", 1000, 1, 5000), ) self._json(payload, status=200 if payload.get("ok") else 404) return if parsed.path == "/api/admin/rules": if not self._require_admin(): return if rule_manager is None: self._json({"error": "rule manager unavailable"}, status=503) else: self._json(rule_manager.content()) return if parsed.path == "/api/admin/rules/sources": if not self._require_admin(): return if rule_manager is None: self._json({"error": "rule manager unavailable"}, status=503) else: payload = rule_manager.source_catalog() self._json(payload, status=200 if payload.get("ok") else 503) return if parsed.path == "/api/admin/rules/sources/queue": if not self._require_admin(): return if rule_manager is None: self._json({"error": "rule manager unavailable"}, status=503) else: self._json(rule_manager.source_queue_status()) return self._json({"error": "not found"}, status=404) def do_POST(self): parsed = urllib.parse.urlparse(self.path) if parsed.path == "/api/auth/login": self._login() return if parsed.path == "/api/auth/logout": if not self._require_session(csrf=True): return self._audit("auth.logout") auth.delete_session_from_cookie(self.headers.get("Cookie", "")) self._json( {"ok": True}, extra_headers={"Set-Cookie": auth.clear_cookie_header()}, ) return if not parsed.path.startswith("/api/admin/"): self._json({"error": "not found"}, status=404) return if not self._require_admin(csrf=True): return body = self._read_json() if body is None: return if parsed.path == "/api/admin/system/backups/create": try: item = backup_manager.create(str(body.get("label") or "manual")) except Exception as exc: self._audit("backup.create", result="error", details={"error": str(exc)}) self._json({"error": f"backup failed: {exc}"}, status=500); return self._audit("backup.create", target=str(item.get("id") or ""), details={"size_bytes": item.get("size_bytes")}) self._json({"ok": True, "backup": item, "message": "Persistent IDS state backup created"}) return if parsed.path == "/api/admin/system/backups/delete": name = str(body.get("id") or "") removed = backup_manager.delete(name) self._audit("backup.delete", target=name, result="ok" if removed else "not-found") self._json({"ok": removed, "message": "Backup deleted" if removed else "Backup not found"}, status=200 if removed else 404) return if parsed.path == "/api/admin/rules/threshold": if rule_manager is None: self._json({"error": "rule manager unavailable"}, status=503); return try: sid = int(body.get("sid")) result = rule_manager.add_threshold(sid, threshold_type=str(body.get("type") or "limit"), track=str(body.get("track") or "by_src"), count=int(body.get("count") or 5), seconds=int(body.get("seconds") or 60)) except (TypeError, ValueError) as exc: self._json({"error": str(exc)}, status=400); return self._audit("rules.threshold", target=str(sid), result="ok" if result.ok else "error", details={"message": result.message}) self._json({"ok": result.ok, "message": result.message}, status=200 if result.ok else 400) return if parsed.path == "/api/admin/rules/snapshot": if rule_manager is None: self._json({"error": "rule manager unavailable"}, status=503); return result = rule_manager.create_snapshot(str(body.get("reason") or "manual")) self._audit("rules.snapshot", result="ok" if result.ok else "error", details={"message": result.message}) self._json({"ok": result.ok, "message": result.message}, status=200 if result.ok else 400) return if parsed.path == "/api/admin/rules/rollback": if rule_manager is None: self._json({"error": "rule manager unavailable"}, status=503); return snapshot_id = str(body.get("id") or "") result = rule_manager.rollback_snapshot(snapshot_id) self._audit("rules.rollback", target=snapshot_id, result="ok" if result.ok else "error", details={"message": result.message}) self._json({"ok": result.ok, "message": result.message}, status=200 if result.ok else 400) return if parsed.path == "/api/admin/alerts/clear": count = store.clear_alerts() self._audit("alerts.clear", details={"rows": count}) self._json({"ok": True, "message": f"Deleted {count} incident rows"}) return if parsed.path == "/api/admin/logs/clear": result = clear_suricata_logs(config.eve_path) self._json({"ok": True, "message": f"Cleared {len(result['files'])} log files; freed {result['bytes_freed']} bytes", **result}) return if parsed.path == "/api/admin/database/vacuum": store.vacuum() self._audit("database.vacuum") self._json({"ok": True, "message": "SQLite VACUUM completed"}) return if parsed.path == "/api/admin/runtime/reset": if stats is None: self._json({"error": "runtime stats unavailable"}, status=503) else: stats.reset() self._audit("runtime.reset") self._json({"ok": True, "message": "Runtime counters reset"}) return if parsed.path == "/api/admin/traffic/clear": try: count = traffic_history.clear() if traffic_history is not None else 0 except RedisUnavailableError as exc: self._json({"error": f"Traffic history unavailable: {exc}"}, status=503) return snapshots = store.clear_traffic_snapshots() self._audit("traffic.clear", details={"events": count, "snapshots": snapshots}) self._json({"ok": True, "message": f"Cleared {count} archived/live traffic events and {snapshots} chart snapshots"}) return if parsed.path == "/api/admin/ndr/incidents/status": try: incident_id = int(body.get("id")) changed = store.set_ndr_incident_status(incident_id, str(body.get("status") or "")) except (TypeError, ValueError) as exc: self._json({"error": str(exc)}, status=400); return self._audit("incident.status", target=str(incident_id), result="ok" if changed else "not-found", details={"status": str(body.get("status") or "")}) self._json({"ok": changed, "message": "Incident status updated" if changed else "Incident not found"}, status=200 if changed else 404) return if parsed.path == "/api/admin/threat-intel/add": try: ioc_id = store.add_ioc( str(body.get("indicator") or ""), str(body.get("type") or ""), source=str(body.get("source") or "manual"), confidence=int(body.get("confidence") or 80), severity=int(body.get("severity") or 1), note=str(body.get("note") or ""), expires_at=str(body.get("expires_at") or "") or None, ) except (ValueError, TypeError) as exc: self._json({"error": str(exc)}, status=400); return sync = threat_intel.sync_suricata_datasets() if threat_intel else {} reload_result = rule_manager.reload() if rule_manager is not None else None self._audit("ioc.add", target=str(ioc_id), details={"type": str(body.get("type") or ""), "indicator": str(body.get("indicator") or "")[:300]}) self._json({"ok": True, "id": ioc_id, "datasets": sync, "message": "IOC saved and Suricata datasets refreshed" + (f"; {reload_result.message}" if reload_result else "")}) return if parsed.path == "/api/admin/threat-intel/import": text = str(body.get("text") or "") if len(text) > 500000: self._json({"error": "IOC import is too large"}, status=413); return added = 0; errors = [] for line_no, raw in enumerate(text.splitlines(), 1): raw = raw.strip() if not raw or raw.startswith("#"): continue parts = [x.strip() for x in re.split(r"[,;\t]", raw, maxsplit=4)] if len(parts) >= 2 and parts[0].lower() in {"ip","domain","sha256","ja3","ja4","hassh"}: kind, indicator = parts[0].lower(), parts[1] confidence = int(parts[2]) if len(parts)>2 and parts[2].isdigit() else 80 source = parts[3] if len(parts)>3 and parts[3] else "import" note = parts[4] if len(parts)>4 else "" else: indicator = parts[0]; kind = self._guess_ioc_type(indicator); confidence=80; source="import"; note="" try: store.add_ioc(indicator, kind, source=source, confidence=confidence, note=note); added += 1 except (ValueError, TypeError) as exc: errors.append(f"line {line_no}: {exc}") if len(errors) >= 20: break sync = threat_intel.sync_suricata_datasets() if threat_intel else {} reload_result = rule_manager.reload() if rule_manager is not None and added else None self._audit("ioc.import", details={"added": added, "rejected": len(errors)}) self._json({"ok": True, "added": added, "errors": errors, "datasets": sync, "message": f"Imported {added} IOC entries" + (f"; {reload_result.message}" if reload_result else "")}) return if parsed.path == "/api/admin/threat-intel/delete": try: ioc_id = int(body.get("id")) except (TypeError, ValueError): self._json({"error": "valid IOC id is required"}, status=400); return removed = store.remove_ioc(ioc_id) sync = threat_intel.sync_suricata_datasets() if threat_intel else {} reload_result = rule_manager.reload() if rule_manager is not None and removed else None self._audit("ioc.delete", target=str(ioc_id), result="ok" if removed else "not-found") self._json({"ok": removed, "datasets": sync, "message": "IOC removed" if removed else "IOC not found"}, status=200 if removed else 404) return if parsed.path == "/api/admin/blocks/add": self._manual_block(body) return if parsed.path == "/api/admin/blocks/remove": self._manual_unblock(body) return if parsed.path.startswith("/api/admin/rules/"): if rule_manager is None: self._json({"error": "rule manager unavailable"}, status=503) return if parsed.path == "/api/admin/rules/custom": result = rule_manager.replace_custom_rules(str(body.get("content", ""))) elif parsed.path == "/api/admin/rules/thresholds": result = rule_manager.replace_threshold_config(str(body.get("content", ""))) elif parsed.path == "/api/admin/rules/suppress": try: sid = int(body.get("sid")) except (TypeError, ValueError): self._json({"error": "valid SID is required"}, status=400) return result = rule_manager.suppress_sid(sid, str(body.get("track") or ""), body.get("ip")) elif parsed.path == "/api/admin/rules/reload": result = rule_manager.reload() elif parsed.path == "/api/admin/rules/update": result = rule_manager.update_vendor_rules() elif parsed.path == "/api/admin/rules/sources/refresh": result = rule_manager.refresh_source_catalog() elif parsed.path == "/api/admin/rules/sources/queue": sources = body.get("sources") if not isinstance(sources, list): self._json({"error": "sources must be an array"}, status=400) return result = rule_manager.queue_sources([str(item) for item in sources]) elif parsed.path == "/api/admin/rules/sources/add": result = rule_manager.add_manual_source( str(body.get("name") or ""), str(body.get("url") or ""), body.get("no_checksum", True) is not False, ) elif parsed.path == "/api/admin/rules/sources/remove": result = rule_manager.remove_manual_source(str(body.get("source") or "")) elif parsed.path in {"/api/admin/rules/sources/enable", "/api/admin/rules/sources/disable"}: result = rule_manager.set_source_enabled( str(body.get("source") or ""), parsed.path.endswith("/enable") ) else: self._json({"error": "not found"}, status=404) return self._audit("rules.action", target=parsed.path.removeprefix("/api/admin/rules/"), result="ok" if result.ok else "error", details={"message": result.message}) self._json({"ok": result.ok, "message": result.message}, status=200 if result.ok else 400) return self._json({"error": "not found"}, status=404) @staticmethod def _guess_ioc_type(value: str) -> str: text = str(value or "").strip() try: ipaddress.ip_address(text); return "ip" except ValueError: pass lower = text.lower() if len(lower) == 64 and all(ch in "0123456789abcdef" for ch in lower): return "sha256" if "_" in lower and len(lower) >= 20: return "ja4" if len(lower) == 32 and all(ch in "0123456789abcdef" for ch in lower): return "ja3" return "domain" def _manual_block(self, body: dict) -> None: if routeros is None or not routeros.configured: self._json({"error": "RouterOS REST integration is not configured"}, status=503) return address = str(body.get("address") or "").strip() try: address = str(ipaddress.ip_address(address)) except ValueError: self._json({"error": "valid IPv4 or IPv6 address is required"}, status=400) return timeout_value = str(body.get("timeout") or config.block_timeout).strip().lower() if not _TIMEOUT_RE.fullmatch(timeout_value): self._json({"error": "timeout must look like 30m, 1h, 2d or 1w"}, status=400) return comment = str(body.get("comment") or "Manual dashboard block").strip()[:180] result = routeros.block_ip(address, timeout_value, comment) if result.success and forensic_pcap is not None: try: forensic_pcap.capture_target(address, label="manual") except Exception as exc: print(f"[forensics] manual block PCAP capture failed: {exc}", flush=True) self._json({"ok": result.success, "message": result.message}, status=200 if result.success else 502) def _manual_unblock(self, body: dict) -> None: if routeros is None or not routeros.configured: self._json({"error": "RouterOS REST integration is not configured"}, status=503) return address = str(body.get("address") or "").strip() try: address = str(ipaddress.ip_address(address)) except ValueError: self._json({"error": "valid IPv4 or IPv6 address is required"}, status=400) return result = routeros.unblock_ip(address) self._json({"ok": result.success, "message": result.message}, status=200 if result.success else 502) def _websocket(self, parsed: urllib.parse.ParseResult) -> None: if event_bus is None or traffic_history is None: self.send_error(503, "live event bus unavailable") return if not self._require_session(websocket=True): return if self.headers.get("Upgrade", "").lower() != "websocket": self.send_error(426, "WebSocket upgrade required") return if self.headers.get("Sec-WebSocket-Version", "13").strip() != "13": self.send_response(426, "Unsupported WebSocket version") self.send_header("Sec-WebSocket-Version", "13") self.end_headers() return origin = self.headers.get("Origin", "").strip() host = self.headers.get("Host", "").strip().lower() if origin: parsed_origin = urllib.parse.urlparse(origin) if parsed_origin.scheme not in {"http", "https"} or parsed_origin.netloc.lower() != host: self.send_error(403, "WebSocket origin is not allowed") return key = self.headers.get("Sec-WebSocket-Key", "").strip() if not key: self.send_error(400, "missing Sec-WebSocket-Key") return accept = base64.b64encode( hashlib.sha1((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode("ascii")).digest() ).decode("ascii") self.send_response(101, "Switching Protocols") self.send_header("Upgrade", "websocket") self.send_header("Connection", "Upgrade") self.send_header("Sec-WebSocket-Accept", accept) self.end_headers() self.wfile.flush() query = urllib.parse.parse_qs(parsed.query) window = self._query_int(query, "window", 3600, 60, config.traffic_retention_hours * 3600) stream = str(query.get("stream", ["0"])[0]).strip().lower() in {"1", "true", "yes", "on"} filters = { "text": self._query_text(query, "q", 256), "event_type": self._query_text(query, "type", 32), "proto": self._query_text(query, "proto", 24), "app_proto": self._query_text(query, "app", 48), "direction": self._query_text(query, "direction", 24), } target = event_bus.subscribe() if stream else None try: recent = [] if stream: try: recent = traffic_history.search( limit=100, since_ms=int((time.time() - min(window, config.traffic_retention_hours * 3600)) * 1000), text=filters["text"], event_type=filters["event_type"], proto=filters["proto"], app_proto=filters["app_proto"], direction=filters["direction"], ) recent.reverse() except RedisUnavailableError: recent = [] try: bootstrap_analytics = outer._analytics_payload(window) except RedisUnavailableError as exc: bootstrap_analytics = { "window_seconds": window, "snapshot_loading": True, "snapshot_error": f"Redis unavailable: {exc}", "timeline": [], } bootstrap = { "type": "bootstrap", "data": { "events": recent, "status": outer._status_payload(), "analytics": bootstrap_analytics, "stream": stream, }, } self._ws_send_json(bootstrap) last_status = time.monotonic() last_analytics = last_status last_throughput = 0.0 while True: if not self._ws_client_control(): return if target is not None: # Render-rate protection: collect a short burst and emit at most # four event frames per second. Under heavy traffic the bounded # EventBus queue deliberately drops browser-only updates rather # than allowing the UI client to back-pressure packet capture. cycle_started = time.monotonic() batch = [] try: batch.append(target.get(timeout=0.25)) except queue.Empty: pass while batch and len(batch) < 512: try: batch.append(target.get_nowait()) except queue.Empty: break if batch: remaining = 0.25 - (time.monotonic() - cycle_started) if remaining > 0: time.sleep(remaining) latest: dict[str, dict] = {} for event in batch: if not event_matches(event, **filters): continue event_id = str(event.get("id") or "") if event_id in latest: latest.pop(event_id, None) latest[event_id or f"anon-{len(latest)}"] = event payload = list(latest.values())[-120:] if payload: self._ws_send_json({"type": "events", "data": payload}) else: time.sleep(0.25) now = time.monotonic() if now - last_throughput >= 1: self._ws_send_json({"type": "throughput", "data": outer._current_throughput_payload(window)}) last_throughput = now if now - last_status >= 30: self._ws_send_json({"type": "status", "data": outer._status_payload()}) last_status = now if now - last_analytics >= 10: try: analytics = outer._analytics_payload(window) except RedisUnavailableError as exc: analytics = { "window_seconds": window, "snapshot_loading": True, "snapshot_error": f"Redis unavailable: {exc}", "timeline": [], } self._ws_send_json({"type": "analytics", "data": analytics}) last_analytics = now except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError, OSError): return finally: if target is not None: event_bus.unsubscribe(target) def _ws_send_json(self, obj: dict) -> None: payload = json.dumps(obj, ensure_ascii=False, separators=(",", ":")).encode("utf-8") self._ws_send_frame(0x1, payload) def _ws_send_frame(self, opcode: int, payload: bytes = b"") -> None: header = bytearray([0x80 | (opcode & 0x0F)]) length = len(payload) if length < 126: header.append(length) elif length <= 0xFFFF: header.append(126) header.extend(struct.pack("!H", length)) else: header.append(127) header.extend(struct.pack("!Q", length)) self.connection.sendall(bytes(header) + payload) def _ws_client_control(self) -> bool: readable, _, _ = select.select([self.connection], [], [], 0) if not readable: return True try: head = self._recv_exact(2) if not head: return False fin = bool(head[0] & 0x80) rsv = head[0] & 0x70 opcode = head[0] & 0x0F masked = bool(head[1] & 0x80) length = head[1] & 0x7F if rsv or not fin or not masked or opcode not in {0x1, 0x2, 0x8, 0x9, 0xA}: self._ws_send_frame(0x8, struct.pack("!H", 1002)) return False if length == 126: length = struct.unpack("!H", self._recv_exact(2))[0] elif length == 127: length = struct.unpack("!Q", self._recv_exact(8))[0] if opcode >= 0x8 and length > 125: self._ws_send_frame(0x8, struct.pack("!H", 1002)) return False if length > 1024 * 1024: self._ws_send_frame(0x8, struct.pack("!H", 1009)) return False mask = self._recv_exact(4) payload = self._recv_exact(length) if length else b"" if payload: payload = bytes(value ^ mask[index % 4] for index, value in enumerate(payload)) if opcode == 0x8: self._ws_send_frame(0x8, payload[:125]) return False if opcode == 0x9: self._ws_send_frame(0xA, payload[:125]) return True except (OSError, socket.timeout, struct.error): return False def _recv_exact(self, length: int) -> bytes: if length <= 0: return b"" previous_timeout = self.connection.gettimeout() self.connection.settimeout(0.5) try: chunks = bytearray() while len(chunks) < length: chunk = self.connection.recv(length - len(chunks)) if not chunk: raise ConnectionResetError("WebSocket peer closed") chunks.extend(chunk) return bytes(chunks) finally: self.connection.settimeout(previous_timeout) def _pcap_files(self) -> list[Path]: root = Path(config.eve_path).resolve().parent try: files = [ p for pattern in ("alert*.pcap*", "block-*.pcap") for p in root.glob(pattern) if p.is_file() and p.resolve().parent == root ] except OSError: return [] return sorted(files, key=lambda p: p.stat().st_mtime, reverse=True)[:max(32, config.forensic_pcap_max_files)] def _send_file(self, path: Path, content_type: str) -> None: try: size = path.stat().st_size handle = path.open("rb") except OSError: self._json({"error": "file unavailable"}, status=404) return try: self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(size)) self.send_header("Content-Disposition", f'attachment; filename="{path.name}"') self.send_header("Cache-Control", "no-store") self.send_header("X-Content-Type-Options", "nosniff") self.send_header("X-Frame-Options", "DENY") self.end_headers() with handle: while True: chunk = handle.read(1024 * 1024) if not chunk: break self.wfile.write(chunk) except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError): self.close_connection = True handle.close() return except OSError as exc: if exc.errno in {errno.EPIPE, errno.ECONNRESET, errno.ECONNABORTED}: self.close_connection = True handle.close() return handle.close() raise def _static(self, request_path: str) -> None: relative = request_path.removeprefix("/static/") try: path = (STATIC_DIR / relative).resolve() path.relative_to(STATIC_DIR.resolve()) except (ValueError, OSError): self.send_error(404) return if not path.is_file(): self.send_error(404) return content_type = mimetypes.guess_type(str(path))[0] or "application/octet-stream" data = path.read_bytes() self._send(200, data, content_type, cache="public, max-age=3600") def _auth_session(self) -> None: session = auth.session_from_cookie(self.headers.get("Cookie", ""), touch=True) self._json({ "auth_enabled": auth.enabled, "authenticated": session is not None, "username": session.get("username") if session else None, "csrf_token": session.get("csrf_token") if session else None, "expires_at": session.get("expires_at") if session else None, "default_username": config.admin_username, }) def _login(self) -> None: if not auth.enabled: self._json( {"error": "login is not configured; set ADMIN_PASSWORD"}, status=503, ) return if not self._same_origin(): self._json({"error": "origin is not allowed"}, status=403) return client_ip = self.client_address[0] if self.client_address else "unknown" if not outer._login_allowed(client_ip): self._json({"error": "too many login attempts; try again later"}, status=429) return body = self._read_json() if body is None: return username = str(body.get("username") or "")[:128] password = str(body.get("password") or "")[:1024] if not auth.authenticate(username, password): store.audit(username, "auth.login", result="denied", remote_ip=client_ip) time.sleep(0.15) self._json({"error": "invalid username or password"}, status=401) return outer._clear_login_attempts(client_ip) token, session = auth.create_session(username) store.audit(username, "auth.login", result="ok", remote_ip=client_ip) self._json( { "ok": True, "username": session["username"], "csrf_token": session["csrf_token"], "expires_at": session["expires_at"], }, extra_headers={"Set-Cookie": auth.cookie_header(token)}, ) def _require_admin(self, *, csrf: bool = False) -> bool: if not auth.enabled: self._json({"error": "admin login is disabled; set ADMIN_PASSWORD"}, status=403) return False return self._require_session(csrf=csrf) def _require_session(self, *, csrf: bool = False, websocket: bool = False) -> bool: if not auth.enabled: return True session = auth.session_from_cookie(self.headers.get("Cookie", ""), touch=True) if session is None: if websocket: self.send_error(401, "authentication required") else: self._json({"error": "authentication required"}, status=401) return False if csrf: supplied = self.headers.get("X-CSRF-Token", "") expected = str(session.get("csrf_token") or "") if not supplied or not hmac.compare_digest(supplied, expected): self._json({"error": "invalid CSRF token"}, status=403) return False if not self._same_origin(): self._json({"error": "origin is not allowed"}, status=403) return False return True def _audit(self, action: str, *, target: str = "", result: str = "ok", details: dict | None = None) -> None: try: session = auth.session_from_cookie(self.headers.get("Cookie", ""), touch=False) username = str(session.get("username") or "") if session else "" remote_ip = self.client_address[0] if self.client_address else "" store.audit(username, action, target=target, result=result, remote_ip=remote_ip, details=details or {}) except Exception: pass def _same_origin(self) -> bool: origin = self.headers.get("Origin", "").strip() if not origin: return True host = self.headers.get("Host", "").strip().lower() parsed_origin = urllib.parse.urlparse(origin) return parsed_origin.scheme in {"http", "https"} and parsed_origin.netloc.lower() == host def _read_json(self): try: length = int(self.headers.get("Content-Length", "0")) except ValueError: length = 0 if length < 0 or length > self.MAX_BODY: self._json({"error": "request body too large"}, status=413) return None raw = self.rfile.read(length) if length else b"{}" try: data = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): self._json({"error": "invalid JSON body"}, status=400) return None if not isinstance(data, dict): self._json({"error": "JSON body must be an object"}, status=400) return None return data @staticmethod def _query_text(query: dict, name: str, max_len: int) -> str: return str(query.get(name, [""])[0])[:max_len] @staticmethod def _query_int(query: dict, name: str, default: int, minimum: int, maximum: int) -> int: try: value = int(query.get(name, [str(default)])[0]) except (TypeError, ValueError): value = default return min(max(value, minimum), maximum) def _json(self, obj, status: int = 200, extra_headers: dict[str, str] | None = None): data = json.dumps(obj, ensure_ascii=False).encode("utf-8") self._send(status, data, "application/json; charset=utf-8", extra_headers=extra_headers) def _send( self, status: int, data: bytes, content_type: str, cache: str = "no-store", extra_headers: dict[str, str] | None = None, ): try: self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(data))) self.send_header("Cache-Control", cache) self.send_header("X-Content-Type-Options", "nosniff") self.send_header("X-Frame-Options", "DENY") self.send_header("Referrer-Policy", "no-referrer") self.send_header( "Content-Security-Policy", "default-src 'self'; script-src 'self'; style-src 'self'; connect-src 'self' ws: wss:; img-src 'self' data:; frame-ancestors 'none'; base-uri 'none'", ) for name, value in (extra_headers or {}).items(): self.send_header(name, value) self.end_headers() self.wfile.write(data) except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError): self.close_connection = True return except OSError as exc: if exc.errno in {errno.EPIPE, errno.ECONNRESET, errno.ECONNABORTED}: self.close_connection = True return raise def log_message(self, fmt, *args): return return Handler def start(self) -> None: self.thread.start() print(f"[web] dashboard on http://{self.config.web_bind}:{self.config.web_port}", flush=True) def stop(self) -> None: self.server.shutdown() self.server.server_close()