netinfo: attempt-based capability detection and backend choice

src/netcheck/netinfo.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import ctypes
+import os
+import platform
+import shutil
+import socket
+
+from netcheck.models import Capabilities
+
+_UNIX_REMEDY = (
+    "ICMP is unavailable without privileges, so latency was measured by TCP connect timing "
+    "(a different metric: 'loss' means connection failures, not dropped packets). "
+    "Remedy: sysctl -w net.ipv4.ping_group_range='0 2147483647'"
+)
+_WINDOWS_REMEDY = (
+    "ICMP is unavailable, so latency was measured by TCP connect timing "
+    "(a different metric: 'loss' means connection failures, not dropped packets). "
+    "Remedy: the Iphlpapi.dll IcmpSendEcho2 API is normally always present; a host firewall "
+    "or security product is blocking it."
+)
+
+
+def choose_latency_backend(caps: Capabilities) -> str:
+    if caps.icmp_win_api:
+        return "icmp_win"
+    if caps.icmp_dgram:
+        return "icmp_dgram"
+    if caps.icmp_raw:
+        return "icmp_raw"
+    return "tcp"
+
+
+def choose_trace_backend(caps: Capabilities) -> str:
+    if caps.mtr_binary:
+        return "mtr_json"
+    if caps.icmp_win_api:
+        return "icmp_win"
+    if caps.icmp_dgram or caps.icmp_raw:
+        return "icmplib"
+    if caps.traceroute_binary:
+        return "system_traceroute"
+    return "none"
+
+
+def degradation_note(caps: Capabilities) -> str | None:
+    if choose_latency_backend(caps) != "tcp":
+        return None
+    return _WINDOWS_REMEDY if caps.os_name == "Windows" else _UNIX_REMEDY
+
+
+def _is_elevated() -> bool:
+    if platform.system() == "Windows":
+        try:
+            return bool(ctypes.windll.shell32.IsUserAnAdmin())
+        except Exception:
+            return False
+    return hasattr(os, "geteuid") and os.geteuid() == 0
+
+
+def _socket_works(family: int, sock_type: int, proto: int) -> bool:
+    try:
+        sock = socket.socket(family, sock_type, proto)
+    except (OSError, AttributeError, PermissionError):
+        return False
+    sock.close()
+    return True
+
+
+def _win_icmp_available() -> bool:
+    if platform.system() != "Windows":
+        return False
+    try:
+        ctypes.WinDLL("Iphlpapi.dll")
+    except OSError:
+        return False
+    return True
+
+
+def detect_capabilities() -> Capabilities:
+    os_name = platform.system()
+    caps = Capabilities(
+        os_name=os_name,
+        is_elevated=_is_elevated(),
+        icmp_dgram=_socket_works(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_ICMP),
+        icmp_raw=_socket_works(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP),
+        icmp_win_api=_win_icmp_available(),
+        mtr_binary=shutil.which("mtr"),
+        traceroute_binary=shutil.which("tracert") if os_name == "Windows" else shutil.which("traceroute"),
+    )
+    caps.chosen_latency_backend = choose_latency_backend(caps)
+    caps.chosen_trace_backend = choose_trace_backend(caps)
+    note = degradation_note(caps)
+    if note:
+        caps.notes.append(note)
+    return caps

tests/test_netinfo.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from netcheck.models import Capabilities
+from netcheck.netinfo import choose_latency_backend, choose_trace_backend, degradation_note
+
+
+def caps(**kw) -> Capabilities:
+    base = dict(os_name="Linux", is_elevated=False, icmp_dgram=False, icmp_raw=False, icmp_win_api=False)
+    base.update(kw)
+    return Capabilities(**base)
+
+
+def test_windows_prefers_the_win32_icmp_api_over_everything():
+    c = caps(os_name="Windows", icmp_win_api=True, icmp_dgram=True, icmp_raw=True)
+    assert choose_latency_backend(c) == "icmp_win"
+
+
+def test_unix_prefers_unprivileged_datagram_icmp():
+    assert choose_latency_backend(caps(icmp_dgram=True, icmp_raw=True)) == "icmp_dgram"
+
+
+def test_unix_falls_back_to_raw_icmp_when_datagram_is_unavailable():
+    assert choose_latency_backend(caps(icmp_raw=True, is_elevated=True)) == "icmp_raw"
+
+
+def test_latency_falls_back_to_tcp_when_no_icmp_is_possible():
+    assert choose_latency_backend(caps()) == "tcp"
+
+
+def test_trace_backend_prefers_mtr_when_the_binary_exists():
+    c = caps(mtr_binary="/usr/bin/mtr", icmp_dgram=True, traceroute_binary="/usr/bin/traceroute")
+    assert choose_trace_backend(c) == "mtr_json"
+
+
+def test_trace_backend_uses_win_api_before_the_system_binary():
+    c = caps(os_name="Windows", icmp_win_api=True, traceroute_binary="C:\\Windows\\System32\\TRACERT.EXE")
+    assert choose_trace_backend(c) == "icmp_win"
+
+
+def test_trace_backend_uses_icmplib_on_unix_before_the_system_binary():
+    c = caps(icmp_dgram=True, traceroute_binary="/usr/bin/traceroute")
+    assert choose_trace_backend(c) == "icmplib"
+
+
+def test_trace_backend_falls_back_to_the_system_binary():
+    assert choose_trace_backend(caps(traceroute_binary="/usr/bin/traceroute")) == "system_traceroute"
+
+
+def test_trace_backend_reports_none_when_nothing_is_available():
+    assert choose_trace_backend(caps()) == "none"
+
+
+def test_degradation_note_is_absent_when_icmp_works():
+    assert degradation_note(caps(icmp_dgram=True, traceroute_binary="/usr/bin/traceroute")) is None
+
+
+def test_degradation_note_explains_the_tcp_fallback_with_a_remedy():
+    note = degradation_note(caps())
+    assert note is not None
+    assert "TCP" in note
+    assert "ping_group_range" in note
+
+
+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