diff --git a/backend/.env.example b/backend/.env.example index 71b295d..597956a 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -17,6 +17,10 @@ BACKEND_SNIFFER_BUFFER_DRAIN_INTERVAL_SECONDS=5.0 BACKEND_SNIFFER_THREAD_JOIN_TIMEOUT_SECONDS=2.0 BACKEND_CAPTURE_INTERFACE= BACKEND_CAPTURE_HEADER_ENABLED=true +BACKEND_CAPTURE_AUTOMANAGE=true +BACKEND_CAPTURE_MIRROR_INTERFACE=mitmcap0 +BACKEND_CAPTURE_AUTOMANAGED_INTERFACE=mitmcap1 +BACKEND_CAPTURE_BUILD_DIR=/tmp/mitm-bpf BACKEND_TELEMETRY_PROCESS_STOP_TIMEOUT_SECONDS=3.0 BACKEND_TELEMETRY_READER_JOIN_TIMEOUT_SECONDS=2.0 BACKEND_NDPI_MAX_FLOWS=200000 diff --git a/backend/src/Models/packets.py b/backend/src/Models/packets.py index 0712271..ac69afd 100644 --- a/backend/src/Models/packets.py +++ b/backend/src/Models/packets.py @@ -3,12 +3,58 @@ from datetime import datetime from typing import Optional, Union -from pydantic import BaseModel, Field, IPvAnyAddress +from pydantic import BaseModel, ConfigDict, Field, IPvAnyAddress class PacketDBModel(BaseModel): """Normalized packet representation used across DB and API layers.""" + model_config = ConfigDict( + json_schema_extra={ + "example": { + "id": 123, + "timestamp": "2026-03-05T12:34:56.789Z", + "updated_at": "2026-03-05T12:34:56.900Z", + "correlation_key": "pid:123456", + "correlation_source": "kernel_mark", + "packet_id": "123456", + "packet_uid": "9f6d3af0d3c81cb20ee8e7d32df7c56414460542", + "skb_mark": 123456, + "ingress_if": "eth0", + "egress_if": "eth1", + "capture_iface": "mitmcap0", + "src_mac": "aa:bb:cc:dd:ee:ff", + "dst_mac": "11:22:33:44:55:66", + "eth_type_raw": 2048, + "eth_type": "IPv4", + "ip_proto_raw": 6, + "ip_proto": "TCP", + "src_ip": "192.168.1.10", + "dst_ip": "192.168.1.1", + "src_port": 54321, + "dst_port": 80, + "vlan_id": None, + "length": 128, + "raw_present": True, + "capture_sources": ["af_packet", "telemetry"], + "raw_b64": "BASE64...", + "app_protocol": "HTTP", + "app_master_protocol": "HTTP", + "app_category": "Web", + "app_confidence": "high", + "app_hostname": "example.org", + "app_is_encrypted": False, + "app_risk_score": 0, + "dpi_metadata": {"method": "GET"}, + "capture_metadata": {"header_magic": "MTCP", "header_version": 1}, + "telemetry_metadata": {"event_type": "egress", "iface": "eth1"}, + "verdict": "accept", + "verdict_reason": "egress-observed", + "verdict_confidence": "high", + } + } + ) + id: Union[int, str] timestamp: datetime = Field(..., description="Packet timestamp in ISO format.") updated_at: Optional[datetime] = Field(None, description="Last DB update time for this row.") @@ -52,48 +98,3 @@ class PacketDBModel(BaseModel): egress_seen_at: Optional[datetime] = None verdict_seen_at: Optional[datetime] = None packets: Optional[int] = None - - class Config: - schema_extra = { - "example": { - "id": 123, - "timestamp": "2026-03-05T12:34:56.789Z", - "updated_at": "2026-03-05T12:34:56.900Z", - "correlation_key": "pid:123456", - "correlation_source": "kernel_mark", - "packet_id": "123456", - "packet_uid": "9f6d3af0d3c81cb20ee8e7d32df7c56414460542", - "skb_mark": 123456, - "ingress_if": "eth0", - "egress_if": "eth1", - "capture_iface": "mitmcap0", - "src_mac": "aa:bb:cc:dd:ee:ff", - "dst_mac": "11:22:33:44:55:66", - "eth_type_raw": 2048, - "eth_type": "IPv4", - "ip_proto_raw": 6, - "ip_proto": "TCP", - "src_ip": "192.168.1.10", - "dst_ip": "192.168.1.1", - "src_port": 54321, - "dst_port": 80, - "vlan_id": None, - "length": 128, - "raw_present": True, - "capture_sources": ["af_packet", "telemetry"], - "raw_b64": "BASE64...", - "app_protocol": "HTTP", - "app_master_protocol": "HTTP", - "app_category": "Web", - "app_confidence": "high", - "app_hostname": "example.org", - "app_is_encrypted": False, - "app_risk_score": 0, - "dpi_metadata": {"method": "GET"}, - "capture_metadata": {"header_magic": "MTCP", "header_version": 1}, - "telemetry_metadata": {"event_type": "egress", "iface": "eth1"}, - "verdict": "accept", - "verdict_reason": "egress-observed", - "verdict_confidence": "high", - } - } diff --git a/backend/src/config.py b/backend/src/config.py index 81f8611..1c3afa0 100644 --- a/backend/src/config.py +++ b/backend/src/config.py @@ -53,6 +53,10 @@ class BackendSettings: sniffer_thread_join_timeout_seconds: float capture_interface: str capture_header_enabled: bool + capture_automanage: bool + capture_automanaged_mirror_if: str + capture_automanaged_capture_if: str + capture_build_dir: str telemetry_process_stop_timeout_seconds: float telemetry_reader_join_timeout_seconds: float ndpi_max_flows: int @@ -84,6 +88,10 @@ def load_settings() -> BackendSettings: sniffer_thread_join_timeout_seconds=_env_float("BACKEND_SNIFFER_THREAD_JOIN_TIMEOUT_SECONDS", 2.0), capture_interface=_env_str("BACKEND_CAPTURE_INTERFACE", ""), capture_header_enabled=_env_bool("BACKEND_CAPTURE_HEADER_ENABLED", True), + capture_automanage=_env_bool("BACKEND_CAPTURE_AUTOMANAGE", True), + capture_automanaged_mirror_if=_env_str("BACKEND_CAPTURE_MIRROR_INTERFACE", "mitmcap0"), + capture_automanaged_capture_if=_env_str("BACKEND_CAPTURE_AUTOMANAGED_INTERFACE", "mitmcap1"), + capture_build_dir=_env_str("BACKEND_CAPTURE_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), diff --git a/backend/src/main.py b/backend/src/main.py index 6ccc57d..e7d81fc 100644 --- a/backend/src/main.py +++ b/backend/src/main.py @@ -75,6 +75,13 @@ async def on_startup() -> None: @app.on_event("shutdown") async def shutdown_event() -> None: """Stop network resources and release shared runtime objects.""" + try: + from src.network_sniffer import stop_afpacket_sniffer + + stop_afpacket_sniffer() + except Exception: + logging.exception("Failed to stop sniffer sessions during shutdown") + try: network_api.shutdown_network_api() except Exception: @@ -87,6 +94,13 @@ async def shutdown_event() -> None: except Exception: logging.exception("Failed to stop bridge telemetry collector") + try: + from src.utilities.capture_pipeline import capture_pipeline_manager + + capture_pipeline_manager.stop_all() + except Exception: + logging.exception("Failed to stop capture pipelines") + try: from src.utilities.packet_tracker import packet_tracker diff --git a/backend/src/network_sniffer.py b/backend/src/network_sniffer.py index 8da240c..fb00b02 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.capture_header import parse_capture_header +from src.utilities.capture_pipeline import capture_pipeline_manager from src.utilities.ndpi_classifier import ndpi_classifier from src.utilities.packet_identity import build_packet_uid from src.utilities.packet_tracker import packet_tracker @@ -567,7 +568,7 @@ def _session_reader_loop(session_id: str) -> None: try: capture_meta = None packet_bytes = raw - if settings.capture_header_enabled and settings.capture_interface and iface == settings.capture_interface: + if settings.capture_header_enabled: capture_meta, packet_bytes = parse_capture_header(raw) pkt = Ether(packet_bytes) pkt.sniffed_on = iface @@ -617,8 +618,8 @@ def start_afpacket_sniffer(target: str, target_is_interface: bool = False) -> st "is_bridge": not target_is_interface, "ports": [], "capture_ifaces": [], + "capture_bridge": None, } - sessions[session_id] = session # determine ports for this session if target_is_interface: @@ -626,19 +627,35 @@ def start_afpacket_sniffer(target: str, target_is_interface: bool = False) -> st capture_ifaces = [target] else: ports = get_bridge_ports_once(target) - capture_ifaces = [settings.capture_interface] if settings.capture_interface else list(ports) + capture_iface = None + if settings.capture_automanage: + capture_iface = capture_pipeline_manager.acquire(target) + session["capture_bridge"] = target + elif settings.capture_interface: + capture_iface = settings.capture_interface + capture_ifaces = [capture_iface] if capture_iface else list(ports) session["ports"] = ports session["capture_ifaces"] = capture_ifaces + sessions[session_id] = session # create sockets for this session only - for iface in capture_ifaces: - if not check_interface_exists(iface): - logger.warning("Snapshot port %s missing for session %s, skipping", iface, session_id) - continue - s = _create_af_packet_socket(iface) - if s: - session["sockets"][iface] = s + try: + for iface in capture_ifaces: + if not check_interface_exists(iface): + logger.warning("Snapshot port %s missing for session %s, skipping", iface, session_id) + continue + s = _create_af_packet_socket(iface) + if s: + session["sockets"][iface] = s + except Exception: + sessions.pop(session_id, None) + if session.get("capture_bridge"): + try: + capture_pipeline_manager.release(session["capture_bridge"]) + except Exception: + logger.exception("Failed to release capture pipeline after startup error") + raise # start dedicated thread t = threading.Thread(target=_session_reader_loop, args=(session_id,), daemon=True) @@ -672,6 +689,12 @@ def stop_afpacket_sniffer(session_id: Optional[str] = None, target: Optional[str t = s.get("thread") if t and isinstance(t, threading.Thread): t.join(timeout=settings.sniffer_thread_join_timeout_seconds) + capture_bridge = s.get("capture_bridge") + if capture_bridge: + try: + capture_pipeline_manager.release(capture_bridge) + except Exception: + logger.exception("Failed to release capture pipeline for bridge %s", capture_bridge) _sync_bridge_telemetry() logger.info("Stopped session %s", session_id) return @@ -681,6 +704,9 @@ def stop_afpacket_sniffer(session_id: Optional[str] = None, target: Optional[str # For interface-mode: close that iface in any session that has it for sid, s in list(sessions.items()): if target in s.get("ports", []): + if s.get("capture_bridge") == target: + stop_afpacket_sniffer(session_id=sid) + continue # close only that iface socket in that session sock = s["sockets"].pop(target, None) if sock: @@ -748,6 +774,7 @@ def get_internal_debug_state() -> dict: sid: { "label": s.get("label"), "is_bridge": s.get("is_bridge"), + "capture_bridge": s.get("capture_bridge"), "ports": list(s.get("ports", [])), "capture_ifaces": list(s.get("capture_ifaces", [])), "sockets": list(s.get("sockets", {}).keys()), @@ -757,5 +784,6 @@ def get_internal_debug_state() -> dict: }, "buffer_len": len(_PACKET_BUFFER), "telemetry_ports": sorted({iface for session in sessions.values() for iface in session.get("ports", [])}), + "capture_pipeline": capture_pipeline_manager.get_debug_snapshot(), "packet_tracker": packet_tracker.get_debug_snapshot(), } diff --git a/backend/src/utilities/bridge_telemetry.py b/backend/src/utilities/bridge_telemetry.py index f1e0173..e79f84a 100644 --- a/backend/src/utilities/bridge_telemetry.py +++ b/backend/src/utilities/bridge_telemetry.py @@ -54,6 +54,9 @@ class BridgeTelemetryManager: 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), "--ifaces", ",".join(sorted(self._interfaces))] logger.info("Starting bridge telemetry collector for interfaces=%s", sorted(self._interfaces)) diff --git a/backend/src/utilities/capture_pipeline.py b/backend/src/utilities/capture_pipeline.py new file mode 100644 index 0000000..86087f4 --- /dev/null +++ b/backend/src/utilities/capture_pipeline.py @@ -0,0 +1,143 @@ +"""Manage the optional tc-based bridge capture pipeline for mirrored packet storage.""" + +from __future__ import annotations + +import logging +import subprocess +import threading +from pathlib import Path +from typing import Dict, Optional + +from src.config import settings + +logger = logging.getLogger("capture_pipeline") + + +class CapturePipelineManager: + """Reference-count mirrored capture pipelines per bridge.""" + + def __init__(self) -> None: + self._lock = threading.Lock() + self._refs: Dict[str, int] = {} + self._active: Dict[str, Dict[str, str]] = {} + + def acquire(self, bridge: str) -> Optional[str]: + """Ensure a bridge capture pipeline exists and return the capture interface name.""" + if not settings.capture_automanage: + return settings.capture_interface or None + + bridge = bridge.strip() + if not bridge: + raise ValueError("Bridge name is required for capture pipeline setup") + + with self._lock: + refcount = self._refs.get(bridge, 0) + if refcount > 0: + self._refs[bridge] = refcount + 1 + active = self._active.get(bridge, {}) + return active.get("capture_if") or settings.capture_interface or settings.capture_automanaged_capture_if + + self._setup_bridge_locked(bridge) + self._refs[bridge] = 1 + active = self._active.get(bridge, {}) + return active.get("capture_if") or settings.capture_interface or settings.capture_automanaged_capture_if + + def release(self, bridge: str) -> None: + """Tear down a bridge capture pipeline when the last session releases it.""" + if not settings.capture_automanage: + return + + bridge = bridge.strip() + if not bridge: + return + + with self._lock: + refcount = self._refs.get(bridge, 0) + if refcount <= 1: + self._refs.pop(bridge, None) + try: + self._teardown_bridge_locked(bridge) + finally: + self._active.pop(bridge, None) + return + + self._refs[bridge] = refcount - 1 + + def stop_all(self) -> None: + """Remove all managed capture pipelines.""" + if not settings.capture_automanage: + return + + with self._lock: + bridges = list(self._active.keys()) + self._refs.clear() + for bridge in bridges: + try: + self._teardown_bridge_locked(bridge) + except Exception: + logger.exception("Failed to tear down capture pipeline for bridge %s", bridge) + self._active.clear() + + def get_debug_snapshot(self) -> Dict[str, object]: + with self._lock: + return { + "automanage_enabled": settings.capture_automanage, + "refs": dict(self._refs), + "active": dict(self._active), + } + + def _setup_bridge_locked(self, bridge: str) -> None: + script = self._repo_root() / "tools" / "setup_bridge_capture.sh" + mirror_if = settings.capture_automanaged_mirror_if + capture_if = settings.capture_interface or settings.capture_automanaged_capture_if + cmd = [ + str(script), + "--bridge", + bridge, + "--mirror-if", + mirror_if, + "--capture-if", + capture_if, + "--build-dir", + settings.capture_build_dir, + ] + logger.info( + "Setting up capture pipeline for bridge=%s mirror_if=%s capture_if=%s", + bridge, + mirror_if, + capture_if, + ) + subprocess.run(cmd, check=True, capture_output=True, text=True) + self._active[bridge] = { + "mirror_if": mirror_if, + "capture_if": capture_if, + "build_dir": settings.capture_build_dir, + } + + def _teardown_bridge_locked(self, bridge: str) -> None: + script = self._repo_root() / "tools" / "teardown_bridge_capture.sh" + active = self._active.get(bridge, {}) + mirror_if = active.get("mirror_if") or settings.capture_automanaged_mirror_if + capture_if = active.get("capture_if") or settings.capture_interface or settings.capture_automanaged_capture_if + cmd = [ + str(script), + "--bridge", + bridge, + "--mirror-if", + mirror_if, + "--capture-if", + capture_if, + ] + logger.info( + "Tearing down capture pipeline for bridge=%s mirror_if=%s capture_if=%s", + bridge, + mirror_if, + capture_if, + ) + subprocess.run(cmd, check=True, capture_output=True, text=True) + + def _repo_root(self) -> Path: + return Path(__file__).resolve().parents[3] + + +capture_pipeline_manager = CapturePipelineManager()