Network Device Discovery

Network Device Discovery Skill

Production-tested patterns for discovering, scanning, and managing network devices over HTTP/HTTPS. Extracted from GUDE Deploy — a commissioning tool for 120+ rack-mounted PDU devices.

---

Section 1: Interface Enumeration

Rule 1: netifaces2 leaks memory on Windows — parse ipconfig instead

The netifaces2 C extension calls GetAdaptersAddresses on Windows and leaks unboundedly (8-22 GB) on machines with AMD GPUs + multiple virtual adapters. It never returns.

Fix: Use subprocess.run(["ipconfig", "/all"]) parsing on Windows. Keep netifaces2 for Linux/macOS.

import platform

def get_network_interfaces() -> list[NetworkInterface]:
    if platform.system() == "Windows":
        return _parse_ipconfig()       # subprocess, no C extension
    else:
        try:
            return _get_interfaces_netifaces()
        except ImportError:
            return _fallback_socket()   # socket.gethostbyname

Rule 2: Adapters can have multiple IPv4 addresses

Windows adapters with temp aliases (cross-subnet scanning) or dual DHCP+static emit multiple "IPv4 Address" lines under the same adapter header. Emit one NetworkInterface per IP+mask pair.

def _parse_ipconfig() -> list[NetworkInterface]:
    # ...
    for line in output.splitlines():
        if "IPv4 Address" in stripped or "IP Address" in stripped:
            # Emit previous IP+mask before starting a new one
            _emit(current_name, current_ip, current_mask)
            current_ip = ""
            current_mask = ""
            parts = stripped.split(":", 1)
            if len(parts) == 2:
                current_ip = parts[1].strip().split("(")[0].strip()
        elif "Subnet Mask" in stripped:
            parts = stripped.split(":", 1)
            if len(parts) == 2:
                current_mask = parts[1].strip()

Rule 3: Use encoding="utf-8-sig" for BOM handling

Windows command output and Excel-generated CSVs may have a UTF-8 BOM. Always open with encoding="utf-8-sig" to strip it transparently. This applies to ipconfig /all output, CSV worklists, and any file that might originate from Windows.

Rule 4: Filter loopback, APIPA, and virtual adapters

Always exclude before presenting to users or scanning:

• Loopback: 127.x.x.x
• APIPA: 169.254.x.x (link-local, no real network)
• Virtual: Hyper-V, Tailscale, WSL, VPN adapters (context-dependent)

DISCOVERY_APIPA_NETWORK = "169.254.0.0/16"

def is_apipa_interface(interface: NetworkInterface) -> bool:
    ip = ipaddress.IPv4Address(interface.ip_address)
    return ip in ipaddress.IPv4Network(DISCOVERY_APIPA_NETWORK)

Rule 5: Sort interfaces by preference

Ethernet-like interfaces (eth, en) are preferred over Wi-Fi, VPN, or virtual adapters for device discovery. Sort before presenting in UI or auto-selecting.

def sort_key(iface: NetworkInterface) -> int:
    name = iface.name.lower()
    if name.startswith(("eth", "en")):
        return 0
    if "ethernet" in name:
        return 1
    return 2

non_loopback.sort(key=sort_key)

---

Section 2: Subnet Scanning

Rule 6: Support /23 and /22 subnets — don't hardcode /24

Production networks use larger subnets. Set HTTP_SCAN_MAX_HOSTS = 1022 to support up to /22. Always compute from CIDR, never assume 254 hosts.

HTTP_SCAN_MAX_HOSTS = 1022  # /22 = 1022, /23 = 510, /24 = 254

network = ipaddress.IPv4Network(f"{ip}/{mask}", strict=False)
num_hosts = network.num_addresses - 2  # exclude network + broadcast
if num_hosts > HTTP_SCAN_MAX_HOSTS:
    log.warning(f"Subnet {network} too large ({num_hosts} hosts), skipping")
    return []
ips = [str(ip) for ip in network.hosts()]

Rule 7: Use ThreadPoolExecutor with 50 workers for parallel probing

50 workers is the sweet spot for HTTP probing a /24 subnet. Below 30 is slow, above 80 risks socket exhaustion on Windows.

HTTP_SCAN_WORKERS = 50

with ThreadPoolExecutor(max_workers=HTTP_SCAN_WORKERS) as executor:
    future_to_ip = {
        executor.submit(_probe_ip, ip, timeout): ip for ip in ips
    }
    for future in as_completed(future_to_ip):
        device = future.result()
        if device is not None:
            devices.append(device)

Rule 8: Use 0.5s timeout for probing, 5s for health checks

Discovery probes dead IPs — most won't respond. Keep probe timeout aggressive (0.5s). Health checks talk to known-alive devices over potentially routed networks — use 5s.

HTTP_SCAN_TIMEOUT = 0.5    # per-IP probe during discovery
DEFAULT_HTTP_TIMEOUT = 10.0  # general API calls
# Health check overrides to 5.0s temporarily

Rule 9: Interface isolation — scan only the selected NIC's subnet

When user selects a specific interface, only scan that interface's subnet + user-supplied extra ranges. Factory default ranges (192.168.0.0/24, 192.168.1.0/24) are only included when scanning "all interfaces". The factory default IP probe (192.168.0.2) always runs.

specific_interface = interface is not None
all_ranges: list[str] = []

if not specific_interface:
    all_ranges.extend(DISCOVERY_SCAN_RANGES)  # factory defaults

# Always add the selected interface's subnet
for iface in interfaces:
    if is_apipa_interface(iface):
        continue
    iface_cidr = str(ipaddress.IPv4Network(
        f"{iface.ip_address}/{iface.netmask}", strict=False
    ))
    if iface_cidr not in all_ranges:
        all_ranges.append(iface_cidr)

Rule 10: Skip factory default ranges when specific interface selected

If a tech selects "Ethernet (172.26.46.1)", they don't want to scan 192.168.0.0/24. Only add factory defaults when scanning all interfaces.

---

Section 3: Device Discovery Strategies

Rule 11: Multi-strategy discovery — run GBL + HTTP + manual in parallel

Never rely on a single method. GBL broadcast finds factory-default devices, HTTP scan finds configured devices, manual connect handles known IPs. Run all in parallel with ThreadPoolExecutor and deduplicate by IP.

def discover_all(interface=None, on_log=None, cancel_check=None, extra_ranges=None):
    all_devices: dict[str, GudeDevice] = {}  # keyed by IP for dedup
    max_workers = min(8, 2 + len(all_ranges) + len(interfaces))

    with ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="discovery") as executor:
        futures = {}

        # Strategy 1: GBL broadcast per interface
        for iface in interfaces:
            futures[executor.submit(discover_gbl, interface=iface, ...)] = f"GBL on {iface.display_name}"

        # Strategy 2: Factory default IP probe
        futures[executor.submit(_probe_ip, "192.168.0.2", timeout)] = "Factory default"

        # Strategy 3: Subnet range scans
        for cidr in all_ranges:
            futures[executor.submit(probe_fixed_range, cidr, ...)] = f"Range {cidr}"

        for future in as_completed(futures):
            result = future.result()
            # Deduplicate by IP
            if isinstance(result, list):
                for device in result:
                    if device.ip_address not in all_devices:
                        all_devices[device.ip_address] = device
            elif result is not None:
                if result.ip_address not in all_devices:
                    all_devices[result.ip_address] = result

    return list(all_devices.values())

Rule 12: Raw UDP sockets for broadcast — no Scapy/Npcap dependency

Scapy and Npcap are heavy, fragile dependencies. Raw UDP sockets work on all platforms for broadcast discovery protocols. Send to the subnet's broadcast address on the protocol port.

with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
    sock.settimeout(timeout)
    sock.sendto(GBL_HEADER, (broadcast_addr, GBL_PORT))

    deadline = time.monotonic() + timeout
    while time.monotonic()  Device | None:
    # Try HTTP first
    try:
        resp = requests.get(f"http://{ip}/status", timeout=timeout)
        return parse_response(resp)
    except requests.Timeout:
        return None  # no host — skip HTTPS too
    except requests.ConnectionError:
        pass  # port 80 refused — try 443

    # Quick TCP check before expensive HTTPS
    try:
        with socket.create_connection((ip, 443), timeout=0.5):
            pass
    except (OSError, socket.timeout):
        return None  # port 443 also closed

    # Now try HTTPS (worth the overhead)
    try:
        resp = requests.get(f"https://{ip}/status", timeout=5.0, verify=False)
        return parse_response(resp)
    except (requests.ConnectionError, requests.Timeout):
        return None

Rule 15: Deduplicate results by IP across all strategies

A device found by both GBL and HTTP scan should appear once. Key by IP address, not MAC (some discovery methods don't return MAC).

all_devices: dict[str, Device] = {}  # keyed by IP
for device in results:
    if device.ip_address not in all_devices:
        all_devices[device.ip_address] = device

Rule 16: Separate quick probe from full info fetch

Discovery probes thousands of IPs — keep it minimal (one lightweight request). After discovery, enrich each found device with a detailed API call. This separates the "is anything there?" check from "what exactly is it?".

def enrich_device(device: Device, client: ApiClient | None = None) -> Device:
    own_client = False
    if client is None:
        client = ApiClient(host=device.ip_address, use_ssl=device.use_ssl)
        own_client = True
    try:
        _populate_device_info(device, client)
    finally:
        if own_client:
            client.close()
    return device

---

Section 4: HTTP API Client Patterns

Rule 17: One requests.Session per client — NOT shared across threads

requests.Session is not thread-safe. Each ApiClient instance creates its own session. In parallel operations, each worker thread creates and closes its own client.

class ApiClient:
    def __init__(self, host, port=80, use_ssl=False, ...):
        self._session = requests.Session()
        if use_ssl:
            self._session.verify = False
        if username:
            self._session.auth = (username, password)

Rule 18: Close sessions in finally blocks

Leaked sessions exhaust connection pools. Use context managers or explicit finally.

# Option 1: Context manager
with ApiClient(host=ip) as client:
    client.health_check()

# Option 2: Explicit finally (for enrichment patterns)
own_client = False
if client is None:
    client = ApiClient(host=device.ip_address, use_ssl=device.use_ssl)
    own_client = True
try:
    _populate_device_info(device, client)
finally:
    if own_client:
        client.close()

Rule 19: Disable SSL verification for self-signed certs

Network devices use self-signed certificates. Set verify=False on the session, not per-request, to avoid missing it on one call.

self._session.verify = False

Rule 20: Suppress urllib3 InsecureRequestWarning globally

Without suppression, every HTTPS request to a self-signed device prints a warning. Do this once at module level, not per-request.

import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

Rule 21: Health check with 5s timeout, not 2s

Routed networks and VPN tunnels add latency. 2s is too aggressive for health checks. Temporarily override the client timeout for health checks.

def health_check(self, timeout: float = 5.0) -> bool:
    saved_timeout = self.timeout
    self.timeout = timeout
    try:
        self._get_json("statusjsn.js", {"components": STATUS_HEALTH_CHECK})
        return True
    except (ConnectionError, AuthenticationError, ConfigLoadError):
        return False
    finally:
        self.timeout = saved_timeout

Rule 22: Handle 401 gracefully — devices may ship with no auth

Many network devices ship with auth disabled. When you get a 401, raise a specific AuthenticationError so the UI can prompt for credentials, not show a generic failure.

if resp.status_code == 401:
    raise AuthenticationError(url=url)

Rule 23: GET-based command API — params via requests library

Many device APIs use GET with query params for commands (e.g. ov.html?cmd=4&ip=x). Let the requests library encode params — don't manually build query strings.

def send_command(self, cmd: int, params: dict | None = None) -> str:
    all_params = {"cmd": cmd}
    if params:
        all_params.update(params)
    resp = self._get("ov.html", all_params)  # requests encodes safely

---

Section 5: Connection Monitoring

Rule 24: Poll at 10s intervals, not 5s

5s polling creates unnecessary load on devices with limited HTTP stacks. 10s is sufficient for connection monitoring during commissioning workflows.

CONNECTION_CHECK_INTERVAL = 10000  # milliseconds

Rule 25: Failure threshold before marking connection lost

Transient network hiccups cause single health check failures. Require 3 consecutive failures before declaring the connection lost.

CONNECTION_FAILURE_THRESHOLD = 3

# In the monitor:
if health_check_failed:
    self._consecutive_failures += 1
    if self._consecutive_failures >= CONNECTION_FAILURE_THRESHOLD:
        self._emit_connection_lost()
else:
    self._consecutive_failures = 0
    if was_disconnected:
        self._emit_connection_restored()

Rule 26: Auto-recover on next successful check

Don't require user intervention to reconnect. If a health check succeeds after failures, automatically restore the connection state.

Rule 27: Stop monitoring during deploy operations

Health check polls interfere with deploy operations (especially IP changes). Pause the monitor before deploy, resume after.

---

Section 6: Parallel Operations

Rule 28: Each worker creates its own ApiClient and closes it in finally

Never share ApiClient or requests.Session across threads. Create, use, close within each worker function.

def _deploy_one_device(device, config):
    client = ApiClient(host=device.ip_address, use_ssl=device.use_ssl)
    try:
        client.configure_ipv4(**config)
        return True
    except Exception as e:
        log.error(f"Deploy failed for {device.ip_address}: {e}")
        return False
    finally:
        client.close()

Rule 29: Stagger parallel operations (3s between firmware uploads)

Simultaneous firmware uploads to 120 devices on the same switch causes packet loss. Stagger with a small delay between submissions.

BATCH_FIRMWARE_STAGGER = 3.0  # seconds between upload starts

for i, device in enumerate(devices):
    if i > 0:
        time.sleep(BATCH_FIRMWARE_STAGGER)
    future = executor.submit(upload_firmware, device, firmware_data)
    futures[future] = device

Rule 30: Progress callbacks from as_completed(), never from pool threads

UI callbacks (Qt signals, tkinter after()) must run on the main thread. Collect results via as_completed() on the main thread, then emit progress.

for future in as_completed(futures):
    device = futures[future]
    try:
        result = future.result()
        on_progress(device, "success")  # main thread — safe for UI
    except Exception as e:
        on_progress(device, f"failed: {e}")

Rule 31: Stabilization sleep after firmware reboot

After firmware upload + reboot command, the device takes 30-90 seconds to come back. Wait before attempting config deployment or health checks.

FIRMWARE_REBOOT_INITIAL_SLEEP = 30  # seconds before first reconnect attempt
FIRMWARE_REBOOT_WAIT = 120.0        # total timeout for device to come back

---

Section 7: IP Management

Rule 32: IPv4 config ALWAYS deployed last

Changing a device's IP disconnects it immediately. Deploy all other config (hostname, SN

…

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: NerdBase-by-Stark
• Source: NerdBase-by-Stark/skill-forge
• License: MIT

Install and usage instructions live in the source repository linked above.