From bef04dbe31c063fd390d000294d386ff93d06816 Mon Sep 17 00:00:00 2001 From: malmert Date: Sun, 12 Apr 2026 19:10:36 +0200 Subject: [PATCH] test new egress ingress detection strategy --- backend/src/network_sniffer.py | 56 +++++++++++++++++++++++-- backend/src/utilities/packet_tracker.py | 47 +++++++++++++++++---- 2 files changed, 92 insertions(+), 11 deletions(-) diff --git a/backend/src/network_sniffer.py b/backend/src/network_sniffer.py index 3c807b8..15d7ee0 100644 --- a/backend/src/network_sniffer.py +++ b/backend/src/network_sniffer.py @@ -211,6 +211,7 @@ def _build_capture_observation( capture_source: str, capture_session_id: Optional[str], capture_metadata: Optional[Dict[str, Any]], + observation_metadata: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: session = sessions.get(capture_session_id) if capture_session_id else None session_is_bridge = bool(session and session.get("is_bridge")) @@ -220,7 +221,7 @@ def _build_capture_observation( if isinstance(session, dict) and session.get("capture_mode") else ("tc_ebpf" if capture_metadata else "af_packet") ) - return { + observation = { "observation_type": "capture", "source": capture_source, "iface": pkt_iface, @@ -230,6 +231,11 @@ def _build_capture_observation( "session_label": session_label, "session_kind": "bridge" if session_is_bridge else "interface", } + if isinstance(observation_metadata, dict): + for key, value in observation_metadata.items(): + if value is not None: + observation[key] = value + return observation def parse_packet( @@ -237,6 +243,7 @@ def parse_packet( bridge_label: str, capture_metadata: Optional[Dict[str, Any]] = None, capture_session_id: Optional[str] = None, + observation_metadata: Optional[Dict[str, Any]] = None, ) -> None: """ Parse a scapy Packet object into a normalized PacketInfo and schedule DB insert. @@ -288,6 +295,7 @@ def parse_packet( capture_source=capture_source, capture_session_id=capture_session_id, capture_metadata=capture_metadata, + observation_metadata=observation_metadata, ), "ip_id": None, "icmp_type": None, @@ -505,11 +513,44 @@ def parse_packet_bytes( iface: str, capture_metadata: Optional[Dict[str, Any]] = None, capture_session_id: Optional[str] = None, + observation_metadata: Optional[Dict[str, Any]] = None, ) -> None: """Parse one raw Ethernet frame using the shared Scapy packet path.""" pkt = Ether(packet_bytes) pkt.sniffed_on = iface - parse_packet(pkt, iface, capture_metadata=capture_metadata, capture_session_id=capture_session_id) + parse_packet( + pkt, + iface, + capture_metadata=capture_metadata, + capture_session_id=capture_session_id, + observation_metadata=observation_metadata, + ) + + +def _decode_af_packet_pkttype(pkttype: Any) -> tuple[Optional[int], Optional[str]]: + try: + normalized = int(pkttype) + except Exception: + return None, None + + packet_outgoing = getattr(socket, "PACKET_OUTGOING", 4) + if normalized == packet_outgoing: + return normalized, "egress" + return normalized, "ingress" + + +def _build_af_packet_observation_metadata(addr: Any) -> Dict[str, Any]: + if not isinstance(addr, tuple): + return {} + + metadata: Dict[str, Any] = {} + if len(addr) >= 3: + pkttype_raw, path_role = _decode_af_packet_pkttype(addr[2]) + if pkttype_raw is not None: + metadata["socket_pkttype"] = pkttype_raw + if path_role is not None: + metadata["path_role"] = path_role + return metadata # ------------------------- @@ -670,7 +711,7 @@ def _session_reader_loop(session_id: str) -> None: sock: socket.socket = key.fileobj iface: str = key.data try: - raw = sock.recv(settings.sniffer_recv_bytes) + raw, addr = sock.recvfrom(settings.sniffer_recv_bytes) if not raw: continue except BlockingIOError: @@ -690,9 +731,16 @@ def _session_reader_loop(session_id: str) -> None: # parse with scapy try: + recv_ts = datetime.now(timezone.utc) pkt = Ether(raw) pkt.sniffed_on = iface - parse_packet(pkt, label, capture_session_id=session_id) + pkt.time = recv_ts.timestamp() + parse_packet( + pkt, + label, + capture_session_id=session_id, + observation_metadata=_build_af_packet_observation_metadata(addr), + ) 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) diff --git a/backend/src/utilities/packet_tracker.py b/backend/src/utilities/packet_tracker.py index 5b72aa6..f5dbe4b 100644 --- a/backend/src/utilities/packet_tracker.py +++ b/backend/src/utilities/packet_tracker.py @@ -35,6 +35,8 @@ def _observation_signature(observation: Dict[str, Any]) -> tuple[Any, ...]: observation.get("session_label"), observation.get("session_kind"), observation.get("reason"), + observation.get("path_role"), + observation.get("socket_pkttype"), ) @@ -104,6 +106,41 @@ def _sorted_bridge_af_packet_observations(payload: Dict[str, Any]) -> Dict[str, return grouped +def _bridge_af_packet_observation_path( + observations: List[Dict[str, Any]], +) -> tuple[Optional[Dict[str, Any]], Optional[Dict[str, Any]]]: + if not observations: + return None, None + + ingress_candidates = [item for item in observations if item.get("path_role") == "ingress"] + egress_candidates = [item for item in observations if item.get("path_role") == "egress"] + + ingress_observation = ingress_candidates[0] if ingress_candidates else observations[0] + ingress_iface = ingress_observation.get("iface") + ingress_ts = _parse_observation_timestamp(ingress_observation.get("timestamp")) + + preferred_egress = next( + ( + item + for item in egress_candidates + if item.get("iface") != ingress_iface and _parse_observation_timestamp(item.get("timestamp")) >= ingress_ts + ), + None, + ) + if preferred_egress is not None: + return ingress_observation, preferred_egress + + fallback_egress = next( + ( + item + for item in observations + if item.get("iface") != ingress_iface and _parse_observation_timestamp(item.get("timestamp")) >= ingress_ts + ), + None, + ) + return ingress_observation, fallback_egress + + class PacketTracker: """Deduplicate packet observations and persist one upserted row per packet.""" @@ -427,7 +464,9 @@ class PacketTracker: if not observations: return False - ingress_observation = observations[0] + ingress_observation, egress_observation = _bridge_af_packet_observation_path(observations) + if ingress_observation is None: + return False ingress_iface = ingress_observation.get("iface") ingress_timestamp = ingress_observation.get("timestamp") @@ -441,12 +480,6 @@ class PacketTracker: except Exception: pass - egress_observation: Optional[Dict[str, Any]] = None - for observation in observations[1:]: - if observation.get("iface") != ingress_iface: - egress_observation = observation - break - if egress_observation is None: return changed