remove ndpi

backend/src/Models/packets.py

@@ -83,12 +83,12 @@ class PacketDBModel(BaseModel):
     raw_b64: Optional[str] = Field(None, description="Base64-encoded packet bytes.")
     app_protocol: Optional[str] = Field(None, description="Detected application protocol.")
     app_master_protocol: Optional[str] = Field(None, description="Detected application master protocol.")
-    app_category: Optional[str] = Field(None, description="nDPI category, if available.")
-    app_confidence: Optional[str] = Field(None, description="nDPI confidence level, if available.")
+    app_category: Optional[str] = Field(None, description="Detected application category, if available.")
+    app_confidence: Optional[str] = Field(None, description="Application detection confidence, if available.")
     app_hostname: Optional[str] = Field(None, description="Detected hostname/SNI, if available.")
     app_is_encrypted: Optional[bool] = Field(None, description="Whether detected protocol appears encrypted.")
-    app_risk_score: Optional[int] = Field(None, description="Count/score of detected nDPI risks.")
-    dpi_metadata: Optional[dict] = Field(None, description="Raw DPI metadata from nDPI.")
+    app_risk_score: Optional[int] = Field(None, description="Count/score of detected application risks.")
+    dpi_metadata: Optional[dict] = Field(None, description="Raw metadata from DPI/flow enrichment.")
     capture_metadata: Optional[dict] = Field(None, description="Raw-capture metadata from the bridge tc ingress exporter.")
     telemetry_metadata: Optional[dict] = Field(None, description="Kernel telemetry details from eBPF collector.")
     verdict: Optional[str] = None

backend/src/config.py

@@ -54,9 +54,6 @@ class BackendSettings:
     bridge_bpf_build_dir: str
     telemetry_process_stop_timeout_seconds: float
     telemetry_reader_join_timeout_seconds: float
-    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
@@ -94,9 +91,6 @@ def load_settings() -> BackendSettings:
         bridge_bpf_build_dir=_env_str("BACKEND_BRIDGE_BPF_BUILD_DIR", "/tmp/mitm-bpf"),
         telemetry_process_stop_timeout_seconds=_env_float("BACKEND_TELEMETRY_PROCESS_STOP_TIMEOUT_SECONDS", 3.0),
         telemetry_reader_join_timeout_seconds=_env_float("BACKEND_TELEMETRY_READER_JOIN_TIMEOUT_SECONDS", 2.0),
-        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),

backend/src/network_sniffer.py

@@ -33,7 +33,6 @@ 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
@@ -380,14 +379,6 @@ def parse_packet(pkt, bridge_label: str, capture_metadata: Optional[Dict[str, An
     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:
-            _merge_enrichment(pkt_info, dpi_info)
-    except Exception:
-        logger.exception("nDPI enrichment failed")
-
     if ICMP in pkt:
         inner = pkt[ICMP].payload
         if inner and IP in inner:

backend/src/utilities/ndpi_classifier.py

@@ -1,277 +0,0 @@
-"""Best-effort nDPI flow classifier wrapper.
-
-This module keeps nDPI integration optional:
-- If the Python nDPI bindings are present, packets are classified per flow.
-- If bindings are missing or fail, callers still receive stable fallback fields.
-"""
-
-from __future__ import annotations
-
-import logging
-import importlib
-import time
-from dataclasses import dataclass
-from typing import Any, Dict, Optional, Tuple
-
-from scapy.all import IP, IPv6, TCP, UDP  # type: ignore
-
-from src.config import settings
-
-logger = logging.getLogger("ndpi_classifier")
-
-ndpi = None  # type: ignore
-_ndpi_import_error: Optional[str] = None
-for _module_name in ("ndpi", "pyndpi", "nDPI"):
-    try:  # pragma: no cover - import depends on host runtime
-        ndpi = importlib.import_module(_module_name)  # type: ignore
-        break
-    except Exception as exc:
-        _ndpi_import_error = f"{_module_name}: {exc!r}"
-
-
-def _is_jsonable_scalar(value: Any) -> bool:
-    return isinstance(value, (str, int, float, bool)) or value is None
-
-
-def _safe_string(value: Any) -> Optional[str]:
-    if value is None:
-        return None
-    text = str(value).strip()
-    return text if text else None
-
-
-@dataclass
-class _FlowState:
-    flow_obj: Any
-    last_seen_ts: float
-    packet_count: int = 0
-    detected_app: Optional[str] = None
-
-
-class NDPIClassifier:
-    """Stateful nDPI classifier for TCP/UDP IPv4/IPv6 traffic."""
-
-    def __init__(self) -> None:
-        self._engine = None
-        self._flows: Dict[Tuple[Any, ...], _FlowState] = {}
-        self._max_flows = settings.ndpi_max_flows
-        self._flow_ttl_seconds = settings.ndpi_flow_ttl_seconds
-        self._cleanup_interval_packets = settings.ndpi_cleanup_interval_packets
-        self._packet_counter = 0
-        self._initialized = False
-
-    @property
-    def enabled(self) -> bool:
-        return self._engine is not None
-
-    def _init_engine(self) -> None:
-        if self._initialized:
-            return
-        self._initialized = True
-
-        if ndpi is None:
-            logger.warning(
-                "nDPI Python bindings unavailable; DPI enrichment disabled (%s)",
-                _ndpi_import_error or "no ndpi module found",
-            )
-            return
-
-        try:
-            engine_cls = getattr(ndpi, "NDPI", None)
-            if engine_cls is None:
-                exported = [name for name in dir(ndpi) if name.lower().startswith("ndpi")]
-                logger.warning(
-                    "nDPI bindings imported but NDPI class missing (module=%s, ndpi* exports=%s); DPI enrichment disabled",
-                    getattr(ndpi, "__name__", "unknown"),
-                    exported[:20],
-                )
-                return
-            self._engine = engine_cls()
-            logger.info("nDPI classifier initialized")
-        except Exception:
-            logger.exception("Failed to initialize nDPI classifier; enrichment disabled")
-            self._engine = None
-
-    def _maybe_cleanup(self, now_ts: float) -> None:
-        self._packet_counter += 1
-        if self._packet_counter % self._cleanup_interval_packets != 0:
-            return
-
-        expiry = now_ts - self._flow_ttl_seconds
-        stale_keys = [k for k, state in self._flows.items() if state.last_seen_ts < expiry]
-        for key in stale_keys:
-            self._flows.pop(key, None)
-
-        if len(self._flows) > self._max_flows:
-            # Drop oldest half of excess flows.
-            sorted_items = sorted(self._flows.items(), key=lambda kv: kv[1].last_seen_ts)
-            trim_count = len(self._flows) - self._max_flows // 2
-            for key, _ in sorted_items[:trim_count]:
-                self._flows.pop(key, None)
-
-    def _flow_key(self, pkt: Any) -> Optional[Tuple[Any, ...]]:
-        if IP in pkt:
-            ip_layer = pkt[IP]
-            src_ip = getattr(ip_layer, "src", None)
-            dst_ip = getattr(ip_layer, "dst", None)
-            proto = int(getattr(ip_layer, "proto", 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)
-            proto = int(getattr(ip_layer, "nh", 0))
-            ip_version = 6
-        else:
-            return None
-
-        src_port = None
-        dst_port = None
-        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)
-
-        left = (src_ip, src_port)
-        right = (dst_ip, dst_port)
-        ep1, ep2 = (left, right) if left <= right else (right, left)
-        return (ip_version, proto, ep1, ep2)
-
-    def _packet_payload_for_ndpi(self, pkt: Any) -> Optional[bytes]:
-        try:
-            if IP in pkt:
-                return bytes(pkt[IP])
-            if IPv6 in pkt:
-                return bytes(pkt[IPv6])
-        except Exception:
-            return None
-        return None
-
-    def _extract_result(self, result: Any) -> Dict[str, Any]:
-        if result is None:
-            return {}
-
-        raw: Dict[str, Any] = {}
-        if isinstance(result, dict):
-            raw.update(result)
-        else:
-            for attr in (
-                "app_protocol",
-                "application_protocol",
-                "master_protocol",
-                "protocol",
-                "category",
-                "confidence",
-                "hostname",
-                "server_name",
-                "sni",
-                "is_encrypted",
-                "risk_score",
-                "risks",
-                "ja3",
-                "ja4",
-                "alpn",
-            ):
-                if hasattr(result, attr):
-                    raw[attr] = getattr(result, attr)
-
-        app_protocol = (
-            _safe_string(raw.get("app_protocol"))
-            or _safe_string(raw.get("application_protocol"))
-            or _safe_string(raw.get("protocol"))
-        )
-        master_protocol = _safe_string(raw.get("master_protocol"))
-        category = _safe_string(raw.get("category"))
-        confidence = _safe_string(raw.get("confidence"))
-        hostname = _safe_string(raw.get("hostname")) or _safe_string(raw.get("server_name")) or _safe_string(raw.get("sni"))
-
-        risks_val = raw.get("risks")
-        if isinstance(risks_val, (list, tuple, set)):
-            risk_score = len(risks_val)
-        else:
-            try:
-                risk_score = int(raw.get("risk_score")) if raw.get("risk_score") is not None else None
-            except Exception:
-                risk_score = None
-
-        encrypted = raw.get("is_encrypted")
-        is_encrypted = bool(encrypted) if isinstance(encrypted, bool) else None
-        if is_encrypted is None and app_protocol:
-            proto_upper = app_protocol.upper()
-            if any(token in proto_upper for token in ("TLS", "SSL", "HTTPS", "QUIC", "VPN")):
-                is_encrypted = True
-
-        metadata: Dict[str, Any] = {}
-        for key, value in raw.items():
-            if _is_jsonable_scalar(value):
-                metadata[key] = value
-            elif isinstance(value, (list, tuple)):
-                metadata[key] = [str(v) for v in value][:32]
-            elif isinstance(value, dict):
-                cleaned = {str(k): str(v) for k, v in list(value.items())[:32]}
-                metadata[key] = cleaned
-            else:
-                metadata[key] = str(value)
-
-        return {
-            "app_protocol": app_protocol,
-            "app_master_protocol": master_protocol,
-            "app_category": category,
-            "app_confidence": confidence,
-            "app_hostname": hostname,
-            "app_is_encrypted": is_encrypted,
-            "app_risk_score": risk_score,
-            "dpi_metadata": metadata if metadata else None,
-        }
-
-    def classify_packet(self, pkt: Any) -> Dict[str, Any]:
-        """Classify packet and return enrichment fields for packet metadata."""
-        self._init_engine()
-        if not self.enabled:
-            return {}
-
-        flow_key = self._flow_key(pkt)
-        if flow_key is None:
-            return {}
-
-        payload = self._packet_payload_for_ndpi(pkt)
-        if not payload:
-            return {}
-
-        flow_cls = getattr(ndpi, "NDPIFlow", None)
-        now_ts = time.time()
-        tick_ms = int(now_ts * 1000)
-
-        flow_state = self._flows.get(flow_key)
-        if flow_state is None:
-            flow_obj = flow_cls() if flow_cls is not None else None
-            flow_state = _FlowState(flow_obj=flow_obj, last_seen_ts=now_ts, packet_count=0)
-            self._flows[flow_key] = flow_state
-
-        flow_state.last_seen_ts = now_ts
-        flow_state.packet_count += 1
-
-        try:
-            if flow_state.flow_obj is not None:
-                result = self._engine.process_packet(flow_state.flow_obj, payload, tick_ms)
-            else:
-                # Fallback for bindings that don't expose per-flow object.
-                result = self._engine.process_packet(payload, tick_ms)
-        except Exception:
-            logger.debug("nDPI process_packet failed for flow=%s", flow_key, exc_info=True)
-            self._maybe_cleanup(now_ts)
-            return {}
-
-        enriched = self._extract_result(result)
-        if enriched.get("app_protocol"):
-            flow_state.detected_app = enriched["app_protocol"]
-        elif flow_state.detected_app:
-            enriched["app_protocol"] = flow_state.detected_app
-
-        self._maybe_cleanup(now_ts)
-        return enriched
-
-
-ndpi_classifier = NDPIClassifier()

setup_database.sh

@@ -86,7 +86,7 @@ CREATE TABLE IF NOT EXISTS packets (
     raw_present BOOLEAN DEFAULT FALSE,
     capture_sources TEXT[],
 
-    -- DPI / nDPI metadata
+    -- DPI / flow enrichment metadata
     app_protocol VARCHAR(128),
     app_master_protocol VARCHAR(128),
     app_category VARCHAR(128),