test ebpf

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}