diff --git a/backend/.env.example b/backend/.env.example
index 629b798..71b295d 100644
--- a/backend/.env.example
+++ b/backend/.env.example
@@ -15,6 +15,8 @@ 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_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 5585092..0712271 100644
--- a/backend/src/Models/packets.py
+++ b/backend/src/Models/packets.py
@@ -11,9 +11,15 @@ class PacketDBModel(BaseModel):
id: Union[int, str]
timestamp: datetime = Field(..., description="Packet timestamp in ISO format.")
- packet_uid: str = Field(..., description="Stable packet identity used for upserts/correlation.")
+ updated_at: Optional[datetime] = Field(None, description="Last DB update time for this row.")
+ correlation_key: str = Field(..., description="Primary upsert key for this packet row.")
+ correlation_source: Optional[str] = Field(None, description="How the correlation key was derived.")
+ packet_id: Optional[str] = Field(None, description="Kernel-side packet identifier derived from skb mark.")
+ packet_uid: Optional[str] = Field(None, description="Legacy hash-based packet identity fallback.")
+ skb_mark: Optional[int] = Field(None, description="Raw skb mark observed in telemetry or capture header.")
ingress_if: Optional[str] = None
egress_if: Optional[str] = None
+ capture_iface: Optional[str] = None
src_mac: Optional[str] = None
dst_mac: Optional[str] = None
eth_type_raw: Optional[int] = Field(None, description="Numeric Ethernet type from the frame header.")
@@ -37,6 +43,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.")
telemetry_metadata: Optional[dict] = Field(None, description="Kernel telemetry details from eBPF collector.")
verdict: Optional[str] = None
verdict_reason: Optional[str] = None
@@ -51,9 +58,15 @@ class PacketDBModel(BaseModel):
"example": {
"id": 123,
"timestamp": "2026-03-05T12:34:56.789Z",
+ "updated_at": "2026-03-05T12:34:56.900Z",
+ "correlation_key": "pid:123456",
+ "correlation_source": "kernel_mark",
+ "packet_id": "123456",
"packet_uid": "9f6d3af0d3c81cb20ee8e7d32df7c56414460542",
+ "skb_mark": 123456,
"ingress_if": "eth0",
"egress_if": "eth1",
+ "capture_iface": "mitmcap0",
"src_mac": "aa:bb:cc:dd:ee:ff",
"dst_mac": "11:22:33:44:55:66",
"eth_type_raw": 2048,
@@ -77,6 +90,7 @@ 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"},
"verdict": "accept",
"verdict_reason": "egress-observed",
diff --git a/backend/src/config.py b/backend/src/config.py
index 36db3c7..81f8611 100644
--- a/backend/src/config.py
+++ b/backend/src/config.py
@@ -25,6 +25,13 @@ 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
@@ -44,6 +51,8 @@ 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
telemetry_process_stop_timeout_seconds: float
telemetry_reader_join_timeout_seconds: float
ndpi_max_flows: int
@@ -73,6 +82,8 @@ 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),
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/network_sniffer.py b/backend/src/network_sniffer.py
index 8f73125..8da240c 100644
--- a/backend/src/network_sniffer.py
+++ b/backend/src/network_sniffer.py
@@ -33,6 +33,7 @@ from src.utilities.interface_bridge_helpers import (
)
from src.config import settings
from src.utilities.bridge_telemetry import bridge_telemetry_manager
+from src.utilities.capture_header import parse_capture_header
from src.utilities.ndpi_classifier import ndpi_classifier
from src.utilities.packet_identity import build_packet_uid
from src.utilities.packet_tracker import packet_tracker
@@ -61,8 +62,13 @@ sessions: Dict[str, Dict[str, Any]] = {}
class PacketInfo(TypedDict, total=False):
"""TypedDict for parsed packet data used by persistence and telemetry."""
+ correlation_key: str
+ correlation_source: str
+ packet_id: Optional[str]
packet_uid: str
+ skb_mark: Optional[int]
iface: str
+ capture_iface: Optional[str]
length: int
raw: bytes
src_mac: Optional[str]
@@ -85,6 +91,7 @@ class PacketInfo(TypedDict, total=False):
app_is_encrypted: Optional[bool]
app_risk_score: Optional[int]
dpi_metadata: Optional[Dict[str, Any]]
+ capture_metadata: Optional[Dict[str, Any]]
ip_id: Optional[int]
icmp_type: Optional[int]
icmp_code: Optional[int]
@@ -155,7 +162,7 @@ def _safe_get_attr(layer, attr: str):
return None
-def parse_packet(pkt, bridge_label: str) -> None:
+def parse_packet(pkt, bridge_label: str, capture_metadata: Optional[Dict[str, Any]] = None) -> None:
"""
Parse a scapy Packet object into a normalized PacketInfo and schedule DB insert.
bridge_label indicates whether the packet was captured as part of a bridge-snapshot or single-interface.
@@ -168,8 +175,11 @@ def parse_packet(pkt, bridge_label: str) -> None:
pkt_info: PacketInfo = {
"iface": pkt_iface,
+ "capture_iface": pkt_iface if capture_metadata else None,
"length": len(pkt),
"raw": bytes(pkt),
+ "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,
"dst_mac": None,
"eth_type_raw": None,
@@ -190,6 +200,7 @@ def parse_packet(pkt, bridge_label: str) -> None:
"app_is_encrypted": None,
"app_risk_score": None,
"dpi_metadata": None,
+ "capture_metadata": capture_metadata,
"ip_id": None,
"icmp_type": None,
"icmp_code": None,
@@ -369,7 +380,13 @@ def parse_packet(pkt, bridge_label: str) -> None:
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)
+ if pkt_info.get("packet_id"):
+ pkt_info["correlation_key"] = f"pid:{pkt_info['packet_id']}"
+ pkt_info["correlation_source"] = "kernel_mark"
+ else:
+ pkt_info["packet_uid"] = build_packet_uid(pkt_info)
+ pkt_info["correlation_key"] = f"uid:{pkt_info['packet_uid']}"
+ pkt_info["correlation_source"] = "legacy_hash"
try:
web_loop = getattr(shared_objects, "web_loop", None)
@@ -548,9 +565,13 @@ def _session_reader_loop(session_id: str) -> None:
# parse with scapy
try:
- pkt = Ether(raw)
+ capture_meta = None
+ packet_bytes = raw
+ if settings.capture_header_enabled and settings.capture_interface and iface == settings.capture_interface:
+ capture_meta, packet_bytes = parse_capture_header(raw)
+ pkt = Ether(packet_bytes)
pkt.sniffed_on = iface
- parse_packet(pkt, label)
+ parse_packet(pkt, label, capture_metadata=capture_meta)
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)
@@ -595,19 +616,23 @@ def start_afpacket_sniffer(target: str, target_is_interface: bool = False) -> st
"label": target,
"is_bridge": not target_is_interface,
"ports": [],
+ "capture_ifaces": [],
}
sessions[session_id] = session
# determine ports for this session
if target_is_interface:
ports = [target]
+ capture_ifaces = [target]
else:
ports = get_bridge_ports_once(target)
+ capture_ifaces = [settings.capture_interface] if settings.capture_interface else list(ports)
session["ports"] = ports
+ session["capture_ifaces"] = capture_ifaces
# create sockets for this session only
- for iface in ports:
+ for iface in capture_ifaces:
if not check_interface_exists(iface):
logger.warning("Snapshot port %s missing for session %s, skipping", iface, session_id)
continue
@@ -620,7 +645,13 @@ def start_afpacket_sniffer(target: str, target_is_interface: bool = False) -> st
session["thread"] = t
t.start()
_sync_bridge_telemetry()
- logger.info("Started sniffer session %s label=%s ports=%s", session_id, target, ports)
+ logger.info(
+ "Started sniffer session %s label=%s ports=%s capture_ifaces=%s",
+ session_id,
+ target,
+ ports,
+ capture_ifaces,
+ )
return session_id
@@ -718,6 +749,7 @@ def get_internal_debug_state() -> dict:
"label": s.get("label"),
"is_bridge": s.get("is_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()),
}
diff --git a/backend/src/utilities/capture_header.py b/backend/src/utilities/capture_header.py
new file mode 100644
index 0000000..5559b74
--- /dev/null
+++ b/backend/src/utilities/capture_header.py
@@ -0,0 +1,37 @@
+"""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/database.py b/backend/src/utilities/database.py
index 1025ba3..ce34505 100644
--- a/backend/src/utilities/database.py
+++ b/backend/src/utilities/database.py
@@ -42,7 +42,7 @@ def _serialize_row_for_broadcast(row: Dict[str, Any]) -> Dict[str, Any]:
def _normalize_json_fields(payload: Dict[str, Any]) -> None:
- for key in ("dpi_metadata", "telemetry_metadata"):
+ for key in ("dpi_metadata", "capture_metadata", "telemetry_metadata"):
value = payload.get(key)
if isinstance(value, str):
try:
@@ -118,7 +118,12 @@ class DatabasePool:
row = await conn.fetchrow(
"""
INSERT INTO packets (
+ correlation_key,
+ packet_id,
packet_uid,
+ correlation_source,
+ skb_mark,
+ capture_iface,
ingress_if,
egress_if,
verdict,
@@ -149,14 +154,21 @@ class DatabasePool:
app_is_encrypted,
app_risk_score,
dpi_metadata,
+ capture_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,$31::jsonb,$32::jsonb,$33
+ $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,$31,$32,$33,$34,$35,$36::jsonb,
+ $37::jsonb,$38::jsonb,$39
)
- ON CONFLICT (packet_uid) DO UPDATE SET
+ ON CONFLICT (correlation_key) DO UPDATE SET
+ updated_at = NOW(),
+ packet_id = COALESCE(EXCLUDED.packet_id, packets.packet_id),
+ packet_uid = COALESCE(EXCLUDED.packet_uid, packets.packet_uid),
+ correlation_source = COALESCE(EXCLUDED.correlation_source, packets.correlation_source),
+ skb_mark = COALESCE(EXCLUDED.skb_mark, packets.skb_mark),
+ capture_iface = COALESCE(EXCLUDED.capture_iface, packets.capture_iface),
ingress_if = COALESCE(EXCLUDED.ingress_if, packets.ingress_if),
egress_if = COALESCE(EXCLUDED.egress_if, packets.egress_if),
verdict = COALESCE(EXCLUDED.verdict, packets.verdict),
@@ -195,11 +207,17 @@ class DatabasePool:
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),
+ capture_metadata = COALESCE(EXCLUDED.capture_metadata, packets.capture_metadata),
telemetry_metadata = COALESCE(EXCLUDED.telemetry_metadata, packets.telemetry_metadata),
raw = COALESCE(EXCLUDED.raw, packets.raw)
RETURNING *
""",
- pkt_info["packet_uid"],
+ pkt_info["correlation_key"],
+ pkt_info.get("packet_id"),
+ pkt_info.get("packet_uid"),
+ pkt_info.get("correlation_source"),
+ pkt_info.get("skb_mark"),
+ pkt_info.get("capture_iface"),
pkt_info.get("ingress_if"),
pkt_info.get("egress_if"),
pkt_info.get("verdict"),
@@ -230,6 +248,7 @@ class DatabasePool:
pkt_info.get("app_is_encrypted"),
pkt_info.get("app_risk_score"),
json.dumps(dpi_metadata) if dpi_metadata is not None else None,
+ json.dumps(pkt_info.get("capture_metadata")) if pkt_info.get("capture_metadata") is not None else None,
json.dumps(telemetry_metadata) if telemetry_metadata is not None else None,
pkt_info.get("raw"),
)
@@ -257,7 +276,7 @@ class DatabasePool:
"""
SELECT *
FROM packets
- ORDER BY id DESC
+ ORDER BY updated_at DESC, id DESC
LIMIT $1
""",
limit,
diff --git a/backend/src/utilities/ebpf_bridge_events.py b/backend/src/utilities/ebpf_bridge_events.py
index 5305618..ce95bb7 100644
--- a/backend/src/utilities/ebpf_bridge_events.py
+++ b/backend/src/utilities/ebpf_bridge_events.py
@@ -20,6 +20,8 @@ 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
+from src.utilities.packet_mark import packet_id_from_mark, verdict_from_mark
+
IDENTITY_FIELDS = (
"src_mac",
@@ -75,6 +77,7 @@ struct arp_eth_ipv4_t {
struct event_t {
__u64 ts_ns;
+ __u32 skb_mark;
__u32 length;
__u32 reason;
__u16 eth_type_raw;
@@ -130,6 +133,7 @@ static __always_inline int parse_skb(struct sk_buff *skb, struct event_t *event)
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);
+ bpf_probe_read_kernel(&event->skb_mark, sizeof(event->skb_mark), &skb->mark);
if (!head) {
return 0;
@@ -302,6 +306,7 @@ TRACEPOINT_PROBE(net, net_dev_queue) {
class Event(ct.Structure):
_fields_ = [
("ts_ns", ct.c_ulonglong),
+ ("skb_mark", ct.c_uint),
("length", ct.c_uint),
("reason", ct.c_uint),
("eth_type_raw", ct.c_ushort),
@@ -371,6 +376,7 @@ def _emit_event(cpu: int, data: int, size: int) -> None:
payload: dict[str, object] = {
"event_type": _event_name(event.event_type),
"iface": iface,
+ "skb_mark": int(event.skb_mark) or None,
"length": int(event.length),
"src_mac": _mac_to_str(event.src_mac),
"dst_mac": _mac_to_str(event.dst_mac),
@@ -391,7 +397,18 @@ def _emit_event(cpu: int, data: int, size: int) -> 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)
+ packet_id = packet_id_from_mark(payload.get("skb_mark"))
+ if packet_id:
+ payload["packet_id"] = packet_id
+ payload["correlation_key"] = f"pid:{packet_id}"
+ payload["correlation_source"] = "kernel_mark"
+ verdict_hint = verdict_from_mark(payload.get("skb_mark"))
+ if verdict_hint:
+ payload["verdict_hint"] = verdict_hint
+ else:
+ payload["packet_uid"] = _build_packet_uid(payload)
+ payload["correlation_key"] = f"uid:{payload['packet_uid']}"
+ payload["correlation_source"] = "legacy_hash"
print(json.dumps(payload, separators=(",", ":")), flush=True)
diff --git a/backend/src/utilities/packet_mark.py b/backend/src/utilities/packet_mark.py
new file mode 100644
index 0000000..dc858cc
--- /dev/null
+++ b/backend/src/utilities/packet_mark.py
@@ -0,0 +1,46 @@
+"""Helpers for the shared skb mark layout used for packet correlation and verdict hints."""
+
+from __future__ import annotations
+
+from typing import Optional
+
+
+PACKET_ID_MASK = 0x0FFFFFFF
+VERDICT_MASK = 0xF0000000
+VERDICT_FLAG_DROP = 0x10000000
+VERDICT_FLAG_REJECT = 0x20000000
+
+
+def normalize_skb_mark(value: object) -> Optional[int]:
+ """Return a positive integer skb mark or `None` when the value is empty."""
+ if value in (None, "", 0, "0"):
+ return None
+ try:
+ mark = int(value)
+ except (TypeError, ValueError):
+ return None
+ if mark < 0:
+ mark &= 0xFFFFFFFF
+ return mark or None
+
+
+def packet_id_from_mark(value: object) -> Optional[str]:
+ """Extract the packet correlation identifier from the shared skb mark layout."""
+ mark = normalize_skb_mark(value)
+ if mark is None:
+ return None
+ packet_id = mark & PACKET_ID_MASK
+ return str(packet_id) if packet_id else None
+
+
+def verdict_from_mark(value: object) -> Optional[str]:
+ """Return a verdict hint encoded into the upper mark bits, if any."""
+ mark = normalize_skb_mark(value)
+ if mark is None:
+ return None
+ verdict_bits = mark & VERDICT_MASK
+ if verdict_bits & VERDICT_FLAG_REJECT:
+ return "reject"
+ if verdict_bits & VERDICT_FLAG_DROP:
+ return "drop"
+ return None
diff --git a/backend/src/utilities/packet_tracker.py b/backend/src/utilities/packet_tracker.py
index e719247..41e328f 100644
--- a/backend/src/utilities/packet_tracker.py
+++ b/backend/src/utilities/packet_tracker.py
@@ -14,6 +14,7 @@ from src.Models.etherType import EtherTypeEnum, ethertype_from_int
from src.Models.ip_protocol import protocol_from_number
from src.config import settings
from src.utilities.packet_identity import build_packet_uid
+from src.utilities.packet_mark import packet_id_from_mark, verdict_from_mark
logger = logging.getLogger("packet_tracker")
@@ -42,6 +43,8 @@ class PacketTracker:
"persisted_merged": 0,
"persisted_with_raw": 0,
"persisted_without_raw": 0,
+ "persisted_kernel_mark": 0,
+ "persisted_legacy_hash": 0,
}
self._lock = threading.Lock()
self._stop_event = threading.Event()
@@ -55,46 +58,46 @@ class PacketTracker:
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
+ correlation_key = self._ensure_correlation(pkt_info)
pkt_info["raw_present"] = pkt_info.get("raw") is not None
pkt_info["capture_sources"] = ["af_packet"]
with self._lock:
- entry = self._entries.get(packet_uid)
+ entry = self._entries.get(correlation_key)
if entry is None:
- entry = self._new_entry(packet_uid, now_ts)
- self._entries[packet_uid] = entry
+ entry = self._new_entry(correlation_key, now_ts)
+ self._entries[correlation_key] = 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
+ return correlation_key
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
+ correlation_key = self._ensure_correlation(event)
+ if not correlation_key:
+ logger.debug("Telemetry event missing packet identity: %s", event)
+ return None
now_ts = time.time()
with self._lock:
- entry = self._entries.get(packet_uid)
+ entry = self._entries.get(correlation_key)
if entry is None:
- entry = self._new_entry(packet_uid, now_ts)
- self._entries[packet_uid] = entry
+ entry = self._new_entry(correlation_key, now_ts)
+ self._entries[correlation_key] = entry
payload = entry["payload"]
- payload["packet_uid"] = packet_uid
+ payload["correlation_key"] = correlation_key
+ payload["packet_id"] = event.get("packet_id") or payload.get("packet_id")
+ payload["packet_uid"] = event.get("packet_uid") or payload.get("packet_uid")
+ payload["correlation_source"] = event.get("correlation_source") or payload.get("correlation_source")
+ payload["skb_mark"] = event.get("skb_mark") or payload.get("skb_mark")
payload["telemetry_metadata"] = event
payload["last_observed_at"] = now_ts
self._add_capture_source(payload, "telemetry")
for key, value in event.items():
- if value is None or key in {"event_type", "reason", "reason_code", "iface", "packet_uid"}:
+ if value is None or key in {"event_type", "reason", "reason_code", "iface", "packet_uid", "correlation_key"}:
continue
if payload.get(key) is None:
payload[key] = value
@@ -112,6 +115,7 @@ class PacketTracker:
payload["protocol"] = int(payload["protocol_raw"])
event_type = event.get("event_type")
iface = event.get("iface")
+ verdict_hint = event.get("verdict_hint")
if event_type == "ingress":
payload["ingress_if"] = iface
@@ -124,11 +128,11 @@ class PacketTracker:
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"] = verdict_hint or "drop"
+ payload["verdict_reason"] = event.get("reason") or ("mark-verdict" if verdict_hint else "kfree_skb")
payload["verdict_confidence"] = "high"
payload["verdict_seen_at"] = _utcnow()
- elif event_type == "reject":
+ elif event_type == "reject" or verdict_hint == "reject":
payload["verdict"] = "reject"
payload["verdict_reason"] = event.get("reason") or "netfilter-reject"
payload["verdict_confidence"] = event.get("verdict_confidence") or "medium"
@@ -137,18 +141,23 @@ class PacketTracker:
entry["last_observed_at"] = now_ts
entry["dirty"] = True
self._maybe_mark_complete(entry)
- return packet_uid
+ return correlation_key
- def _new_entry(self, packet_uid: str, now_ts: float) -> Dict[str, Any]:
+ def _new_entry(self, correlation_key: str, now_ts: float) -> Dict[str, Any]:
return {
- "packet_uid": packet_uid,
+ "correlation_key": correlation_key,
"payload": {
- "packet_uid": packet_uid,
+ "correlation_key": correlation_key,
+ "correlation_source": None,
+ "packet_id": None,
+ "packet_uid": None,
+ "skb_mark": None,
"verdict": "pending",
"verdict_reason": None,
"verdict_confidence": None,
"raw_present": False,
"capture_sources": [],
+ "capture_metadata": None,
"telemetry_metadata": None,
},
"persisted": False,
@@ -160,6 +169,34 @@ class PacketTracker:
"last_persisted_at": 0.0,
}
+ def _ensure_correlation(self, payload: Dict[str, Any]) -> Optional[str]:
+ skb_mark = payload.get("skb_mark")
+ if payload.get("packet_id") is None and skb_mark is not None:
+ payload["packet_id"] = packet_id_from_mark(skb_mark)
+ if payload.get("verdict_hint") is None and skb_mark is not None:
+ payload["verdict_hint"] = verdict_from_mark(skb_mark)
+
+ packet_id = payload.get("packet_id")
+ if packet_id not in (None, ""):
+ packet_id = str(packet_id)
+ payload["packet_id"] = packet_id
+ payload["correlation_key"] = f"pid:{packet_id}"
+ if not payload.get("correlation_source"):
+ payload["correlation_source"] = "kernel_mark"
+ return payload["correlation_key"]
+
+ packet_uid = payload.get("packet_uid")
+ if packet_uid in (None, ""):
+ try:
+ packet_uid = build_packet_uid(payload)
+ except Exception:
+ return None
+ payload["packet_uid"] = packet_uid
+ payload["correlation_key"] = f"uid:{packet_uid}"
+ if not payload.get("correlation_source"):
+ payload["correlation_source"] = "legacy_hash"
+ return payload["correlation_key"]
+
def _add_capture_source(self, payload: Dict[str, Any], source: str) -> None:
capture_sources = payload.setdefault("capture_sources", [])
if source not in capture_sources:
@@ -182,7 +219,14 @@ class PacketTracker:
changed = True
iface = pkt_info.get("iface")
- if iface and not payload.get("ingress_if"):
+ if iface and pkt_info.get("capture_iface") and not payload.get("capture_iface"):
+ payload["capture_iface"] = pkt_info.get("capture_iface")
+ changed = True
+ elif iface and pkt_info.get("capture_metadata") and not payload.get("capture_iface"):
+ payload["capture_iface"] = iface
+ changed = True
+
+ if iface and not pkt_info.get("capture_metadata") and not payload.get("ingress_if"):
payload["ingress_if"] = iface
payload["ingress_seen_at"] = _utcnow()
changed = True
@@ -267,11 +311,11 @@ class PacketTracker:
while not self._stop_event.is_set():
time.sleep(0.05)
due_entries: List[Dict[str, Any]] = []
- expired_uids: List[str] = []
+ expired_keys: List[str] = []
now_ts = time.time()
with self._lock:
- for packet_uid, entry in list(self._entries.items()):
+ for correlation_key, 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
@@ -290,7 +334,7 @@ class PacketTracker:
if should_flush:
due_entries.append(
{
- "packet_uid": entry["packet_uid"],
+ "correlation_key": entry["correlation_key"],
"payload": dict(entry["payload"]),
}
)
@@ -298,10 +342,10 @@ class PacketTracker:
if entry["finalized"] and not entry["stats_recorded"]:
self._record_stats(entry["payload"])
entry["stats_recorded"] = True
- expired_uids.append(packet_uid)
+ expired_keys.append(correlation_key)
- for packet_uid in expired_uids:
- self._entries.pop(packet_uid, None)
+ for correlation_key in expired_keys:
+ self._entries.pop(correlation_key, None)
for entry in due_entries:
self._persist(entry)
@@ -318,13 +362,13 @@ class PacketTracker:
fut = asyncio.run_coroutine_threadsafe(web_db.upsert_packet(payload), web_loop)
fut.result(timeout=settings.packet_tracker_persist_timeout_seconds)
with self._lock:
- current = self._entries.get(entry["packet_uid"])
+ current = self._entries.get(entry["correlation_key"])
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"])
+ logger.exception("Failed to persist packet %s", entry["correlation_key"])
def _record_stats(self, payload: Dict[str, Any]) -> None:
capture_sources = set(payload.get("capture_sources") or [])
@@ -333,6 +377,10 @@ class PacketTracker:
self._stats["persisted_with_raw"] += 1
else:
self._stats["persisted_without_raw"] += 1
+ if payload.get("correlation_source") == "kernel_mark":
+ self._stats["persisted_kernel_mark"] = self._stats.get("persisted_kernel_mark", 0) + 1
+ 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
@@ -352,6 +400,8 @@ class PacketTracker:
active_af_packet_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"}:
@@ -360,6 +410,10 @@ class PacketTracker:
active_telemetry_only += 1
else:
active_merged += 1
+ if entry["payload"].get("correlation_source") == "kernel_mark":
+ active_kernel_mark += 1
+ else:
+ active_legacy_hash += 1
return {
"active_total": active_total,
@@ -368,6 +422,8 @@ class PacketTracker:
"active_af_packet_only": active_af_packet_only,
"active_telemetry_only": active_telemetry_only,
"active_merged": active_merged,
+ "active_kernel_mark": active_kernel_mark,
+ "active_legacy_hash": active_legacy_hash,
"cumulative": stats,
}
diff --git a/frontend/src/components/PacketViewer.tsx b/frontend/src/components/PacketViewer.tsx
index 6d020e3..f053331 100644
--- a/frontend/src/components/PacketViewer.tsx
+++ b/frontend/src/components/PacketViewer.tsx
@@ -85,7 +85,7 @@ function colorForName(name: string) {
}
function packetKey(packet: PacketRow) {
- return String(packet.packet_uid ?? packet.id ?? `${packet.ingress_if ?? 'if'}:${packet.timestamp ?? ''}`);
+ return String(packet.correlation_key ?? packet.packet_id ?? packet.packet_uid ?? packet.id ?? `${packet.ingress_if ?? 'if'}:${packet.timestamp ?? ''}`);
}
function formatEthType(packet: PacketRow) {
@@ -302,6 +302,19 @@ export default function PacketViewer(): ReactElement {
width: 120,
render: (val: any) => {formatTimestamp(val)},
},
+ {
+ title: 'Corr',
+ key: 'correlation',
+ width: 140,
+ render: (_: any, rec: PacketRow) => (
+
+ {rec.packet_id ?? rec.packet_uid?.slice(0, 10) ?? '-'}
+
+ {rec.correlation_source ?? '-'}
+
+
+ ),
+ },
{
title: 'Path',
key: 'path',
diff --git a/frontend/src/types/packets.ts b/frontend/src/types/packets.ts
index 773a0df..95cc1e4 100644
--- a/frontend/src/types/packets.ts
+++ b/frontend/src/types/packets.ts
@@ -1,7 +1,13 @@
export interface PacketRow {
id?: number | string;
timestamp?: string;
+ updated_at?: string | null;
+ correlation_key?: string;
+ correlation_source?: string | null;
+ packet_id?: string | null;
packet_uid?: string;
+ skb_mark?: number | null;
+ capture_iface?: string | null;
ingress_if?: string | null;
egress_if?: string | null;
src_mac?: string | null;
@@ -27,6 +33,7 @@ export interface PacketRow {
app_is_encrypted?: boolean | null;
app_risk_score?: number | null;
dpi_metadata?: Record | null;
+ capture_metadata?: Record | null;
telemetry_metadata?: Record | null;
verdict?: string | null;
verdict_reason?: string | null;
diff --git a/setup_build_server.sh b/setup_build_server.sh
index 7ff51d9..286c0fc 100755
--- a/setup_build_server.sh
+++ b/setup_build_server.sh
@@ -26,7 +26,8 @@ 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 \
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
+ libpcre2-dev libmaxminddb-dev librrd-dev python3-bpfcc bpfcc-tools linux-headers-generic \
+ clang llvm libelf-dev libbpf-dev iproute2
apt install -y linux-headers-$(uname -r) || apt install -y linux-headers-generic
# -----------------------------
diff --git a/setup_database.sh b/setup_database.sh
index 5e7c007..3476ee1 100755
--- a/setup_database.sh
+++ b/setup_database.sh
@@ -44,9 +44,15 @@ sudo -u postgres psql -d "$DB_NAME" < '' THEN 'pid:' || packet_id
+ WHEN packet_uid IS NOT NULL AND packet_uid <> '' THEN 'uid:' || packet_uid
+ ELSE 'row:' || id::text
+END
+WHERE correlation_key IS NULL OR correlation_key = '';
+UPDATE packets
+SET correlation_source = CASE
+ WHEN packet_id IS NOT NULL AND packet_id <> '' THEN 'kernel_mark'
+ ELSE 'legacy_hash'
+END
+WHERE correlation_source IS NULL OR correlation_source = '';
+UPDATE packets
SET capture_sources = ARRAY_REMOVE(ARRAY[
CASE WHEN raw IS NOT NULL THEN 'af_packet' END,
CASE WHEN telemetry_metadata IS NOT NULL THEN 'telemetry' END
], NULL)
WHERE capture_sources IS NULL OR array_length(capture_sources, 1) IS NULL;
+UPDATE packets
+SET updated_at = COALESCE(verdict_seen_at, egress_seen_at, ingress_seen_at, timestamp, NOW())
+WHERE updated_at IS NULL;
+CREATE UNIQUE INDEX IF NOT EXISTS idx_packets_correlation_key ON packets(correlation_key);
CREATE UNIQUE INDEX IF NOT EXISTS idx_packets_packet_uid ON packets(packet_uid);
EOF
@@ -163,6 +200,8 @@ ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT USAGE, SELECT, UPDATE ON SEQUENC
-- Create indexes for faster queries
CREATE INDEX IF NOT EXISTS idx_packets_timestamp ON packets(timestamp DESC);
+CREATE INDEX IF NOT EXISTS idx_packets_updated_at ON packets(updated_at DESC);
+CREATE INDEX IF NOT EXISTS idx_packets_packet_id ON packets(packet_id);
CREATE INDEX IF NOT EXISTS idx_packets_src_ip ON packets(src_ip);
CREATE INDEX IF NOT EXISTS idx_packets_app_protocol ON packets(app_protocol);
CREATE INDEX IF NOT EXISTS idx_packets_app_hostname ON packets(app_hostname);
diff --git a/tools/ebpf/mark_packet_id.c b/tools/ebpf/mark_packet_id.c
new file mode 100644
index 0000000..aadfe7b
--- /dev/null
+++ b/tools/ebpf/mark_packet_id.c
@@ -0,0 +1,29 @@
+#include
+#include
+#include
+
+#define PACKET_ID_MASK 0x0FFFFFFF
+#define VERDICT_MASK 0xF0000000
+
+struct {
+ __uint(type, BPF_MAP_TYPE_ARRAY);
+ __uint(max_entries, 1);
+ __type(key, __u32);
+ __type(value, __u32);
+} packet_counter SEC(".maps");
+
+SEC("classifier")
+int classifier(struct __sk_buff *skb) {
+ __u32 key = 0;
+ __u32 *counter = bpf_map_lookup_elem(&packet_counter, &key);
+ __u32 next = 1;
+
+ if (counter) {
+ next = __sync_add_and_fetch(counter, 1);
+ }
+
+ skb->mark = (skb->mark & VERDICT_MASK) | (next & PACKET_ID_MASK);
+ return TC_ACT_OK;
+}
+
+char LICENSE[] SEC("license") = "GPL";
diff --git a/tools/ebpf/prepend_capture_header.c b/tools/ebpf/prepend_capture_header.c
new file mode 100644
index 0000000..fa23ba4
--- /dev/null
+++ b/tools/ebpf/prepend_capture_header.c
@@ -0,0 +1,41 @@
+#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 classifier(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
new file mode 100755
index 0000000..30ea623
--- /dev/null
+++ b/tools/setup_bridge_capture.sh
@@ -0,0 +1,84 @@
+#!/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; then
+ echo "Missing required tools: tc and clang must be installed." >&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"
+
+clang -O2 -g -target bpf -c "$SRC_DIR/mark_packet_id.c" -o "$MARK_OBJ"
+clang -O2 -g -target bpf -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
+
+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 "$CAPTURE_IF" clsact
+tc filter replace dev "$CAPTURE_IF" ingress 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 "$CAPTURE_IF" clsact 2>/dev/null || true
+
+if ip link show "$MIRROR_IF" >/dev/null 2>&1; then
+ ip link del "$MIRROR_IF"
+fi