diff --git a/backend/requirements.txt b/backend/requirements.txt index 5377e25..79ba04c 100644 Binary files a/backend/requirements.txt and b/backend/requirements.txt differ diff --git a/backend/src/Models/packets.py b/backend/src/Models/packets.py index 09731d4..b91ef32 100644 --- a/backend/src/Models/packets.py +++ b/backend/src/Models/packets.py @@ -11,7 +11,11 @@ class PacketDBModel(BaseModel): id: Union[int, str] timestamp: datetime = Field(..., description="Packet timestamp in ISO format.") - iface: str + packet_uid: str = Field(..., description="Stable packet identity used for upserts/correlation.") + iface: Optional[str] = None + ingress_if: Optional[str] = None + egress_if: Optional[str] = None + observed_ifaces: Optional[list[str]] = None src_mac: Optional[str] = None dst_mac: Optional[str] = None eth_type: Optional[Union[int, str]] = None @@ -31,7 +35,14 @@ 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.") + telemetry_metadata: Optional[dict] = Field(None, description="Kernel telemetry details from eBPF collector.") direction: Optional[str] = None + verdict: Optional[str] = None + verdict_reason: Optional[str] = None + verdict_confidence: Optional[str] = None + ingress_seen_at: Optional[datetime] = None + egress_seen_at: Optional[datetime] = None + verdict_seen_at: Optional[datetime] = None packets: Optional[int] = None class Config: @@ -39,7 +50,11 @@ class PacketDBModel(BaseModel): "example": { "id": 123, "timestamp": "2026-03-05T12:34:56.789Z", + "packet_uid": "9f6d3af0d3c81cb20ee8e7d32df7c56414460542", "iface": "eth0", + "ingress_if": "eth0", + "egress_if": "eth1", + "observed_ifaces": ["eth0", "eth1"], "src_mac": "aa:bb:cc:dd:ee:ff", "dst_mac": "11:22:33:44:55:66", "eth_type": "IPv4", @@ -59,5 +74,10 @@ class PacketDBModel(BaseModel): "app_is_encrypted": False, "app_risk_score": 0, "dpi_metadata": {"method": "GET"}, + "telemetry_metadata": {"event_type": "egress", "iface": "eth1"}, + "direction": "forwarded", + "verdict": "accept", + "verdict_reason": "egress-observed", + "verdict_confidence": "high", } } diff --git a/backend/src/main.py b/backend/src/main.py index 79b98ec..00d4491 100644 --- a/backend/src/main.py +++ b/backend/src/main.py @@ -77,6 +77,26 @@ async def shutdown_event() -> None: except Exception: logging.exception("Error shutting down network API") + try: + from src.utilities.bridge_telemetry import bridge_telemetry_manager + + bridge_telemetry_manager.stop() + except Exception: + logging.exception("Failed to stop bridge telemetry collector") + + try: + from src.utilities.packet_tracker import packet_tracker + + packet_tracker.stop() + except Exception: + logging.exception("Failed to stop packet tracker") + + try: + if shared_objects.broadcaster is not None: + await shared_objects.broadcaster.close() + except Exception: + logging.exception("Failed to close packet broadcaster during shutdown") + try: web_db = getattr(shared_objects, "db", None) if web_db is not None: diff --git a/backend/src/network_sniffer.py b/backend/src/network_sniffer.py index 25ed73e..9f47178 100644 --- a/backend/src/network_sniffer.py +++ b/backend/src/network_sniffer.py @@ -31,7 +31,10 @@ from src.utilities.interface_bridge_helpers import ( check_interface_up, get_bridge_ports_once, ) +from src.utilities.bridge_telemetry import bridge_telemetry_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 from src.Models.etherType import EtherTypeEnum, ethertype_from_int from src.Models.ip_protocol import IPProtocolEnum, protocol_from_number @@ -55,13 +58,12 @@ sessions: Dict[str, Dict[str, Any]] = {} # PacketInfo typing # ------------------------- class PacketInfo(TypedDict, total=False): - """ - TypedDict for the parsed packet info produced by parse_packet. - Fields marked optional (total=False) for flexibility across contexts. - """ + """TypedDict for parsed packet data used by persistence and telemetry.""" + + packet_uid: str iface: str length: int - raw: bytes # original raw bytes (kept for buffering; DB helper may convert to base64) + raw: bytes src_mac: Optional[str] dst_mac: Optional[str] eth_type_raw: Optional[int] @@ -82,6 +84,18 @@ class PacketInfo(TypedDict, total=False): app_is_encrypted: Optional[bool] app_risk_score: Optional[int] dpi_metadata: Optional[Dict[str, Any]] + ip_id: Optional[int] + icmp_type: Optional[int] + icmp_code: Optional[int] + arp_op: Optional[int] + tcp_seq: Optional[int] + tcp_ack: Optional[int] + tcp_flags: Optional[int] + icmp_embedded_src_ip: Optional[str] + icmp_embedded_dst_ip: Optional[str] + icmp_embedded_protocol: Optional[int] + icmp_embedded_src_port: Optional[int] + icmp_embedded_dst_port: Optional[int] # small bounded buffer for packets produced before shared_objects is ready @@ -105,24 +119,19 @@ threading.Thread(target=_start_async_loop, args=(async_loop,), daemon=True).star # Helpers for buffer draining # ------------------------- def drain_buffer_to_shared_db() -> None: - """ - Attempt to schedule buffered packets for insertion on shared_objects.web_loop. - Call this from main.py after shared_objects.db and shared_objects.web_loop are initialized. - """ + """Replay buffered packets once the shared DB loop becomes available.""" try: web_loop = getattr(shared_objects, "web_loop", None) web_db = getattr(shared_objects, "db", None) if web_db is None or web_loop is None: return - # schedule draining on the web loop to avoid blocking this thread def _drain() -> None: while _PACKET_BUFFER: pkt = _PACKET_BUFFER.pop(0) try: - asyncio.run_coroutine_threadsafe(web_db.insert_packet(pkt), web_loop) + packet_tracker.observe_packet(pkt) except Exception: - # re-buffer first element and stop to avoid busy loop _PACKET_BUFFER.insert(0, pkt) break @@ -180,6 +189,18 @@ def parse_packet(pkt, bridge_label: str) -> None: "app_is_encrypted": None, "app_risk_score": None, "dpi_metadata": None, + "ip_id": None, + "icmp_type": None, + "icmp_code": None, + "arp_op": None, + "tcp_seq": None, + "tcp_ack": None, + "tcp_flags": None, + "icmp_embedded_src_ip": None, + "icmp_embedded_dst_ip": None, + "icmp_embedded_protocol": None, + "icmp_embedded_src_port": None, + "icmp_embedded_dst_port": None, } # Ethernet layer @@ -220,6 +241,10 @@ def parse_packet(pkt, bridge_label: str) -> None: pkt_info["protocol_name"] = "ARP" pkt_info["src_ip"] = _safe_get_attr(arp, "psrc") pkt_info["dst_ip"] = _safe_get_attr(arp, "pdst") + try: + pkt_info["arp_op"] = int(_safe_get_attr(arp, "op")) + except Exception: + pkt_info["arp_op"] = None pkt_info["src_port"] = None pkt_info["dst_port"] = None @@ -228,6 +253,10 @@ def parse_packet(pkt, bridge_label: str) -> None: ip = pkt[IP] pkt_info["src_ip"] = _safe_get_attr(ip, "src") pkt_info["dst_ip"] = _safe_get_attr(ip, "dst") + try: + pkt_info["ip_id"] = int(_safe_get_attr(ip, "id")) + except Exception: + pkt_info["ip_id"] = None try: proto_num = int(_safe_get_attr(ip, "proto")) @@ -245,12 +274,23 @@ def parse_packet(pkt, bridge_label: str) -> None: pkt_info["protocol_name"] = "TCP" pkt_info["src_port"] = _safe_get_attr(pkt[TCP], "sport") pkt_info["dst_port"] = _safe_get_attr(pkt[TCP], "dport") + try: + pkt_info["tcp_seq"] = int(_safe_get_attr(pkt[TCP], "seq")) + pkt_info["tcp_ack"] = int(_safe_get_attr(pkt[TCP], "ack")) + pkt_info["tcp_flags"] = int(_safe_get_attr(pkt[TCP], "flags")) + except Exception: + pass elif proto_num == 17 and UDP in pkt: pkt_info["protocol_name"] = "UDP" pkt_info["src_port"] = _safe_get_attr(pkt[UDP], "sport") pkt_info["dst_port"] = _safe_get_attr(pkt[UDP], "dport") elif proto_num == 1 and ICMP in pkt: pkt_info["protocol_name"] = "ICMP" + try: + pkt_info["icmp_type"] = int(_safe_get_attr(pkt[ICMP], "type")) + pkt_info["icmp_code"] = int(_safe_get_attr(pkt[ICMP], "code")) + except Exception: + pass else: if pkt_info.get("protocol_name") is None: pkt_info["protocol_name"] = f"IP_PROTO_{proto_num}" if proto_num is not None else None @@ -277,12 +317,23 @@ def parse_packet(pkt, bridge_label: str) -> None: pkt_info["protocol_name"] = "TCP" pkt_info["src_port"] = _safe_get_attr(pkt[TCP], "sport") pkt_info["dst_port"] = _safe_get_attr(pkt[TCP], "dport") + try: + pkt_info["tcp_seq"] = int(_safe_get_attr(pkt[TCP], "seq")) + pkt_info["tcp_ack"] = int(_safe_get_attr(pkt[TCP], "ack")) + pkt_info["tcp_flags"] = int(_safe_get_attr(pkt[TCP], "flags")) + except Exception: + pass elif nh == 17 and UDP in pkt: pkt_info["protocol_name"] = "UDP" pkt_info["src_port"] = _safe_get_attr(pkt[UDP], "sport") pkt_info["dst_port"] = _safe_get_attr(pkt[UDP], "dport") elif ICMPv6Unknown in pkt: pkt_info["protocol_name"] = "ICMPv6" + try: + pkt_info["icmp_type"] = int(_safe_get_attr(pkt[ICMPv6Unknown], "type")) + pkt_info["icmp_code"] = int(_safe_get_attr(pkt[ICMPv6Unknown], "code")) + except Exception: + pass else: if pkt_info.get("protocol_name") is None: pkt_info["protocol_name"] = f"IPV6_PROTO_{nh}" if nh is not None else None @@ -299,21 +350,35 @@ def parse_packet(pkt, bridge_label: str) -> None: except Exception: logger.exception("nDPI enrichment failed") - # Submit DB insert to shared web loop if available, otherwise buffer + if ICMP in pkt: + inner = pkt[ICMP].payload + if inner and IP in inner: + inner_ip = inner[IP] + pkt_info["icmp_embedded_src_ip"] = _safe_get_attr(inner_ip, "src") + pkt_info["icmp_embedded_dst_ip"] = _safe_get_attr(inner_ip, "dst") + try: + pkt_info["icmp_embedded_protocol"] = int(_safe_get_attr(inner_ip, "proto")) + except Exception: + pkt_info["icmp_embedded_protocol"] = None + + if TCP in inner: + pkt_info["icmp_embedded_src_port"] = _safe_get_attr(inner[TCP], "sport") + pkt_info["icmp_embedded_dst_port"] = _safe_get_attr(inner[TCP], "dport") + elif UDP in inner: + pkt_info["icmp_embedded_src_port"] = _safe_get_attr(inner[UDP], "sport") + pkt_info["icmp_embedded_dst_port"] = _safe_get_attr(inner[UDP], "dport") + + pkt_info["packet_uid"] = build_packet_uid(pkt_info) + try: web_loop = getattr(shared_objects, "web_loop", None) web_db = getattr(shared_objects, "db", None) if web_db is not None and web_loop is not None: try: - fut = asyncio.run_coroutine_threadsafe(web_db.insert_packet(pkt_info), web_loop) - # best-effort non-blocking check for immediate errors - try: - fut.result(timeout=0.005) - except Exception: - pass - logger.debug("Scheduled insert for packet on %s (len=%d)", pkt_info.get("iface"), pkt_info.get("length")) + packet_tracker.observe_packet(pkt_info) + logger.debug("Tracked packet on %s (len=%d)", pkt_info.get("iface"), pkt_info.get("length")) except Exception as e: - logger.exception("Failed to schedule insert_packet for %s — buffering: %s", pkt_info.get("iface"), e) + logger.exception("Failed to track packet for %s — buffering: %s", pkt_info.get("iface"), e) _PACKET_BUFFER.append(pkt_info) if len(_PACKET_BUFFER) > _BUFFER_CAPACITY: _PACKET_BUFFER.pop(0) @@ -399,6 +464,14 @@ def _ensure_socket_for_session(sockets: Dict[str, socket.socket], iface: str, br logger.warning("Failed to create AF_PACKET socket for %s (label=%s)", iface, bridge_label) +def _sync_bridge_telemetry() -> None: + interfaces = sorted({iface for session in sessions.values() for iface in session.get("ports", [])}) + try: + bridge_telemetry_manager.update_interfaces(interfaces) + except Exception: + logger.exception("Failed to update bridge telemetry collector") + + # ------------------------- # Per-session reader loop # ------------------------- @@ -545,6 +618,7 @@ def start_afpacket_sniffer(target: str, target_is_interface: bool = False) -> st t = threading.Thread(target=_session_reader_loop, args=(session_id,), daemon=True) session["thread"] = t t.start() + _sync_bridge_telemetry() logger.info("Started sniffer session %s label=%s ports=%s", session_id, target, ports) return session_id @@ -566,6 +640,7 @@ 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=2) + _sync_bridge_telemetry() logger.info("Stopped session %s", session_id) return @@ -586,6 +661,7 @@ def stop_afpacket_sniffer(session_id: Optional[str] = None, target: Optional[str except Exception: pass logger.info("Removed target %s from session %s", target, sid) + _sync_bridge_telemetry() return # Global stop: stop all sessions @@ -602,6 +678,10 @@ def stop_afpacket_sniffer(session_id: Optional[str] = None, target: Optional[str logger.exception("Failed to schedule DB pool close") logger.info("All sniffer sessions stopped") + try: + bridge_telemetry_manager.stop() + except Exception: + logger.exception("Failed to stop bridge telemetry collector") def get_sniffer_status() -> Dict[str, Dict[str, object]]: @@ -643,4 +723,5 @@ def get_internal_debug_state() -> dict: for sid, s in sessions.items() }, "buffer_len": len(_PACKET_BUFFER), + "telemetry_ports": sorted({iface for session in sessions.values() for iface in session.get("ports", [])}), } diff --git a/backend/src/utilities/bridge_telemetry.py b/backend/src/utilities/bridge_telemetry.py new file mode 100644 index 0000000..0959289 --- /dev/null +++ b/backend/src/utilities/bridge_telemetry.py @@ -0,0 +1,128 @@ +"""Manage the eBPF telemetry subprocess used for bridge direction and verdict events.""" + +from __future__ import annotations + +import json +import logging +import os +import signal +import subprocess +import sys +import threading +from pathlib import Path +from typing import Iterable, Optional + +from src.utilities.packet_tracker import packet_tracker + +logger = logging.getLogger("bridge_telemetry") + + +class BridgeTelemetryManager: + """Run a single eBPF collector process for the active sniffed interfaces.""" + + def __init__(self) -> None: + self._interfaces: set[str] = set() + self._process: Optional[subprocess.Popen[str]] = None + self._reader_thread: Optional[threading.Thread] = None + self._lock = threading.Lock() + + def update_interfaces(self, interfaces: Iterable[str]) -> None: + """Restart the collector when the active interface set changes.""" + normalized = {iface.strip() for iface in interfaces if iface and iface.strip()} + with self._lock: + if normalized == self._interfaces: + return + self._interfaces = normalized + self._restart_locked() + + def stop(self) -> None: + """Stop the collector process and reader thread.""" + with self._lock: + self._interfaces = set() + self._stop_process_locked() + + def _restart_locked(self) -> None: + self._stop_process_locked() + if not self._interfaces: + return + + helper = Path(__file__).with_name("ebpf_bridge_events.py") + python_bin = sys.executable or "python3" + if "venv" in python_bin: + python_bin = "python3" + + env = os.environ.copy() + env["PYTHONUNBUFFERED"] = "1" + + cmd = [python_bin, str(helper), "--ifaces", ",".join(sorted(self._interfaces))] + logger.info("Starting bridge telemetry collector for interfaces=%s", sorted(self._interfaces)) + try: + self._process = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + env=env, + ) + except Exception: + logger.exception("Failed to start bridge telemetry collector") + self._process = None + return + + self._reader_thread = threading.Thread( + target=self._read_loop, + args=(self._process,), + daemon=True, + name="bridge-telemetry-reader", + ) + self._reader_thread.start() + + def _stop_process_locked(self) -> None: + process = self._process + reader = self._reader_thread + self._process = None + self._reader_thread = None + + if process is not None and process.poll() is None: + try: + process.send_signal(signal.SIGTERM) + process.wait(timeout=3) + except subprocess.TimeoutExpired: + process.kill() + except Exception: + logger.exception("Failed to stop bridge telemetry collector cleanly") + + if reader is not None and reader.is_alive(): + reader.join(timeout=2) + + def _read_loop(self, process: subprocess.Popen[str]) -> None: + stdout = process.stdout + if stdout is None: + return + + for line in stdout: + text = line.strip() + if not text: + continue + try: + event = json.loads(text) + except json.JSONDecodeError: + logger.info("bridge-telemetry: %s", text) + continue + + if "event_type" not in event: + logger.info("bridge-telemetry: %s", event) + continue + + try: + packet_tracker.observe_telemetry(event) + except Exception: + logger.exception("Failed to process telemetry event: %s", event) + + rc = process.poll() + if rc not in (0, None): + logger.warning("Bridge telemetry collector exited with code %s", rc) + + +bridge_telemetry_manager = BridgeTelemetryManager() diff --git a/backend/src/utilities/database.py b/backend/src/utilities/database.py index 82f3868..ee8e0be 100644 --- a/backend/src/utilities/database.py +++ b/backend/src/utilities/database.py @@ -15,6 +15,16 @@ from src.Models.packets import PacketDBModel logger = logging.getLogger("af_packet_sniffer") +def _db_text(value: Any) -> Any: + if value is None: + return None + if isinstance(value, (str, int)): + return value + if hasattr(value, "value"): + return getattr(value, "value") + return str(value) + + class DatabasePool: """Asyncpg connection pool wrapper used by the packet APIs.""" @@ -65,15 +75,33 @@ class DatabasePool: async def insert_packet(self, pkt_info: Dict[str, Any]) -> None: """Insert one packet record and publish it to subscribers.""" + await self.upsert_packet(pkt_info) + + async def upsert_packet(self, pkt_info: Dict[str, Any]) -> None: + """Insert or update one packet record and publish it to subscribers.""" if self._pool is None: await self.init_pool() + dpi_metadata = pkt_info.get("dpi_metadata") + telemetry_metadata = pkt_info.get("telemetry_metadata") + try: async with self._pool.acquire() as conn: new_row = await conn.fetchrow( """ - INSERT INTO packets( + INSERT INTO packets ( + packet_uid, iface, + ingress_if, + egress_if, + observed_ifaces, + direction, + verdict, + verdict_reason, + verdict_confidence, + ingress_seen_at, + egress_seen_at, + verdict_seen_at, src_mac, dst_mac, eth_type, @@ -84,7 +112,6 @@ class DatabasePool: src_port, dst_port, length, - raw, app_protocol, app_master_protocol, app_category, @@ -92,22 +119,70 @@ class DatabasePool: app_hostname, app_is_encrypted, app_risk_score, - dpi_metadata - ) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20::jsonb) + dpi_metadata, + telemetry_metadata, + raw + ) VALUES( + $1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12, + $13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23, + $24,$25,$26,$27,$28,$29,$30::jsonb,$31::jsonb,$32 + ) + ON CONFLICT (packet_uid) DO UPDATE SET + iface = COALESCE(EXCLUDED.iface, packets.iface), + ingress_if = COALESCE(EXCLUDED.ingress_if, packets.ingress_if), + egress_if = COALESCE(EXCLUDED.egress_if, packets.egress_if), + observed_ifaces = COALESCE(EXCLUDED.observed_ifaces, packets.observed_ifaces), + direction = COALESCE(EXCLUDED.direction, packets.direction), + verdict = COALESCE(EXCLUDED.verdict, packets.verdict), + verdict_reason = COALESCE(EXCLUDED.verdict_reason, packets.verdict_reason), + verdict_confidence = COALESCE(EXCLUDED.verdict_confidence, packets.verdict_confidence), + ingress_seen_at = COALESCE(EXCLUDED.ingress_seen_at, packets.ingress_seen_at), + egress_seen_at = COALESCE(EXCLUDED.egress_seen_at, packets.egress_seen_at), + verdict_seen_at = COALESCE(EXCLUDED.verdict_seen_at, packets.verdict_seen_at), + src_mac = COALESCE(EXCLUDED.src_mac, packets.src_mac), + dst_mac = COALESCE(EXCLUDED.dst_mac, packets.dst_mac), + eth_type = COALESCE(EXCLUDED.eth_type, packets.eth_type), + vlan_id = COALESCE(EXCLUDED.vlan_id, packets.vlan_id), + src_ip = COALESCE(EXCLUDED.src_ip, packets.src_ip), + dst_ip = COALESCE(EXCLUDED.dst_ip, packets.dst_ip), + ip_proto = COALESCE(EXCLUDED.ip_proto, packets.ip_proto), + src_port = COALESCE(EXCLUDED.src_port, packets.src_port), + dst_port = COALESCE(EXCLUDED.dst_port, packets.dst_port), + length = COALESCE(EXCLUDED.length, packets.length), + app_protocol = COALESCE(EXCLUDED.app_protocol, packets.app_protocol), + app_master_protocol = COALESCE(EXCLUDED.app_master_protocol, packets.app_master_protocol), + app_category = COALESCE(EXCLUDED.app_category, packets.app_category), + app_confidence = COALESCE(EXCLUDED.app_confidence, packets.app_confidence), + app_hostname = COALESCE(EXCLUDED.app_hostname, packets.app_hostname), + app_is_encrypted = COALESCE(EXCLUDED.app_is_encrypted, packets.app_is_encrypted), + app_risk_score = COALESCE(EXCLUDED.app_risk_score, packets.app_risk_score), + dpi_metadata = COALESCE(EXCLUDED.dpi_metadata, packets.dpi_metadata), + telemetry_metadata = COALESCE(EXCLUDED.telemetry_metadata, packets.telemetry_metadata), + raw = COALESCE(EXCLUDED.raw, packets.raw) RETURNING id, timestamp """, - pkt_info["iface"], + pkt_info["packet_uid"], + pkt_info.get("iface"), + pkt_info.get("ingress_if"), + pkt_info.get("egress_if"), + pkt_info.get("observed_ifaces"), + pkt_info.get("direction"), + pkt_info.get("verdict"), + pkt_info.get("verdict_reason"), + pkt_info.get("verdict_confidence"), + pkt_info.get("ingress_seen_at"), + pkt_info.get("egress_seen_at"), + pkt_info.get("verdict_seen_at"), pkt_info.get("src_mac"), pkt_info.get("dst_mac"), - pkt_info.get("eth_type"), + _db_text(pkt_info.get("eth_type")), pkt_info.get("vlan_id"), pkt_info.get("src_ip"), pkt_info.get("dst_ip"), - pkt_info.get("protocol"), + _db_text(pkt_info.get("protocol_name") or pkt_info.get("protocol")), pkt_info.get("src_port"), pkt_info.get("dst_port"), - pkt_info["length"], - pkt_info["raw"], + pkt_info.get("length"), pkt_info.get("app_protocol"), pkt_info.get("app_master_protocol"), pkt_info.get("app_category"), @@ -115,10 +190,12 @@ class DatabasePool: pkt_info.get("app_hostname"), pkt_info.get("app_is_encrypted"), pkt_info.get("app_risk_score"), - json.dumps(pkt_info.get("dpi_metadata")) if pkt_info.get("dpi_metadata") is not None else None, + json.dumps(dpi_metadata) if dpi_metadata is not None else None, + json.dumps(telemetry_metadata) if telemetry_metadata is not None else None, + pkt_info.get("raw"), ) except Exception: - logger.exception("DB insert failed") + logger.exception("DB upsert failed") return if new_row: diff --git a/backend/src/utilities/ebpf_bridge_events.py b/backend/src/utilities/ebpf_bridge_events.py new file mode 100644 index 0000000..19534a4 --- /dev/null +++ b/backend/src/utilities/ebpf_bridge_events.py @@ -0,0 +1,443 @@ +#!/usr/bin/env python3 +"""Emit bridge ingress, egress, and drop telemetry events via eBPF.""" + +from __future__ import annotations + +import argparse +import ctypes as ct +import hashlib +import ipaddress +import json +import signal +import socket +import sys +from typing import Iterable + +try: + from bcc import BPF # type: ignore +except Exception as exc: # pragma: no cover - depends on host runtime + print(f"Failed to import python3-bpfcc: {exc}", file=sys.stderr, flush=True) + raise + + +IDENTITY_FIELDS = ( + "src_mac", + "dst_mac", + "eth_type_raw", + "vlan_id", + "src_ip", + "dst_ip", + "protocol_raw", + "src_port", + "dst_port", + "length", + "ip_id", + "icmp_type", + "icmp_code", + "arp_op", + "tcp_seq", + "tcp_ack", + "tcp_flags", +) + +BPF_SOURCE = r""" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#define EVENT_INGRESS 1 +#define EVENT_EGRESS 2 +#define EVENT_DROP 3 + +struct vlan_hdr_t { + __be16 h_vlan_TCI; + __be16 h_vlan_encapsulated_proto; +}; + +struct arp_eth_ipv4_t { + __u8 sha[6]; + __u8 spa[4]; + __u8 tha[6]; + __u8 tpa[4]; +}; + +struct event_t { + __u64 ts_ns; + __u32 length; + __u32 reason; + __u16 eth_type_raw; + __u16 vlan_id; + __u16 src_port; + __u16 dst_port; + __u16 ip_id; + __u16 arp_op; + __u32 protocol_raw; + __u32 tcp_seq; + __u32 tcp_ack; + __u8 event_type; + __u8 ip_version; + __u8 icmp_type; + __u8 icmp_code; + __u8 tcp_flags; + char ifname[IFNAMSIZ]; + unsigned char src_mac[6]; + unsigned char dst_mac[6]; + unsigned char src_ip[16]; + unsigned char dst_ip[16]; +}; + +BPF_PERF_OUTPUT(events); + +static __always_inline int fill_ifname(struct sk_buff *skb, struct event_t *event) { + if (!skb) { + return 0; + } + + struct net_device *dev = NULL; + bpf_probe_read_kernel(&dev, sizeof(dev), &skb->dev); + if (!dev) { + return 0; + } + + 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) { + unsigned char *head = NULL; + __u16 mac_header = 0; + __u16 network_header = 0; + __u16 transport_header = 0; + + if (!skb) { + return 0; + } + + bpf_probe_read_kernel(&head, sizeof(head), &skb->head); + bpf_probe_read_kernel(&mac_header, sizeof(mac_header), &skb->mac_header); + bpf_probe_read_kernel(&network_header, sizeof(network_header), &skb->network_header); + bpf_probe_read_kernel(&transport_header, sizeof(transport_header), &skb->transport_header); + bpf_probe_read_kernel(&event->length, sizeof(event->length), &skb->len); + + if (!head) { + return 0; + } + + 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); + + __be16 eth_proto = eth.h_proto; + unsigned char *l3_ptr = head + network_header; + if (eth_proto == htons(ETH_P_8021Q) || eth_proto == htons(ETH_P_8021AD)) { + struct vlan_hdr_t vlan = {}; + 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); + } + + if (eth_proto == htons(ETH_P_ARP)) { + struct arphdr arph = {}; + struct arp_eth_ipv4_t arp_body = {}; + bpf_probe_read_kernel(&arph, sizeof(arph), l3_ptr); + event->arp_op = ntohs(arph.ar_op); + if (arph.ar_hrd == htons(ARPHRD_ETHER) && arph.ar_pro == htons(ETH_P_IP) && + arph.ar_hln == ETH_ALEN && arph.ar_pln == 4) { + bpf_probe_read_kernel(&arp_body, sizeof(arp_body), l3_ptr + sizeof(struct arphdr)); + __builtin_memcpy(event->src_ip, arp_body.spa, 4); + __builtin_memcpy(event->dst_ip, arp_body.tpa, 4); + event->ip_version = 4; + } + return 1; + } + + if (eth_proto == htons(ETH_P_IP)) { + struct iphdr iph = {}; + bpf_probe_read_kernel(&iph, sizeof(iph), l3_ptr); + event->ip_version = 4; + event->protocol_raw = iph.protocol; + event->ip_id = ntohs(iph.id); + bpf_probe_read_kernel(event->src_ip, 4, &iph.saddr); + bpf_probe_read_kernel(event->dst_ip, 4, &iph.daddr); + + if (iph.protocol == IPPROTO_TCP) { + struct tcphdr tcph = {}; + unsigned char flags = 0; + bpf_probe_read_kernel(&tcph, sizeof(tcph), head + transport_header); + event->src_port = ntohs(tcph.source); + event->dst_port = ntohs(tcph.dest); + event->tcp_seq = ntohl(tcph.seq); + event->tcp_ack = ntohl(tcph.ack_seq); + bpf_probe_read_kernel(&flags, sizeof(flags), (void *)(head + transport_header + 13)); + event->tcp_flags = flags; + } else if (iph.protocol == IPPROTO_UDP) { + struct udphdr udph = {}; + bpf_probe_read_kernel(&udph, sizeof(udph), head + transport_header); + event->src_port = ntohs(udph.source); + event->dst_port = ntohs(udph.dest); + } else if (iph.protocol == IPPROTO_ICMP) { + struct icmphdr icmph = {}; + bpf_probe_read_kernel(&icmph, sizeof(icmph), head + transport_header); + event->icmp_type = icmph.type; + event->icmp_code = icmph.code; + } + return 1; + } + + if (eth_proto == htons(ETH_P_IPV6)) { + struct ipv6hdr ip6h = {}; + bpf_probe_read_kernel(&ip6h, sizeof(ip6h), l3_ptr); + 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 = {}; + unsigned char flags = 0; + bpf_probe_read_kernel(&tcph, sizeof(tcph), head + transport_header); + event->src_port = ntohs(tcph.source); + event->dst_port = ntohs(tcph.dest); + event->tcp_seq = ntohl(tcph.seq); + event->tcp_ack = ntohl(tcph.ack_seq); + bpf_probe_read_kernel(&flags, sizeof(flags), (void *)(head + transport_header + 13)); + event->tcp_flags = flags; + } else if (ip6h.nexthdr == IPPROTO_UDP) { + struct udphdr udph = {}; + bpf_probe_read_kernel(&udph, sizeof(udph), head + transport_header); + event->src_port = ntohs(udph.source); + event->dst_port = ntohs(udph.dest); + } else if (ip6h.nexthdr == IPPROTO_ICMPV6) { + struct icmp6hdr icmp6 = {}; + bpf_probe_read_kernel(&icmp6, sizeof(icmp6), head + transport_header); + event->icmp_type = icmp6.icmp6_type; + event->icmp_code = icmp6.icmp6_code; + } + return 1; + } + + return 1; +} + +static __always_inline int emit_event(struct pt_regs *ctx, struct sk_buff *skb, __u8 event_type, __u32 reason) { + struct event_t event = {}; + event.ts_ns = bpf_ktime_get_ns(); + event.event_type = event_type; + event.reason = reason; + + if (!fill_ifname(skb, &event)) { + return 0; + } + if (!parse_skb(skb, &event)) { + return 0; + } + + events.perf_submit(ctx, &event, sizeof(event)); + return 0; +} + +int trace_ingress(struct pt_regs *ctx, struct sk_buff *skb) { + return emit_event(ctx, skb, EVENT_INGRESS, 0); +} + +int trace_egress(struct pt_regs *ctx, struct sk_buff *skb) { + return emit_event(ctx, skb, EVENT_EGRESS, 0); +} + +TRACEPOINT_PROBE(skb, kfree_skb) { + struct sk_buff *skb = (struct sk_buff *)args->skbaddr; + struct event_t event = {}; + 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)) { + return 0; + } + + events.perf_submit(args, &event, sizeof(event)); + return 0; +} +""" + + +class Event(ct.Structure): + _fields_ = [ + ("ts_ns", ct.c_ulonglong), + ("length", ct.c_uint), + ("reason", ct.c_uint), + ("eth_type_raw", ct.c_ushort), + ("vlan_id", ct.c_ushort), + ("src_port", ct.c_ushort), + ("dst_port", ct.c_ushort), + ("ip_id", ct.c_ushort), + ("arp_op", ct.c_ushort), + ("protocol_raw", ct.c_uint), + ("tcp_seq", ct.c_uint), + ("tcp_ack", ct.c_uint), + ("event_type", ct.c_ubyte), + ("ip_version", ct.c_ubyte), + ("icmp_type", ct.c_ubyte), + ("icmp_code", ct.c_ubyte), + ("tcp_flags", ct.c_ubyte), + ("ifname", ct.c_char * 16), + ("src_mac", ct.c_ubyte * 6), + ("dst_mac", ct.c_ubyte * 6), + ("src_ip", ct.c_ubyte * 16), + ("dst_ip", ct.c_ubyte * 16), + ] + + +def _mac_to_str(value: Iterable[int]) -> str: + return ":".join(f"{byte:02x}" for byte in value) + + +def _ip_to_str(ip_version: int, raw: Iterable[int]) -> str | None: + data = bytes(raw) + if ip_version == 4: + try: + return str(ipaddress.IPv4Address(data[:4])) + except ipaddress.AddressValueError: + return None + if ip_version == 6: + try: + return str(ipaddress.IPv6Address(data[:16])) + except ipaddress.AddressValueError: + return None + return None + + +def _build_packet_uid(payload: dict[str, object]) -> str: + normalized = [] + for field in IDENTITY_FIELDS: + value = payload.get(field) + normalized.append("" if value is None else str(value)) + return hashlib.sha1("|".join(normalized).encode("utf-8")).hexdigest() + + +def _event_name(value: int) -> str: + return {1: "ingress", 2: "egress", 3: "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") + if iface not in TARGET_INTERFACES: + return + + payload: dict[str, object] = { + "event_type": _event_name(event.event_type), + "iface": iface, + "length": int(event.length), + "src_mac": _mac_to_str(event.src_mac), + "dst_mac": _mac_to_str(event.dst_mac), + "eth_type_raw": int(event.eth_type_raw) or None, + "vlan_id": int(event.vlan_id) or None, + "src_ip": _ip_to_str(int(event.ip_version), event.src_ip), + "dst_ip": _ip_to_str(int(event.ip_version), event.dst_ip), + "protocol_raw": int(event.protocol_raw) or None, + "src_port": int(event.src_port) or None, + "dst_port": int(event.dst_port) or None, + "ip_id": int(event.ip_id) or None, + "arp_op": int(event.arp_op) or None, + "icmp_type": int(event.icmp_type) or None, + "icmp_code": int(event.icmp_code) or None, + "tcp_seq": int(event.tcp_seq) or None, + "tcp_ack": int(event.tcp_ack) or None, + "tcp_flags": int(event.tcp_flags) 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, + } + payload["packet_uid"] = _build_packet_uid(payload) + print(json.dumps(payload, separators=(",", ":")), flush=True) + + +def _attach_kprobe_first(bpf: BPF, symbols: list[str], fn_name: str) -> str: + for symbol in symbols: + 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 {symbols}") + + +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") + return parser.parse_args() + + +def _sigterm(_signum: int, _frame: object) -> None: + raise KeyboardInterrupt + + +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) + 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_kprobe_first(bpf, ["dev_queue_xmit"], "trace_egress") + + 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) + try: + while True: + bpf.perf_buffer_poll() + except KeyboardInterrupt: + return 0 + + +TARGET_INTERFACES: set[str] = set() + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/src/utilities/packet_identity.py b/backend/src/utilities/packet_identity.py new file mode 100644 index 0000000..1e3f1bc --- /dev/null +++ b/backend/src/utilities/packet_identity.py @@ -0,0 +1,44 @@ +"""Helpers for stable packet identity across AF_PACKET and eBPF events.""" + +from __future__ import annotations + +import hashlib +from typing import Any, Dict + + +IDENTITY_FIELDS = ( + "src_mac", + "dst_mac", + "eth_type_raw", + "vlan_id", + "src_ip", + "dst_ip", + "protocol_raw", + "src_port", + "dst_port", + "length", + "ip_id", + "icmp_type", + "icmp_code", + "arp_op", + "tcp_seq", + "tcp_ack", + "tcp_flags", +) + + +def build_packet_uid(fields: Dict[str, Any]) -> str: + """Build a deterministic packet identifier from selected L2-L4 fields.""" + normalized = [] + for field in IDENTITY_FIELDS: + value = fields.get(field) + if isinstance(value, bytes): + value = value.hex() + normalized.append("" if value is None else str(value)) + digest = hashlib.sha1("|".join(normalized).encode("utf-8")).hexdigest() + return digest + + +def minimal_identity_dict(fields: Dict[str, Any]) -> Dict[str, Any]: + """Return only the identity-relevant subset of packet fields.""" + return {field: fields.get(field) for field in IDENTITY_FIELDS} diff --git a/backend/src/utilities/packet_tracker.py b/backend/src/utilities/packet_tracker.py new file mode 100644 index 0000000..d14bb40 --- /dev/null +++ b/backend/src/utilities/packet_tracker.py @@ -0,0 +1,312 @@ +"""Aggregate AF_PACKET observations and kernel telemetry into one packet record.""" + +from __future__ import annotations + +import asyncio +import logging +import threading +import time +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +import src.shared_objects as shared_objects +from src.utilities.packet_identity import build_packet_uid + +logger = logging.getLogger("packet_tracker") + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc) + + +class PacketTracker: + """Deduplicate packet observations and persist one upserted row per packet.""" + + def __init__( + self, + finalize_delay_seconds: float = 0.25, + retention_seconds: float = 10.0, + min_flush_interval_seconds: float = 0.05, + ): + self._finalize_delay_seconds = finalize_delay_seconds + self._retention_seconds = retention_seconds + self._min_flush_interval_seconds = min_flush_interval_seconds + self._entries: Dict[str, Dict[str, Any]] = {} + self._lock = threading.Lock() + self._stop_event = threading.Event() + self._thread = threading.Thread(target=self._run, daemon=True, name="packet-tracker") + self._thread.start() + + def stop(self) -> None: + self._stop_event.set() + self._thread.join(timeout=2) + + def observe_packet(self, pkt_info: Dict[str, Any]) -> str: + """Merge parsed packet information into a pending packet entry.""" + now_ts = time.time() + packet_uid = pkt_info.get("packet_uid") or build_packet_uid(pkt_info) + pkt_info["packet_uid"] = packet_uid + + with self._lock: + entry = self._entries.get(packet_uid) + if entry is None: + entry = self._new_entry(packet_uid, now_ts) + self._entries[packet_uid] = entry + + self._merge_packet_info(entry, pkt_info, now_ts) + self._maybe_promote_reject_from_reply(pkt_info, now_ts) + self._maybe_mark_complete(entry) + return packet_uid + + def observe_telemetry(self, event: Dict[str, Any]) -> Optional[str]: + """Merge ingress/egress/verdict telemetry into a pending packet entry.""" + packet_uid = event.get("packet_uid") + if not packet_uid: + try: + packet_uid = build_packet_uid(event) + except Exception: + logger.debug("Telemetry event missing packet identity: %s", event) + return None + + now_ts = time.time() + with self._lock: + entry = self._entries.get(packet_uid) + if entry is None: + entry = self._new_entry(packet_uid, now_ts) + self._entries[packet_uid] = entry + + payload = entry["payload"] + payload["packet_uid"] = packet_uid + payload["telemetry_metadata"] = event + payload["last_observed_at"] = now_ts + for key, value in event.items(): + if value is None or key in {"event_type", "reason", "reason_code", "iface", "packet_uid"}: + continue + if payload.get(key) is None: + payload[key] = value + event_type = event.get("event_type") + iface = event.get("iface") + if iface: + observed_ifaces = payload.setdefault("observed_ifaces", []) + if iface not in observed_ifaces: + observed_ifaces.append(iface) + + if event_type == "ingress": + payload["ingress_if"] = iface + payload["iface"] = iface + payload["ingress_seen_at"] = _utcnow() + payload["direction"] = "ingress" + elif event_type == "egress": + payload["egress_if"] = iface + payload["egress_seen_at"] = _utcnow() + payload["direction"] = "forwarded" + payload["verdict"] = "accept" + payload["verdict_reason"] = "egress-observed" + payload["verdict_confidence"] = "high" + payload["verdict_seen_at"] = _utcnow() + elif event_type == "drop": + payload["verdict"] = "drop" + payload["verdict_reason"] = event.get("reason") or "kfree_skb" + payload["verdict_confidence"] = "high" + payload["verdict_seen_at"] = _utcnow() + payload["direction"] = "dropped" + elif event_type == "reject": + payload["verdict"] = "reject" + payload["verdict_reason"] = event.get("reason") or "netfilter-reject" + payload["verdict_confidence"] = event.get("verdict_confidence") or "medium" + payload["verdict_seen_at"] = _utcnow() + payload["direction"] = "rejected" + + entry["last_observed_at"] = now_ts + entry["dirty"] = True + self._maybe_mark_complete(entry) + return packet_uid + + def _new_entry(self, packet_uid: str, now_ts: float) -> Dict[str, Any]: + return { + "packet_uid": packet_uid, + "payload": { + "packet_uid": packet_uid, + "observed_ifaces": [], + "verdict": "pending", + "verdict_reason": None, + "verdict_confidence": None, + "telemetry_metadata": None, + }, + "persisted": False, + "dirty": True, + "finalized": False, + "created_at": now_ts, + "last_observed_at": now_ts, + "last_persisted_at": 0.0, + } + + def _merge_packet_info(self, entry: Dict[str, Any], pkt_info: Dict[str, Any], now_ts: float) -> None: + payload = entry["payload"] + changed = False + for key, value in pkt_info.items(): + if key == "observed_ifaces": + continue + if value is None: + continue + if key == "raw" and payload.get("raw") is not None: + continue + if payload.get(key) == value: + continue + payload[key] = value + changed = True + + iface = pkt_info.get("iface") + if iface: + observed_ifaces = payload.setdefault("observed_ifaces", []) + if iface not in observed_ifaces: + observed_ifaces.append(iface) + changed = True + if not payload.get("ingress_if"): + payload["ingress_if"] = iface + payload["iface"] = iface + payload["ingress_seen_at"] = _utcnow() + changed = True + + payload["last_observed_at"] = now_ts + entry["last_observed_at"] = now_ts + entry["dirty"] = entry["dirty"] or changed + + def _maybe_promote_reject_from_reply(self, pkt_info: Dict[str, Any], now_ts: float) -> None: + reject_reason = None + + tcp_flags = pkt_info.get("tcp_flags") + if pkt_info.get("protocol_raw") == 6 and tcp_flags is not None and int(tcp_flags) & 0x04: + reject_reason = "tcp-rst-observed" + match = self._find_recent_drop( + src_ip=pkt_info.get("dst_ip"), + dst_ip=pkt_info.get("src_ip"), + protocol_raw=6, + src_port=pkt_info.get("dst_port"), + dst_port=pkt_info.get("src_port"), + ) + elif pkt_info.get("protocol_raw") == 1 and pkt_info.get("icmp_type") == 3: + reject_reason = "icmp-unreachable-observed" + match = self._find_recent_drop( + src_ip=pkt_info.get("icmp_embedded_src_ip"), + dst_ip=pkt_info.get("icmp_embedded_dst_ip"), + protocol_raw=pkt_info.get("icmp_embedded_protocol"), + src_port=pkt_info.get("icmp_embedded_src_port"), + dst_port=pkt_info.get("icmp_embedded_dst_port"), + ) + else: + return + + if match is None: + return + + payload = match["payload"] + if payload.get("verdict") != "drop": + return + + payload["verdict"] = "reject" + payload["verdict_reason"] = reject_reason + payload["verdict_confidence"] = "medium" + payload["verdict_seen_at"] = _utcnow() + payload["direction"] = "rejected" + match["last_observed_at"] = now_ts + match["dirty"] = True + match["finalized"] = True + + def _find_recent_drop( + self, + src_ip: Any, + dst_ip: Any, + protocol_raw: Any, + src_port: Any, + dst_port: Any, + ) -> Optional[Dict[str, Any]]: + if not src_ip or not dst_ip or protocol_raw is None: + return None + + cutoff = time.time() - 1.0 + for entry in self._entries.values(): + payload = entry["payload"] + if entry["last_observed_at"] < cutoff: + continue + if payload.get("verdict") != "drop": + continue + if payload.get("src_ip") != src_ip or payload.get("dst_ip") != dst_ip: + continue + if payload.get("protocol_raw") != protocol_raw: + continue + if payload.get("src_port") != src_port or payload.get("dst_port") != dst_port: + continue + return entry + return None + + def _maybe_mark_complete(self, entry: Dict[str, Any]) -> None: + payload = entry["payload"] + if payload.get("verdict") in {"accept", "drop", "reject"}: + entry["finalized"] = True + + def _run(self) -> None: + while not self._stop_event.is_set(): + time.sleep(0.05) + due_entries: List[Dict[str, Any]] = [] + expired_uids: List[str] = [] + now_ts = time.time() + + with self._lock: + for packet_uid, entry in list(self._entries.items()): + age = now_ts - entry["last_observed_at"] + if not entry["finalized"] and age >= self._finalize_delay_seconds: + entry["finalized"] = True + if entry["payload"].get("verdict") == "pending": + entry["payload"]["verdict"] = "unknown" + entry["payload"]["verdict_reason"] = "timeout" + entry["payload"]["verdict_confidence"] = "low" + entry["payload"]["verdict_seen_at"] = _utcnow() + entry["payload"]["direction"] = "observed" + entry["dirty"] = True + + should_flush = entry["dirty"] and ( + not entry["persisted"] + or entry["finalized"] + or (now_ts - entry["last_persisted_at"]) >= self._min_flush_interval_seconds + ) + if should_flush: + due_entries.append( + { + "packet_uid": entry["packet_uid"], + "payload": dict(entry["payload"]), + } + ) + elif entry["persisted"] and age >= self._retention_seconds: + expired_uids.append(packet_uid) + + for packet_uid in expired_uids: + self._entries.pop(packet_uid, None) + + for entry in due_entries: + self._persist(entry) + + def _persist(self, entry: Dict[str, Any]) -> None: + payload = dict(entry["payload"]) + if payload.get("observed_ifaces") == []: + payload["observed_ifaces"] = None + + web_loop = getattr(shared_objects, "web_loop", None) + web_db = getattr(shared_objects, "db", None) + if web_loop is None or web_db is None: + return + + try: + fut = asyncio.run_coroutine_threadsafe(web_db.upsert_packet(payload), web_loop) + fut.result(timeout=2.0) + with self._lock: + current = self._entries.get(entry["packet_uid"]) + if current is not None: + current["persisted"] = True + current["dirty"] = False + current["last_persisted_at"] = time.time() + except Exception: + logger.exception("Failed to persist packet %s", entry["packet_uid"]) + + +packet_tracker = PacketTracker() diff --git a/frontend/src/components/PacketViewer.tsx b/frontend/src/components/PacketViewer.tsx index b671793..14678cf 100644 --- a/frontend/src/components/PacketViewer.tsx +++ b/frontend/src/components/PacketViewer.tsx @@ -84,8 +84,12 @@ function colorForName(name: string) { return result; } -/** Parse iface field into an array of strings */ -function parseIfaces(ifaceField?: string | string[]) { +function packetKey(packet: PacketRow) { + return String(packet.packet_uid ?? packet.id ?? `${packet.iface ?? 'if'}:${packet.timestamp ?? ''}`); +} + +function parseIfaces(ifaceField?: string | string[] | null, observedIfaces?: string[] | null) { + if (Array.isArray(observedIfaces) && observedIfaces.length > 0) return observedIfaces; if (!ifaceField) return []; if (Array.isArray(ifaceField)) return ifaceField; return String(ifaceField) @@ -131,14 +135,32 @@ export default function PacketViewer(): ReactElement { const [selectedPacket, setSelectedPacket] = useState(null); const queuedDuringPause = useRef([]); - // helper: prepend new packets (newest first). Keep list capped. const pushNew = useCallback((newPkts: PacketRow[] | PacketRow) => { setPackets((prev) => { const arr = Array.isArray(newPkts) ? newPkts : [newPkts]; - const normalized = arr.map((p) => p); - const merged = [...normalized, ...prev]; - if (merged.length > MAX_PACKETS) return merged.slice(0, MAX_PACKETS); - return merged; + const next = [...prev]; + + for (const packet of arr) { + const key = packetKey(packet); + const index = next.findIndex((item) => packetKey(item) === key); + if (index >= 0) { + next[index] = { ...next[index], ...packet }; + } else { + next.unshift(packet); + } + } + + next.sort((left, right) => { + const leftId = Number(left.id ?? 0); + const rightId = Number(right.id ?? 0); + if (!Number.isNaN(leftId) && !Number.isNaN(rightId) && leftId !== rightId) { + return rightId - leftId; + } + return String(right.timestamp ?? '').localeCompare(String(left.timestamp ?? '')); + }); + + if (next.length > MAX_PACKETS) return next.slice(0, MAX_PACKETS); + return next; }); }, []); @@ -284,7 +306,7 @@ export default function PacketViewer(): ReactElement { key: 'iface', width: 100, render: (_: any, rec: PacketRow) => { - const ifaces = parseIfaces(rec.iface); + const ifaces = parseIfaces(rec.iface, rec.observed_ifaces); if (ifaces.length === 0) return -; return ( @@ -301,10 +323,38 @@ export default function PacketViewer(): ReactElement { ); }, }, + { + title: 'Path', + key: 'path', + width: 150, + render: (_: any, rec: PacketRow) => ( + + {rec.ingress_if ?? '-'} + {rec.egress_if ?? '-'} + + ), + }, + { + title: 'Verdict', + key: 'verdict', + width: 120, + render: (_: any, rec: PacketRow) => { + const verdict = rec.verdict ?? '-'; + const confidence = rec.verdict_confidence ? ` (${rec.verdict_confidence})` : ''; + return ( + + {verdict} + + {(rec.verdict_reason ?? '') + confidence} + + + ); + }, + }, { title: 'MAC (src → dst)', key: 'macs', - width: 100, + width: 180, render: (_: any, rec: PacketRow) => (
@@ -445,7 +495,14 @@ export default function PacketViewer(): ReactElement { handlePauseToggle(!checked ? true : false)} />
) : ( - String(rec.id ?? `${rec.iface ?? 'if'}:${rec.timestamp ?? ''}:${Math.random()}`) - } + rowKey={(rec: PacketRow) => packetKey(rec)} dataSource={packets} columns={columns as any} pagination={{ pageSize: 25 }} diff --git a/frontend/src/types/packets.ts b/frontend/src/types/packets.ts index 8ac5b78..d8e9fed 100644 --- a/frontend/src/types/packets.ts +++ b/frontend/src/types/packets.ts @@ -1,7 +1,11 @@ export interface PacketRow { id?: number | string; timestamp?: string; + packet_uid?: string; iface?: string | string[]; + ingress_if?: string | null; + egress_if?: string | null; + observed_ifaces?: string[] | null; src_mac?: string | null; dst_mac?: string | null; eth_type?: string | number | null; @@ -21,6 +25,14 @@ export interface PacketRow { app_is_encrypted?: boolean | null; app_risk_score?: number | null; dpi_metadata?: Record | null; + telemetry_metadata?: Record | null; + direction?: string | null; + verdict?: string | null; + verdict_reason?: string | null; + verdict_confidence?: string | null; + ingress_seen_at?: string | null; + egress_seen_at?: string | null; + verdict_seen_at?: string | null; [key: string]: unknown; } diff --git a/setup_build_server.sh b/setup_build_server.sh index 1c48f5e..58bdc64 100755 --- a/setup_build_server.sh +++ b/setup_build_server.sh @@ -25,8 +25,9 @@ PYTHON_VERSION="3" # aktuelle Python 3 Version echo "==> Update & Upgrade" apt update && apt upgrade -y apt install -y git curl build-essential nginx python3 python3-pip python3-venv unzip wget python3-dev \ - libnetfilter-queue-dev libnfnetlink-dev libpcap-dev autoconf automake libtool pkg-config libjson-c-dev \ - gettext flex bison libnuma-dev libpcre2-dev libmaxminddb-dev librrd-dev + libpcap-dev autoconf automake libtool pkg-config libjson-c-dev gettext flex bison libnuma-dev \ + libpcre2-dev libmaxminddb-dev librrd-dev python3-bpfcc bpfcc-tools linux-headers-generic +apt install -y linux-headers-$(uname -r) || apt install -y linux-headers-generic # ----------------------------- # Node.js installieren (LTS) @@ -64,81 +65,6 @@ python3 -m venv venv source venv/bin/activate pip install --upgrade pip -# ----------------------------- -# nDPI installieren (system lib + python bindings) -# ----------------------------- -NDPIDIR="/tmp/nDPI" - -patch_ndpi_pack_on_if_needed() { - local py_dir="$1" - if [ -f "$py_dir/pyproject.toml" ] && grep -q "PACK_ON" "$py_dir/pyproject.toml"; then - sed -i 's/PACK_ON/"0.0.0"/g' "$py_dir/pyproject.toml" - echo "==> Applied PACK_ON workaround in pyproject.toml" - fi - if [ -f "$py_dir/setup.cfg" ] && grep -q "PACK_ON" "$py_dir/setup.cfg"; then - sed -i 's/PACK_ON/0.0.0/g' "$py_dir/setup.cfg" - echo "==> Applied PACK_ON workaround in setup.cfg" - fi - if [ -f "$py_dir/setup.py" ] && grep -q "PACK_ON" "$py_dir/setup.py"; then - sed -i "s/PACK_ON/'0.0.0'/g" "$py_dir/setup.py" - echo "==> Applied PACK_ON workaround in setup.py" - fi -} - -if ! ldconfig -p | grep -q "libndpi"; then - echo "==> Build and install nDPI library" - rm -rf "$NDPIDIR" - git clone --depth 1 --branch dev https://github.com/ntop/nDPI.git "$NDPIDIR" - cd "$NDPIDIR" - ./autogen.sh - ./configure - make -j"$(nproc)" - make install - ldconfig -else - echo "==> nDPI library already installed" - # Keep python bindings source in sync with official instructions. - if [ ! -d "$NDPIDIR/python" ]; then - rm -rf "$NDPIDIR" - git clone --depth 1 --branch dev https://github.com/ntop/nDPI.git "$NDPIDIR" - fi -fi - -if [ -d "$NDPIDIR/python" ]; then - echo "==> Install nDPI Python bindings" - cd "$NDPIDIR/python" - python3 -m pip install --upgrade pip - if ! python3 -m pip install -r dev_requirements.txt; then - echo "==> WARNING: failed to install nDPI python dev_requirements" - fi - if ! python3 -m pip install .; then - echo "==> Initial nDPI pip install failed; retrying with PACK_ON workaround" - patch_ndpi_pack_on_if_needed "$NDPIDIR/python" - if ! python3 -m pip install .; then - echo "==> WARNING: nDPI Python bindings install failed, backend will run without DPI enrichment" - fi - fi - if python - <<'PY' -import importlib -mods = [] -for name in ("ndpi", "pyndpi", "nDPI"): - try: - m = importlib.import_module(name) - mods.append((name, getattr(m, "__file__", "unknown"))) - except Exception: - pass -if not mods: - raise SystemExit(1) -print("nDPI Python modules:", mods) -PY - then - echo "==> nDPI Python import check passed" - else - echo "==> WARNING: nDPI installed but import failed in backend venv" - fi - cd "$BACKEND_DIR" -fi - cd "$BACKEND_DIR" pip install -r requirements.txt deactivate @@ -266,19 +192,6 @@ FRONTEND_DIR="$APP_DIR/frontend" BACKEND_DIR="$APP_DIR/backend" BACKEND_SERVICE="mitm-backend" -patch_ndpi_pack_on_if_needed() { - local py_dir="$1" - if [ -f "$py_dir/pyproject.toml" ] && grep -q "PACK_ON" "$py_dir/pyproject.toml"; then - sed -i 's/PACK_ON/"0.0.0"/g' "$py_dir/pyproject.toml" - fi - if [ -f "$py_dir/setup.cfg" ] && grep -q "PACK_ON" "$py_dir/setup.cfg"; then - sed -i 's/PACK_ON/0.0.0/g' "$py_dir/setup.cfg" - fi - if [ -f "$py_dir/setup.py" ] && grep -q "PACK_ON" "$py_dir/setup.py"; then - sed -i "s/PACK_ON/'0.0.0'/g" "$py_dir/setup.py" - fi -} - cd $APP_DIR git reset --hard git pull @@ -291,35 +204,6 @@ npm run build # Backend Dependencies cd $BACKEND_DIR source venv/bin/activate -NDPIDIR="/tmp/nDPI" -if [ ! -d "$NDPIDIR/python" ]; then - rm -rf "$NDPIDIR" - git clone --depth 1 https://github.com/ntop/nDPI.git "$NDPIDIR" -fi -if [ -d "/tmp/nDPI/python" ]; then - cd "$NDPIDIR/python" - python3 -m pip install --upgrade pip - python3 -m pip install -r dev_requirements.txt || true - if ! python3 -m pip install .; then - echo "WARNING: Initial nDPI install failed during runner build; retry with PACK_ON workaround" - patch_ndpi_pack_on_if_needed "$NDPIDIR/python" - if ! python3 -m pip install .; then - echo "WARNING: nDPI Python bindings install failed during runner build" - fi - fi - python - <<'PY' || echo "WARNING: nDPI import check failed during runner build" -import importlib -for name in ("ndpi", "pyndpi", "nDPI"): - try: - m = importlib.import_module(name) - print("nDPI import OK:", name, getattr(m, "__file__", "unknown")) - raise SystemExit(0) - except Exception: - pass -raise SystemExit(1) -PY - cd $BACKEND_DIR -fi pip install -r requirements.txt deactivate diff --git a/setup_database.sh b/setup_database.sh index 7f3b889..f4227d8 100755 --- a/setup_database.sh +++ b/setup_database.sh @@ -31,29 +31,33 @@ echo "[4] Restarting PostgreSQL…" sudo systemctl restart postgresql echo "[5] Creating database + user (idempotent)…" -sudo -u postgres psql <