"""Helpers for stable packet identity across capture and telemetry 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}