netinfo: local interface, gateway, mtu and per-adapter resolvers

src/netcheck/netinfo.py

@@ -3,10 +3,21 @@ from __future__ import annotations
 import ctypes
 import os
 import platform
+import re
 import shutil
 import socket
+import subprocess
 
-from netcheck.models import Capabilities
+import psutil
+
+from netcheck.models import Capabilities, LocalNet
+
+_TUNNEL_PATTERNS = (
+    re.compile(r"^(tun|tap|utun|ppp|wg|nordlynx|proton|ipsec|gpd)\d*", re.IGNORECASE),
+    re.compile(r"wireguard", re.IGNORECASE),
+    re.compile(r"tap-windows", re.IGNORECASE),
+    re.compile(r"openvpn", re.IGNORECASE),
+)
 
 _UNIX_REMEDY = (
     "ICMP is unavailable without privileges, so latency was measured by TCP connect timing "
@@ -94,3 +105,156 @@ def detect_capabilities() -> Capabilities:
     if note:
         caps.notes.append(note)
     return caps
+
+
+def is_tunnel_iface(name: str) -> bool:
+    return any(pattern.search(name) for pattern in _TUNNEL_PATTERNS)
+
+
+def iface_for_ip(ip: str, addrs_by_iface: dict[str, list[tuple[int, str]]]) -> str | None:
+    for iface, addrs in addrs_by_iface.items():
+        for _family, address in addrs:
+            if address == ip:
+                return iface
+    return None
+
+
+def mtu_anomaly(mtu: int | None) -> str | None:
+    if mtu is None or mtu >= 1500:
+        return None
+    if 1405 <= mtu <= 1440:
+        return "wireguard"
+    if 1350 <= mtu <= 1404:
+        return "ipsec"
+    return "small"
+
+
+def primary_interface_ip(target: str = "1.1.1.1", family: int = socket.AF_INET) -> str | None:
+    # Connecting a UDP socket sends nothing; it only asks the kernel which local
+    # address the route to `target` would use. This is the reliable way to pick
+    # the *active* interface when several are up.
+    probe = "2606:4700:4700::1111" if family == socket.AF_INET6 else target
+    sock = socket.socket(family, socket.SOCK_DGRAM)
+    try:
+        sock.connect((probe, 53))
+        return sock.getsockname()[0]
+    except OSError:
+        return None
+    finally:
+        sock.close()
+
+
+def _resolvers_per_adapter() -> dict[str, list[str]]:
+    if platform.system() == "Windows":
+        return _resolvers_windows()
+    return _resolvers_unix()
+
+
+def _resolvers_windows() -> dict[str, list[str]]:
+    script = (
+        "Get-DnsClientServerAddress -AddressFamily IPv4,IPv6 | "
+        "ForEach-Object { $_.InterfaceAlias + '|' + ($_.ServerAddresses -join ',') }"
+    )
+    try:
+        out = subprocess.run(
+            ["powershell", "-NoProfile", "-Command", script],
+            capture_output=True,
+            text=True,
+            errors="replace",
+            timeout=15,
+        ).stdout
+    except (OSError, subprocess.SubprocessError):
+        return {}
+    adapters: dict[str, list[str]] = {}
+    for line in out.splitlines():
+        if "|" not in line:
+            continue
+        alias, _, servers = line.partition("|")
+        found = [s.strip() for s in servers.split(",") if s.strip()]
+        if found:
+            adapters.setdefault(alias.strip(), []).extend(found)
+    return adapters
+
+
+def _resolvers_unix() -> dict[str, list[str]]:
+    adapters: dict[str, list[str]] = {}
+    try:
+        text = open("/etc/resolv.conf", encoding="utf-8", errors="replace").read()
+    except OSError:
+        text = ""
+    servers = [line.split()[1] for line in text.splitlines() if line.startswith("nameserver") and len(line.split()) > 1]
+    if servers:
+        adapters["system"] = servers
+    try:
+        scutil = subprocess.run(
+            ["scutil", "--dns"], capture_output=True, text=True, errors="replace", timeout=15
+        ).stdout
+    except (OSError, subprocess.SubprocessError):
+        scutil = ""
+    current = None
+    for line in scutil.splitlines():
+        stripped = line.strip()
+        if stripped.startswith("if_index"):
+            current = stripped.split("(")[-1].rstrip(")") or None
+        elif stripped.startswith("nameserver[") and current:
+            adapters.setdefault(current, []).append(stripped.split(":", 1)[1].strip())
+    return adapters
+
+
+def collect_local_net() -> LocalNet:
+    v4 = primary_interface_ip()
+    v6 = primary_interface_ip(family=socket.AF_INET6)
+    addrs_by_iface = {
+        name: [(a.family, a.address.split("%")[0]) for a in addrs]
+        for name, addrs in psutil.net_if_addrs().items()
+    }
+    iface = iface_for_ip(v4, addrs_by_iface) if v4 else None
+    stats = psutil.net_if_stats()
+    return LocalNet(
+        iface_name=iface,
+        local_ipv4=v4,
+        local_ipv6=v6,
+        iface_mtu=stats[iface].mtu if iface and iface in stats else None,
+        default_gateway_v4=_default_gateway(socket.AF_INET),
+        default_gateway_v6=_default_gateway(socket.AF_INET6),
+        dns_servers_per_adapter=_resolvers_per_adapter(),
+        is_dual_stack=bool(v4 and v6),
+    )
+
+
+def _default_gateway(family: int) -> str | None:
+    if platform.system() == "Windows":
+        args = ["route", "print", "-6" if family == socket.AF_INET6 else "-4"]
+        needle = "::/0" if family == socket.AF_INET6 else "0.0.0.0"
+    else:
+        args = ["ip", "-6", "route", "show", "default"] if family == socket.AF_INET6 else ["ip", "route", "show", "default"]
+        needle = "via"
+    try:
+        out = subprocess.run(
+            args, capture_output=True, text=True, errors="replace", timeout=15
+        ).stdout
+    except (OSError, subprocess.SubprocessError):
+        return None
+    for line in out.splitlines():
+        if needle not in line:
+            continue
+        tokens = line.split()
+        if needle == "via" and "via" in tokens:
+            return tokens[tokens.index("via") + 1]
+        candidates = [t for t in tokens if _looks_like_ip(t)]
+        if len(candidates) >= 3:
+            return candidates[2]
+    return None
+
+
+def _looks_like_ip(token: str) -> bool:
+    try:
+        socket.inet_pton(socket.AF_INET, token)
+        return True
+    except OSError:
+        pass
+    try:
+        socket.inet_pton(socket.AF_INET6, token)
+        return True
+    except OSError:
+        return False

tests/test_netinfo.py

@@ -65,3 +65,49 @@ def test_degradation_note_on_windows_mentions_the_api_rather_than_sysctl():
     note = degradation_note(caps(os_name="Windows"))
     assert note is not None
     assert "Iphlpapi" in note
+
+
+import pytest
+
+from netcheck.netinfo import iface_for_ip, is_tunnel_iface, mtu_anomaly
+
+
+def test_iface_for_ip_matches_the_owning_adapter():
+    addrs = {
+        "lo": [(2, "127.0.0.1")],
+        "eth0": [(2, "192.168.1.34"), (23, "fe80::1")],
+        "wg0": [(2, "10.7.0.2")],
+    }
+    assert iface_for_ip("192.168.1.34", addrs) == "eth0"
+    assert iface_for_ip("10.7.0.2", addrs) == "wg0"
+    assert iface_for_ip("203.0.113.1", addrs) is None
+
+
+@pytest.mark.parametrize(
+    "name",
+    ["tun0", "tap0", "wg0", "utun3", "ppp0", "WireGuard Tunnel", "TAP-Windows Adapter V9", "nordlynx"],
+)
+def test_tunnel_interfaces_are_recognised(name):
+    assert is_tunnel_iface(name) is True
+
+
+@pytest.mark.parametrize("name", ["eth0", "en0", "Wi-Fi", "Ethernet 2", "lo", "Loopback Pseudo-Interface 1"])
+def test_ordinary_interfaces_are_not_flagged_as_tunnels(name):
+    assert is_tunnel_iface(name) is False
+
+
+def test_mtu_anomaly_recognises_wireguard_and_ipsec_sizes():
+    assert mtu_anomaly(1420) == "wireguard"
+    assert mtu_anomaly(1412) == "wireguard"
+    assert mtu_anomaly(1400) == "ipsec"
+    assert mtu_anomaly(1380) == "ipsec"
+
+
+def test_mtu_anomaly_ignores_normal_and_unknown_values():
+    assert mtu_anomaly(1500) is None
+    assert mtu_anomaly(9000) is None
+    assert mtu_anomaly(None) is None
+
+
+def test_mtu_anomaly_flags_unusually_small_links_generically():
+    assert mtu_anomaly(1200) == "small"