fix: enhance bridge port handling and sniffer management with improved error logging and interface checks
All checks were successful
Build and Deploy MITM Webserver / build (push) Successful in 7s
All checks were successful
Build and Deploy MITM Webserver / build (push) Successful in 7s
This commit is contained in:
@@ -4,6 +4,7 @@ import asyncpg
|
|||||||
import threading
|
import threading
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
|
import socket
|
||||||
from typing import List, Dict
|
from typing import List, Dict
|
||||||
|
|
||||||
logging.basicConfig(level=logging.INFO)
|
logging.basicConfig(level=logging.INFO)
|
||||||
@@ -23,29 +24,55 @@ bridge_ports_cache = {}
|
|||||||
# -------------------------------------------------------------------
|
# -------------------------------------------------------------------
|
||||||
async_loop = asyncio.new_event_loop()
|
async_loop = asyncio.new_event_loop()
|
||||||
|
|
||||||
|
|
||||||
def start_async_loop(loop):
|
def start_async_loop(loop):
|
||||||
asyncio.set_event_loop(loop)
|
asyncio.set_event_loop(loop)
|
||||||
loop.run_forever()
|
loop.run_forever()
|
||||||
|
|
||||||
|
|
||||||
threading.Thread(target=start_async_loop, args=(async_loop,), daemon=True).start()
|
threading.Thread(target=start_async_loop, args=(async_loop,), daemon=True).start()
|
||||||
|
|
||||||
|
|
||||||
# -------------------------------------------------------------------
|
# -------------------------------------------------------------------
|
||||||
# BRIDGE PORT HANDLING
|
# 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]:
|
def get_bridge_ports(bridge: str) -> List[str]:
|
||||||
if bridge in bridge_ports_cache:
|
if bridge in bridge_ports_cache:
|
||||||
return bridge_ports_cache[bridge]
|
return bridge_ports_cache[bridge]
|
||||||
|
|
||||||
base = f"/sys/class/net/{bridge}/brif/"
|
base = f"/sys/class/net/{bridge}/brif/"
|
||||||
if not os.path.isdir(base):
|
if not os.path.isdir(base):
|
||||||
logger.error(f"Bridge {bridge} does not exist")
|
logger.error(f"[ERROR] Bridge '{bridge}' does not exist.")
|
||||||
return []
|
return []
|
||||||
|
|
||||||
ports = os.listdir(base)
|
try:
|
||||||
bridge_ports_cache[bridge] = ports
|
ports = os.listdir(base)
|
||||||
logger.info(f"Bridge {bridge} ports: {ports}")
|
except PermissionError:
|
||||||
return ports
|
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):
|
def determine_direction(pkt_iface: str, bridge: str):
|
||||||
@@ -59,6 +86,7 @@ def determine_direction(pkt_iface: str, bridge: str):
|
|||||||
# DATABASE INSERTION
|
# DATABASE INSERTION
|
||||||
# -------------------------------------------------------------------
|
# -------------------------------------------------------------------
|
||||||
async def db_insert_packet(pkt_info: dict):
|
async def db_insert_packet(pkt_info: dict):
|
||||||
|
conn = None
|
||||||
try:
|
try:
|
||||||
conn = await asyncpg.connect(DB_DSN)
|
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)
|
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)
|
||||||
""",
|
""",
|
||||||
pkt_info["ingress"],
|
pkt_info["ingress"],
|
||||||
"unknown",
|
"unknown",
|
||||||
pkt_info["src_mac"],
|
pkt_info["src_mac"],
|
||||||
pkt_info["dst_mac"],
|
pkt_info["dst_mac"],
|
||||||
pkt_info["eth_type"],
|
pkt_info["eth_type"],
|
||||||
pkt_info["src_ip"],
|
pkt_info["src_ip"],
|
||||||
pkt_info["dst_ip"],
|
pkt_info["dst_ip"],
|
||||||
pkt_info["protocol"],
|
pkt_info["protocol"],
|
||||||
pkt_info["length"],
|
pkt_info["length"],
|
||||||
str(pkt_info["egress"]),
|
str(pkt_info["egress"]),
|
||||||
pkt_info["raw"]
|
pkt_info["raw"]
|
||||||
)
|
)
|
||||||
|
|
||||||
except Exception as e:
|
except (asyncpg.PostgresError, ConnectionError, OSError) as e:
|
||||||
logger.exception(f"DB insert failed: {e}")
|
logger.exception(f"DB insert failed: {e}")
|
||||||
|
|
||||||
finally:
|
finally:
|
||||||
if 'conn' in locals():
|
if conn:
|
||||||
await conn.close()
|
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):
|
def start_sniffer_thread(bridge: str):
|
||||||
ports = get_bridge_ports(bridge)
|
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:
|
for iface in ports:
|
||||||
if iface in sniffer_threads:
|
if iface in sniffer_threads:
|
||||||
|
logger.info(f"[INFO] Sniffer on {iface} is already running")
|
||||||
continue
|
continue
|
||||||
|
|
||||||
stop_event = threading.Event()
|
stop_event = threading.Event()
|
||||||
thread_stop_flags[iface] = stop_event
|
thread_stop_flags[iface] = stop_event
|
||||||
|
|
||||||
def sniff_blocking(ifname=iface):
|
thread = threading.Thread(
|
||||||
logger.info(f"Sniffer STARTED on {ifname}")
|
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
|
sniffer_threads[iface] = thread
|
||||||
thread.start()
|
thread.start()
|
||||||
|
|
||||||
return sniffer_threads
|
return sniffer_threads
|
||||||
|
|
||||||
|
|
||||||
|
# -------------------------------------------------------------------
|
||||||
|
# PUBLIC ASYNC START/STOP METHODS
|
||||||
|
# -------------------------------------------------------------------
|
||||||
async def start_sniffing(bridge: str):
|
async def start_sniffing(bridge: str):
|
||||||
|
logger.info(f"Starting sniffing for bridge {bridge}")
|
||||||
start_sniffer_thread(bridge)
|
start_sniffer_thread(bridge)
|
||||||
return True
|
return True
|
||||||
|
|
||||||
@@ -179,16 +240,14 @@ async def stop_sniffing():
|
|||||||
|
|
||||||
|
|
||||||
# -------------------------------------------------------------------
|
# -------------------------------------------------------------------
|
||||||
# STATUS HELPER (for API endpoint)
|
# STATUS HELPER
|
||||||
# -------------------------------------------------------------------
|
# -------------------------------------------------------------------
|
||||||
def get_sniffer_status():
|
def get_sniffer_status():
|
||||||
"""
|
out = {}
|
||||||
Return a dict of interface → running/stopped status.
|
|
||||||
Useful for an API /status endpoint.
|
|
||||||
"""
|
|
||||||
status = {}
|
|
||||||
|
|
||||||
for iface, t in sniffer_threads.items():
|
for iface, t in sniffer_threads.items():
|
||||||
status[iface] = "running" if t.is_alive() else "stopped"
|
out[iface] = {
|
||||||
|
"running": t.is_alive(),
|
||||||
return status
|
"exists": check_interface_exists(iface),
|
||||||
|
"up": check_interface_up(iface)
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
|||||||
Reference in New Issue
Block a user