From b151a934b7874fb76288cf57590e3284054a832d Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Sun, 9 Aug 2026 02:31:25 +0300 Subject: [PATCH] traceroute: mtr to icmp_win to icmplib to system-binary cascade --- src/netcheck/probes/traceroute.py | 164 ++++++++++++++++++++++++++++++ tests/test_traceroute_cascade.py | 148 +++++++++++++++++++++++++++ 2 files changed, 312 insertions(+) create mode 100644 src/netcheck/probes/traceroute.py create mode 100644 tests/test_traceroute_cascade.py diff --git a/src/netcheck/probes/traceroute.py b/src/netcheck/probes/traceroute.py new file mode 100644 index 0000000..c718489 --- /dev/null +++ b/src/netcheck/probes/traceroute.py @@ -0,0 +1,164 @@ +from __future__ import annotations + +import asyncio +import json +import platform +import shutil +from typing import Awaitable, Callable + +from netcheck.models import Capabilities, TraceHop, TraceResult +from netcheck.probes.icmp_win import IcmpReply, classify_status +from netcheck.traceparse import build_trace_result, finalize_hop + + +def tier_order(caps: Capabilities) -> list[str]: + order: list[str] = [] + if caps.mtr_binary: + order.append("mtr_json") + if caps.icmp_win_api: + order.append("icmp_win") + elif caps.icmp_dgram or caps.icmp_raw: + order.append("icmplib") + if caps.traceroute_binary: + order.append("system_traceroute") + return order + + +async def run_cascade(tiers: list[tuple[str, Callable[[], Awaitable[TraceResult]]]]) -> TraceResult: + for _name, tier in tiers: + try: + result = await tier() + except asyncio.CancelledError: + raise + except BaseException: # noqa: BLE001 - a dead tier is data, the next tier gets its turn + continue + if result.hops: + return result + return TraceResult(backend="none", hops=[], completed=False) + + +def hops_from_win_replies(replies: list[tuple[int, IcmpReply]]) -> list[TraceHop]: + hops: list[TraceHop] = [] + for ttl, reply in replies: + kind = classify_status(reply.status) + rtt = reply.rtt_ms if kind in ("ok", "ttl_expired") else None + hops.append(finalize_hop(TraceHop(ttl=ttl, ip=reply.address, probes=[rtt]))) + return hops + + +async def _run(args: list[str], timeout: float) -> str: + process = await asyncio.create_subprocess_exec( + *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL + ) + try: + stdout, _ = await asyncio.wait_for(process.communicate(), timeout=timeout) + except asyncio.TimeoutError: + process.kill() + raise + return stdout.decode(_console_encoding(), errors="replace") + + +def _console_encoding() -> str: + # Windows console tools emit the OEM code page (cp866 on Russian Windows), + # which is why every parser here works on decoded text and never on bytes. + if platform.system() != "Windows": + return "utf-8" + import ctypes + + try: + return f"cp{ctypes.windll.kernel32.GetOEMCP()}" + except Exception: + return "utf-8" + + +async def _tier_mtr(target: str, binary: str, cycles: int, max_hops: int, timeout: float) -> TraceResult: + text = await _run( + [binary, "--json", "--report-cycles", str(cycles), "--max-ttl", str(max_hops), target], timeout + ) + payload = json.loads(text) + report = payload.get("report") or {} + hops = [] + for entry in report.get("hubs") or []: + hop = TraceHop( + ttl=entry.get("count", 0), + ip=None if entry.get("host") in ("???", None) else entry.get("host"), + probes=[], + loss_pct=float(entry.get("Loss%", 0.0)), + min_ms=entry.get("Best"), + avg_ms=entry.get("Avg"), + max_ms=entry.get("Wrst"), + jitter_ms=entry.get("StDev"), + ) + hops.append(hop) + return TraceResult( + target=target, + backend="mtr_json", + hops=hops, + cycles=cycles, + completed=bool(hops) and hops[-1].ip is not None, + ) + + +async def _tier_icmp_win(target: str, max_hops: int, timeout: float) -> TraceResult: + from netcheck.probes.icmp_win import trace_hops_win + + replies = await asyncio.to_thread(trace_hops_win, target, max_hops, int(timeout * 1000)) + hops = hops_from_win_replies(replies) + return TraceResult( + target=target, + backend="icmp_win", + hops=hops, + cycles=1, + completed=bool(hops) and classify_status(replies[-1][1].status) == "ok", + max_hops_reached=bool(hops) and hops[-1].ttl >= max_hops, + ) + + +async def _tier_icmplib(target: str, max_hops: int, timeout: float, privileged: bool) -> TraceResult: + from icmplib import async_traceroute + + raw = await async_traceroute(target, max_hops=max_hops, timeout=timeout, privileged=privileged) + hops = [ + finalize_hop(TraceHop(ttl=h.distance, ip=h.address, probes=list(h.rtts))) + for h in raw + ] + return TraceResult( + target=target, + backend="icmplib", + hops=hops, + cycles=1, + completed=bool(hops) and hops[-1].ip is not None, + max_hops_reached=bool(hops) and hops[-1].ttl >= max_hops, + ) + + +async def _tier_system(target: str, binary: str, max_hops: int, timeout: float) -> TraceResult: + os_name = platform.system() + args = ( + [binary, "-h", str(max_hops), "-w", "2", target] + if os_name == "Windows" + else [binary, "-m", str(max_hops), "-w", "2", target] + ) + text = await _run(args, timeout) + return build_trace_result(text, os_name, target=target, resolved_ip=None, max_hops=max_hops) + + +async def traceroute( + target: str, + caps: Capabilities, + max_hops: int, + cycles: int, + timeout: float, + semaphore: asyncio.Semaphore | None = None, +) -> TraceResult: + builders = { + "mtr_json": lambda: _tier_mtr(target, caps.mtr_binary or shutil.which("mtr") or "mtr", cycles, max_hops, timeout), + "icmp_win": lambda: _tier_icmp_win(target, max_hops, timeout), + "icmplib": lambda: _tier_icmplib(target, max_hops, timeout, privileged=not caps.icmp_dgram), + "system_traceroute": lambda: _tier_system(target, caps.traceroute_binary or "traceroute", max_hops, timeout), + } + tiers = [(name, builders[name]) for name in tier_order(caps)] + if semaphore is None: + return await run_cascade(tiers) + async with semaphore: + return await run_cascade(tiers) diff --git a/tests/test_traceroute_cascade.py b/tests/test_traceroute_cascade.py new file mode 100644 index 0000000..75ffbd1 --- /dev/null +++ b/tests/test_traceroute_cascade.py @@ -0,0 +1,148 @@ +from __future__ import annotations + +import asyncio + +import pytest + +from netcheck.models import Capabilities, TraceHop, TraceResult +from netcheck.probes.icmp_win import IP_REQ_TIMED_OUT, IP_SUCCESS, IP_TTL_EXPIRED_TRANSIT, IcmpReply +from netcheck.probes.traceroute import hops_from_win_replies, run_cascade, tier_order + + +def caps(**kw) -> Capabilities: + base = dict(os_name="Linux", icmp_dgram=False, icmp_raw=False, icmp_win_api=False) + base.update(kw) + return Capabilities(**base) + + +def test_tier_order_puts_mtr_first_when_available(): + order = tier_order(caps(mtr_binary="/usr/bin/mtr", icmp_dgram=True, traceroute_binary="/usr/bin/traceroute")) + assert order == ["mtr_json", "icmplib", "system_traceroute"] + + +def test_tier_order_on_windows_uses_the_win_api_before_the_binary(): + order = tier_order(caps(os_name="Windows", icmp_win_api=True, traceroute_binary="C:\\tracert.exe")) + assert order == ["icmp_win", "system_traceroute"] + + +def test_tier_order_on_windows_still_prefers_mtr_if_someone_installed_it(): + order = tier_order(caps(os_name="Windows", mtr_binary="C:\\mtr.exe", icmp_win_api=True)) + assert order[0] == "mtr_json" + assert order[1] == "icmp_win" + + +def test_tier_order_without_any_icmp_falls_back_to_the_system_binary(): + assert tier_order(caps(traceroute_binary="/usr/bin/traceroute")) == ["system_traceroute"] + + +def test_tier_order_with_nothing_available_is_empty(): + assert tier_order(caps()) == [] + + +def good(name: str) -> TraceResult: + return TraceResult(target="1.1.1.1", backend=name, hops=[TraceHop(ttl=1, ip="192.168.1.1")], completed=True) + + +async def test_cascade_uses_the_first_tier_that_works(): + calls: list[str] = [] + + async def tier1() -> TraceResult: + calls.append("t1") + return good("mtr_json") + + async def tier2() -> TraceResult: + calls.append("t2") + return good("icmplib") + + result = await run_cascade([("mtr_json", tier1), ("icmplib", tier2)]) + assert result.backend == "mtr_json" + assert calls == ["t1"] + + +async def test_cascade_falls_through_when_a_tier_raises(): + calls: list[str] = [] + + async def tier1() -> TraceResult: + calls.append("t1") + raise FileNotFoundError("mtr not on PATH") + + async def tier2() -> TraceResult: + calls.append("t2") + return good("icmplib") + + result = await run_cascade([("mtr_json", tier1), ("icmplib", tier2)]) + assert result.backend == "icmplib" + assert calls == ["t1", "t2"] + + +async def test_cascade_falls_through_when_a_tier_returns_no_hops(): + async def tier1() -> TraceResult: + return TraceResult(target="1.1.1.1", backend="mtr_json", hops=[]) + + async def tier2() -> TraceResult: + return good("system_traceroute") + + result = await run_cascade([("mtr_json", tier1), ("system_traceroute", tier2)]) + assert result.backend == "system_traceroute" + + +async def test_cascade_tries_every_tier_in_order(): + calls: list[str] = [] + + def failing(name: str): + async def tier() -> TraceResult: + calls.append(name) + raise OSError(name) + + return tier + + async def last() -> TraceResult: + calls.append("system") + return good("system_traceroute") + + result = await run_cascade( + [("mtr_json", failing("mtr")), ("icmp_win", failing("win")), ("icmplib", failing("lib")), ("system_traceroute", last)] + ) + assert calls == ["mtr", "win", "lib", "system"] + assert result.backend == "system_traceroute" + + +async def test_cascade_exhaustion_yields_a_none_backend_not_an_exception(): + async def boom() -> TraceResult: + raise OSError("nope") + + result = await run_cascade([("mtr_json", boom), ("icmplib", boom)]) + assert isinstance(result, TraceResult) + assert result.backend == "none" + assert result.hops == [] + assert result.completed is False + + +async def test_cascade_with_no_tiers_at_all_yields_a_none_backend(): + result = await run_cascade([]) + assert result.backend == "none" + + +async def test_cascade_never_swallows_cancellation(): + async def cancelled() -> TraceResult: + raise asyncio.CancelledError() + + with pytest.raises(asyncio.CancelledError): + await run_cascade([("mtr_json", cancelled)]) + + +def test_win_replies_become_hops_with_ttl_expiry_handled(): + replies = [ + (1, IcmpReply(address="192.168.1.1", status=IP_TTL_EXPIRED_TRANSIT, rtt_ms=1.0, ttl=64)), + (2, IcmpReply(address=None, status=IP_REQ_TIMED_OUT, rtt_ms=None, ttl=0)), + (3, IcmpReply(address="1.1.1.1", status=IP_SUCCESS, rtt_ms=12.0, ttl=57)), + ] + hops = hops_from_win_replies(replies) + assert [h.ttl for h in hops] == [1, 2, 3] + assert hops[0].ip == "192.168.1.1" + assert hops[0].probes == [1.0] + assert hops[1].ip is None + assert hops[1].probes == [None] + assert hops[1].loss_pct == 100.0 + assert hops[2].ip == "1.1.1.1" + assert hops[2].avg_ms == 12.0