tc full packet test

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:]

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,

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)
 
 

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

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