1787 lines
77 KiB
Python
1787 lines
77 KiB
Python
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",
|
|
}
|
|
MAX_ANALYTICS_DIMENSION_KEYS = 4096
|
|
|
|
|
|
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 _bounded_counter_add(
|
|
counter: collections.Counter[str],
|
|
key: str,
|
|
amount: int = 1,
|
|
*,
|
|
max_keys: int = MAX_ANALYTICS_DIMENSION_KEYS,
|
|
) -> None:
|
|
"""Update a dashboard Counter without retaining unbounded unique values."""
|
|
if not key:
|
|
return
|
|
if key in counter or len(counter) < max_keys:
|
|
counter[key] += amount
|
|
|
|
|
|
def _parse_networks(value: str) -> list[ipaddress._BaseNetwork]:
|
|
result: list[ipaddress._BaseNetwork] = []
|
|
for raw in (value or "").split(","):
|
|
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()
|
|
interval = 1.0 if self.max_events > 0 else 15.0
|
|
if now - self._last_trim > interval:
|
|
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:
|
|
_bounded_counter_add(self.signatures, signature)
|
|
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)
|
|
_bounded_counter_add(self.file_activity, f"{filename}{' · ' + digest if digest else ''}")
|
|
direction = _text(item.get("direction"), 24) or "unknown"
|
|
src_ip = _text(item.get("src_ip"), 64)
|
|
dest_ip = _text(item.get("dest_ip"), 64)
|
|
ether_src = _text(item.get("ether_src"), 32)
|
|
ether_dest = _text(item.get("ether_dest"), 32)
|
|
if direction in {"outbound", "internal"} and src_ip and ether_src:
|
|
_bounded_counter_add(self.assets, f"{src_ip} · {ether_src}")
|
|
if direction in {"inbound", "internal"} and dest_ip and ether_dest:
|
|
_bounded_counter_add(self.assets, f"{dest_ip} · {ether_dest}")
|
|
if item.get("type") == "dhcp":
|
|
asset_ip = _text(item.get("dhcp_assigned_ip") or item.get("src_ip"), 64)
|
|
identity = _text(item.get("dhcp_hostname") or item.get("dhcp_client_mac"), 160)
|
|
if asset_ip or identity:
|
|
_bounded_counter_add(self.assets, f"{asset_ip}{' · ' if asset_ip and identity else ''}{identity}")
|
|
elif item.get("type") == "arp":
|
|
asset_ip = _text(item.get("arp_src_ip") or item.get("src_ip"), 64)
|
|
mac = _text(item.get("arp_src_mac"), 32)
|
|
if asset_ip or mac:
|
|
_bounded_counter_add(self.assets, f"{asset_ip}{' · ' if asset_ip and mac else ''}{mac}")
|
|
app_proto = _valid_app_proto(item.get("app_proto"))
|
|
if item.get("type") in {"tls", "quic", "ssh"} or app_proto in {"tls", "quic", "ssh"}:
|
|
self.encrypted += 1
|
|
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:
|
|
_bounded_counter_add(self.fingerprints, f"{label} {value}")
|
|
if item.get("type") in {"http", "ftp", "smtp"} or app_proto in {"http", "ftp", "smtp", "telnet"}:
|
|
self.cleartext += 1
|
|
if is_flow:
|
|
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()
|
|
|