mirror of
https://github.com/Horizontal42/netsleuth.git
synced 2026-11-10 16:52:03 +03:00
fd72604240
Every probe used to follow whatever the OS default route picked, with no way to point the tool at a specific adapter for comparison (VPN vs raw) or to confirm which one actually carries traffic on a multi-adapter machine. New iface.py resolves an adapter name or IP to a BindTarget (pure, tested): exact/case-insensitive/unique-substring name match, rejects APIPA-only, link-local-only and down adapters. Wired through every httpx provider client, the Ookla binary tier, ndt7, and the whole ping/traceroute cascade (icmplib, mtr, unix traceroute, TCP connect, scapy) plus --watch. Windows icmp_win now uses IcmpSendEcho2Ex (adds SourceAddress) only when a bind is forced, so the unforced path is unchanged. Windows tracert has no IPv4 source-address option (verified from its own usage text) so that tier is dropped when forced, noted in the report rather than silently measuring the wrong path. An unresolvable/down adapter aborts with every adapter listed; a resolvable adapter with no route to the internet degrades gracefully instead of aborting - that's a real diagnostic answer, not an error. netinfo's observation of the OS's actual default route stays untouched, so the report can show the forced adapter and the OS default side by side. Verified live: forcing Tailscale (no internet route on this box) triggered the preflight warning and degraded correctly; forcing Ethernet ran clean end-to-end including a full speedtest+MTR cycle. Also noted, not fixed: probes/traceroute.py's icmplib tier imports async_traceroute, which doesn't exist in icmplib 3.0.4 - that tier has always raised ImportError and silently fallen through. Found while verifying the icmplib source-binding wiring live.
149 lines
4.7 KiB
Python
149 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
import socket
|
|
|
|
from netsleuth.iface import available_interfaces_hint, resolve_bind_target, usable_addresses
|
|
|
|
|
|
def addrs(**kw) -> dict[str, list[tuple[int, str]]]:
|
|
return kw
|
|
|
|
|
|
def up(**kw) -> dict[str, bool]:
|
|
return kw
|
|
|
|
|
|
def test_usable_addresses_picks_first_non_loopback_v4_and_v6():
|
|
v4, v6 = usable_addresses(
|
|
[
|
|
(socket.AF_INET, "127.0.0.1"),
|
|
(socket.AF_INET, "192.168.1.34"),
|
|
(socket.AF_INET6, "::1"),
|
|
(socket.AF_INET6, "fe80::1"),
|
|
(socket.AF_INET6, "2001:db8::1"),
|
|
]
|
|
)
|
|
assert v4 == "192.168.1.34"
|
|
assert v6 == "2001:db8::1"
|
|
|
|
|
|
def test_usable_addresses_rejects_apipa_v4():
|
|
v4, v6 = usable_addresses([(socket.AF_INET, "169.254.1.2")])
|
|
assert v4 is None
|
|
assert v6 is None
|
|
|
|
|
|
def test_usable_addresses_rejects_link_local_v6():
|
|
v4, v6 = usable_addresses([(socket.AF_INET6, "fe80::1")])
|
|
assert v6 is None
|
|
|
|
|
|
def test_usable_addresses_accepts_link_local_only_when_nothing_else_present():
|
|
# A link-local-only adapter (typical for an interface with no DHCP/global
|
|
# IPv6) is a real, common case, not an error -- it just means ipv6=None.
|
|
v4, v6 = usable_addresses([(socket.AF_INET, "192.168.1.34"), (socket.AF_INET6, "fe80::1")])
|
|
assert v4 == "192.168.1.34"
|
|
assert v6 is None
|
|
|
|
|
|
def test_resolve_bind_target_by_exact_name():
|
|
result = resolve_bind_target(
|
|
"Ethernet",
|
|
addrs(Ethernet=[(socket.AF_INET, "192.168.3.72")], VPN=[(socket.AF_INET, "10.0.0.5")]),
|
|
up(Ethernet=True, VPN=True),
|
|
)
|
|
assert result.iface_name == "Ethernet"
|
|
assert result.ipv4 == "192.168.3.72"
|
|
assert result.error is None
|
|
|
|
|
|
def test_resolve_bind_target_by_ip_finds_owning_adapter():
|
|
result = resolve_bind_target(
|
|
"192.168.3.72",
|
|
addrs(Ethernet=[(socket.AF_INET, "192.168.3.72")]),
|
|
up(Ethernet=True),
|
|
)
|
|
assert result.iface_name == "Ethernet"
|
|
assert result.ipv4 == "192.168.3.72"
|
|
|
|
|
|
def test_resolve_bind_target_exact_name_beats_substring_match():
|
|
result = resolve_bind_target(
|
|
"Ethernet",
|
|
addrs(
|
|
Ethernet=[(socket.AF_INET, "192.168.3.72")],
|
|
**{"vEthernet (Default Switch)": [(socket.AF_INET, "172.20.0.1")]},
|
|
),
|
|
up(**{"Ethernet": True, "vEthernet (Default Switch)": True}),
|
|
)
|
|
assert result.iface_name == "Ethernet"
|
|
|
|
|
|
def test_resolve_bind_target_case_insensitive_exact_match():
|
|
result = resolve_bind_target(
|
|
"ethernet",
|
|
addrs(Ethernet=[(socket.AF_INET, "192.168.3.72")]),
|
|
up(Ethernet=True),
|
|
)
|
|
assert result.iface_name == "Ethernet"
|
|
|
|
|
|
def test_resolve_bind_target_unique_substring_match():
|
|
result = resolve_bind_target(
|
|
"radmin",
|
|
addrs(**{"Radmin VPN": [(socket.AF_INET, "26.0.0.5")]}),
|
|
up(**{"Radmin VPN": True}),
|
|
)
|
|
assert result.iface_name == "Radmin VPN"
|
|
|
|
|
|
def test_resolve_bind_target_ambiguous_substring_is_an_error():
|
|
result = resolve_bind_target(
|
|
"eth",
|
|
addrs(Ethernet=[(socket.AF_INET, "192.168.3.72")], **{"Ethernet 2": [(socket.AF_INET, "10.0.0.9")]}),
|
|
up(Ethernet=True, **{"Ethernet 2": True}),
|
|
)
|
|
assert result.error is not None
|
|
assert "Ethernet" in result.error and "Ethernet 2" in result.error
|
|
|
|
|
|
def test_resolve_bind_target_unknown_name_is_an_error_listing_candidates():
|
|
result = resolve_bind_target("Nope", addrs(Ethernet=[(socket.AF_INET, "192.168.3.72")]), up(Ethernet=True))
|
|
assert result.error is not None
|
|
assert "Ethernet" in result.error
|
|
|
|
|
|
def test_resolve_bind_target_down_adapter_is_an_error():
|
|
result = resolve_bind_target(
|
|
"Ethernet", addrs(Ethernet=[(socket.AF_INET, "192.168.3.72")]), up(Ethernet=False)
|
|
)
|
|
assert result.error is not None
|
|
assert result.is_up is False
|
|
|
|
|
|
def test_resolve_bind_target_apipa_only_adapter_is_an_error():
|
|
result = resolve_bind_target(
|
|
"PdaNet", addrs(PdaNet=[(socket.AF_INET, "169.254.1.2")]), up(PdaNet=True)
|
|
)
|
|
assert result.error is not None
|
|
|
|
|
|
def test_resolve_bind_target_ipv6_comes_from_the_same_matched_adapter():
|
|
result = resolve_bind_target(
|
|
"Ethernet",
|
|
addrs(Ethernet=[(socket.AF_INET, "192.168.3.72"), (socket.AF_INET6, "2001:db8::1")]),
|
|
up(Ethernet=True),
|
|
)
|
|
assert result.ipv4 == "192.168.3.72"
|
|
assert result.ipv6 == "2001:db8::1"
|
|
|
|
|
|
def test_available_interfaces_hint_lists_every_adapter_with_its_state():
|
|
hint = available_interfaces_hint(
|
|
addrs(Ethernet=[(socket.AF_INET, "192.168.3.72")], PdaNet=[(socket.AF_INET, "169.254.1.2")]),
|
|
up(Ethernet=True, PdaNet=False),
|
|
)
|
|
assert "Ethernet" in hint
|
|
assert "192.168.3.72" in hint
|
|
assert "PdaNet" in hint
|
|
assert "down" in hint.lower()
|