diff --git a/backend/src/network_sniffer.py b/backend/src/network_sniffer.py index d070417..da81b23 100644 --- a/backend/src/network_sniffer.py +++ b/backend/src/network_sniffer.py @@ -4,6 +4,7 @@ import asyncpg import threading import logging import os +import socket from typing import List, Dict logging.basicConfig(level=logging.INFO) @@ -23,29 +24,55 @@ bridge_ports_cache = {} # ------------------------------------------------------------------- async_loop = asyncio.new_event_loop() + def start_async_loop(loop): asyncio.set_event_loop(loop) loop.run_forever() + threading.Thread(target=start_async_loop, args=(async_loop,), daemon=True).start() # ------------------------------------------------------------------- # BRIDGE PORT HANDLING # ------------------------------------------------------------------- +def check_interface_exists(iface: str) -> bool: + return os.path.isdir(f"/sys/class/net/{iface}") + + +def check_interface_up(iface: str) -> bool: + try: + with open(f"/sys/class/net/{iface}/operstate", "r") as f: + return f.read().strip() == "up" + except FileNotFoundError: + return False + + def get_bridge_ports(bridge: str) -> List[str]: if bridge in bridge_ports_cache: return bridge_ports_cache[bridge] base = f"/sys/class/net/{bridge}/brif/" if not os.path.isdir(base): - logger.error(f"Bridge {bridge} does not exist") + logger.error(f"[ERROR] Bridge '{bridge}' does not exist.") return [] - ports = os.listdir(base) - bridge_ports_cache[bridge] = ports - logger.info(f"Bridge {bridge} ports: {ports}") - return ports + try: + ports = os.listdir(base) + except PermissionError: + logger.error(f"[ERROR] No permissions to read bridge ports for '{bridge}'.") + return [] + + ok_ports = [] + for p in ports: + if not check_interface_exists(p): + logger.warning(f"[WARN] Port '{p}' in bridge but does not exist in /sys/class/net") + continue + ok_ports.append(p) + + bridge_ports_cache[bridge] = ok_ports + logger.info(f"[INFO] Bridge {bridge} ports: {ok_ports}") + return ok_ports def determine_direction(pkt_iface: str, bridge: str): @@ -59,6 +86,7 @@ def determine_direction(pkt_iface: str, bridge: str): # DATABASE INSERTION # ------------------------------------------------------------------- async def db_insert_packet(pkt_info: dict): + conn = None try: conn = await asyncpg.connect(DB_DSN) @@ -78,23 +106,24 @@ async def db_insert_packet(pkt_info: dict): ) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) """, - pkt_info["ingress"], - "unknown", - pkt_info["src_mac"], - pkt_info["dst_mac"], - pkt_info["eth_type"], - pkt_info["src_ip"], - pkt_info["dst_ip"], - pkt_info["protocol"], - pkt_info["length"], - str(pkt_info["egress"]), - pkt_info["raw"] + pkt_info["ingress"], + "unknown", + pkt_info["src_mac"], + pkt_info["dst_mac"], + pkt_info["eth_type"], + pkt_info["src_ip"], + pkt_info["dst_ip"], + pkt_info["protocol"], + pkt_info["length"], + str(pkt_info["egress"]), + pkt_info["raw"] ) - except Exception as e: + except (asyncpg.PostgresError, ConnectionError, OSError) as e: logger.exception(f"DB insert failed: {e}") + finally: - if 'conn' in locals(): + if conn: await conn.close() @@ -125,39 +154,71 @@ def handle_packet(pkt, bridge: str): # ------------------------------------------------------------------- -# SNIFFING +# CLEAN STOPPING SNIFFERS # ------------------------------------------------------------------- +def sniffer_loop(ifname: str, stop_event: threading.Event, bridge: str): + logger.info(f"Sniffer STARTED on {ifname}") + + if not check_interface_exists(ifname): + logger.error(f"[ERROR] Interface {ifname} does not exist. Stopping sniffer.") + return + + if not check_interface_up(ifname): + logger.error(f"[ERROR] Interface {ifname} is DOWN. Stopping sniffer.") + return + + while not stop_event.is_set(): + try: + sniff( + iface=ifname, + prn=lambda pkt: handle_packet(pkt, bridge), + store=False, + timeout=1 # periodic return so we can check stop_event + ) + except PermissionError: + logger.error(f"[ERROR] Permission denied sniffing on {ifname}. Run as root.") + break + except OSError as e: + logger.error(f"[ERROR] Sniffer error on {ifname}: {e}") + break + except Exception as e: + logger.exception(f"Unexpected sniffer error on {ifname}: {e}") + break + + logger.info(f"Sniffer STOPPED on {ifname}") + + def start_sniffer_thread(bridge: str): ports = get_bridge_ports(bridge) + if not ports: + logger.error(f"[ERROR] Could not start sniffer: No valid ports found for bridge {bridge}") + return {} for iface in ports: if iface in sniffer_threads: + logger.info(f"[INFO] Sniffer on {iface} is already running") continue stop_event = threading.Event() thread_stop_flags[iface] = stop_event - def sniff_blocking(ifname=iface): - logger.info(f"Sniffer STARTED on {ifname}") + thread = threading.Thread( + target=sniffer_loop, + args=(iface, stop_event, bridge), + daemon=True + ) - sniff( - iface=ifname, - prn=lambda pkt: handle_packet(pkt, bridge), - store=False, - stop_filter=lambda _: stop_event.is_set(), - timeout=1 # ensure periodic stop checks - ) - - logger.info(f"Sniffer STOPPED on {ifname}") - - thread = threading.Thread(target=sniff_blocking, daemon=True) sniffer_threads[iface] = thread thread.start() return sniffer_threads +# ------------------------------------------------------------------- +# PUBLIC ASYNC START/STOP METHODS +# ------------------------------------------------------------------- async def start_sniffing(bridge: str): + logger.info(f"Starting sniffing for bridge {bridge}") start_sniffer_thread(bridge) return True @@ -179,16 +240,14 @@ async def stop_sniffing(): # ------------------------------------------------------------------- -# STATUS HELPER (for API endpoint) +# STATUS HELPER # ------------------------------------------------------------------- def get_sniffer_status(): - """ - Return a dict of interface → running/stopped status. - Useful for an API /status endpoint. - """ - status = {} - + out = {} for iface, t in sniffer_threads.items(): - status[iface] = "running" if t.is_alive() else "stopped" - - return status + out[iface] = { + "running": t.is_alive(), + "exists": check_interface_exists(iface), + "up": check_interface_up(iface) + } + return out