diff --git a/backend/src/Models/packets.py b/backend/src/Models/packets.py index 816eb4b..1975d60 100644 --- a/backend/src/Models/packets.py +++ b/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 diff --git a/backend/src/config.py b/backend/src/config.py index 69de592..95e60f1 100644 --- a/backend/src/config.py +++ b/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), diff --git a/backend/src/network_sniffer.py b/backend/src/network_sniffer.py index baba608..bb899c7 100644 --- a/backend/src/network_sniffer.py +++ b/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: diff --git a/backend/src/utilities/ndpi_classifier.py b/backend/src/utilities/ndpi_classifier.py deleted file mode 100644 index d7ca36c..0000000 --- a/backend/src/utilities/ndpi_classifier.py +++ /dev/null @@ -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() diff --git a/setup_database.sh b/setup_database.sh index 3476ee1..630b9de 100755 --- a/setup_database.sh +++ b/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),