diff --git a/backend/src/network_sniffer.py b/backend/src/network_sniffer.py index 1e8aa41..8bc85a8 100644 --- a/backend/src/network_sniffer.py +++ b/backend/src/network_sniffer.py @@ -41,8 +41,10 @@ af_selector: Optional[selectors.BaseSelector] = None af_thread: Optional[threading.Thread] = None af_stop_event: Optional[threading.Event] = None -# Cache of bridge -> ports -bridge_ports_cache: Dict[str, List[str]] = {} +# Snapshot of bridge -> ports taken once when sniffer starts +fixed_bridge_ports: Dict[str, List[str]] = {} +# Remember which bridge the sniffer is using (single active sniffer model) +current_bridge: Optional[str] = None # --------------------------------------------------------------------- # Async loop used to schedule DB inserts from packet callback threads. @@ -88,40 +90,41 @@ def check_interface_up(iface: str) -> bool: return False -def get_bridge_ports(bridge: str) -> List[str]: +def _read_bridge_ports_from_sysfs(bridge: str) -> List[str]: """ - Read the bridge member interfaces from sysfs (/sys/class/net//brif/). - This function refreshes the cache each call (cache entry is updated), - which keeps behavior predictable for dynamic topologies. - If the bridge does not exist, an empty list is returned. + Read bridge member interfaces from sysfs. Internal helper that always reads. """ base = f"/sys/class/net/{bridge}/brif/" if not os.path.isdir(base): - logger.error("Bridge '%s' does not exist", bridge) return [] try: ports = [p for p in os.listdir(base) if check_interface_exists(p)] - except Exception as e: - logger.exception("Error listing bridge ports for %s: %s", bridge, e) + except Exception: + logger.exception("Error reading bridge ports for %s", bridge) ports = [] - - # update local cache and log discovered ports - bridge_ports_cache[bridge] = ports - logger.info("Bridge %s ports: %s", bridge, ports) return ports +def get_bridge_ports_once(bridge: str) -> List[str]: + """ + Convenience: read bridge ports (no caching here) used when preparing the fixed snapshot. + """ + return _read_bridge_ports_from_sysfs(bridge) + + def determine_direction(pkt_iface: str, bridge: str): """ - Determine ingress and egress information for a packet based on the - interface it was sniffed on and the bridge port list. + Determine ingress/egress using the fixed snapshot of bridge ports + taken when the sniffer started. This avoids reading sysfs for every + packet and prevents noisy logs when bridge ports are static. - Returns: - ingress: the interface where the packet was observed - egress: list of other bridge ports (possible egress ports) + If snapshot is missing (defensive), fall back to a direct sysfs read. """ - ports = bridge_ports_cache.get(bridge) or get_bridge_ports(bridge) + ports = fixed_bridge_ports.get(bridge) + if ports is None: + # Defensive fallback (should not happen if start_afpacket_sniffer was used) + ports = _read_bridge_ports_from_sysfs(bridge) ingress = pkt_iface egress = [p for p in ports if p != pkt_iface] return ingress, egress @@ -184,7 +187,7 @@ async def db_insert_packet(pkt_info: dict) -> None: # ------------------------------------------------------------------- -# Packet parsing using your existing parse_packet logic +# Packet parsing callback used by scapy-like parser # ------------------------------------------------------------------- def parse_packet(pkt, bridge: str) -> None: """ @@ -348,55 +351,34 @@ def _close_socket(ifname: str) -> None: pass -def _ensure_sockets_for_bridge(bridge: str) -> None: +def _ensure_sockets_for_bridge_from_snapshot(bridge: str) -> None: """ - Ensure there is an AF_PACKET socket bound for every current bridge port. - We keep sockets for interfaces even if they are DOWN (they will start receiving when link comes up). + Create sockets for ports listed in the fixed snapshot. + Only creates sockets for ports that don't already have one. """ - ports = get_bridge_ports(bridge) + ports = fixed_bridge_ports.get(bridge, []) for p in ports: if p in af_sockets: continue if not check_interface_exists(p): + logger.warning("Snapshot port %s does not exist in sysfs (skipping)", p) continue s = _create_af_packet_socket(p) if s: af_sockets[p] = s - # register later in selector by reader thread - - -def _rebind_if_needed(ifname: str) -> None: - """ - Try to re-create a socket for an interface if it is missing (e.g. after deletion). - """ - if ifname in af_sockets: - return - if not check_interface_exists(ifname): - return - s = _create_af_packet_socket(ifname) - if s: - af_sockets[ifname] = s - if af_selector: - try: - af_selector.register(s, selectors.EVENT_READ, data=ifname) - except Exception: - # ignore duplicate registration / race - pass def afpacket_reader_loop(bridge: str, stop_event: threading.Event) -> None: """ Single reader thread that multiplexes all AF_PACKET sockets for the bridge using a selector. When data arrives, we parse into a Scapy packet and call parse_packet(). - - This design uses a single thread + selector rather than one thread per interface. """ global af_selector logger.info("AF_PACKET reader starting for bridge %s", bridge) af_selector = selectors.DefaultSelector() - # ensure sockets exist for current ports - _ensure_sockets_for_bridge(bridge) + # Ensure sockets exist based on the fixed snapshot (created at start) + _ensure_sockets_for_bridge_from_snapshot(bridge) # register sockets we have for ifname, s in list(af_sockets.items()): @@ -410,20 +392,17 @@ def afpacket_reader_loop(bridge: str, stop_event: threading.Event) -> None: # main loop while not stop_event.is_set(): - # refresh sockets for any new bridge ports (cheap) + # In this static snapshot mode we only occasionally try to register + # newly created sockets (e.g., interfaces that existed but socket creation failed earlier). try: - _ensure_sockets_for_bridge(bridge) - # register any new sockets with selector for ifname, s in list(af_sockets.items()): try: - # register only if not registered if not any(k.fileobj is s for k in af_selector.get_map().values()): af_selector.register(s, selectors.EVENT_READ, data=ifname) except Exception: - # ignore races pass except Exception: - logger.exception("Error ensuring sockets") + logger.exception("Error ensuring selector registrations") # wait for events with short timeout to remain responsive try: @@ -434,30 +413,27 @@ def afpacket_reader_loop(bridge: str, stop_event: threading.Event) -> None: continue if not events: - # no events; loop will re-ensure sockets again + # no events; loop will re-check registrations continue for key, mask in events: sock: socket.socket = key.fileobj ifname: str = key.data try: - # read raw frame - # using a single large buffer; AF_PACKET will give full frame raw = sock.recv(65536) if not raw: continue except BlockingIOError: continue except OSError as e: - # handle interface removal (ENODEV) or other errors: close socket and attempt rebind later + # handle interface removal (ENODEV) or other errors: close socket and continue if e.errno in (errno.ENODEV, errno.ENETDOWN, errno.EBADF): - logger.warning("Socket error on %s: %s — closing socket and will attempt to rebind later", ifname, e) + logger.warning("Socket error on %s: %s — closing socket", ifname, e) try: af_selector.unregister(sock) except Exception: pass _close_socket(ifname) - # schedule rebind attempt on next loop iteration continue else: logger.exception("Recv error on %s: %s", ifname, e) @@ -502,15 +478,19 @@ def afpacket_reader_loop(bridge: str, stop_event: threading.Event) -> None: def start_afpacket_sniffer(bridge: str) -> None: """ Start the optimized AF_PACKET sniffer for the provided bridge. - This will bind sockets to all current bridge ports and start a single reader thread. + This reads bridge ports once (snapshot) and starts a single reader thread. """ - global af_thread, af_stop_event + global af_thread, af_stop_event, current_bridge, fixed_bridge_ports if af_thread and af_thread.is_alive(): logger.info("AF_PACKET sniffer already running") return - # ensure we have initial ports - get_bridge_ports(bridge) + # Warm and freeze the bridge ports snapshot. We read sysfs once here. + ports = get_bridge_ports_once(bridge) + fixed_bridge_ports[bridge] = ports + current_bridge = bridge + logger.info("Fixed bridge ports for %s: %s", bridge, ports) + af_stop_event = threading.Event() af_thread = threading.Thread(target=afpacket_reader_loop, args=(bridge, af_stop_event), daemon=True) af_thread.start() @@ -519,9 +499,9 @@ def start_afpacket_sniffer(bridge: str) -> None: def stop_afpacket_sniffer() -> None: """ - Stop the AF_PACKET sniffer thread and close sockets. + Stop the AF_PACKET sniffer thread and close sockets. Clear the fixed snapshot. """ - global af_thread, af_stop_event + global af_thread, af_stop_event, current_bridge, fixed_bridge_ports if not af_thread: return if af_stop_event: @@ -529,6 +509,12 @@ def stop_afpacket_sniffer() -> None: af_thread.join(timeout=2) af_thread = None af_stop_event = None + + # clear fixed snapshot(s) + if current_bridge: + fixed_bridge_ports.pop(current_bridge, None) + current_bridge = None + logger.info("AF_PACKET sniffer stopped")