malmert
f9bbe9b73c
All checks were successful
Build and Deploy MITM Webserver / build (push) Successful in 1m40s
45 lines
1.1 KiB
Python
45 lines
1.1 KiB
Python
"""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}
|