refactor utils

backend/src/utilities/interface_bridge_helpers.py

@@ -0,0 +1,54 @@
+from typing import List, Dict, Optional
+import logging
+import os
+
+# ---- Logging ----------------------------------------------------------
+logging.basicConfig(level=logging.INFO)
+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/<iface> 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/<iface>/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_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)