traceroute: mtr to icmp_win to icmplib to system-binary cascade

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)

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