diff --git a/src/netcheck/netinfo.py b/src/netcheck/netinfo.py new file mode 100644 index 0000000..7067aac --- /dev/null +++ b/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 diff --git a/tests/test_netinfo.py b/tests/test_netinfo.py new file mode 100644 index 0000000..4f35274 --- /dev/null +++ b/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