manag veth directly with sniffer

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

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",
-            }
-        }

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

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
 

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,12 +627,20 @@ 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
+    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)
@@ -639,6 +648,14 @@ def start_afpacket_sniffer(target: str, target_is_interface: bool = False) -> st
             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(),
     }

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

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