from __future__ import annotations import collections import functools import hashlib import ipaddress import json import queue import socket import threading import time import urllib.parse import uuid from dataclasses import dataclass from datetime import datetime, timezone from typing import Any, Iterable SUPPORTED_EVENT_TYPES = { "flow", "dns", "mdns", "http", "http2", "doh2", "tls", "alert", "fileinfo", "anomaly", "ssh", "rdp", "smb", "quic", "dhcp", "arp", "ike", "mqtt", "ftp", "ftp_data", "smtp", "websocket", "nfs", "tftp", "dcerpc", "krb5", "snmp", "rfb", "sip", "ldap", "pop3", } def _utc_now() -> str: return datetime.now(timezone.utc).isoformat() def _epoch_ms(value: Any) -> int: if value: try: parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return int(parsed.timestamp() * 1000) except (TypeError, ValueError): pass return int(time.time() * 1000) def _safe_int(value: Any, default: int = 0) -> int: try: return int(value) except (TypeError, ValueError): return default def _text(value: Any, max_len: int = 512) -> str: if value is None: return "" return str(value)[:max_len] def _parse_networks(value: str) -> list[ipaddress._BaseNetwork]: result: list[ipaddress._BaseNetwork] = [] for raw in (value or "").split(","): raw = raw.strip() if not raw: continue try: result.append(ipaddress.ip_network(raw, strict=False)) except ValueError: continue return result class TrafficNormalizer: """Convert large EVE objects to small, stable records for UI/history.""" def __init__(self, monitored_networks: str) -> None: self.networks = _parse_networks(monitored_networks) def normalize(self, event: dict[str, Any], **extra: Any) -> dict[str, Any] | None: event_type = _text(event.get("event_type"), 32).lower() if event_type not in SUPPORTED_EVENT_TYPES: return None src_ip = _text(event.get("src_ip"), 64) dst_ip = _text(event.get("dest_ip"), 64) ether = event.get("ether") if isinstance(event.get("ether"), dict) else {} record: dict[str, Any] = { "id": uuid.uuid4().hex, "timestamp": _text(event.get("timestamp"), 64) or _utc_now(), "ts_ms": _epoch_ms(event.get("timestamp")), "type": event_type, "flow_id": _text(event.get("flow_id"), 48), "community_id": _text(event.get("community_id"), 96), "tx_id": _text(event.get("tx_id"), 48), "vlan": event.get("vlan") if isinstance(event.get("vlan"), list) else [], "src_ip": src_ip, "src_port": _safe_int(event.get("src_port")) or None, "dest_ip": dst_ip, "dest_port": _safe_int(event.get("dest_port")) or None, "proto": _text(event.get("proto"), 24).upper(), "app_proto": _text(event.get("app_proto"), 48).lower(), "pkt_src": _text(event.get("pkt_src"), 64), "ether_src": _text(ether.get("src_mac"), 32), "ether_dest": _text(ether.get("dest_mac"), 32), "direction": self._direction(src_ip, dst_ip), } flow = event.get("flow") if isinstance(event.get("flow"), dict) else {} to_server = _safe_int(flow.get("bytes_toserver")) to_client = _safe_int(flow.get("bytes_toclient")) record["bytes_toserver"] = max(to_server, 0) record["bytes_toclient"] = max(to_client, 0) record["bytes"] = record["bytes_toserver"] + record["bytes_toclient"] record["bytes_out"] = record["bytes_toserver"] if record["direction"] == "outbound" else record["bytes_toclient"] if record["direction"] == "inbound" else 0 record["bytes_in"] = record["bytes_toclient"] if record["direction"] == "outbound" else record["bytes_toserver"] if record["direction"] == "inbound" else 0 record["packets"] = max(_safe_int(flow.get("pkts_toserver")), 0) + max( _safe_int(flow.get("pkts_toclient")), 0 ) if flow: record["flow_state"] = _text(flow.get("state"), 32) record["flow_reason"] = _text(flow.get("reason"), 64) if event_type == "dns": dns = event.get("dns") if isinstance(event.get("dns"), dict) else {} queries = dns.get("queries") if isinstance(dns.get("queries"), list) else [] first_query = queries[0] if queries and isinstance(queries[0], dict) else {} record["dns_query"] = _text( dns.get("rrname") or dns.get("query") or first_query.get("rrname"), 255 ) record["dns_type"] = _text( dns.get("rrtype") or first_query.get("rrtype") or dns.get("type"), 32 ) record["dns_rcode"] = _text(dns.get("rcode"), 32) record["dns_message_type"] = _text(dns.get("type"), 24) elif event_type == "http": http = event.get("http") if isinstance(event.get("http"), dict) else {} record["http_host"] = _text(http.get("hostname") or http.get("host"), 255) record["http_url"] = _text(http.get("url"), 512) record["http_method"] = _text(http.get("http_method"), 16) record["http_status"] = _safe_int(http.get("status")) or None record["http_user_agent"] = _text(http.get("http_user_agent"), 255) elif event_type == "tls": tls = event.get("tls") if isinstance(event.get("tls"), dict) else {} record["tls_sni"] = _text(tls.get("sni"), 255) record["tls_subject"] = _text(tls.get("subject"), 255) record["tls_issuer"] = _text(tls.get("issuerdn"), 255) record["tls_version"] = _text(tls.get("version"), 32) record["tls_fingerprint"] = _text(tls.get("fingerprint"), 160) record["tls_alpn"] = _text(tls.get("alpn") or tls.get("next_protocol"), 96) record["tls_ja3"] = _text((tls.get("ja3") or {}).get("hash") if isinstance(tls.get("ja3"), dict) else tls.get("ja3"), 96) record["tls_ja4"] = _text((tls.get("ja4") or {}).get("hash") if isinstance(tls.get("ja4"), dict) else tls.get("ja4"), 128) elif event_type == "alert": alert = event.get("alert") if isinstance(event.get("alert"), dict) else {} record.update( { "signature_id": _safe_int(alert.get("signature_id")) or None, "signature": _text(alert.get("signature"), 300), "category": _text(alert.get("category"), 160), "severity": _safe_int(alert.get("severity")) or None, "action": _text(alert.get("action"), 48), } ) elif event_type == "fileinfo": fileinfo = event.get("fileinfo") if isinstance(event.get("fileinfo"), dict) else {} record["filename"] = _text(fileinfo.get("filename"), 255) record["file_size"] = _safe_int(fileinfo.get("size")) or None record["file_state"] = _text(fileinfo.get("state"), 32) record["file_md5"] = _text(fileinfo.get("md5"), 64) record["file_sha1"] = _text(fileinfo.get("sha1"), 64) record["file_sha256"] = _text(fileinfo.get("sha256"), 96) elif event_type == "anomaly": anomaly = event.get("anomaly") if isinstance(event.get("anomaly"), dict) else {} record["anomaly_event"] = _text(anomaly.get("event"), 160) record["anomaly_layer"] = _text(anomaly.get("layer"), 64) elif event_type == "ssh": ssh = event.get("ssh") if isinstance(event.get("ssh"), dict) else {} client = ssh.get("client") if isinstance(ssh.get("client"), dict) else {} server = ssh.get("server") if isinstance(ssh.get("server"), dict) else {} client_hassh = client.get("hassh") if isinstance(client.get("hassh"), dict) else {} server_hassh = server.get("hassh") if isinstance(server.get("hassh"), dict) else {} record["ssh_client"] = _text( client.get("software_version") or ssh.get("software_client"), 160 ) record["ssh_server"] = _text( server.get("software_version") or ssh.get("software_server"), 160 ) record["ssh_proto"] = _text( client.get("proto_version") or server.get("proto_version") or ssh.get("proto_version"), 32 ) record["ssh_hassh_client"] = _text(client_hassh.get("hash"), 96) record["ssh_hassh_server"] = _text(server_hassh.get("hash"), 96) elif event_type == "rdp": rdp = event.get("rdp") if isinstance(event.get("rdp"), dict) else {} client = rdp.get("client") if isinstance(rdp.get("client"), dict) else {} record["rdp_event_type"] = _text(rdp.get("event_type"), 48) record["rdp_cookie"] = _text(rdp.get("cookie"), 160) record["rdp_protocol"] = _text(rdp.get("protocol"), 48) record["rdp_client_name"] = _text(client.get("client_name"), 160) record["rdp_client_build"] = _text(client.get("build"), 160) if not record["tx_id"]: record["tx_id"] = _text(rdp.get("tx_id"), 48) elif event_type == "smb": smb = event.get("smb") if isinstance(event.get("smb"), dict) else {} ntlm = smb.get("ntlmssp") if isinstance(smb.get("ntlmssp"), dict) else {} record["smb_command"] = _text(smb.get("command"), 96) record["smb_share"] = _text(smb.get("share"), 160) record["smb_dialect"] = _text(smb.get("dialect"), 64) record["smb_filename"] = _text(smb.get("filename"), 255) record["smb_status"] = _text(smb.get("status"), 96) record["smb_client_guid"] = _text(smb.get("client_guid"), 96) record["smb_user"] = _text(ntlm.get("user"), 160) record["smb_domain"] = _text(ntlm.get("domain"), 160) elif event_type == "quic": quic = event.get("quic") if isinstance(event.get("quic"), dict) else {} ja3 = quic.get("ja3") if isinstance(quic.get("ja3"), dict) else {} record["quic_sni"] = _text(quic.get("sni"), 255) record["quic_version"] = _text(quic.get("version"), 64) record["quic_ja3"] = _text(ja3.get("hash"), 96) record["quic_ja4"] = _text(quic.get("ja4"), 128) elif event_type == "dhcp": dhcp = event.get("dhcp") if isinstance(event.get("dhcp"), dict) else {} record["dhcp_event_type"] = _text(dhcp.get("type"), 48) record["dhcp_type"] = _text(dhcp.get("dhcp_type") or dhcp.get("message_type"), 48) record["dhcp_hostname"] = _text(dhcp.get("hostname"), 255) record["dhcp_client_mac"] = _text(dhcp.get("client_mac") or dhcp.get("mac"), 32) record["dhcp_assigned_ip"] = _text(dhcp.get("assigned_ip") or dhcp.get("client_ip"), 64) record["dhcp_requested_ip"] = _text(dhcp.get("requested_ip"), 64) elif event_type == "arp": arp = event.get("arp") if isinstance(event.get("arp"), dict) else {} record["arp_opcode"] = _text(arp.get("opcode"), 32) record["arp_src_mac"] = _text(arp.get("src_mac"), 32) record["arp_dest_mac"] = _text(arp.get("dest_mac"), 32) record["arp_src_ip"] = _text(arp.get("src_ip"), 64) record["arp_dest_ip"] = _text(arp.get("dest_ip"), 64) if not record["src_ip"] and record["arp_src_ip"]: record["src_ip"] = record["arp_src_ip"] if not record["dest_ip"] and record["arp_dest_ip"]: record["dest_ip"] = record["arp_dest_ip"] record["direction"] = self._direction(record["src_ip"], record["dest_ip"]) elif event_type in { "ike", "mqtt", "ftp", "ftp_data", "smtp", "mdns", "http2", "doh2", "websocket", "nfs", "tftp", "dcerpc", "krb5", "snmp", "rfb", "sip", "ldap", "pop3", }: app = event.get(event_type) if isinstance(event.get(event_type), dict) else {} safe_summary_fields = ( "command", "subject", "version", "msg_type", "realm", "cname", "sname", "operation", "service", "hostname", "event_type", ) parts = [] for field in safe_summary_fields: value = app.get(field) if isinstance(value, (str, int, float)) and str(value): parts.append(str(value)) if len(parts) >= 4: break record["app_summary"] = _text(" · ".join(parts), 255) for key, value in extra.items(): if value is not None: record[key] = value return record def _direction(self, src: str, dst: str) -> str: src_local = self._local(src) dst_local = self._local(dst) if src_local and not dst_local: return "outbound" if dst_local and not src_local: return "inbound" if src_local and dst_local: return "internal" return "external" @functools.lru_cache(maxsize=8192) def _local(self, value: str) -> bool: try: ip = ipaddress.ip_address(value) except ValueError: return False return any(ip.version == net.version and ip in net for net in self.networks) class EventBus: """Thread-safe fan-out with bounded queues so slow browsers cannot block EVE.""" def __init__(self, history_size: int = 0, subscriber_queue_size: int = 512) -> None: self._history: collections.deque[dict[str, Any]] = collections.deque( maxlen=max(0, int(history_size)) ) self._queue_size = max(64, int(subscriber_queue_size)) self._lock = threading.RLock() self._subscribers: set[queue.Queue[dict[str, Any]]] = set() self._dropped = 0 def publish(self, event: dict[str, Any]) -> None: with self._lock: if self._history.maxlen: self._history.append(event) dead: list[queue.Queue[dict[str, Any]]] = [] for target in self._subscribers: try: target.put_nowait(event) except queue.Full: # Drop the oldest event for this browser only. Capture must never block. try: target.get_nowait() target.put_nowait(event) self._dropped += 1 except (queue.Empty, queue.Full): dead.append(target) for target in dead: self._subscribers.discard(target) def subscribe(self) -> queue.Queue[dict[str, Any]]: target: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=self._queue_size) with self._lock: self._subscribers.add(target) return target def unsubscribe(self, target: queue.Queue[dict[str, Any]]) -> None: with self._lock: self._subscribers.discard(target) def recent(self, limit: int = 250) -> list[dict[str, Any]]: limit = min(max(int(limit), 1), 5000) with self._lock: return list(self._history)[-limit:][::-1] def has_subscribers(self) -> bool: # Reading set truthiness is atomic under CPython's GIL and avoids taking # the event-bus lock for every captured packet in FlowTracker. return bool(self._subscribers) def status(self) -> dict[str, Any]: with self._lock: return { "history_events": len(self._history), "subscribers": len(self._subscribers), "subscriber_dropped_events": self._dropped, } class RedisProtocolError(RuntimeError): pass class RedisUnavailableError(RuntimeError): pass @dataclass(frozen=True) class RedisEndpoint: host: str port: int db: int password: str | None @classmethod def parse(cls, url: str) -> "RedisEndpoint": parsed = urllib.parse.urlparse(url) if parsed.scheme not in {"redis", "rediss"}: raise ValueError("REDIS_URL must use redis:// or rediss://") if parsed.scheme == "rediss": raise ValueError("rediss:// is not supported by the dependency-free Redis client") db_text = (parsed.path or "/0").lstrip("/") or "0" return cls( host=parsed.hostname or "127.0.0.1", port=parsed.port or 6379, db=int(db_text), password=urllib.parse.unquote(parsed.password) if parsed.password else None, ) class RedisConnection: """Small persistent RESP2 connection for the commands used by traffic history.""" def __init__(self, endpoint: RedisEndpoint, timeout: float = 1.5) -> None: self.endpoint = endpoint self.timeout = timeout self._lock = threading.RLock() self._sock: socket.socket | None = None self._stream = None def execute(self, *parts: Any) -> Any: with self._lock: try: self._ensure_connected() self._write(self._stream, *parts) return self._read(self._stream) except (OSError, RedisProtocolError): self.close() raise def close(self) -> None: with self._lock: stream, sock = self._stream, self._sock self._stream = None self._sock = None if stream is not None: try: stream.close() except OSError: pass if sock is not None: try: sock.close() except OSError: pass def _ensure_connected(self) -> None: if self._sock is not None and self._stream is not None: return sock = socket.create_connection((self.endpoint.host, self.endpoint.port), self.timeout) sock.settimeout(self.timeout) stream = sock.makefile("rwb", buffering=0) self._sock = sock self._stream = stream try: if self.endpoint.password: self._write(stream, "AUTH", self.endpoint.password) self._read(stream) if self.endpoint.db: self._write(stream, "SELECT", str(self.endpoint.db)) self._read(stream) except Exception: self.close() raise @staticmethod def _write(stream, *parts: Any) -> None: encoded = [str(part).encode("utf-8") if not isinstance(part, bytes) else part for part in parts] stream.write(f"*{len(encoded)}\r\n".encode("ascii")) for part in encoded: stream.write(f"${len(part)}\r\n".encode("ascii")) stream.write(part + b"\r\n") def _read(self, stream) -> Any: prefix = stream.read(1) if not prefix: raise RedisProtocolError("Redis closed the connection") line = stream.readline() if not line.endswith(b"\r\n"): raise RedisProtocolError("Malformed Redis response") payload = line[:-2] if prefix == b"+": return payload.decode("utf-8", "replace") if prefix == b"-": raise RedisProtocolError(payload.decode("utf-8", "replace")) if prefix == b":": return int(payload) if prefix == b"$": length = int(payload) if length < 0: return None data = stream.read(length) if stream.read(2) != b"\r\n": raise RedisProtocolError("Malformed bulk response") return data if prefix == b"*": length = int(payload) if length < 0: return None return [self._read(stream) for _ in range(length)] raise RedisProtocolError(f"Unknown Redis response prefix {prefix!r}") class TrafficHistory: """Traffic history with Redis ingest buffering and optional SQLite archive. Production requires Redis for the live ingest queue and drains committed history to SQLite. The optional memory mode is retained only for development/unit tests. """ REDIS_KEY = "suricata:traffic:v2" LEGACY_REDIS_KEY = "suricata:traffic:v1" THROUGHPUT_KEY = "suricata:throughput:v1" SNAPSHOT_PREFIX = "suricata:analytics:v3:" LEGACY_SNAPSHOT_PREFIX = "suricata:analytics:v2:" def __init__( self, redis_url: str, retention_hours: int, max_events: int, memory_events: int = 0, *, require_redis: bool = False, allow_memory_fallback: bool = True, archive_store: Any | None = None, ) -> None: self.retention_hours = max(1, int(retention_hours)) # 0 means no count cap. Time retention is the authoritative bound. self.max_events = max(0, int(max_events)) self.allow_memory_fallback = bool(allow_memory_fallback) self.archive_store = archive_store self._archived_events = 0 self._archived_throughput = 0 self._archive_batches = 0 self._archive_errors = 0 self._last_archive_at = 0.0 memory_capacity = max(0, int(memory_events)) if self.allow_memory_fallback else 0 self._memory: collections.deque[dict[str, Any]] = collections.deque(maxlen=memory_capacity) self._throughput_memory: collections.deque[dict[str, Any]] = collections.deque( maxlen=max(0, min(memory_capacity, self.retention_hours * 3600)) ) self._snapshot_memory: dict[int, dict[str, Any]] = {} self._lock = threading.RLock() self._redis_url = redis_url.strip() self._redis: RedisConnection | None = None self._redis_error = "disabled" self._last_retry = 0.0 self._last_trim = 0.0 if self._redis_url: try: self._redis = RedisConnection(RedisEndpoint.parse(self._redis_url)) self._redis.execute("PING") self._redis_error = "" self._migrate_legacy_key() except Exception as exc: self._redis = None self._redis_error = str(exc) if require_redis and self._redis is None: raise RedisUnavailableError(f"Redis traffic ingest buffer is required: {self._redis_error}") def add(self, event: dict[str, Any]) -> None: self.add_many([event]) def add_many(self, events: Iterable[dict[str, Any]]) -> None: batch = list(events) if not batch: return if self.allow_memory_fallback and self._memory.maxlen: with self._lock: self._memory.extend(batch) redis = self._redis_or_retry() if redis is None: if self.allow_memory_fallback: return raise RedisUnavailableError(self._redis_error or "Redis is unavailable") args: list[Any] = ["ZADD", self.REDIS_KEY] for event in batch: member = f"{event.get('id','')}|".encode("ascii", "ignore") + json.dumps( event, ensure_ascii=False, separators=(",", ":") ).encode("utf-8") args.extend((int(event.get("ts_ms") or _epoch_ms(None)), member)) try: # One multi-member ZADD drastically reduces socket/Redis command # overhead under high EVE rates while preserving every event. redis.execute(*args) self._redis_error = "" self._trim_if_due(redis) except Exception as exc: self._mark_redis_down(exc) if not self.allow_memory_fallback: raise RedisUnavailableError(str(exc)) from exc def add_throughput_sample(self, sample: dict[str, Any]) -> None: sample = dict(sample) ts_ms = int(sample.get("ts_ms") or _epoch_ms(None)) sample["ts_ms"] = ts_ms if self.allow_memory_fallback and self._throughput_memory.maxlen: with self._lock: self._throughput_memory.append(sample) redis = self._redis_or_retry() if redis is None: if self.allow_memory_fallback: return raise RedisUnavailableError(self._redis_error or "Redis is unavailable") encoded = json.dumps(sample, ensure_ascii=False, separators=(",", ":")).encode("utf-8") digest = hashlib.blake2s(encoded, digest_size=6).hexdigest() member = f"{ts_ms}|{digest}|".encode("ascii") + encoded try: redis.execute("ZADD", self.THROUGHPUT_KEY, ts_ms, member) self._redis_error = "" self._trim_if_due(redis) except Exception as exc: self._mark_redis_down(exc) if not self.allow_memory_fallback: raise RedisUnavailableError(str(exc)) from exc def search( self, *, limit: int = 250, since_ms: int | None = None, until_ms: int | None = None, text: str = "", event_type: str = "", proto: str = "", app_proto: str = "", direction: str = "", ) -> list[dict[str, Any]]: limit = min(max(int(limit), 1), 2000) since_ms = since_ms or int((time.time() - self.retention_hours * 3600) * 1000) until_ms = until_ms or int(time.time() * 1000) + 1000 filters = { "text": text.strip().lower(), "event_type": event_type.strip().lower(), "proto": proto.strip().lower(), "app_proto": app_proto.strip().lower(), "direction": direction.strip().lower(), } combined: list[dict[str, Any]] = [] seen: set[str] = set() if self.archive_store is not None: offset = 0 page_size = min(2000, max(limit * 3, 250)) while len(combined) < limit: page = self.archive_store.traffic_event_page( since_ms, until_ms, offset=offset, limit=page_size, event_type=filters["event_type"], proto=filters["proto"], app_proto=filters["app_proto"], direction=filters["direction"], text=filters["text"], ) if not page: break for item in page: archive_key = str(item.pop("_archive_key", "")) if not _matches_search(item, filters): continue key = _event_identity(item) if key in seen: continue seen.add(key) combined.append(item) offset += len(page) if len(page) < page_size: break redis = self._redis_or_retry() if redis is not None: remote = self._redis_search(redis, since_ms, until_ms, limit, filters) if remote is not None: for item in remote: key = _event_identity(item) if key in seen: continue seen.add(key) combined.append(item) elif self.archive_store is None and not self.allow_memory_fallback: raise RedisUnavailableError(self._redis_error or "Redis is unavailable") if self.allow_memory_fallback and self.archive_store is None: with self._lock: candidates = [ item for item in reversed(self._memory) if since_ms <= int(item.get("ts_ms") or 0) <= until_ms ] for item in candidates: if _matches_search(item, filters): key = _event_identity(item) if key not in seen: seen.add(key) combined.append(item) combined.sort(key=lambda item: int(item.get("ts_ms") or 0), reverse=True) return combined[:limit] def latest_throughput(self) -> dict[str, Any] | None: """Return only the newest persisted TZSP rate sample (constant-cost Redis read).""" redis = self._redis_or_retry() if redis is None: if not self.allow_memory_fallback: raise RedisUnavailableError(self._redis_error or "Redis is unavailable") with self._lock: return dict(self._throughput_memory[-1]) if self._throughput_memory else None try: raw = redis.execute("ZREVRANGE", self.THROUGHPUT_KEY, 0, 0) self._redis_error = "" if not raw: return None return _decode_throughput_member(raw[0]) except Exception as exc: self._mark_redis_down(exc) if not self.allow_memory_fallback: raise RedisUnavailableError(str(exc)) from exc return None def throughput_analytics(self, window_seconds: int = 3600) -> dict[str, Any]: """Build the speed/volume chart without scanning long-lived Redis history.""" window_seconds = min(max(int(window_seconds), 60), self.retention_hours * 3600) if self.archive_store is not None: payload = self._archive_analytics_many((window_seconds,), include_events=False)[window_seconds] payload["throughput_only"] = True return payload now_ms = int(time.time() * 1000) since_ms = now_ms - window_seconds * 1000 throughput = self._redis_throughput_candidates(since_ms, now_ms + 1000) if throughput is None: if not self.allow_memory_fallback: raise RedisUnavailableError(self._redis_error or "Redis is unavailable") with self._lock: throughput = [item for item in self._throughput_memory if int(item.get("ts_ms") or 0) >= since_ms] source = "memory-dev" else: source = "redis" payload = _analytics([], since_ms, now_ms, window_seconds, throughput) payload["analytics_source"] = source payload["throughput_samples_scanned"] = len(throughput) payload["throughput_only"] = True return payload def analytics(self, window_seconds: int = 3600, sample_limit: int | None = None) -> dict[str, Any]: window_seconds = min(max(int(window_seconds), 60), self.retention_hours * 3600) if self.archive_store is not None: return self._archive_analytics_many((window_seconds,))[window_seconds] now_ms = int(time.time() * 1000) since_ms = now_ms - window_seconds * 1000 limit = None if sample_limit is None else max(int(sample_limit), 1) events = self._redis_candidates(since_ms, now_ms + 1000, limit) throughput = self._redis_throughput_candidates(since_ms, now_ms + 1000) if events is None or throughput is None: if not self.allow_memory_fallback: raise RedisUnavailableError(self._redis_error or "Redis is unavailable") with self._lock: events = [item for item in self._memory if int(item.get("ts_ms") or 0) >= since_ms] events.reverse() throughput = [item for item in self._throughput_memory if int(item.get("ts_ms") or 0) >= since_ms] source = "memory-dev" complete = False else: source = "redis" complete = limit is None or len(events) < limit payload = _analytics(events, since_ms, now_ms, window_seconds, throughput) payload["analytics_source"] = source payload["analytics_complete"] = complete payload["retained_events_scanned"] = len(events) payload["throughput_samples_scanned"] = len(throughput) return payload def analytics_many(self, windows: Iterable[int]) -> dict[int, dict[str, Any]]: normalized = sorted({ min(max(int(window), 60), self.retention_hours * 3600) for window in windows }) if not normalized: return {} if self.archive_store is not None: return self._archive_analytics_many(normalized) now_ms = int(time.time() * 1000) max_window = max(normalized) oldest_ms = now_ms - max_window * 1000 events = self._redis_candidates(oldest_ms, now_ms + 1000, None) throughput = self._redis_throughput_candidates(oldest_ms, now_ms + 1000) if events is None or throughput is None: if not self.allow_memory_fallback: raise RedisUnavailableError(self._redis_error or "Redis is unavailable") with self._lock: events = [item for item in self._memory if int(item.get("ts_ms") or 0) >= oldest_ms] events.reverse() throughput = [item for item in self._throughput_memory if int(item.get("ts_ms") or 0) >= oldest_ms] source = "memory-dev" complete = False else: source = "redis" complete = True result: dict[int, dict[str, Any]] = {} for window in normalized: since_ms = now_ms - window * 1000 payload = _analytics(events, since_ms, now_ms, window, throughput) payload["analytics_source"] = source payload["analytics_complete"] = complete payload["retained_events_scanned"] = sum( 1 for item in events if int(item.get("ts_ms") or 0) >= since_ms ) payload["throughput_samples_scanned"] = sum( 1 for item in throughput if int(item.get("ts_ms") or 0) >= since_ms ) result[window] = payload return result def _archive_analytics_many( self, windows: Iterable[int], *, include_events: bool = True, ) -> dict[int, dict[str, Any]]: if self.archive_store is None: return {} normalized = sorted({ min(max(int(window), 60), self.retention_hours * 3600) for window in windows }) if not normalized: return {} now_ms = int(time.time() * 1000) archive_status = self.archive_store.traffic_archive_status() newest_archived_ms = max( int(archive_status.get("newest_event_ms") or 0), int(archive_status.get("newest_throughput_ms") or 0), ) # Freeze the upper bound for this calculation. The archive worker can # keep appending newer rows through WAL without shifting OFFSET-based # pages underneath the snapshot worker. read_until_ms = min(now_ms + 1000, newest_archived_ms) if newest_archived_ms else now_ms + 1000 result: dict[int, dict[str, Any]] = {} # Deliberately build one window at a time. The worker therefore has a # bounded Python memory footprint even when the SQLite archive contains # millions of rows. High-cardinality endpoint/application aggregations # are delegated to SQL below instead of retaining large sets/counters. for window in normalized: since_ms = now_ms - window * 1000 accumulator = _AnalyticsAccumulator( since_ms, now_ms, window, track_high_cardinality=False, ) event_count = 0 throughput_count = 0 if include_events: offset = 0 page_size = 2000 while True: page = self.archive_store.traffic_event_page( since_ms, read_until_ms, offset=offset, limit=page_size ) if not page: break for item in page: item.pop("_archive_key", None) accumulator.add_event(item) event_count += 1 offset += len(page) if len(page) < page_size: break offset = 0 page_size = 5000 while True: page = self.archive_store.traffic_throughput_page( since_ms, read_until_ms, offset=offset, limit=page_size ) if not page: break for sample in page: sample.pop("_archive_key", None) accumulator.add_throughput(sample) throughput_count += 1 offset += len(page) if len(page) < page_size: break payload = accumulator.finish() if include_events: payload.update( self.archive_store.traffic_dimension_summary( since_ms, read_until_ms, limit=10, ) ) payload["analytics_source"] = "sqlite-archive" payload["analytics_complete"] = True payload["retained_events_scanned"] = event_count payload["throughput_samples_scanned"] = throughput_count result[window] = payload return result def save_snapshot(self, window_seconds: int, payload: dict[str, Any]) -> None: window = int(window_seconds) stored = dict(payload) stored["window_seconds"] = window stored["generated_at"] = _utc_now() redis = self._redis_or_retry() if redis is None: if self.allow_memory_fallback: with self._lock: self._snapshot_memory[window] = stored return raise RedisUnavailableError(self._redis_error or "Redis is unavailable") try: redis.execute("SET", f"{self.SNAPSHOT_PREFIX}{window}", json.dumps(stored, ensure_ascii=False, separators=(",", ":"))) self._redis_error = "" except Exception as exc: self._mark_redis_down(exc) if not self.allow_memory_fallback: raise RedisUnavailableError(str(exc)) from exc def snapshot(self, window_seconds: int) -> dict[str, Any] | None: window = int(window_seconds) redis = self._redis_or_retry() if redis is not None: try: raw = redis.execute("GET", f"{self.SNAPSHOT_PREFIX}{window}") self._redis_error = "" if raw is None: return None data = json.loads(raw.decode("utf-8") if isinstance(raw, bytes) else str(raw)) return data if isinstance(data, dict) else None except Exception as exc: self._mark_redis_down(exc) if not self.allow_memory_fallback: raise RedisUnavailableError(str(exc)) from exc if self.allow_memory_fallback: with self._lock: data = self._snapshot_memory.get(window) return dict(data) if data is not None else None raise RedisUnavailableError(self._redis_error or "Redis is unavailable") def snapshot_status(self, windows: Iterable[int]) -> list[dict[str, Any]]: rows = [] for window in windows: try: snap = self.snapshot(int(window)) except RedisUnavailableError: break if snap is not None: rows.append({"window_seconds": int(window), "generated_at": snap.get("generated_at")}) return rows def clear_snapshots(self, windows: Iterable[int]) -> int: windows = [int(window) for window in windows] with self._lock: local_count = sum(1 for window in windows if window in self._snapshot_memory) for window in windows: self._snapshot_memory.pop(window, None) redis = self._redis_or_retry() if redis is None: if self.allow_memory_fallback: return local_count raise RedisUnavailableError(self._redis_error or "Redis is unavailable") keys = [f"{self.SNAPSHOT_PREFIX}{window}" for window in windows] try: remote = _safe_int(redis.execute("DEL", *keys)) if keys else 0 return max(local_count, remote) except Exception as exc: self._mark_redis_down(exc) if not self.allow_memory_fallback: raise RedisUnavailableError(str(exc)) from exc return local_count def archive_redis_to_store( self, cutoff_ms: int, *, batch_size: int = 1000, max_batches: int = 0, ) -> dict[str, int]: """Move old Redis queue entries into the disk-backed SQLite archive. Redis is only the ingestion buffer. A batch is removed from Redis only after SQLite commits it, so retries after a crash are safe through the archive tables' stable primary keys. """ if self.archive_store is None: return {"events": 0, "throughput_samples": 0, "batches": 0} redis = self._redis_or_retry() if redis is None: raise RedisUnavailableError(self._redis_error or "Redis is unavailable") batch_size = max(50, min(int(batch_size), 5000)) max_batches = max(0, int(max_batches)) moved_events = 0 moved_throughput = 0 batches = 0 try: streams = [ [self.REDIS_KEY, _decode_redis_member, self.archive_store.archive_traffic_events, True], [self.THROUGHPUT_KEY, _decode_throughput_member, self.archive_store.archive_traffic_throughput, True], ] # Alternate event and throughput batches so a large legacy EVE # backlog cannot starve rate samples in Redis during migration. while any(bool(stream[3]) for stream in streams) and (max_batches == 0 or batches < max_batches): for stream in streams: if not stream[3] or (max_batches and batches >= max_batches): continue key, decoder, writer, _active = stream raw = redis.execute( "ZRANGEBYSCORE", key, "-inf", int(cutoff_ms), "LIMIT", 0, batch_size, ) if not raw: stream[3] = False continue records: list[tuple[str, dict[str, Any]]] = [] for member in raw: payload = decoder(member) if payload is None: continue digest = hashlib.blake2s(bytes(member), digest_size=16).hexdigest() records.append((digest, payload)) writer(records) redis.execute("ZREM", key, *raw) if key == self.REDIS_KEY: moved_events += len(raw) else: moved_throughput += len(raw) batches += 1 if len(raw) < batch_size: stream[3] = False self._redis_error = "" with self._lock: self._archived_events += moved_events self._archived_throughput += moved_throughput self._archive_batches += batches self._last_archive_at = time.time() return { "events": moved_events, "throughput_samples": moved_throughput, "batches": batches, } except RedisUnavailableError: raise except Exception as exc: with self._lock: self._archive_errors += 1 if isinstance(exc, (RedisProtocolError, OSError, ConnectionError)): self._mark_redis_down(exc) raise RedisUnavailableError(str(exc)) from exc raise def purge_archive(self) -> dict[str, int]: if self.archive_store is None: return {"events": 0, "throughput_samples": 0} cutoff_ms = int((time.time() - self.retention_hours * 3600) * 1000) return self.archive_store.purge_traffic_archive_before(cutoff_ms) def clear(self) -> int: with self._lock: count = len(self._memory) self._memory.clear() self._throughput_memory.clear() self._snapshot_memory.clear() archived_events = 0 if self.archive_store is not None: archived = self.archive_store.clear_traffic_archive() archived_events = int(archived.get("events") or 0) redis = self._redis_or_retry() if redis is None: if self.allow_memory_fallback or self.archive_store is not None: return count + archived_events raise RedisUnavailableError(self._redis_error or "Redis is unavailable") try: remote = _safe_int(redis.execute("ZCARD", self.REDIS_KEY)) keys = [self.REDIS_KEY, self.LEGACY_REDIS_KEY, self.THROUGHPUT_KEY] keys.extend(f"{self.SNAPSHOT_PREFIX}{window}" for window in (900, 3600, 18000, 21600, 86400)) redis.execute("DEL", *keys) return count + archived_events + remote except Exception as exc: self._mark_redis_down(exc) if not self.allow_memory_fallback and self.archive_store is None: raise RedisUnavailableError(str(exc)) from exc return count + archived_events def status(self) -> dict[str, Any]: with self._lock: memory_count = len(self._memory) archive_stats = { "archived_events_total": self._archived_events, "archived_throughput_total": self._archived_throughput, "archive_batches": self._archive_batches, "archive_errors": self._archive_errors, "last_archive_at": self._last_archive_at, } redis = self._redis_or_retry() remote_count = None throughput_count = None if redis is not None: try: remote_count = _safe_int(redis.execute("ZCARD", self.REDIS_KEY)) throughput_count = _safe_int(redis.execute("ZCARD", self.THROUGHPUT_KEY)) self._redis_error = "" except Exception as exc: self._mark_redis_down(exc) if self.archive_store is not None: try: archive_stats.update(self.archive_store.traffic_archive_status()) except Exception: archive_stats["archive_errors"] = int(archive_stats.get("archive_errors") or 0) + 1 if self._redis_url and self.archive_store is not None: backend = "redis-buffer+sqlite" elif self._redis_url and not self.allow_memory_fallback: backend = "redis" elif self._redis_url: backend = "redis+memory-dev" else: backend = "memory-dev" return { "backend": backend, "redis_configured": bool(self._redis_url), "redis_ok": self._redis is not None if self._redis_url else None, "redis_error": self._redis_error, "redis_events": remote_count, "throughput_samples": throughput_count, "memory_fallback": self.allow_memory_fallback, "memory_events": memory_count if self.allow_memory_fallback else 0, "memory_capacity": self._memory.maxlen if self.allow_memory_fallback else 0, "retention_hours": self.retention_hours, "max_events": self.max_events, "archive": archive_stats if self.archive_store is not None else None, } def _redis_search( self, redis: RedisConnection, since_ms: int, until_ms: int, limit: int, filters: dict[str, str], ) -> list[dict[str, Any]] | None: page_size = 2000 offset = 0 result: list[dict[str, Any]] = [] try: while len(result) < limit and (self.max_events == 0 or offset < self.max_events): count = page_size if self.max_events == 0 else min(page_size, self.max_events - offset) if count <= 0: break raw = redis.execute( "ZREVRANGEBYSCORE", self.REDIS_KEY, until_ms, since_ms, "LIMIT", offset, count, ) self._redis_error = "" if not raw: break for member in raw: item = _decode_redis_member(member) if item is not None and _matches_search(item, filters): result.append(item) if len(result) >= limit: break offset += len(raw) if len(raw) < count: break return result except Exception as exc: self._mark_redis_down(exc) return None def _redis_candidates( self, since_ms: int, until_ms: int, limit: int | None = None, ) -> list[dict[str, Any]] | None: redis = self._redis_or_retry() if redis is None: return None return self._redis_zset_candidates(redis, self.REDIS_KEY, since_ms, until_ms, limit, _decode_redis_member) def _redis_throughput_candidates(self, since_ms: int, until_ms: int) -> list[dict[str, Any]] | None: redis = self._redis_or_retry() if redis is None: return None return self._redis_zset_candidates(redis, self.THROUGHPUT_KEY, since_ms, until_ms, None, _decode_throughput_member) def _redis_zset_candidates( self, redis: RedisConnection, key: str, since_ms: int, until_ms: int, limit: int | None, decoder, ) -> list[dict[str, Any]] | None: page_size = 5000 offset = 0 result: list[dict[str, Any]] = [] try: while limit is None or len(result) < limit: count = page_size if limit is None else min(page_size, limit - len(result)) if count <= 0: break raw = redis.execute( "ZREVRANGEBYSCORE", key, until_ms, since_ms, "LIMIT", offset, count, ) if not raw: break for member in raw: item = decoder(member) if item is not None: result.append(item) offset += len(raw) if len(raw) < count: break self._redis_error = "" except Exception as exc: self._mark_redis_down(exc) return None return result def _redis_or_retry(self) -> RedisConnection | None: if not self._redis_url: return None if self._redis is not None: return self._redis now = time.monotonic() if now - self._last_retry < 1: return None self._last_retry = now try: redis = RedisConnection(RedisEndpoint.parse(self._redis_url)) redis.execute("PING") self._redis = redis self._redis_error = "" self._migrate_legacy_key() return redis except Exception as exc: self._redis_error = str(exc) return None def _mark_redis_down(self, exc: Exception) -> None: redis = self._redis self._redis = None if redis is not None: redis.close() self._redis_error = str(exc) self._last_retry = time.monotonic() def _trim_if_due(self, redis: RedisConnection) -> None: now = time.monotonic() if now - self._last_trim > 15: self._trim(redis) self._last_trim = now def _trim(self, redis: RedisConnection) -> None: cutoff = int((time.time() - self.retention_hours * 3600) * 1000) redis.execute("ZREMRANGEBYSCORE", self.REDIS_KEY, "-inf", cutoff) redis.execute("ZREMRANGEBYSCORE", self.THROUGHPUT_KEY, "-inf", cutoff) if self.max_events > 0: count = _safe_int(redis.execute("ZCARD", self.REDIS_KEY)) excess = count - self.max_events if excess > 0: redis.execute("ZREMRANGEBYRANK", self.REDIS_KEY, 0, excess - 1) def _migrate_legacy_key(self) -> None: redis = self._redis if redis is None: return try: new_count = _safe_int(redis.execute("ZCARD", self.REDIS_KEY)) old_count = _safe_int(redis.execute("ZCARD", self.LEGACY_REDIS_KEY)) if new_count == 0 and old_count > 0: redis.execute("RENAME", self.LEGACY_REDIS_KEY, self.REDIS_KEY) # Dashboard snapshots are SQLite-backed now. Remove both generations # of obsolete Redis snapshot keys during upgrade; the raw queue is # preserved here and drained transactionally by the archive worker. keys = [ *(f"{self.LEGACY_SNAPSHOT_PREFIX}{window}" for window in (900, 3600, 21600, 86400)), *(f"{self.SNAPSHOT_PREFIX}{window}" for window in (900, 3600, 18000, 21600, 86400)), ] redis.execute("DEL", *keys) except Exception: # Migration is best-effort; a missing legacy key is normal. pass def _decode_redis_member(member: bytes) -> dict[str, Any] | None: try: data = member.split(b"|", 1)[1] if b"|" in member else member item = json.loads(data.decode("utf-8")) return item if isinstance(item, dict) else None except (UnicodeDecodeError, json.JSONDecodeError, IndexError, AttributeError): return None def _event_identity(item: dict[str, Any]) -> str: event_id = _text(item.get("id"), 128) if event_id: return event_id raw = json.dumps(item, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") return hashlib.blake2s(raw, digest_size=16).hexdigest() def _decode_throughput_member(member: bytes) -> dict[str, Any] | None: try: parts = member.split(b"|", 2) data = parts[2] if len(parts) == 3 else member item = json.loads(data.decode("utf-8")) return item if isinstance(item, dict) else None except (UnicodeDecodeError, json.JSONDecodeError, IndexError, AttributeError): return None IGNORED_APPLICATION_PROTOCOLS = {"", "failed", "unknown", "none", "null", "notset"} IGNORED_DASHBOARD_ALERT_SIGNATURES = { "suricata ipv4 truncated packet", "suricata ipv6 truncated packet", } def is_dashboard_noise(item: dict[str, Any]) -> bool: """Return True for sensor/decoder noise that should not enter UI analytics.""" if _text(item.get("type"), 32).lower() != "alert": return False signature = _text(item.get("signature"), 300).strip().casefold() return signature in IGNORED_DASHBOARD_ALERT_SIGNATURES def _valid_app_proto(value: Any) -> str: name = _text(value, 48).strip().lower() return "" if name in IGNORED_APPLICATION_PROTOCOLS else name def event_matches( item: dict[str, Any], *, text: str = "", event_type: str = "", proto: str = "", app_proto: str = "", direction: str = "", ) -> bool: """Cheap normalized-event filter shared by history search and WebSocket streaming.""" filters = { "text": text.strip().lower(), "event_type": event_type.strip().lower(), "proto": proto.strip().lower(), "app_proto": app_proto.strip().lower(), "direction": direction.strip().lower(), } return _matches_search(item, filters) def _matches_search(item: dict[str, Any], filters: dict[str, str]) -> bool: if is_dashboard_noise(item): return False if filters["event_type"] and _text(item.get("type")).lower() != filters["event_type"]: return False if filters["proto"] and _text(item.get("proto")).lower() != filters["proto"]: return False if filters["app_proto"] and _text(item.get("app_proto")).lower() != filters["app_proto"]: return False if filters["direction"] and _text(item.get("direction")).lower() != filters["direction"]: return False if filters["text"] and filters["text"] not in _search_blob(item): return False return True def _search_blob(item: dict[str, Any]) -> str: fields = ( "id", "flow_id", "community_id", "tx_id", "src_ip", "src_port", "dest_ip", "dest_port", "proto", "app_proto", "pkt_src", "ether_src", "ether_dest", "signature", "signature_id", "category", "action", "block_target", "dns_query", "dns_type", "http_host", "http_url", "http_method", "http_status", "http_user_agent", "tls_sni", "tls_subject", "tls_issuer", "tls_fingerprint", "tls_alpn", "tls_ja3", "tls_ja4", "filename", "file_md5", "file_sha1", "file_sha256", "direction", "ssh_client", "ssh_server", "ssh_proto", "ssh_hassh_client", "ssh_hassh_server", "rdp_event_type", "rdp_cookie", "rdp_protocol", "rdp_client_name", "rdp_client_build", "smb_command", "smb_share", "smb_dialect", "smb_filename", "smb_status", "smb_client_guid", "smb_user", "smb_domain", "quic_sni", "quic_version", "quic_ja3", "quic_ja4", "dhcp_event_type", "dhcp_type", "dhcp_hostname", "dhcp_client_mac", "dhcp_assigned_ip", "dhcp_requested_ip", "arp_src_mac", "arp_dest_mac", "arp_src_ip", "arp_dest_ip", "app_summary", ) return " ".join(_text(item.get(name)).lower() for name in fields) class _AnalyticsAccumulator: """Streaming analytics builder used for both Redis/dev and SQLite archive reads.""" def __init__( self, since_ms: int, now_ms: int, window_seconds: int, *, track_high_cardinality: bool = True, ) -> None: self.since_ms = int(since_ms) self.now_ms = int(now_ms) self.window_seconds = int(window_seconds) self.track_high_cardinality = bool(track_high_cardinality) self.bins_count = 60 self.bin_ms = max(1000, int(self.window_seconds * 1000 / self.bins_count)) self.bins = [ { "ts_ms": self.since_ms + idx * self.bin_ms, "events": 0, "bytes": 0, "bytes_in": 0, "bytes_out": 0, "packets": 0, "alerts": 0, "rate_bytes": 0, "rate_bytes_in": 0, "rate_bytes_out": 0, "rate_packets": 0, } for idx in range(self.bins_count) ] self.apps: collections.Counter[str] = collections.Counter() self.protocols: collections.Counter[str] = collections.Counter() self.sources: collections.Counter[str] = collections.Counter() self.destinations: collections.Counter[str] = collections.Counter() self.local_clients: collections.Counter[str] = collections.Counter() self.remote_peers: collections.Counter[str] = collections.Counter() self.local_client_bytes: collections.Counter[str] = collections.Counter() self.remote_peer_bytes: collections.Counter[str] = collections.Counter() self.app_bytes: collections.Counter[str] = collections.Counter() self.directions: collections.Counter[str] = collections.Counter() self.types: collections.Counter[str] = collections.Counter() self.signatures: collections.Counter[str] = collections.Counter() self.severities: collections.Counter[str] = collections.Counter() self.fingerprints: collections.Counter[str] = collections.Counter() self.assets: collections.Counter[str] = collections.Counter() self.file_activity: collections.Counter[str] = collections.Counter() self.app_flow_seen: set[tuple[str, str]] = set() self.included_events = 0 self.eve_flow_bytes = 0 self.alerts = 0 self.blocked = 0 self.anomalies = 0 self.dns_nxdomain = 0 self.files = 0 self.encrypted = 0 self.cleartext = 0 self.throughput_bytes = 0 self.throughput_classified_bytes = 0 self.throughput_packets = 0 self.throughput_samples = 0 self.latest_sample: dict[str, Any] | None = None def add_event(self, item: dict[str, Any]) -> None: ts = _safe_int(item.get("ts_ms")) if ts < self.since_ms or ts > self.now_ms + 1000 or is_dashboard_noise(item): return self.included_events += 1 idx = min(max((ts - self.since_ms) // self.bin_ms, 0), self.bins_count - 1) is_flow = _text(item.get("type"), 32).lower() == "flow" size = max(_safe_int(item.get("bytes")), 0) if is_flow else 0 bytes_in = max(_safe_int(item.get("bytes_in")), 0) if is_flow else 0 bytes_out = max(_safe_int(item.get("bytes_out")), 0) if is_flow else 0 packets = max(_safe_int(item.get("packets")), 0) if is_flow else 0 bucket = self.bins[idx] bucket["events"] += 1 bucket["bytes"] += size bucket["bytes_in"] += bytes_in bucket["bytes_out"] += bytes_out bucket["packets"] += packets if item.get("type") == "alert": bucket["alerts"] += 1 self.alerts += 1 signature = _text(item.get("signature"), 160) if signature: self.signatures[signature] += 1 severity = item.get("severity") if severity not in (None, ""): self.severities[f"S{severity}"] += 1 if item.get("blocked"): self.blocked += 1 if item.get("type") == "anomaly": self.anomalies += 1 if item.get("type") == "dns" and _text(item.get("dns_rcode"), 32).upper() == "NXDOMAIN": self.dns_nxdomain += 1 if item.get("type") == "fileinfo": 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 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 if direction in {"inbound", "internal"} and dest_ip and ether_dest: self.assets[f"{dest_ip} · {ether_dest}"] += 1 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 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 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 for label, key in ( ("JA4", "tls_ja4"), ("JA3", "tls_ja3"), ("QUIC JA4", "quic_ja4"), ("QUIC JA3", "quic_ja3"), ("HASSH-C", "ssh_hassh_client"), ("HASSH-S", "ssh_hassh_server"), ): value = _text(item.get(key), 160) if value: self.fingerprints[f"{label} {value}"] += 1 if item.get("type") in {"http", "ftp", "smtp"} or app_proto in {"http", "ftp", "smtp", "telnet"}: self.cleartext += 1 if is_flow: self.eve_flow_bytes += size if app_proto and self.track_high_cardinality: flow_identity = _text(item.get("flow_id") or item.get("community_id") or item.get("id"), 128) app_key = (app_proto, flow_identity) if app_key not in self.app_flow_seen: self.app_flow_seen.add(app_key) self.apps[app_proto] += 1 if is_flow: self.app_bytes[app_proto] += size if item.get("proto"): self.protocols[_text(item.get("proto"), 24)] += 1 if self.track_high_cardinality: if item.get("src_ip"): self.sources[_text(item.get("src_ip"), 64)] += 1 if item.get("dest_ip"): self.destinations[_text(item.get("dest_ip"), 64)] += 1 if direction == "outbound": if src_ip: self.local_clients[src_ip] += 1 self.local_client_bytes[src_ip] += size if dest_ip: self.remote_peers[dest_ip] += 1 self.remote_peer_bytes[dest_ip] += size elif direction == "inbound": if dest_ip: self.local_clients[dest_ip] += 1 self.local_client_bytes[dest_ip] += size if src_ip: self.remote_peers[src_ip] += 1 self.remote_peer_bytes[src_ip] += size elif direction == "internal": if src_ip: self.local_clients[src_ip] += 1 self.local_client_bytes[src_ip] += size if dest_ip and dest_ip != src_ip: self.local_clients[dest_ip] += 1 self.local_client_bytes[dest_ip] += size else: if src_ip: self.remote_peers[src_ip] += 1 self.remote_peer_bytes[src_ip] += size if dest_ip and dest_ip != src_ip: self.remote_peers[dest_ip] += 1 self.remote_peer_bytes[dest_ip] += size self.directions[direction] += 1 self.types[_text(item.get("type"), 32)] += 1 def add_throughput(self, sample: dict[str, Any]) -> None: ts = _safe_int(sample.get("ts_ms")) if ts < self.since_ms or ts > self.now_ms + 1000: return idx = min(max((ts - self.since_ms) // self.bin_ms, 0), self.bins_count - 1) bytes_in = max(_safe_int(sample.get("bytes_in")), 0) bytes_out = max(_safe_int(sample.get("bytes_out")), 0) bytes_total = max( _safe_int(sample.get("bytes_total")), bytes_in + bytes_out + max(_safe_int(sample.get("bytes_internal")), 0) + max(_safe_int(sample.get("bytes_external")), 0), ) packets_total = max(_safe_int(sample.get("packets_total")), 0) bucket = self.bins[idx] bucket["rate_bytes"] += bytes_total bucket["rate_bytes_in"] += bytes_in bucket["rate_bytes_out"] += bytes_out bucket["rate_packets"] += packets_total self.throughput_bytes += bytes_total self.throughput_classified_bytes += bytes_in + bytes_out self.throughput_packets += packets_total self.throughput_samples += 1 if self.latest_sample is None or ts > _safe_int(self.latest_sample.get("ts_ms")): self.latest_sample = sample def finish(self) -> dict[str, Any]: bucket_seconds = max(self.window_seconds / self.bins_count, 1) has_throughput = self.throughput_samples > 0 for bucket in self.bins: if has_throughput: bucket["bps"] = round(bucket.pop("rate_bytes") * 8 / bucket_seconds) bucket["in_bps"] = round(bucket.pop("rate_bytes_in") * 8 / bucket_seconds) bucket["out_bps"] = round(bucket.pop("rate_bytes_out") * 8 / bucket_seconds) bucket["pps"] = round(bucket.pop("rate_packets") / bucket_seconds, 2) else: bucket.pop("rate_bytes", None) bucket.pop("rate_bytes_in", None) bucket.pop("rate_bytes_out", None) bucket.pop("rate_packets", None) bucket["bps"] = round(bucket["bytes"] * 8 / bucket_seconds) bucket["in_bps"] = round(bucket["bytes_in"] * 8 / bucket_seconds) bucket["out_bps"] = round(bucket["bytes_out"] * 8 / bucket_seconds) bucket["pps"] = round(bucket["packets"] / bucket_seconds, 2) bucket["other_bps"] = max(0, bucket["bps"] - bucket["in_bps"] - bucket["out_bps"]) latest_sample = self.latest_sample if latest_sample is not None: interval = max(float(latest_sample.get("interval_ms") or 1000) / 1000.0, 0.001) sample_age_ms = max(0, self.now_ms - _safe_int(latest_sample.get("ts_ms"))) if sample_age_ms > max(3000, round(interval * 3000)): current_bps = current_in_bps = current_out_bps = current_pps = 0 else: current_bps = round(max(_safe_int(latest_sample.get("bytes_total")), 0) * 8 / interval) current_in_bps = round(max(_safe_int(latest_sample.get("bytes_in")), 0) * 8 / interval) current_out_bps = round(max(_safe_int(latest_sample.get("bytes_out")), 0) * 8 / interval) current_pps = round(max(_safe_int(latest_sample.get("packets_total")), 0) / interval, 2) else: current_bps = self.bins[-1]["bps"] if self.bins else 0 current_in_bps = self.bins[-1]["in_bps"] if self.bins else 0 current_out_bps = self.bins[-1]["out_bps"] if self.bins else 0 current_pps = self.bins[-1]["pps"] if self.bins else 0 current_other_bps = max(0, current_bps - current_in_bps - current_out_bps) direction_coverage_pct = round( (self.throughput_classified_bytes / self.throughput_bytes) * 100.0, 1 ) if self.throughput_bytes else 0.0 observed_bytes = self.throughput_bytes if has_throughput else self.eve_flow_bytes return { "window_seconds": self.window_seconds, "events": self.included_events, "bytes": observed_bytes, "eve_flow_bytes": self.eve_flow_bytes, "throughput_bytes": self.throughput_bytes, "throughput_packets": self.throughput_packets, "current_bps": current_bps, "current_in_bps": current_in_bps, "current_out_bps": current_out_bps, "current_other_bps": current_other_bps, "throughput_direction_coverage_pct": direction_coverage_pct, "current_pps": current_pps, "avg_bps": round(observed_bytes * 8 / max(self.window_seconds, 1)), "peak_bps": max((bucket["bps"] for bucket in self.bins), default=0), "peak_in_bps": max((bucket["in_bps"] for bucket in self.bins), default=0), "peak_out_bps": max((bucket["out_bps"] for bucket in self.bins), default=0), "alerts": self.alerts, "blocked": self.blocked, "anomalies": self.anomalies, "dns_nxdomain": self.dns_nxdomain, "files": self.files, "encrypted_sessions": self.encrypted, "cleartext_sessions": self.cleartext, "unique_local_clients": len(self.local_clients), "unique_remote_peers": len(self.remote_peers), "timeline": self.bins, "top_apps": _counter_rows(self.apps), "protocols": _counter_rows(self.protocols), "top_sources": _counter_rows(self.sources), "top_destinations": _counter_rows(self.destinations), "top_local_clients": _counter_rows(self.local_clients), "top_remote_peers": _counter_rows(self.remote_peers), "top_local_clients_by_bytes": _counter_rows_metric(self.local_client_bytes, "bytes"), "top_remote_peers_by_bytes": _counter_rows_metric(self.remote_peer_bytes, "bytes"), "top_apps_by_bytes": _counter_rows_metric(self.app_bytes, "bytes"), "directions": _counter_rows(self.directions), "event_types": _counter_rows(self.types), "top_signatures": _counter_rows(self.signatures), "severities": _counter_rows(self.severities), "top_fingerprints": _counter_rows(self.fingerprints), "top_assets": _counter_rows(self.assets), "top_files": _counter_rows(self.file_activity), } def _analytics( events: Iterable[dict[str, Any]], since_ms: int, now_ms: int, window_seconds: int, throughput_samples: Iterable[dict[str, Any]] | None = None, ) -> dict[str, Any]: accumulator = _AnalyticsAccumulator(since_ms, now_ms, window_seconds) for item in events: accumulator.add_event(item) for sample in throughput_samples or (): accumulator.add_throughput(sample) return accumulator.finish() def _counter_rows(counter: collections.Counter[str], limit: int = 10) -> list[dict[str, Any]]: return [{"name": name, "count": count} for name, count in counter.most_common(limit)] def _counter_rows_metric( counter: collections.Counter[str], key: str, limit: int = 10 ) -> list[dict[str, Any]]: return [{"name": name, key: value} for name, value in counter.most_common(limit)] class LiveEventPipeline: """Immediate WebSocket fan-out plus asynchronous Redis-buffer persistence.""" def __init__(self, bus: EventBus, history: TrafficHistory, queue_size: int = 10000) -> None: self.bus = bus self.history = history self._queue: queue.Queue[tuple[str, dict[str, Any]]] = queue.Queue(maxsize=max(1000, queue_size)) self._stop = threading.Event() self._thread = threading.Thread(target=self._run, name="traffic-history-writer", daemon=True) self._dropped = 0 self._written = 0 self._throughput_written = 0 self._redis_errors = 0 self._batches_written = 0 def start(self) -> None: if not self._thread.is_alive(): self._thread.start() def publish(self, event: dict[str, Any], persist: bool = True) -> None: self.bus.publish(event) if persist: self._enqueue("event", event) def publish_throughput(self, sample: dict[str, Any]) -> None: self._enqueue("throughput", sample) def has_live_subscribers(self) -> bool: return self.bus.has_subscribers() def _enqueue(self, kind: str, payload: dict[str, Any]) -> None: item = (kind, payload) try: self._queue.put_nowait(item) except queue.Full: try: self._queue.get_nowait() self._queue.task_done() self._queue.put_nowait(item) self._dropped += 1 except (queue.Empty, queue.Full): self._dropped += 1 def stop(self, timeout: float = 2.0) -> None: self._stop.set() if self._thread.is_alive(): self._thread.join(timeout=timeout) def status(self) -> dict[str, Any]: return { "writer_queue": self._queue.qsize(), "writer_dropped": self._dropped, "writer_written": self._written, "throughput_written": self._throughput_written, "writer_batches": self._batches_written, "writer_redis_errors": self._redis_errors, } def _run(self) -> None: max_batch = 128 while not self._stop.is_set() or not self._queue.empty(): try: first = self._queue.get(timeout=0.25) except queue.Empty: continue batch = [first] while len(batch) < max_batch: try: batch.append(self._queue.get_nowait()) except queue.Empty: break events = [payload for kind, payload in batch if kind == "event"] throughput = [payload for kind, payload in batch if kind == "throughput"] try: while True: try: if events: self.history.add_many(events) for sample in throughput: self.history.add_throughput_sample(sample) self._written += len(events) self._throughput_written += len(throughput) self._batches_written += 1 break except RedisUnavailableError: self._redis_errors += 1 # Redis is supervised in the same container. Keep this bounded # writer batch pending until it is ready; never switch history # reads to RAM or present partial data as complete. if self._stop.wait(0.5): break finally: for _ in batch: self._queue.task_done()