Robust interface detection — retry fallback chain for headless boot (#206)

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

@@ -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)

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:

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")

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

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 (

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():