try nfstream

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
 

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

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),
     )
 
 

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
 

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(),
     }

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)

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())

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()