diff --git a/backend/.env.example b/backend/.env.example index 597956a..285908d 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -15,12 +15,7 @@ BACKEND_SNIFFER_SELECTOR_TIMEOUT_SECONDS=1.0 BACKEND_SNIFFER_RECV_BYTES=65536 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_BRIDGE_BPF_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 ac69afd..816eb4b 100644 --- a/backend/src/Models/packets.py +++ b/backend/src/Models/packets.py @@ -22,7 +22,7 @@ class PacketDBModel(BaseModel): "skb_mark": 123456, "ingress_if": "eth0", "egress_if": "eth1", - "capture_iface": "mitmcap0", + "capture_iface": None, "src_mac": "aa:bb:cc:dd:ee:ff", "dst_mac": "11:22:33:44:55:66", "eth_type_raw": 2048, @@ -36,7 +36,7 @@ class PacketDBModel(BaseModel): "vlan_id": None, "length": 128, "raw_present": True, - "capture_sources": ["af_packet", "telemetry"], + "capture_sources": ["tc_ingress_raw", "telemetry"], "raw_b64": "BASE64...", "app_protocol": "HTTP", "app_master_protocol": "HTTP", @@ -46,8 +46,8 @@ class PacketDBModel(BaseModel): "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"}, + "capture_metadata": {"capture_mode": "tc_ingress", "packet_id": "123456"}, + "telemetry_metadata": {"event_type": "egress", "iface": "eth1", "packet_id": "123456"}, "verdict": "accept", "verdict_reason": "egress-observed", "verdict_confidence": "high", @@ -89,7 +89,7 @@ class PacketDBModel(BaseModel): 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.") - capture_metadata: Optional[dict] = Field(None, description="Capture-side metadata from the mirrored packet shim.") + 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 verdict_reason: Optional[str] = None diff --git a/backend/src/config.py b/backend/src/config.py index 1c3afa0..faec3bc 100644 --- a/backend/src/config.py +++ b/backend/src/config.py @@ -25,13 +25,6 @@ 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 @@ -51,12 +44,7 @@ class BackendSettings: sniffer_recv_bytes: int sniffer_buffer_drain_interval_seconds: float 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 + bridge_bpf_build_dir: str telemetry_process_stop_timeout_seconds: float telemetry_reader_join_timeout_seconds: float ndpi_max_flows: int @@ -86,12 +74,7 @@ def load_settings() -> BackendSettings: sniffer_recv_bytes=_env_int("BACKEND_SNIFFER_RECV_BYTES", 65_536), sniffer_buffer_drain_interval_seconds=_env_float("BACKEND_SNIFFER_BUFFER_DRAIN_INTERVAL_SECONDS", 5.0), 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"), + 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), diff --git a/backend/src/main.py b/backend/src/main.py index e7d81fc..004cb84 100644 --- a/backend/src/main.py +++ b/backend/src/main.py @@ -94,13 +94,6 @@ 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 8f9ab7d..545bfdd 100644 --- a/backend/src/network_sniffer.py +++ b/backend/src/network_sniffer.py @@ -33,8 +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.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 @@ -68,6 +66,7 @@ class PacketInfo(TypedDict, total=False): packet_id: Optional[str] packet_uid: str skb_mark: Optional[int] + capture_source: str iface: str capture_iface: Optional[str] length: int @@ -176,9 +175,10 @@ def parse_packet(pkt, bridge_label: str, capture_metadata: Optional[Dict[str, An pkt_info: PacketInfo = { "iface": pkt_iface, - "capture_iface": pkt_iface if capture_metadata else None, + "capture_iface": None, "length": len(pkt), "raw": bytes(pkt), + "capture_source": "tc_ingress_raw" if capture_metadata else "af_packet", "packet_id": capture_metadata.get("packet_id") if capture_metadata else None, "skb_mark": capture_metadata.get("skb_mark") if capture_metadata else None, "src_mac": None, @@ -412,6 +412,17 @@ def parse_packet(pkt, bridge_label: str, capture_metadata: Optional[Dict[str, An logger.exception("Failed to schedule DB insert (unexpected)") +def parse_packet_bytes( + packet_bytes: bytes, + iface: str, + capture_metadata: Optional[Dict[str, Any]] = None, +) -> None: + """Parse one raw Ethernet frame using the shared Scapy packet path.""" + pkt = Ether(packet_bytes) + pkt.sniffed_on = iface + parse_packet(pkt, iface, capture_metadata=capture_metadata) + + # ------------------------- # AF_PACKET socket utilities # ------------------------- @@ -484,22 +495,20 @@ def _ensure_socket_for_session(sockets: Dict[str, socket.socket], iface: str, br def _sync_bridge_telemetry() -> None: - interfaces = sorted({iface for session in sessions.values() for iface in session.get("ports", [])}) + interfaces = sorted( + { + iface + for session in sessions.values() + if session.get("is_bridge") + for iface in session.get("ports", []) + } + ) try: bridge_telemetry_manager.update_interfaces(interfaces) except Exception: logger.exception("Failed to update bridge telemetry collector") -def _managed_capture_ifaces() -> set[str]: - return { - iface - for session in sessions.values() - if session.get("is_bridge") - for iface in session.get("capture_ifaces", []) - } - - # ------------------------- # Per-session reader loop # ------------------------- @@ -575,16 +584,9 @@ def _session_reader_loop(session_id: str) -> None: # parse with scapy try: - capture_meta = None - packet_bytes = raw - if settings.capture_header_enabled: - capture_meta, packet_bytes = parse_capture_header(raw) - if capture_meta is None and iface in _managed_capture_ifaces(): - logger.debug("Skipping unshimmed packet on managed capture interface %s", iface) - continue - pkt = Ether(packet_bytes) + pkt = Ether(raw) pkt.sniffed_on = iface - parse_packet(pkt, label, capture_metadata=capture_meta) + parse_packet(pkt, label) logger.debug("Captured packet on %s in session %s (len=%d)", iface, session_id, len(raw)) except Exception: logger.exception("Failed to parse/process packet from %s in session %s", iface, session_id) @@ -630,7 +632,6 @@ def start_afpacket_sniffer(target: str, target_is_interface: bool = False) -> st "is_bridge": not target_is_interface, "ports": [], "capture_ifaces": [], - "capture_bridge": None, } # determine ports for this session @@ -639,13 +640,7 @@ def start_afpacket_sniffer(target: str, target_is_interface: bool = False) -> st capture_ifaces = [target] else: ports = get_bridge_ports_once(target) - 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) + capture_ifaces = [] session["ports"] = ports session["capture_ifaces"] = capture_ifaces @@ -662,17 +657,15 @@ def start_afpacket_sniffer(target: str, target_is_interface: bool = False) -> st 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) - session["thread"] = t - t.start() + if capture_ifaces: + t = threading.Thread(target=_session_reader_loop, args=(session_id,), daemon=True) + session["thread"] = t + t.start() + else: + session["thread"] = None _sync_bridge_telemetry() logger.info( "Started sniffer session %s label=%s ports=%s capture_ifaces=%s", @@ -701,12 +694,6 @@ 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 @@ -715,10 +702,13 @@ def stop_afpacket_sniffer(session_id: Optional[str] = None, target: Optional[str if target: # For interface-mode: close that iface in any session that has it for sid, s in list(sessions.items()): + if not target_is_interface and s.get("is_bridge") and s.get("label") == target: + stop_afpacket_sniffer(session_id=sid) + continue + if target_is_interface and not s.get("is_bridge") and s.get("label") == target: + stop_afpacket_sniffer(session_id=sid) + continue 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: @@ -774,6 +764,15 @@ def get_sniffer_status() -> Dict[str, Dict[str, object]]: "session_id": sid, "session_label": s.get("label"), } + if not s.get("sockets") and s.get("is_bridge"): + for iface in s.get("ports", []): + out[iface] = { + "running": not s.get("stop_event").is_set(), + "exists": check_interface_exists(iface), + "up": check_interface_up(iface), + "session_id": sid, + "session_label": s.get("label"), + } return out @@ -786,16 +785,16 @@ 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()), "thread_alive": bool(s.get("thread") and s.get("thread").is_alive()), + "active": not s.get("stop_event").is_set(), } for sid, s in sessions.items() }, "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", [])}), - "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 e79f84a..1897fc4 100644 --- a/backend/src/utilities/bridge_telemetry.py +++ b/backend/src/utilities/bridge_telemetry.py @@ -1,7 +1,8 @@ -"""Manage the eBPF telemetry subprocess used for bridge direction and verdict events.""" +"""Manage the eBPF/tc telemetry subprocess used for bridge packet capture and verdict events.""" from __future__ import annotations +import base64 import json import logging import os @@ -19,7 +20,7 @@ logger = logging.getLogger("bridge_telemetry") class BridgeTelemetryManager: - """Run a single eBPF collector process for the active sniffed interfaces.""" + """Run one tc/eBPF collector for the currently sniffed bridge interfaces.""" def __init__(self) -> None: self._interfaces: set[str] = set() @@ -28,7 +29,7 @@ class BridgeTelemetryManager: self._lock = threading.Lock() def update_interfaces(self, interfaces: Iterable[str]) -> None: - """Restart the collector when the active interface set changes.""" + """Restart the collector when the active bridge interface set changes.""" normalized = {iface.strip() for iface in interfaces if iface and iface.strip()} with self._lock: if normalized == self._interfaces: @@ -58,7 +59,14 @@ class BridgeTelemetryManager: 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))] + cmd = [ + python_bin, + str(helper), + "--ifaces", + ",".join(sorted(self._interfaces)), + "--build-dir", + settings.bridge_bpf_build_dir, + ] logger.info("Starting bridge telemetry collector for interfaces=%s", sorted(self._interfaces)) try: self._process = subprocess.Popen( @@ -100,6 +108,31 @@ class BridgeTelemetryManager: if reader is not None and reader.is_alive(): reader.join(timeout=settings.telemetry_reader_join_timeout_seconds) + def _handle_ingress_packet(self, event: dict[str, object]) -> None: + raw_b64 = event.pop("raw_b64", None) + if not isinstance(raw_b64, str) or not raw_b64: + return + + try: + packet_bytes = base64.b64decode(raw_b64) + except Exception: + logger.exception("Failed to decode ingress raw packet") + return + + capture_metadata = { + "capture_source": "tc_ingress_raw", + "packet_id": event.get("packet_id"), + "skb_mark": event.get("skb_mark"), + "capture_mode": "tc_ingress", + } + + try: + from src.network_sniffer import parse_packet_bytes + + parse_packet_bytes(packet_bytes, str(event.get("iface")), capture_metadata=capture_metadata) + except Exception: + logger.exception("Failed to process ingress raw packet event") + def _read_loop(self, process: subprocess.Popen[str]) -> None: stdout = process.stdout if stdout is None: @@ -119,6 +152,9 @@ class BridgeTelemetryManager: logger.info("bridge-telemetry: %s", event) continue + if event.get("event_type") == "ingress": + self._handle_ingress_packet(event) + try: packet_tracker.observe_telemetry(event) except Exception: diff --git a/backend/src/utilities/capture_header.py b/backend/src/utilities/capture_header.py deleted file mode 100644 index 5559b74..0000000 --- a/backend/src/utilities/capture_header.py +++ /dev/null @@ -1,37 +0,0 @@ -"""Parse the capture-side shim that carries the skb mark to AF_PACKET userspace.""" - -from __future__ import annotations - -import struct -from typing import Dict, Optional, Tuple - -from src.utilities.packet_mark import packet_id_from_mark - - -CAPTURE_HEADER_MAGIC = b"MTCP" -CAPTURE_HEADER_VERSION = 1 -CAPTURE_HEADER_STRUCT = struct.Struct("!4sBBHII") -CAPTURE_HEADER_SIZE = CAPTURE_HEADER_STRUCT.size - - -def parse_capture_header(frame: bytes) -> Tuple[Optional[Dict[str, object]], bytes]: - """Return parsed capture metadata and the stripped Ethernet frame.""" - if len(frame) < CAPTURE_HEADER_SIZE: - return None, frame - - magic, version, flags, header_len, skb_mark, aux_value = CAPTURE_HEADER_STRUCT.unpack_from(frame) - if magic != CAPTURE_HEADER_MAGIC or version != CAPTURE_HEADER_VERSION: - return None, frame - if header_len < CAPTURE_HEADER_SIZE or len(frame) < header_len: - return None, frame - - metadata: Dict[str, object] = { - "header_magic": magic.decode("ascii", "replace"), - "header_version": int(version), - "header_flags": int(flags), - "header_len": int(header_len), - "skb_mark": int(skb_mark), - "aux_value": int(aux_value), - "packet_id": packet_id_from_mark(skb_mark), - } - return metadata, frame[header_len:] diff --git a/backend/src/utilities/capture_pipeline.py b/backend/src/utilities/capture_pipeline.py deleted file mode 100644 index df96bde..0000000 --- a/backend/src/utilities/capture_pipeline.py +++ /dev/null @@ -1,158 +0,0 @@ -"""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") - - -def _run_checked(cmd: list[str]) -> subprocess.CompletedProcess[str]: - """Run a command and surface stdout/stderr in raised errors.""" - try: - return subprocess.run(cmd, check=True, capture_output=True, text=True) - except subprocess.CalledProcessError as exc: - stdout = (exc.stdout or "").strip() - stderr = (exc.stderr or "").strip() - if stdout: - logger.error("Command stdout for %s:\n%s", cmd[0], stdout) - if stderr: - logger.error("Command stderr for %s:\n%s", cmd[0], stderr) - detail = stderr or stdout or str(exc) - raise RuntimeError(detail) from exc - - -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, - ) - _run_checked(cmd) - 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, - ) - _run_checked(cmd) - - def _repo_root(self) -> Path: - return Path(__file__).resolve().parents[3] - - -capture_pipeline_manager = CapturePipelineManager() diff --git a/backend/src/utilities/ebpf_bridge_events.py b/backend/src/utilities/ebpf_bridge_events.py index 899e3e1..2c0415b 100644 --- a/backend/src/utilities/ebpf_bridge_events.py +++ b/backend/src/utilities/ebpf_bridge_events.py @@ -1,19 +1,24 @@ #!/usr/bin/env python3 -"""Emit bridge ingress, egress, and drop telemetry events via eBPF.""" +"""Emit bridge ingress raw packets plus egress/drop telemetry via tc/eBPF.""" from __future__ import annotations import argparse +import base64 import ctypes as ct import hashlib import ipaddress import json -import os import signal +import shutil import socket +import subprocess import sys +from pathlib import Path from typing import Iterable +from pyroute2 import IPRoute + try: from bcc import BPF # type: ignore except Exception as exc: # pragma: no cover - depends on host runtime @@ -23,6 +28,15 @@ except Exception as exc: # pragma: no cover - depends on host runtime from src.utilities.packet_mark import packet_id_from_mark, verdict_from_mark +EVENT_INGRESS = 1 +EVENT_EGRESS = 2 +EVENT_DROP = 3 +INGRESS_PARENT = "ffff:fff2" +EGRESS_PARENT = "ffff:fff3" +MARK_FILTER_PREF = 10 +INGRESS_FILTER_HANDLE = ":20" +EGRESS_FILTER_HANDLE = ":30" + IDENTITY_FIELDS = ( "src_mac", "dst_mac", @@ -38,9 +52,10 @@ IDENTITY_FIELDS = ( BPF_SOURCE = r""" #include +#include +#include #include #include -#include #include #include #include @@ -48,7 +63,6 @@ BPF_SOURCE = r""" #include #include #include -#include #define EVENT_INGRESS 1 #define EVENT_EGRESS 2 @@ -71,6 +85,7 @@ struct event_t { __u32 skb_mark; __u32 length; __u32 reason; + __u32 ifindex; __u16 eth_type_raw; __u16 vlan_id; __u16 src_port; @@ -78,31 +93,119 @@ struct event_t { __u32 protocol_raw; __u8 event_type; __u8 ip_version; - char ifname[IFNAMSIZ]; + __u8 reserved[2]; unsigned char src_mac[6]; unsigned char dst_mac[6]; unsigned char src_ip[16]; unsigned char dst_ip[16]; }; -BPF_PERF_OUTPUT(events); +BPF_PERF_OUTPUT(ingress_events); +BPF_PERF_OUTPUT(meta_events); -static __always_inline int fill_ifname(struct sk_buff *skb, struct event_t *event) { - if (!skb) { +static __always_inline int parse_l3_l4_direct(struct event_t *event, void *data, void *data_end) { + struct ethhdr *eth = data; + __be16 eth_proto; + void *l3; + + if ((void *)(eth + 1) > data_end) { return 0; } - struct net_device *dev = NULL; - bpf_probe_read_kernel(&dev, sizeof(dev), &skb->dev); - if (!dev) { - return 0; + __builtin_memcpy(event->src_mac, eth->h_source, ETH_ALEN); + __builtin_memcpy(event->dst_mac, eth->h_dest, ETH_ALEN); + eth_proto = eth->h_proto; + l3 = eth + 1; + + if (eth_proto == htons(ETH_P_8021Q) || eth_proto == htons(ETH_P_8021AD)) { + struct vlan_hdr_t *vlan = l3; + if ((void *)(vlan + 1) > data_end) { + return 0; + } + event->vlan_id = ntohs(vlan->h_vlan_TCI) & 0x0fff; + eth_proto = vlan->h_vlan_encapsulated_proto; + l3 = vlan + 1; + } + + event->eth_type_raw = ntohs(eth_proto); + + if (eth_proto == htons(ETH_P_ARP)) { + struct arphdr *arph = l3; + struct arp_eth_ipv4_t *body = (void *)(arph + 1); + if ((void *)(body + 1) > data_end) { + return 1; + } + if (arph->ar_hrd == htons(ARPHRD_ETHER) && arph->ar_pro == htons(ETH_P_IP) && + arph->ar_hln == ETH_ALEN && arph->ar_pln == 4) { + __builtin_memcpy(event->src_ip, body->spa, 4); + __builtin_memcpy(event->dst_ip, body->tpa, 4); + event->ip_version = 4; + } + return 1; + } + + if (eth_proto == htons(ETH_P_IP)) { + struct iphdr *iph = l3; + if ((void *)(iph + 1) > data_end) { + return 0; + } + + event->ip_version = 4; + event->protocol_raw = iph->protocol; + __builtin_memcpy(event->src_ip, &iph->saddr, 4); + __builtin_memcpy(event->dst_ip, &iph->daddr, 4); + + if (iph->protocol == IPPROTO_TCP) { + struct tcphdr *tcph = (void *)iph + (iph->ihl * 4); + if ((void *)(tcph + 1) > data_end) { + return 1; + } + event->src_port = ntohs(tcph->source); + event->dst_port = ntohs(tcph->dest); + } else if (iph->protocol == IPPROTO_UDP) { + struct udphdr *udph = (void *)iph + (iph->ihl * 4); + if ((void *)(udph + 1) > data_end) { + return 1; + } + event->src_port = ntohs(udph->source); + event->dst_port = ntohs(udph->dest); + } + return 1; + } + + if (eth_proto == htons(ETH_P_IPV6)) { + struct ipv6hdr *ip6h = l3; + if ((void *)(ip6h + 1) > data_end) { + return 0; + } + + event->ip_version = 6; + event->protocol_raw = ip6h->nexthdr; + __builtin_memcpy(event->src_ip, &ip6h->saddr, 16); + __builtin_memcpy(event->dst_ip, &ip6h->daddr, 16); + + if (ip6h->nexthdr == IPPROTO_TCP) { + struct tcphdr *tcph = ip6h + 1; + if ((void *)(tcph + 1) > data_end) { + return 1; + } + event->src_port = ntohs(tcph->source); + event->dst_port = ntohs(tcph->dest); + } else if (ip6h->nexthdr == IPPROTO_UDP) { + struct udphdr *udph = ip6h + 1; + if ((void *)(udph + 1) > data_end) { + return 1; + } + event->src_port = ntohs(udph->source); + event->dst_port = ntohs(udph->dest); + } + return 1; } - bpf_probe_read_kernel(event->ifname, sizeof(event->ifname), dev->name); return 1; } -static __always_inline int parse_skb(struct sk_buff *skb, struct event_t *event) { +static __always_inline int parse_skb_linear(struct event_t *event, struct sk_buff *skb) { unsigned char *head = NULL; __u16 mac_header = 0; __u16 network_header = 0; @@ -126,8 +229,8 @@ static __always_inline int parse_skb(struct sk_buff *skb, struct event_t *event) struct ethhdr eth = {}; unsigned char *eth_ptr = head + mac_header; bpf_probe_read_kernel(ð, sizeof(eth), eth_ptr); - __builtin_memcpy(event->src_mac, eth.h_source, 6); - __builtin_memcpy(event->dst_mac, eth.h_dest, 6); + __builtin_memcpy(event->src_mac, eth.h_source, ETH_ALEN); + __builtin_memcpy(event->dst_mac, eth.h_dest, ETH_ALEN); __be16 eth_proto = eth.h_proto; unsigned char *l3_ptr = head + network_header; @@ -136,10 +239,8 @@ static __always_inline int parse_skb(struct sk_buff *skb, struct event_t *event) bpf_probe_read_kernel(&vlan, sizeof(vlan), eth_ptr + sizeof(struct ethhdr)); event->vlan_id = ntohs(vlan.h_vlan_TCI) & 0x0fff; eth_proto = vlan.h_vlan_encapsulated_proto; - event->eth_type_raw = ntohs(eth_proto); - } else { - event->eth_type_raw = ntohs(eth_proto); } + event->eth_type_raw = ntohs(eth_proto); if (eth_proto == htons(ETH_P_ARP)) { struct arphdr arph = {}; @@ -202,64 +303,76 @@ static __always_inline int parse_skb(struct sk_buff *skb, struct event_t *event) return 1; } -static __always_inline int emit_event(struct pt_regs *ctx, struct sk_buff *skb, __u8 event_type, __u32 reason) { +int handle_ingress(struct __sk_buff *skb) { struct event_t event = {}; + void *data = (void *)(long)skb->data; + void *data_end = (void *)(long)skb->data_end; + event.ts_ns = bpf_ktime_get_ns(); - event.event_type = event_type; - event.reason = reason; + event.event_type = EVENT_INGRESS; + event.ifindex = skb->ifindex; + event.length = skb->len; + event.skb_mark = skb->mark; - if (!fill_ifname(skb, &event)) { - return 0; + if (!event.skb_mark) { + return TC_ACT_OK; } - if (!parse_skb(skb, &event)) { - return 0; + if (!parse_l3_l4_direct(&event, data, data_end)) { + return TC_ACT_OK; } - events.perf_submit(ctx, &event, sizeof(event)); - return 0; + ingress_events.perf_submit_skb(skb, skb->len, &event, sizeof(event)); + return TC_ACT_OK; } -int trace_ingress(struct pt_regs *ctx, struct sk_buff *skb) { - return emit_event(ctx, skb, EVENT_INGRESS, 0); -} +int handle_egress(struct __sk_buff *skb) { + struct event_t event = {}; + void *data = (void *)(long)skb->data; + void *data_end = (void *)(long)skb->data_end; -int trace_egress(struct pt_regs *ctx, struct sk_buff *skb) { - return emit_event(ctx, skb, EVENT_EGRESS, 0); + event.ts_ns = bpf_ktime_get_ns(); + event.event_type = EVENT_EGRESS; + event.ifindex = skb->ifindex; + event.length = skb->len; + event.skb_mark = skb->mark; + + if (!event.skb_mark) { + return TC_ACT_OK; + } + if (!parse_l3_l4_direct(&event, data, data_end)) { + return TC_ACT_OK; + } + + meta_events.perf_submit(skb, &event, sizeof(event)); + return TC_ACT_OK; } TRACEPOINT_PROBE(skb, kfree_skb) { struct sk_buff *skb = (struct sk_buff *)args->skbaddr; struct event_t event = {}; + struct net_device *dev = NULL; + event.ts_ns = bpf_ktime_get_ns(); event.event_type = EVENT_DROP; event.reason = args->reason; - if (!fill_ifname(skb, &event)) { - return 0; - } - if (!parse_skb(skb, &event)) { + if (!skb) { return 0; } - events.perf_submit(args, &event, sizeof(event)); - return 0; -} - -TRACEPOINT_PROBE(net, net_dev_queue) { - struct sk_buff *skb = (struct sk_buff *)args->skbaddr; - struct event_t event = {}; - event.ts_ns = bpf_ktime_get_ns(); - event.event_type = EVENT_EGRESS; - event.reason = 0; - - if (!fill_ifname(skb, &event)) { + bpf_probe_read_kernel(&dev, sizeof(dev), &skb->dev); + if (!dev) { return 0; } - if (!parse_skb(skb, &event)) { + bpf_probe_read_kernel(&event.ifindex, sizeof(event.ifindex), &dev->ifindex); + if (!parse_skb_linear(&event, skb)) { + return 0; + } + if (!event.skb_mark) { return 0; } - events.perf_submit(args, &event, sizeof(event)); + meta_events.perf_submit(args, &event, sizeof(event)); return 0; } """ @@ -271,6 +384,7 @@ class Event(ct.Structure): ("skb_mark", ct.c_uint), ("length", ct.c_uint), ("reason", ct.c_uint), + ("ifindex", ct.c_uint), ("eth_type_raw", ct.c_ushort), ("vlan_id", ct.c_ushort), ("src_port", ct.c_ushort), @@ -278,7 +392,7 @@ class Event(ct.Structure): ("protocol_raw", ct.c_uint), ("event_type", ct.c_ubyte), ("ip_version", ct.c_ubyte), - ("ifname", ct.c_char * 16), + ("reserved", ct.c_ubyte * 2), ("src_mac", ct.c_ubyte * 6), ("dst_mac", ct.c_ubyte * 6), ("src_ip", ct.c_ubyte * 16), @@ -286,6 +400,51 @@ class Event(ct.Structure): ] +TARGET_INTERFACES: set[str] = set() +IPR: IPRoute | None = None + + +def _run_checked(cmd: list[str]) -> None: + subprocess.run(cmd, check=True, capture_output=True, text=True) + + +def _multiarch_include_flag() -> list[str]: + if shutil.which("gcc") is None: + return [] + try: + triple = subprocess.run(["gcc", "-dumpmachine"], check=True, capture_output=True, text=True).stdout.strip() + except Exception: + return [] + include_dir = Path("/usr/include") / triple + return [f"-I{include_dir}"] if include_dir.is_dir() else [] + + +def _repo_root() -> Path: + return Path(__file__).resolve().parents[3] + + +def _ensure_mark_object(build_dir: str) -> Path: + build_path = Path(build_dir) + build_path.mkdir(parents=True, exist_ok=True) + src = _repo_root() / "tools" / "ebpf" / "mark_packet_id.c" + obj = build_path / "mark_packet_id.o" + if obj.exists() and obj.stat().st_mtime >= src.stat().st_mtime: + return obj + + cmd = ["clang", "-O2", "-g", "-target", "bpf", *_multiarch_include_flag(), "-c", str(src), "-o", str(obj)] + _run_checked(cmd) + return obj + + +def _ifname_from_index(ifindex: int) -> str | None: + if ifindex <= 0: + return None + try: + return socket.if_indextoname(ifindex) + except OSError: + return None + + def _mac_to_str(value: Iterable[int]) -> str: return ":".join(f"{byte:02x}" for byte in value) @@ -314,22 +473,20 @@ def _build_packet_uid(payload: dict[str, object]) -> str: def _event_name(value: int) -> str: - return {1: "ingress", 2: "egress", 3: "drop"}.get(value, "unknown") + return {EVENT_INGRESS: "ingress", EVENT_EGRESS: "egress", EVENT_DROP: "drop"}.get(value, "unknown") def _reason_name(reason: int) -> str: return f"skb_drop_reason_{reason}" -def _emit_event(cpu: int, data: int, size: int) -> None: - del cpu, size - event = ct.cast(data, ct.POINTER(Event)).contents - iface = bytes(event.ifname).split(b"\x00", 1)[0].decode("utf-8", "replace") +def _build_payload(event: Event) -> dict[str, object] | None: + iface = _ifname_from_index(int(event.ifindex)) if iface not in TARGET_INTERFACES: - return + return None payload: dict[str, object] = { - "event_type": _event_name(event.event_type), + "event_type": _event_name(int(event.event_type)), "iface": iface, "skb_mark": int(event.skb_mark) or None, "length": int(event.length), @@ -342,9 +499,10 @@ def _emit_event(cpu: int, data: int, size: int) -> None: "protocol_raw": int(event.protocol_raw) or None, "src_port": int(event.src_port) or None, "dst_port": int(event.dst_port) or None, - "reason": _reason_name(int(event.reason)) if event.event_type == 3 else None, - "reason_code": int(event.reason) if event.event_type == 3 else None, + "reason": _reason_name(int(event.reason)) if int(event.event_type) == EVENT_DROP else None, + "reason_code": int(event.reason) if int(event.event_type) == EVENT_DROP else None, } + packet_id = packet_id_from_mark(payload.get("skb_mark")) if packet_id: payload["packet_id"] = packet_id @@ -357,67 +515,37 @@ def _emit_event(cpu: int, data: int, size: int) -> None: payload["packet_uid"] = _build_packet_uid(payload) payload["correlation_key"] = f"uid:{payload['packet_uid']}" payload["correlation_source"] = "legacy_hash" + return payload + + +def _emit_ingress_event(cpu: int, data: int, size: int) -> None: + del cpu + event = ct.cast(data, ct.POINTER(Event)).contents + payload = _build_payload(event) + if payload is None: + return + + raw_size = size - ct.sizeof(Event) + if raw_size > 0: + raw = ct.string_at(data + ct.sizeof(Event), raw_size) + payload["raw_b64"] = base64.b64encode(raw).decode("ascii") + print(json.dumps(payload, separators=(",", ":")), flush=True) -def _attach_kprobe_first(bpf: BPF, symbols: list[str], fn_name: str) -> str: - supported = _supported_kprobe_symbols(symbols) - if not supported: - raise RuntimeError(f"No supported kprobe symbols found for {fn_name}: {symbols}") - - for symbol in supported: - try: - bpf.attach_kprobe(event=symbol, fn_name=fn_name) - return symbol - except Exception: - continue - raise RuntimeError(f"Failed to attach {fn_name} to any of {supported}") - - -def _supported_kprobe_symbols(symbols: list[str]) -> list[str]: - try: - available = set() - for symbol in symbols: - for candidate in BPF.get_kprobe_functions(symbol.encode()): - decoded = candidate.decode("utf-8", "replace") - if decoded == symbol: - available.add(symbol) - if available: - return [symbol for symbol in symbols if symbol in available] - except Exception: - pass - - if os.path.exists("/proc/kallsyms"): - try: - with open("/proc/kallsyms", "r", encoding="utf-8", errors="replace") as handle: - names = {line.rsplit(" ", 1)[-1].strip() for line in handle} - return [symbol for symbol in symbols if symbol in names] - except Exception: - pass - - return symbols - - -def _attach_egress_probe(bpf: BPF) -> str: - symbols = ["__dev_queue_xmit", "dev_queue_xmit"] - try: - return _attach_kprobe_first(bpf, symbols, "trace_egress") - except Exception: - pass - - try: - bpf.attach_tracepoint(tp="net:net_dev_queue", fn_name="tracepoint__net__net_dev_queue") - return "tracepoint:net:net_dev_queue" - except Exception as exc: - raise RuntimeError( - "Failed to attach egress telemetry to any of " - f"{symbols} or tracepoint net:net_dev_queue" - ) from exc +def _emit_meta_event(cpu: int, data: int, size: int) -> None: + del cpu, size + event = ct.cast(data, ct.POINTER(Event)).contents + payload = _build_payload(event) + if payload is None: + return + print(json.dumps(payload, separators=(",", ":")), flush=True) def _parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser(description="eBPF bridge telemetry collector") - parser.add_argument("--ifaces", required=True, help="Comma-separated list of interfaces to keep") + parser = argparse.ArgumentParser(description="tc/eBPF bridge telemetry collector") + parser.add_argument("--ifaces", required=True, help="Comma-separated list of interfaces to instrument") + parser.add_argument("--build-dir", required=True, help="Directory for compiled tc BPF objects") return parser.parse_args() @@ -425,47 +553,131 @@ def _sigterm(_signum: int, _frame: object) -> None: raise KeyboardInterrupt +def _ensure_clean_clsact(iface: str) -> None: + subprocess.run(["tc", "qdisc", "del", "dev", iface, "clsact"], check=False, capture_output=True, text=True) + _run_checked(["tc", "qdisc", "add", "dev", iface, "clsact"]) + + +def _attach_mark_filter(iface: str, obj_path: Path) -> None: + _run_checked( + [ + "tc", + "filter", + "replace", + "dev", + iface, + "ingress", + "pref", + str(MARK_FILTER_PREF), + "protocol", + "all", + "bpf", + "direct-action", + "obj", + str(obj_path), + "sec", + "classifier", + ] + ) + + +def _attach_tc_programs(bpf: BPF, ifaces: list[str], build_dir: str) -> tuple[str, str]: + global IPR + obj_path = _ensure_mark_object(build_dir) + ingress_fn = bpf.load_func("handle_ingress", BPF.SCHED_CLS) + egress_fn = bpf.load_func("handle_egress", BPF.SCHED_CLS) + + ipr = IPRoute() + try: + for iface in ifaces: + matches = ipr.link_lookup(ifname=iface) + if not matches: + raise RuntimeError(f"Interface not found: {iface}") + ifindex = matches[0] + _ensure_clean_clsact(iface) + _attach_mark_filter(iface, obj_path) + ipr.tc( + "add-filter", + "bpf", + ifindex, + INGRESS_FILTER_HANDLE, + fd=ingress_fn.fd, + name=ingress_fn.name, + parent=INGRESS_PARENT, + classid=1, + direct_action=True, + ) + ipr.tc( + "add-filter", + "bpf", + ifindex, + EGRESS_FILTER_HANDLE, + fd=egress_fn.fd, + name=egress_fn.name, + parent=EGRESS_PARENT, + classid=1, + direct_action=True, + ) + except Exception: + for iface in ifaces: + subprocess.run(["tc", "qdisc", "del", "dev", iface, "clsact"], check=False, capture_output=True, text=True) + if ipr is not None: + ipr.close() + IPR = None + raise + + return ingress_fn.name, egress_fn.name + + +def _cleanup_tc(ifaces: Iterable[str]) -> None: + for iface in ifaces: + subprocess.run(["tc", "qdisc", "del", "dev", iface, "clsact"], check=False, capture_output=True, text=True) + global IPR + if IPR is not None: + try: + IPR.close() + finally: + IPR = None + + def main() -> int: args = _parse_args() global TARGET_INTERFACES TARGET_INTERFACES = {iface.strip() for iface in args.ifaces.split(",") if iface.strip()} if not TARGET_INTERFACES: - print("No interfaces provided", file=sys.stderr) + print("No interfaces provided", file=sys.stderr, flush=True) return 1 signal.signal(signal.SIGTERM, _sigterm) signal.signal(signal.SIGINT, _sigterm) bpf = BPF(text=BPF_SOURCE) - ingress_symbol = _attach_kprobe_first( - bpf, - ["__netif_receive_skb_core", "netif_receive_skb", "__netif_receive_skb_one_core"], - "trace_ingress", - ) - egress_symbol = _attach_egress_probe(bpf) - - print( - json.dumps( - { - "status": "collector_started", - "ifaces": sorted(TARGET_INTERFACES), - "ingress_symbol": ingress_symbol, - "egress_symbol": egress_symbol, - }, - separators=(",", ":"), - ), - flush=True, - ) - - bpf["events"].open_perf_buffer(_emit_event, page_cnt=128) + ingress_prog_name = "" + egress_prog_name = "" try: + ingress_prog_name, egress_prog_name = _attach_tc_programs(bpf, sorted(TARGET_INTERFACES), args.build_dir) + print( + json.dumps( + { + "status": "collector_started", + "ifaces": sorted(TARGET_INTERFACES), + "ingress_program": ingress_prog_name, + "egress_program": egress_prog_name, + "build_dir": args.build_dir, + }, + separators=(",", ":"), + ), + flush=True, + ) + + bpf["ingress_events"].open_perf_buffer(_emit_ingress_event, page_cnt=256) + bpf["meta_events"].open_perf_buffer(_emit_meta_event, page_cnt=128) while True: bpf.perf_buffer_poll() except KeyboardInterrupt: return 0 - - -TARGET_INTERFACES: set[str] = set() + finally: + _cleanup_tc(sorted(TARGET_INTERFACES)) if __name__ == "__main__": diff --git a/backend/src/utilities/packet_tracker.py b/backend/src/utilities/packet_tracker.py index 41e328f..f205d38 100644 --- a/backend/src/utilities/packet_tracker.py +++ b/backend/src/utilities/packet_tracker.py @@ -38,7 +38,7 @@ class PacketTracker: self._entries: Dict[str, Dict[str, Any]] = {} self._stats: Dict[str, int] = { "persisted_total": 0, - "persisted_af_packet_only": 0, + "persisted_capture_only": 0, "persisted_telemetry_only": 0, "persisted_merged": 0, "persisted_with_raw": 0, @@ -60,7 +60,7 @@ class PacketTracker: now_ts = time.time() correlation_key = self._ensure_correlation(pkt_info) pkt_info["raw_present"] = pkt_info.get("raw") is not None - pkt_info["capture_sources"] = ["af_packet"] + pkt_info["capture_sources"] = [pkt_info.get("capture_source") or "af_packet"] with self._lock: entry = self._entries.get(correlation_key) @@ -205,9 +205,9 @@ class PacketTracker: def _merge_packet_info(self, entry: Dict[str, Any], pkt_info: Dict[str, Any], now_ts: float) -> None: payload = entry["payload"] changed = False - self._add_capture_source(payload, "af_packet") + self._add_capture_source(payload, pkt_info.get("capture_source") or "af_packet") for key, value in pkt_info.items(): - if key == "iface": + if key in {"iface", "capture_source"}: continue if value is None: continue @@ -382,8 +382,8 @@ class PacketTracker: else: self._stats["persisted_legacy_hash"] = self._stats.get("persisted_legacy_hash", 0) + 1 - if capture_sources == {"af_packet"}: - self._stats["persisted_af_packet_only"] += 1 + if capture_sources and capture_sources != {"telemetry"} and "telemetry" not in capture_sources: + self._stats["persisted_capture_only"] += 1 elif capture_sources == {"telemetry"}: self._stats["persisted_telemetry_only"] += 1 else: @@ -397,15 +397,15 @@ class PacketTracker: active_total = len(active_entries) active_with_raw = sum(1 for entry in active_entries if entry["payload"].get("raw_present")) active_without_raw = active_total - active_with_raw - active_af_packet_only = 0 + active_capture_only = 0 active_telemetry_only = 0 active_merged = 0 active_kernel_mark = 0 active_legacy_hash = 0 for entry in active_entries: capture_sources = set(entry["payload"].get("capture_sources") or []) - if capture_sources == {"af_packet"}: - active_af_packet_only += 1 + if capture_sources and capture_sources != {"telemetry"} and "telemetry" not in capture_sources: + active_capture_only += 1 elif capture_sources == {"telemetry"}: active_telemetry_only += 1 else: @@ -419,7 +419,7 @@ class PacketTracker: "active_total": active_total, "active_with_raw": active_with_raw, "active_without_raw": active_without_raw, - "active_af_packet_only": active_af_packet_only, + "active_capture_only": active_capture_only, "active_telemetry_only": active_telemetry_only, "active_merged": active_merged, "active_kernel_mark": active_kernel_mark, diff --git a/tools/ebpf/prepend_capture_header.c b/tools/ebpf/prepend_capture_header.c deleted file mode 100644 index f6bfdce..0000000 --- a/tools/ebpf/prepend_capture_header.c +++ /dev/null @@ -1,41 +0,0 @@ -#include -#include -#include -#include - -#define CAPTURE_MAGIC 0x4d544350 -#define CAPTURE_VERSION 1 - -struct capture_header { - __be32 magic; - __u8 version; - __u8 flags; - __be16 header_len; - __be32 skb_mark; - __be32 aux_value; -} __attribute__((packed)); - -SEC("classifier") -int prepend_capture_header_prog(struct __sk_buff *skb) { - struct capture_header hdr = { - .magic = bpf_htonl(CAPTURE_MAGIC), - .version = CAPTURE_VERSION, - .flags = 0, - .header_len = bpf_htons(sizeof(struct capture_header)), - .skb_mark = bpf_htonl(skb->mark), - .aux_value = 0, - }; - - if (!skb->mark) { - return TC_ACT_OK; - } - if (bpf_skb_adjust_room(skb, sizeof(hdr), BPF_ADJ_ROOM_MAC, 0)) { - return TC_ACT_OK; - } - if (bpf_skb_store_bytes(skb, 0, &hdr, sizeof(hdr), 0)) { - return TC_ACT_OK; - } - return TC_ACT_OK; -} - -char LICENSE[] SEC("license") = "GPL"; diff --git a/tools/setup_bridge_capture.sh b/tools/setup_bridge_capture.sh deleted file mode 100755 index 72640f5..0000000 --- a/tools/setup_bridge_capture.sh +++ /dev/null @@ -1,98 +0,0 @@ -#!/usr/bin/env bash -set -euo pipefail - -usage() { - cat <<'EOF' -Usage: tools/setup_bridge_capture.sh --bridge [--mirror-if mitmcap0] [--capture-if mitmcap1] [--build-dir /tmp/mitm-bpf] - -Sets up: -1. A veth pair used as a capture mirror target -2. tc ingress packet-id marking on each bridge slave -3. tc mirroring from each bridge slave into the mirror interface -4. A capture-header injector on the capture-side interface - -Set BACKEND_CAPTURE_INTERFACE to the capture interface printed at the end. -EOF -} - -BRIDGE="" -MIRROR_IF="mitmcap0" -CAPTURE_IF="mitmcap1" -BUILD_DIR="/tmp/mitm-bpf" - -while [[ $# -gt 0 ]]; do - case "$1" in - --bridge) BRIDGE="$2"; shift 2 ;; - --mirror-if) MIRROR_IF="$2"; shift 2 ;; - --capture-if) CAPTURE_IF="$2"; shift 2 ;; - --build-dir) BUILD_DIR="$2"; shift 2 ;; - -h|--help) usage; exit 0 ;; - *) echo "Unknown argument: $1" >&2; usage; exit 1 ;; - esac -done - -if [[ -z "$BRIDGE" ]]; then - usage - exit 1 -fi - -if ! command -v tc >/dev/null 2>&1 || ! command -v clang >/dev/null 2>&1 || ! command -v ip >/dev/null 2>&1 || ! command -v bridge >/dev/null 2>&1; then - echo "Missing required tools: tc, clang, ip, and bridge must be installed." >&2 - exit 1 -fi - -if ! ip link show "$BRIDGE" >/dev/null 2>&1; then - echo "Bridge interface not found: $BRIDGE" >&2 - exit 1 -fi - -mkdir -p "$BUILD_DIR" -SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" -SRC_DIR="$SCRIPT_DIR/ebpf" -MARK_OBJ="$BUILD_DIR/mark_packet_id.o" -CAPTURE_OBJ="$BUILD_DIR/prepend_capture_header.o" -MULTIARCH_INCLUDE="" -if command -v gcc >/dev/null 2>&1; then - GCC_TRIPLE="$(gcc -dumpmachine 2>/dev/null || true)" - if [[ -n "$GCC_TRIPLE" && -d "/usr/include/$GCC_TRIPLE" ]]; then - MULTIARCH_INCLUDE="-I/usr/include/$GCC_TRIPLE" - fi -fi - -clang -O2 -g -target bpf ${MULTIARCH_INCLUDE:+$MULTIARCH_INCLUDE} -c "$SRC_DIR/mark_packet_id.c" -o "$MARK_OBJ" -clang -O2 -g -target bpf ${MULTIARCH_INCLUDE:+$MULTIARCH_INCLUDE} -c "$SRC_DIR/prepend_capture_header.c" -o "$CAPTURE_OBJ" - -if ! ip link show "$MIRROR_IF" >/dev/null 2>&1; then - ip link add "$MIRROR_IF" type veth peer name "$CAPTURE_IF" -fi - -ip link set "$MIRROR_IF" up -ip link set "$CAPTURE_IF" up -sysctl -q -w "net.ipv6.conf.$MIRROR_IF.disable_ipv6=1" >/dev/null || true -sysctl -q -w "net.ipv6.conf.$CAPTURE_IF.disable_ipv6=1" >/dev/null || true - -mapfile -t PORTS < <(bridge link show master "$BRIDGE" | awk -F': ' '{print $2}' | awk '{print $1}') -if [[ ${#PORTS[@]} -eq 0 ]]; then - echo "No bridge slave interfaces found for $BRIDGE" >&2 - exit 1 -fi - -for port in "${PORTS[@]}"; do - tc qdisc replace dev "$port" clsact - tc filter replace dev "$port" ingress pref 10 protocol all bpf direct-action obj "$MARK_OBJ" sec classifier - tc filter replace dev "$port" ingress pref 20 protocol all matchall action mirred egress mirror dev "$MIRROR_IF" -done - -tc qdisc replace dev "$MIRROR_IF" clsact -tc filter replace dev "$MIRROR_IF" egress pref 10 protocol all bpf direct-action obj "$CAPTURE_OBJ" sec classifier - -cat < [--mirror-if mitmcap0] [--capture-if mitmcap1]" -} - -BRIDGE="" -MIRROR_IF="mitmcap0" -CAPTURE_IF="mitmcap1" - -while [[ $# -gt 0 ]]; do - case "$1" in - --bridge) BRIDGE="$2"; shift 2 ;; - --mirror-if) MIRROR_IF="$2"; shift 2 ;; - --capture-if) CAPTURE_IF="$2"; shift 2 ;; - -h|--help) usage; exit 0 ;; - *) echo "Unknown argument: $1" >&2; usage; exit 1 ;; - esac -done - -if [[ -n "$BRIDGE" ]] && ip link show "$BRIDGE" >/dev/null 2>&1; then - while read -r port; do - [[ -n "$port" ]] || continue - tc qdisc del dev "$port" clsact 2>/dev/null || true - done < <(bridge link show master "$BRIDGE" | awk -F': ' '{print $2}' | awk '{print $1}') -fi - -tc qdisc del dev "$MIRROR_IF" clsact 2>/dev/null || true -tc qdisc del dev "$CAPTURE_IF" clsact 2>/dev/null || true - -if ip link show "$MIRROR_IF" >/dev/null 2>&1; then - ip link del "$MIRROR_IF" -fi