From 3153b7e8a3d9866e8419b4b61a676f3bec011114 Mon Sep 17 00:00:00 2001 From: Cobra Date: Wed, 8 Apr 2026 20:58:43 -0400 Subject: [PATCH] =?UTF-8?q?Robust=20interface=20detection=20=E2=80=94=20re?= =?UTF-8?q?try=20fallback=20chain=20for=20headless=20boot=20(#206)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes interface auto-detection failures on Pi Zero 2W where wlan0 doesn't exist at boot. Implementation: - Added detect_interface_with_retry() function with 3x retry fallback chain - Config-supplied interface preferred, then /proc/net/route, then UP interface - Exponential backoff: 5s, 10s, 20s delays between retries - Filters excluded prefixes: lo, tailscale*, wg*, tun*, docker*, veth*, br-* - Used in Engine.__init__(), PacketCapture.start(), MacManager.start() - Full test coverage: 10 tests for detection, backoff, filtering Addresses issue where network stack initializes after boot and interfaces become available after 5-10 second delay. --- core/engine.py | 13 +++- modules/passive/packet_capture.py | 9 ++- modules/stealth/mac_manager.py | 9 ++- tests/test_interface_detection.py | 119 ++++++++++++++++++++++++++++++ utils/__init__.py | 1 + utils/networking.py | 92 +++++++++++++++++++++++ 6 files changed, 235 insertions(+), 8 deletions(-) create mode 100644 tests/test_interface_detection.py diff --git a/core/engine.py b/core/engine.py index f787316..b956c30 100644 --- a/core/engine.py +++ b/core/engine.py @@ -24,7 +24,7 @@ from typing import Optional, Type from core.bus import EventBus, Event from core.state import StateManager from modules.base import BaseModule -from utils.networking import get_primary_interface +from utils.networking import detect_interface_with_retry logger = logging.getLogger("sensor.engine") @@ -240,12 +240,17 @@ class Engine: self._scope = config.get("scope", {}) self._phase = config.get("engagement_phase", {}).get("mode", "passive_only") - # Resolve "auto" interface to actual interface name + # Resolve "auto" interface to actual interface name with retry fallback if self.config.get("network", {}).get("primary_interface") == "auto": - actual_iface = get_primary_interface() + actual_iface = detect_interface_with_retry( + max_retries=3, + retry_delay=5, + exponential=True, + config_interface=None + ) if actual_iface: self.config.setdefault("network", {})["primary_interface"] = actual_iface - logger.debug("Resolved auto interface to: %s", actual_iface) + logger.info("Resolved auto interface to: %s (with retry)", actual_iface) self.capture_bus = None self._lock = multiprocessing.Lock() logger.info("Engine initialized (tier=%s, phase=%s)", self._tier, self._phase) diff --git a/modules/passive/packet_capture.py b/modules/passive/packet_capture.py index d4be53d..d37bbfe 100644 --- a/modules/passive/packet_capture.py +++ b/modules/passive/packet_capture.py @@ -23,7 +23,7 @@ from pathlib import Path from typing import Optional from modules.base import BaseModule -from utils.networking import get_primary_interface +from utils.networking import detect_interface_with_retry logger = logging.getLogger("sensor.passive.packet_capture") @@ -64,7 +64,12 @@ class PacketCapture(BaseModule): if self._running: return - iface = self.config.get("interface") or get_primary_interface() or self.DEFAULT_INTERFACE + iface = self.config.get("interface") or detect_interface_with_retry( + max_retries=3, + retry_delay=5, + exponential=True, + config_interface=self.DEFAULT_INTERFACE + ) snap_len = self.config.get("snap_length", self.DEFAULT_SNAP_LEN) rotation = self.config.get("rotation_minutes", 60) * 60 if rotation <= 0: diff --git a/modules/stealth/mac_manager.py b/modules/stealth/mac_manager.py index 63e1fa1..53805fe 100644 --- a/modules/stealth/mac_manager.py +++ b/modules/stealth/mac_manager.py @@ -20,7 +20,7 @@ from typing import Optional from modules.base import BaseModule from utils.networking import ( get_mac, - get_primary_interface, + detect_interface_with_retry, get_wifi_interfaces, set_mac, ) @@ -74,7 +74,12 @@ class MacManager(BaseModule): "primary_interface", "auto" ) if self._eth_iface == "auto": - self._eth_iface = get_primary_interface() + self._eth_iface = detect_interface_with_retry( + max_retries=3, + retry_delay=5, + exponential=True, + config_interface=None + ) wifi_ifaces = get_wifi_interfaces() wifi_cfg = self.config.get("network", {}).get("wifi", {}).get("interface", "wlan0") diff --git a/tests/test_interface_detection.py b/tests/test_interface_detection.py new file mode 100644 index 0000000..6e5c104 --- /dev/null +++ b/tests/test_interface_detection.py @@ -0,0 +1,119 @@ +"""Tests for interface detection with retry fallback chain.""" + +import time +from pathlib import Path +from unittest.mock import patch, MagicMock +from tempfile import TemporaryDirectory + +import pytest + +from utils.networking import get_primary_interface, detect_interface_with_retry + + +class TestInterfaceDetection: + """Test robust interface detection with retry fallback.""" + + def test_get_primary_interface_from_route_table(self): + """get_primary_interface reads default route from /proc/net/route.""" + route_content = """Iface Destination Gateway Flags RefCnt Use Metric Mask MTU Window IRTT +eth0 00000000 0101010A 0003 0 0 100 00000000 0 0 0 +eth0 0101010A 00000000 0001 0 0 100 FFFFFFFF 0 0 0 +""" + with patch("builtins.open", create=True) as mock_open: + mock_open.return_value.__enter__.return_value.readlines.return_value = route_content.split("\n") + result = get_primary_interface() + assert result == "eth0" + + def test_get_primary_interface_fallback_eth(self): + """get_primary_interface falls back to eth* interfaces if /proc/net/route is empty.""" + with patch("builtins.open", side_effect=IOError): + with patch("utils.networking.get_all_interfaces", return_value=["eth0", "eth1"]): + result = get_primary_interface() + assert result == "eth0" + + def test_get_primary_interface_fallback_wlan(self): + """get_primary_interface falls back to first interface if no eth/en.""" + with patch("builtins.open", side_effect=IOError): + with patch("utils.networking.get_all_interfaces", return_value=["wlan0"]): + result = get_primary_interface() + assert result == "wlan0" + + def test_get_primary_interface_no_interfaces(self): + """get_primary_interface returns None if no interfaces available.""" + with patch("builtins.open", side_effect=IOError): + with patch("utils.networking.get_all_interfaces", return_value=[]): + result = get_primary_interface() + assert result is None + + def test_detect_interface_with_retry_immediate_success(self): + """detect_interface_with_retry returns interface immediately if found.""" + with patch("utils.networking.get_primary_interface", return_value="eth0"): + result = detect_interface_with_retry(max_retries=3, retry_delay=1) + assert result == "eth0" + + def test_detect_interface_with_retry_eventual_success(self): + """detect_interface_with_retry retries and succeeds after N attempts.""" + call_count = [0] + + def get_iface_side_effect(): + call_count[0] += 1 + if call_count[0] <= 2: + return None # Fail first 2 times + return "wlan0" # Succeed on 3rd call + + with patch("utils.networking.get_primary_interface", side_effect=get_iface_side_effect): + with patch("utils.networking._get_up_interface", return_value=None): + result = detect_interface_with_retry(max_retries=3, retry_delay=0.01) + assert result == "wlan0" + assert call_count[0] == 3 + + def test_detect_interface_with_retry_exhausted(self): + """detect_interface_with_retry returns None if all retries exhausted.""" + with patch("utils.networking.get_primary_interface", return_value=None): + with patch("utils.networking._get_up_interface", return_value=None): + result = detect_interface_with_retry(max_retries=2, retry_delay=0.01) + assert result is None + + def test_detect_interface_with_retry_uses_config_fallback(self): + """detect_interface_with_retry accepts config-set interface as ultimate fallback.""" + with patch("utils.networking.get_primary_interface", return_value=None): + with patch("utils.networking._get_up_interface", return_value=None): + result = detect_interface_with_retry( + max_retries=1, + retry_delay=0.01, + config_interface="configured_iface" + ) + # Should use config fallback when detection fails + assert result == "configured_iface" + + def test_detect_interface_exponential_backoff(self): + """detect_interface_with_retry uses exponential backoff (5s, 10s, 20s).""" + call_times = [] + + def get_iface_with_timing(): + call_times.append(time.time()) + return None + + with patch("utils.networking.get_primary_interface", side_effect=get_iface_with_timing): + with patch("utils.networking._get_up_interface", return_value=None): + with patch("time.sleep") as mock_sleep: + result = detect_interface_with_retry( + max_retries=3, + retry_delay=5, # Base delay + exponential=True + ) + # Should have called sleep with 5, 10, 20 + assert mock_sleep.call_count == 3 + # Check exponential backoff values (5 * 2^0, 5 * 2^1, 5 * 2^2) + sleep_calls = [call[0][0] for call in mock_sleep.call_args_list] + assert sleep_calls == [5, 10, 20] + + def test_detect_interface_filters_excluded(self): + """detect_interface_with_retry excludes lo, tailscale*, wg*, tun*, docker*, veth*, br-*.""" + excluded_ifaces = ["lo", "tailscale0", "wg0", "tun0", "docker0", "veth123", "br-abc"] + + with patch("utils.networking.get_primary_interface", return_value=None): + # Mock get_all_interfaces to return only excluded ones + with patch("utils.networking.get_all_interfaces", return_value=excluded_ifaces): + result = detect_interface_with_retry(max_retries=1, retry_delay=0.01) + assert result is None diff --git a/utils/__init__.py b/utils/__init__.py index da438a8..72858a7 100644 --- a/utils/__init__.py +++ b/utils/__init__.py @@ -3,6 +3,7 @@ from utils.networking import ( get_primary_interface, get_bridge_candidates, get_wifi_interfaces, get_mac, set_mac, get_ip, compile_bpf, send_gratuitous_arp, interface_up, interface_down, create_vlan_interface, + detect_interface_with_retry, ) from utils.logging import BBLogger from utils.stealth import ( diff --git a/utils/networking.py b/utils/networking.py index de17811..1468fea 100644 --- a/utils/networking.py +++ b/utils/networking.py @@ -8,9 +8,13 @@ import os import socket import struct import subprocess +import time +import logging from pathlib import Path from typing import List, Optional, Tuple +logger = logging.getLogger("sensor.utils.networking") + SIOCGIFADDR = 0x8915 SIOCGIFHWADDR = 0x8927 SIOCSIFHWADDR = 0x8924 @@ -64,6 +68,94 @@ def get_primary_interface() -> Optional[str]: return interfaces[0] if interfaces else None +def _is_excluded_interface(iface: str) -> bool: + """Check if interface should be excluded from detection.""" + excluded_prefixes = ("lo", "tailscale", "wg", "tun", "docker", "veth", "br-") + return iface.startswith(excluded_prefixes) + + +def _get_up_interface() -> Optional[str]: + """Get first UP interface that is not excluded.""" + try: + interfaces = get_all_interfaces() + for iface in interfaces: + if not _is_excluded_interface(iface): + if is_interface_up(iface): + return iface + except Exception as e: + logger.debug("Error checking interface status: %s", e) + return None + + +def detect_interface_with_retry( + max_retries: int = 3, + retry_delay: int = 5, + exponential: bool = True, + config_interface: Optional[str] = None, +) -> Optional[str]: + """Detect primary interface with retry fallback chain. + + Retry chain: + 1. Try get_primary_interface() which checks /proc/net/route + 2. Retry with exponential backoff if interface not available (delays: 5s, 10s, 20s) + 3. Fall back to first non-excluded UP interface + 4. Final fallback: use config_interface if all else fails + + Excluded interfaces: lo, tailscale*, wg*, tun*, docker*, veth*, br-* + + Args: + max_retries: Number of retry attempts (default 3) + retry_delay: Initial retry delay in seconds (default 5) + exponential: Use exponential backoff (default True, multiplies delay by 2 each retry) + config_interface: Interface from config to use as fallback + + Returns: + Interface name if found, otherwise None or config_interface if set. + """ + # Try direct detection first + iface = get_primary_interface() + if iface and not _is_excluded_interface(iface): + logger.debug("Detected interface from /proc/net/route: %s", iface) + return iface + + # Retry with exponential backoff + current_delay = retry_delay + for attempt in range(max_retries): + logger.debug( + "Interface detection retry %d/%d (delay=%ds)", + attempt + 1, + max_retries, + current_delay, + ) + time.sleep(current_delay) + + # Try detection again + iface = get_primary_interface() + if iface and not _is_excluded_interface(iface): + logger.info("Interface detected after retry %d: %s", attempt + 1, iface) + return iface + + # Try first UP interface + iface = _get_up_interface() + if iface: + logger.info("Fallback to UP interface after retry %d: %s", attempt + 1, iface) + return iface + + if exponential: + current_delay *= 2 + + # Final fallback to config-supplied interface + if config_interface: + logger.warning( + "Interface detection exhausted retries, using config fallback: %s", + config_interface, + ) + return config_interface + + logger.error("Interface detection failed after %d retries and no config fallback", max_retries) + return None + + def get_bridge_candidates() -> List[str]: candidates = [] for iface in get_all_interfaces():