diff --git a/backend/Dockerfile b/backend/Dockerfile index 90e1863..2ffc1ab 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -2,6 +2,10 @@ FROM python:3.11-slim WORKDIR /app +RUN apt-get update \ + && apt-get install -y --no-install-recommends build-essential libpcap-dev pkg-config \ + && rm -rf /var/lib/apt/lists/* + COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt diff --git a/backend/requirements.txt b/backend/requirements.txt index 79ba04c..0b13e45 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -12,6 +12,7 @@ pydantic==2.12.3 pydantic_core==2.41.4 pyroute2==0.9.5 python-multipart==0.0.22 +nfstream==6.5.4 scapy==2.6.1 sniffio==1.3.1 starlette==0.48.0 diff --git a/backend/src/config.py b/backend/src/config.py index faec3bc..69de592 100644 --- a/backend/src/config.py +++ b/backend/src/config.py @@ -25,6 +25,13 @@ def _env_float(name: str, default: float) -> float: return float(value) +def _env_bool(name: str, default: bool) -> bool: + value = os.getenv(name) + if value in (None, ""): + return default + return value.strip().lower() in {"1", "true", "yes", "on"} + + @dataclass(frozen=True) class BackendSettings: db_dsn: str @@ -50,6 +57,16 @@ class BackendSettings: ndpi_max_flows: int ndpi_flow_ttl_seconds: float ndpi_cleanup_interval_packets: int + nfstream_enabled: bool + nfstream_promiscuous_mode: bool + nfstream_idle_timeout_seconds: int + nfstream_active_timeout_seconds: int + nfstream_snapshot_length: int + nfstream_n_dissections: int + nfstream_cache_ttl_seconds: float + nfstream_lookup_window_ms: int + nfstream_reader_join_timeout_seconds: float + nfstream_process_stop_timeout_seconds: float def load_settings() -> BackendSettings: @@ -80,6 +97,16 @@ def load_settings() -> BackendSettings: ndpi_max_flows=_env_int("BACKEND_NDPI_MAX_FLOWS", 200_000), ndpi_flow_ttl_seconds=_env_float("BACKEND_NDPI_FLOW_TTL_SECONDS", 120.0), ndpi_cleanup_interval_packets=_env_int("BACKEND_NDPI_CLEANUP_INTERVAL_PACKETS", 10_000), + nfstream_enabled=_env_bool("BACKEND_NFSTREAM_ENABLED", True), + nfstream_promiscuous_mode=_env_bool("BACKEND_NFSTREAM_PROMISCUOUS_MODE", True), + nfstream_idle_timeout_seconds=_env_int("BACKEND_NFSTREAM_IDLE_TIMEOUT_SECONDS", 120), + nfstream_active_timeout_seconds=_env_int("BACKEND_NFSTREAM_ACTIVE_TIMEOUT_SECONDS", 1800), + nfstream_snapshot_length=_env_int("BACKEND_NFSTREAM_SNAPSHOT_LENGTH", 1536), + nfstream_n_dissections=_env_int("BACKEND_NFSTREAM_N_DISSECTIONS", 20), + nfstream_cache_ttl_seconds=_env_float("BACKEND_NFSTREAM_CACHE_TTL_SECONDS", 10.0), + nfstream_lookup_window_ms=_env_int("BACKEND_NFSTREAM_LOOKUP_WINDOW_MS", 5_000), + nfstream_reader_join_timeout_seconds=_env_float("BACKEND_NFSTREAM_READER_JOIN_TIMEOUT_SECONDS", 2.0), + nfstream_process_stop_timeout_seconds=_env_float("BACKEND_NFSTREAM_PROCESS_STOP_TIMEOUT_SECONDS", 3.0), ) diff --git a/backend/src/main.py b/backend/src/main.py index 004cb84..b93687e 100644 --- a/backend/src/main.py +++ b/backend/src/main.py @@ -94,6 +94,13 @@ async def shutdown_event() -> None: except Exception: logging.exception("Failed to stop bridge telemetry collector") + try: + from src.utilities.nfstream_manager import nfstream_manager + + nfstream_manager.stop() + except Exception: + logging.exception("Failed to stop NFStream workers") + try: from src.utilities.packet_tracker import packet_tracker diff --git a/backend/src/network_sniffer.py b/backend/src/network_sniffer.py index c110347..baba608 100644 --- a/backend/src/network_sniffer.py +++ b/backend/src/network_sniffer.py @@ -34,6 +34,7 @@ from src.utilities.interface_bridge_helpers import ( from src.config import settings from src.utilities.bridge_telemetry import bridge_telemetry_manager from src.utilities.ndpi_classifier import ndpi_classifier +from src.utilities.nfstream_manager import nfstream_manager from src.utilities.packet_identity import build_packet_uid from src.utilities.packet_tracker import packet_tracker from src.Models.etherType import EtherTypeEnum, ethertype_from_int @@ -162,6 +163,22 @@ def _safe_get_attr(layer, attr: str): return None +def _merge_enrichment(pkt_info: PacketInfo, enrichment: Dict[str, Any]) -> None: + """Populate enrichment fields without discarding existing metadata.""" + for key, value in enrichment.items(): + if value is None: + continue + if key == "dpi_metadata": + current = pkt_info.get("dpi_metadata") + if isinstance(current, dict) and isinstance(value, dict): + pkt_info["dpi_metadata"] = {**current, **value} + elif current is None: + pkt_info["dpi_metadata"] = value + continue + if pkt_info.get(key) is None: + pkt_info[key] = value + + def parse_packet(pkt, bridge_label: str, capture_metadata: Optional[Dict[str, Any]] = None) -> None: """ Parse a scapy Packet object into a normalized PacketInfo and schedule DB insert. @@ -355,11 +372,19 @@ def parse_packet(pkt, bridge_label: str, capture_metadata: Optional[Dict[str, An if Raw in pkt and not pkt_info.get("protocol_name"): pkt_info["protocol_name"] = "RAW" + # Flow-level enrichment using NFStream, if available. + try: + flow_info = nfstream_manager.lookup_packet(pkt, pkt_iface) + if flow_info: + _merge_enrichment(pkt_info, flow_info) + except Exception: + logger.exception("NFStream enrichment failed") + # Best-effort DPI enrichment using nDPI (optional dependency). try: dpi_info = ndpi_classifier.classify_packet(pkt) if dpi_info: - pkt_info.update(dpi_info) + _merge_enrichment(pkt_info, dpi_info) except Exception: logger.exception("nDPI enrichment failed") @@ -508,6 +533,22 @@ def _sync_bridge_telemetry() -> None: except Exception: logger.exception("Failed to update bridge telemetry collector") + active_enrichment_ifaces = sorted( + { + iface + for session in sessions.values() + for iface in ( + list(session.get("capture_ifaces", [])) + + (list(session.get("ports", [])) if session.get("is_bridge") else []) + ) + if iface + } + ) + try: + nfstream_manager.update_interfaces(active_enrichment_ifaces) + except Exception: + logger.exception("Failed to update NFStream enrichment workers") + # ------------------------- # Per-session reader loop @@ -727,6 +768,10 @@ def stop_afpacket_sniffer(session_id: Optional[str] = None, target: Optional[str s["ports"].remove(target) except Exception: pass + try: + s["capture_ifaces"].remove(target) + except Exception: + pass logger.info("Removed target %s from session %s", target, sid) _sync_bridge_telemetry() return @@ -803,5 +848,6 @@ def get_internal_debug_state() -> dict: "buffer_len": len(_PACKET_BUFFER), "bridge_capture_mode": "tc_ingress_raw" if any(s.get("is_bridge") for s in sessions.values()) else "af_packet", "telemetry_ports": sorted({iface for session in sessions.values() for iface in session.get("ports", [])}), + "nfstream": nfstream_manager.get_debug_snapshot(), "packet_tracker": packet_tracker.get_debug_snapshot(), } diff --git a/backend/src/utilities/flow_identity.py b/backend/src/utilities/flow_identity.py new file mode 100644 index 0000000..66ef650 --- /dev/null +++ b/backend/src/utilities/flow_identity.py @@ -0,0 +1,77 @@ +"""Helpers for canonical bidirectional flow keys used by enrichers.""" + +from __future__ import annotations + +import time +from typing import Any, Optional, Tuple + +from scapy.all import IP, IPv6, TCP, UDP # type: ignore + + +def flow_key_from_endpoints( + ip_version: int, + protocol: int, + src_ip: Any, + src_port: Any, + dst_ip: Any, + dst_port: Any, +) -> Optional[str]: + """Build a stable bidirectional key for IPv4/IPv6 TCP/UDP flows.""" + if src_ip in (None, "") or dst_ip in (None, ""): + return None + + try: + left = (str(src_ip), int(src_port or 0)) + right = (str(dst_ip), int(dst_port or 0)) + ep1, ep2 = (left, right) if left <= right else (right, left) + return f"{int(ip_version)}|{int(protocol)}|{ep1[0]}|{ep1[1]}|{ep2[0]}|{ep2[1]}" + except Exception: + return None + + +def flow_key_from_packet(pkt: Any) -> Optional[str]: + """Extract a canonical flow key from a Scapy packet.""" + if IP in pkt: + ip_layer = pkt[IP] + src_ip = getattr(ip_layer, "src", None) + dst_ip = getattr(ip_layer, "dst", None) + protocol = int(getattr(ip_layer, "proto", 0) or 0) + ip_version = 4 + elif IPv6 in pkt: + ip_layer = pkt[IPv6] + src_ip = getattr(ip_layer, "src", None) + dst_ip = getattr(ip_layer, "dst", None) + protocol = int(getattr(ip_layer, "nh", 0) or 0) + ip_version = 6 + else: + return None + + if protocol not in (6, 17): + return None + + src_port = 0 + dst_port = 0 + if TCP in pkt: + src_port = int(getattr(pkt[TCP], "sport", 0) or 0) + dst_port = int(getattr(pkt[TCP], "dport", 0) or 0) + elif UDP in pkt: + src_port = int(getattr(pkt[UDP], "sport", 0) or 0) + dst_port = int(getattr(pkt[UDP], "dport", 0) or 0) + + return flow_key_from_endpoints(ip_version, protocol, src_ip, src_port, dst_ip, dst_port) + + +def packet_observed_at_ms(pkt: Any) -> int: + """Return packet timestamp in milliseconds with a real-time fallback.""" + try: + packet_time = float(getattr(pkt, "time", 0.0) or 0.0) + if packet_time > 0: + return int(packet_time * 1000) + except Exception: + pass + return int(time.time() * 1000) + + +def flow_cache_key(iface: str, flow_key: str) -> Tuple[str, str]: + """Key active flow enrichment by interface and canonical flow ID.""" + return (str(iface or ""), flow_key) diff --git a/backend/src/utilities/nfstream_flow_worker.py b/backend/src/utilities/nfstream_flow_worker.py new file mode 100644 index 0000000..d995366 --- /dev/null +++ b/backend/src/utilities/nfstream_flow_worker.py @@ -0,0 +1,166 @@ +"""Run NFStream on one interface and emit flow metadata as JSON lines.""" + +from __future__ import annotations + +import argparse +import json +import logging +import signal +import sys +import time +from typing import Any, Optional + +logger = logging.getLogger("nfstream_flow_worker") +_STOP = False + + +def _handle_signal(signum: int, _frame: Any) -> None: + global _STOP + _STOP = True + logger.info("Received signal %s, stopping NFStream worker", signum) + + +def _safe_scalar(value: Any) -> Any: + if isinstance(value, (str, int, float, bool)) or value is None: + return value + return str(value) + + +def _safe_int(value: Any) -> Optional[int]: + try: + if value is None: + return None + return int(value) + except Exception: + return None + + +def _build_event(iface: str, flow: Any, event_name: str) -> Optional[dict[str, Any]]: + ip_version = _safe_int(getattr(flow, "ip_version", None)) + protocol = _safe_int(getattr(flow, "protocol", None)) + src_ip = getattr(flow, "src_ip", None) + dst_ip = getattr(flow, "dst_ip", None) + src_port = _safe_int(getattr(flow, "src_port", None)) + dst_port = _safe_int(getattr(flow, "dst_port", None)) + if ip_version is None or protocol is None or not src_ip or not dst_ip: + return None + + return { + "type": "flow_update", + "event": event_name, + "iface": iface, + "ip_version": ip_version, + "protocol": protocol, + "src_ip": str(src_ip), + "dst_ip": str(dst_ip), + "src_port": src_port or 0, + "dst_port": dst_port or 0, + "first_seen_ms": _safe_int(getattr(flow, "bidirectional_first_seen_ms", None)) + or _safe_int(getattr(flow, "src2dst_first_seen_ms", None)) + or int(time.time() * 1000), + "last_seen_ms": _safe_int(getattr(flow, "bidirectional_last_seen_ms", None)) + or _safe_int(getattr(flow, "src2dst_last_seen_ms", None)) + or int(time.time() * 1000), + "application_name": _safe_scalar(getattr(flow, "application_name", None)), + "application_category_name": _safe_scalar(getattr(flow, "application_category_name", None)), + "application_confidence": _safe_scalar(getattr(flow, "application_confidence", None)), + "requested_server_name": _safe_scalar(getattr(flow, "requested_server_name", None)), + "client_fingerprint": _safe_scalar(getattr(flow, "client_fingerprint", None)), + "server_fingerprint": _safe_scalar(getattr(flow, "server_fingerprint", None)), + "user_agent": _safe_scalar(getattr(flow, "user_agent", None)), + "content_type": _safe_scalar(getattr(flow, "content_type", None)), + "bidirectional_packets": _safe_int(getattr(flow, "bidirectional_packets", None)), + "bidirectional_bytes": _safe_int(getattr(flow, "bidirectional_bytes", None)), + } + + +def _emit(payload: dict[str, Any]) -> None: + sys.stdout.write(json.dumps(payload, separators=(",", ":")) + "\n") + sys.stdout.flush() + + +def main() -> int: + parser = argparse.ArgumentParser(description="NFStream flow worker") + parser.add_argument("--iface", required=True) + parser.add_argument("--idle-timeout", type=int, required=True) + parser.add_argument("--active-timeout", type=int, required=True) + parser.add_argument("--snapshot-length", type=int, required=True) + parser.add_argument("--n-dissections", type=int, required=True) + parser.add_argument("--promiscuous-mode", action="store_true") + args = parser.parse_args() + + signal.signal(signal.SIGTERM, _handle_signal) + signal.signal(signal.SIGINT, _handle_signal) + + try: + from nfstream import NFPlugin, NFStreamer + except Exception as exc: + _emit( + { + "type": "worker_error", + "iface": args.iface, + "message": f"Failed to import nfstream: {exc!r}", + } + ) + return 1 + + class EmitFlowMetadata(NFPlugin): + def on_init(self, _packet: Any, flow: Any) -> None: + payload = _build_event(args.iface, flow, "init") + if payload: + _emit(payload) + + def on_update(self, _packet: Any, flow: Any) -> None: + payload = _build_event(args.iface, flow, "update") + if payload: + _emit(payload) + + def on_expire(self, flow: Any) -> None: + payload = _build_event(args.iface, flow, "expire") + if payload: + _emit(payload) + + try: + streamer = NFStreamer( + source=args.iface, + promiscuous_mode=bool(args.promiscuous_mode), + snapshot_length=args.snapshot_length, + idle_timeout=args.idle_timeout, + active_timeout=args.active_timeout, + n_dissections=args.n_dissections, + statistical_analysis=False, + accounting_mode=0, + udps=EmitFlowMetadata(), + ) + except Exception as exc: + _emit( + { + "type": "worker_error", + "iface": args.iface, + "message": f"Failed to start NFStreamer: {exc!r}", + } + ) + return 1 + + _emit({"type": "worker_ready", "iface": args.iface}) + + try: + for _flow in streamer: + if _STOP: + break + except Exception as exc: + _emit( + { + "type": "worker_error", + "iface": args.iface, + "message": f"NFStreamer runtime error: {exc!r}", + } + ) + return 1 + + return 0 + + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO) + raise SystemExit(main()) diff --git a/backend/src/utilities/nfstream_manager.py b/backend/src/utilities/nfstream_manager.py new file mode 100644 index 0000000..6b474a7 --- /dev/null +++ b/backend/src/utilities/nfstream_manager.py @@ -0,0 +1,284 @@ +"""Manage optional NFStream flow enrichment workers and packet lookups.""" + +from __future__ import annotations + +import json +import logging +import os +import signal +import subprocess +import sys +import threading +import time +from pathlib import Path +from typing import Any, Dict, Iterable, Optional + +from src.config import settings +from src.utilities.flow_identity import flow_cache_key, flow_key_from_endpoints, flow_key_from_packet, packet_observed_at_ms + +logger = logging.getLogger("nfstream_manager") + + +def _merge_metadata(base: Optional[Dict[str, Any]], extra: Dict[str, Any]) -> Dict[str, Any]: + merged = dict(base or {}) + for key, value in extra.items(): + if value not in (None, "", [], {}): + merged[key] = value + return merged + + +def _infer_is_encrypted(payload: Dict[str, Any]) -> Optional[bool]: + application_name = str(payload.get("application_name") or "").upper() + if any(token in application_name for token in ("TLS", "HTTPS", "QUIC", "SSL")): + return True + if payload.get("requested_server_name") or payload.get("client_fingerprint") or payload.get("server_fingerprint"): + return True + return None + + +class NFStreamManager: + """Own long-lived NFStream subprocesses and a short-lived flow metadata cache.""" + + def __init__(self) -> None: + self._enabled = settings.nfstream_enabled + self._workers: Dict[str, Dict[str, Any]] = {} + self._cache: Dict[tuple[str, str], Dict[str, Any]] = {} + self._last_error_by_iface: Dict[str, str] = {} + self._lock = threading.Lock() + + @property + def enabled(self) -> bool: + return self._enabled + + def update_interfaces(self, interfaces: Iterable[str]) -> None: + """Align worker set with currently active interfaces.""" + if not self._enabled: + return + + targets = {iface.strip() for iface in interfaces if iface and iface.strip()} + with self._lock: + current = set(self._workers.keys()) + for iface in sorted(current - targets): + self._stop_worker_locked(iface) + for iface in sorted(targets - current): + self._start_worker_locked(iface) + self._purge_cache_locked() + + def stop(self) -> None: + """Stop all workers and clear transient state.""" + with self._lock: + for iface in list(self._workers.keys()): + self._stop_worker_locked(iface) + self._cache.clear() + + def lookup_packet(self, pkt: Any, iface: Optional[str]) -> Dict[str, Any]: + """Return NFStream-derived enrichment for the packet if a recent flow is known.""" + if not self._enabled or not iface: + return {} + + flow_key = flow_key_from_packet(pkt) + if flow_key is None: + return {} + + now_ms = packet_observed_at_ms(pkt) + with self._lock: + self._purge_cache_locked(now_ms=now_ms) + payload = self._cache.get(flow_cache_key(iface, flow_key)) + if payload is None: + return {} + + first_seen_ms = int(payload.get("first_seen_ms") or 0) + last_seen_ms = int(payload.get("last_seen_ms") or 0) + window_ms = settings.nfstream_lookup_window_ms + if now_ms + window_ms < first_seen_ms or now_ms - window_ms > last_seen_ms: + return {} + + metadata = { + "iface": iface, + "flow_key": flow_key, + "first_seen_ms": first_seen_ms or None, + "last_seen_ms": last_seen_ms or None, + "event": payload.get("event"), + "bidirectional_packets": payload.get("bidirectional_packets"), + "bidirectional_bytes": payload.get("bidirectional_bytes"), + } + metadata = _merge_metadata( + metadata, + { + "requested_server_name": payload.get("requested_server_name"), + "client_fingerprint": payload.get("client_fingerprint"), + "server_fingerprint": payload.get("server_fingerprint"), + "user_agent": payload.get("user_agent"), + "content_type": payload.get("content_type"), + }, + ) + + return { + "app_protocol": payload.get("application_name"), + "app_master_protocol": payload.get("application_name"), + "app_category": payload.get("application_category_name"), + "app_confidence": payload.get("application_confidence"), + "app_hostname": payload.get("requested_server_name"), + "app_is_encrypted": _infer_is_encrypted(payload), + "dpi_metadata": {"nfstream": metadata}, + } + + def get_debug_snapshot(self) -> Dict[str, Any]: + """Expose current worker/cache state for API debugging.""" + with self._lock: + self._purge_cache_locked() + return { + "enabled": self._enabled, + "workers": { + iface: { + "running": bool(worker.get("process") and worker["process"].poll() is None), + "thread_alive": bool(worker.get("reader") and worker["reader"].is_alive()), + "last_error": self._last_error_by_iface.get(iface), + } + for iface, worker in self._workers.items() + }, + "cache_entries": len(self._cache), + "last_error_by_iface": dict(self._last_error_by_iface), + } + + def _start_worker_locked(self, iface: str) -> None: + helper = Path(__file__).with_name("nfstream_flow_worker.py") + python_bin = sys.executable or "python3" + + env = os.environ.copy() + env["PYTHONUNBUFFERED"] = "1" + backend_root = str(helper.parents[2]) + existing_pythonpath = env.get("PYTHONPATH", "") + env["PYTHONPATH"] = backend_root if not existing_pythonpath else f"{backend_root}:{existing_pythonpath}" + + cmd = [ + python_bin, + str(helper), + "--iface", + iface, + "--idle-timeout", + str(settings.nfstream_idle_timeout_seconds), + "--active-timeout", + str(settings.nfstream_active_timeout_seconds), + "--snapshot-length", + str(settings.nfstream_snapshot_length), + "--n-dissections", + str(settings.nfstream_n_dissections), + ] + if settings.nfstream_promiscuous_mode: + cmd.append("--promiscuous-mode") + + logger.info("Starting NFStream worker for %s", iface) + try: + process = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + env=env, + ) + except Exception: + logger.exception("Failed to start NFStream worker for %s", iface) + self._last_error_by_iface[iface] = "failed_to_start_process" + return + + reader = threading.Thread( + target=self._read_loop, + args=(iface, process), + daemon=True, + name=f"nfstream-reader-{iface}", + ) + self._workers[iface] = { + "process": process, + "reader": reader, + } + reader.start() + + def _stop_worker_locked(self, iface: str) -> None: + worker = self._workers.pop(iface, None) + if worker is None: + return + + process = worker.get("process") + reader = worker.get("reader") + + if process is not None and process.poll() is None: + try: + process.send_signal(signal.SIGTERM) + process.wait(timeout=settings.nfstream_process_stop_timeout_seconds) + except subprocess.TimeoutExpired: + process.kill() + except Exception: + logger.exception("Failed to stop NFStream worker for %s cleanly", iface) + + if reader is not None and reader.is_alive(): + reader.join(timeout=settings.nfstream_reader_join_timeout_seconds) + + for key in [cache_key for cache_key in self._cache.keys() if cache_key[0] == iface]: + self._cache.pop(key, None) + + def _read_loop(self, iface: str, process: subprocess.Popen[str]) -> None: + stdout = process.stdout + if stdout is None: + return + + for line in stdout: + text = line.strip() + if not text: + continue + try: + payload = json.loads(text) + except json.JSONDecodeError: + logger.info("nfstream[%s]: %s", iface, text) + continue + + payload_type = payload.get("type") + if payload_type == "worker_ready": + logger.info("NFStream worker ready on %s", iface) + continue + if payload_type == "worker_error": + message = str(payload.get("message") or "unknown error") + logger.warning("NFStream worker error on %s: %s", iface, message) + with self._lock: + self._last_error_by_iface[iface] = message + continue + if payload_type != "flow_update": + logger.debug("Ignoring NFStream payload on %s: %s", iface, payload) + continue + + flow_key = flow_key_from_endpoints( + payload.get("ip_version"), + payload.get("protocol"), + payload.get("src_ip"), + payload.get("src_port"), + payload.get("dst_ip"), + payload.get("dst_port"), + ) + if flow_key is None: + continue + + with self._lock: + self._cache[flow_cache_key(iface, flow_key)] = dict(payload) + self._purge_cache_locked() + + rc = process.poll() + if rc not in (0, None): + logger.warning("NFStream worker for %s exited with code %s", iface, rc) + + def _purge_cache_locked(self, now_ms: Optional[int] = None) -> None: + if now_ms is None: + now_ms = int(time.time() * 1000) + expiry_ms = int(settings.nfstream_cache_ttl_seconds * 1000) + + stale_keys = [] + for key, payload in self._cache.items(): + last_seen_ms = int(payload.get("last_seen_ms") or 0) + if last_seen_ms and now_ms - last_seen_ms > expiry_ms: + stale_keys.append(key) + + for key in stale_keys: + self._cache.pop(key, None) + + +nfstream_manager = NFStreamManager()