fix
All checks were successful
Build and Deploy MITM Webserver / traffic_target (push) Successful in 1s
Build and Deploy MITM Webserver / build (push) Successful in 10s

This commit is contained in:
2026-03-09 18:01:35 +01:00
parent a6825dd790
commit 234827f5bc

View File

@@ -1,11 +1,15 @@
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional
import errno
import logging import logging
import os import os
import subprocess import subprocess
import time
# ---- Logging ---------------------------------------------------------- # ---- Logging ----------------------------------------------------------
logger = logging.getLogger("af_packet_sniffer") logger = logging.getLogger("af_packet_sniffer")
_ETHTOOL_BIN = "/usr/sbin/ethtool" if os.path.exists("/usr/sbin/ethtool") else "ethtool" _ETHTOOL_BIN = "/usr/sbin/ethtool" if os.path.exists("/usr/sbin/ethtool") else "ethtool"
_ETHERNET_PROFILE_CACHE_TTL_SECONDS = 3.0
_ETHERNET_PROFILE_CACHE: Dict[str, tuple[float, Dict[str, Any]]] = {}
# ------------------------- # -------------------------
# Interface / bridge helpers # Interface / bridge helpers
@@ -39,6 +43,12 @@ def _read_sysfs_text(path: str) -> Optional[str]:
return f.read().strip() return f.read().strip()
except FileNotFoundError: except FileNotFoundError:
return None return None
except OSError as exc:
if exc.errno in {errno.EINVAL, errno.EIO, errno.ENODEV}:
logger.debug("Transient sysfs read failure for %s: %s", path, exc)
return None
logger.exception("Error reading sysfs path %s", path)
return None
except Exception: except Exception:
logger.exception("Error reading sysfs path %s", path) logger.exception("Error reading sysfs path %s", path)
return None return None
@@ -98,7 +108,7 @@ def read_interface_ethernet_profile(iface: str) -> Optional[Dict[str, Any]]:
check=True, check=True,
) )
except (FileNotFoundError, subprocess.CalledProcessError): except (FileNotFoundError, subprocess.CalledProcessError):
return None return _get_cached_ethernet_profile(iface)
values: dict[str, str] = {} values: dict[str, str] = {}
for line in result.stdout.splitlines(): for line in result.stdout.splitlines():
@@ -128,13 +138,29 @@ def read_interface_ethernet_profile(iface: str) -> Optional[Dict[str, Any]]:
autoneg = lowered == "on" autoneg = lowered == "on"
if speed_mbps is None and duplex is None and autoneg is None: if speed_mbps is None and duplex is None and autoneg is None:
return None return _get_cached_ethernet_profile(iface)
return { profile = {
"speed_mbps": speed_mbps, "speed_mbps": speed_mbps,
"duplex": duplex, "duplex": duplex,
"autoneg": autoneg, "autoneg": autoneg,
} }
_ETHERNET_PROFILE_CACHE[iface] = (time.time(), profile)
return profile
def _get_cached_ethernet_profile(iface: str) -> Optional[Dict[str, Any]]:
"""Return a recent ethtool snapshot to smooth short renegotiation gaps."""
cached = _ETHERNET_PROFILE_CACHE.get(iface)
if cached is None:
return None
ts, profile = cached
if time.time() - ts > _ETHERNET_PROFILE_CACHE_TTL_SECONDS:
_ETHERNET_PROFILE_CACHE.pop(iface, None)
return None
return dict(profile)
def _read_bridge_ports_from_sysfs(bridge: str) -> List[str]: def _read_bridge_ports_from_sysfs(bridge: str) -> List[str]: