from typing import List, Optional import logging import os # ---- Logging ---------------------------------------------------------- logger = logging.getLogger("af_packet_sniffer") # ------------------------- # Interface / bridge helpers # ------------------------- def check_interface_exists(iface: str) -> bool: """ Check for the presence of a network interface by testing sysfs. Returns True if /sys/class/net/ exists. """ return os.path.isdir(f"/sys/class/net/{iface}") def check_interface_up(iface: str) -> bool: """ Check whether the given interface is administratively/operationally up by reading /sys/class/net//operstate. Returns False if the path does not exist. """ try: with open(f"/sys/class/net/{iface}/operstate", "r") as f: return f.read().strip() == "up" except FileNotFoundError: return False def _read_sysfs_text(path: str) -> Optional[str]: """Read one sysfs file and return stripped text, or `None` if unavailable.""" try: with open(path, "r") as f: return f.read().strip() except FileNotFoundError: return None except Exception: logger.exception("Error reading sysfs path %s", path) return None def read_interface_operstate(iface: str) -> Optional[str]: """Return the kernel operstate string for an interface.""" return _read_sysfs_text(f"/sys/class/net/{iface}/operstate") def read_interface_carrier(iface: str) -> Optional[bool]: """Return the carrier state for an interface if available.""" value = _read_sysfs_text(f"/sys/class/net/{iface}/carrier") if value is None: return None try: return int(value, 10) == 1 except ValueError: logger.warning("Unexpected carrier value for %s: %r", iface, value) return None def read_interface_admin_up(iface: str) -> Optional[bool]: """Return whether the interface has the IFF_UP flag set.""" value = _read_sysfs_text(f"/sys/class/net/{iface}/flags") if value is None: return None try: return bool(int(value, 0) & 0x1) except ValueError: logger.warning("Unexpected flags value for %s: %r", iface, value) return None def _read_bridge_ports_from_sysfs(bridge: str) -> List[str]: """ Read bridge member interfaces from sysfs. Internal helper that always reads. """ base = f"/sys/class/net/{bridge}/brif/" if not os.path.isdir(base): return [] try: ports = [p for p in os.listdir(base) if check_interface_exists(p)] except Exception: logger.exception("Error reading bridge ports for %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)