traceparse: gnu traceroute parser with per-hop stats

src/netcheck/traceparse.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import re
+
+from netcheck.models import TraceHop
+from netcheck.stats import rtt_stats
+
+IPV4_RE = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b")
+IPV6_RE = re.compile(r"\b(?:[0-9a-fA-F]{0,4}:){2,7}[0-9a-fA-F]{0,4}\b")
+ANNOTATION_RE = re.compile(r"!\w*")
+_HOP_LINE_RE = re.compile(r"^\s*(\d{1,3})\s+(.*)$")
+_UNIX_PROBE_RE = re.compile(r"(?P<rtt>\d+(?:\.\d+)?)\s*ms|(?P<star>\*)")
+_UNIX_HOSTIP_RE = re.compile(r"(?P<name>[A-Za-z0-9._-]+)\s+\((?P<ip>[0-9a-fA-F:.]+)\)")
+
+
+def finalize_hop(hop: TraceHop) -> TraceHop:
+    s = rtt_stats(hop.probes)
+    hop.loss_pct = s.loss_pct
+    hop.min_ms = s.min_ms
+    hop.avg_ms = s.avg_ms
+    hop.max_ms = s.max_ms
+    hop.jitter_ms = s.jitter_ms
+    return hop
+
+
+def _extract_ip(text: str) -> str | None:
+    v4 = IPV4_RE.search(text)
+    if v4:
+        return v4.group(0)
+    v6 = IPV6_RE.search(text)
+    return v6.group(0) if v6 else None
+
+
+def _unix_host_and_ip(body: str) -> tuple[str | None, str | None]:
+    match = _UNIX_HOSTIP_RE.search(body)
+    if match:
+        name, ip = match.group("name"), match.group("ip")
+        return (None if name == ip else name), ip
+    return None, _extract_ip(body)
+
+
+def _unix_probes(body: str) -> list[float | None]:
+    probes: list[float | None] = []
+    for match in _UNIX_PROBE_RE.finditer(body):
+        probes.append(None if match.group("star") else float(match.group("rtt")))
+    return probes
+
+
+def _unix_annotations(body: str) -> list[str]:
+    seen: list[str] = []
+    for token in ANNOTATION_RE.findall(body):
+        if token not in seen:
+            seen.append(token)
+    return seen
+
+
+def parse_linux(text: str) -> list[TraceHop]:
+    hops: list[TraceHop] = []
+    for line in text.splitlines():
+        match = _HOP_LINE_RE.match(line)
+        if not match:
+            continue
+        ttl, body = int(match.group(1)), match.group(2)
+        rdns, ip = _unix_host_and_ip(body)
+        hop = TraceHop(
+            ttl=ttl,
+            ip=ip,
+            reverse_dns=rdns,
+            probes=_unix_probes(body),
+            annotations=_unix_annotations(body),
+        )
+        hops.append(finalize_hop(hop))
+    return hops

tests/fixtures/traceroute/linux/gnu_basic.txt

@@ -0,0 +1,7 @@
+traceroute to 1.1.1.1 (1.1.1.1), 30 hops max, 60 byte packets
+ 1  _gateway (192.168.1.1)  1.234 ms  1.102 ms  1.045 ms
+ 2  10.64.0.1 (10.64.0.1)  8.421 ms  8.502 ms  8.377 ms
+ 3  * * *
+ 4  ae-1.r01.ams.example.net (77.88.1.5)  12.004 ms  11.883 ms  12.111 ms
+ 5  * 13.402 ms *
+ 6  one.one.one.one (1.1.1.1)  13.552 ms  13.401 ms  13.298 ms

tests/fixtures/traceroute/linux/gnu_maxhops.txt

@@ -0,0 +1,6 @@
+traceroute to 203.0.113.200 (203.0.113.200), 5 hops max, 60 byte packets
+ 1  _gateway (192.168.1.1)  1.100 ms  1.050 ms  1.030 ms
+ 2  * * *
+ 3  * * *
+ 4  * * *
+ 5  * * *

tests/fixtures/traceroute/linux/gnu_unreachable.txt

@@ -0,0 +1,4 @@
+traceroute to 203.0.113.9 (203.0.113.9), 30 hops max, 60 byte packets
+ 1  _gateway (192.168.1.1)  1.100 ms  1.050 ms  1.030 ms
+ 2  10.64.0.1 (10.64.0.1)  8.200 ms !N  8.300 ms !N  8.250 ms !N
+ 3  border.example.net (198.51.100.7)  15.100 ms !H  15.200 ms  15.050 ms

tests/test_traceparse.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import pytest
+
+from netcheck.traceparse import parse_linux
+
+
+@pytest.fixture()
+def basic(trace_fixture):
+    return parse_linux(trace_fixture("linux", "gnu_basic.txt"))
+
+
+def test_linux_hop_count_ignores_the_header_line(basic):
+    assert [hop.ttl for hop in basic] == [1, 2, 3, 4, 5, 6]
+
+
+def test_linux_extracts_reverse_dns_and_ip(basic):
+    assert basic[0].ip == "192.168.1.1"
+    assert basic[0].reverse_dns == "_gateway"
+    assert basic[3].ip == "77.88.1.5"
+    assert basic[3].reverse_dns == "ae-1.r01.ams.example.net"
+
+
+def test_linux_bare_ip_hop_has_no_reverse_dns(basic):
+    assert basic[1].ip == "10.64.0.1"
+    assert basic[1].reverse_dns is None
+
+
+def test_linux_probe_values_are_floats_in_order(basic):
+    assert basic[0].probes == [1.234, 1.102, 1.045]
+    assert basic[3].probes == [12.004, 11.883, 12.111]
+
+
+def test_linux_full_timeout_hop_has_three_none_probes_and_no_ip(basic):
+    hop = basic[2]
+    assert hop.probes == [None, None, None]
+    assert hop.ip is None
+    assert hop.loss_pct == 100.0
+    assert hop.avg_ms is None
+
+
+def test_linux_partial_timeout_hop_keeps_probe_positions(basic):
+    hop = basic[4]
+    assert hop.probes == [None, 13.402, None]
+    assert hop.loss_pct == pytest.approx(66.667)
+    assert hop.avg_ms == 13.402
+
+
+def test_linux_computes_per_hop_statistics(basic):
+    hop = basic[0]
+    assert hop.min_ms == 1.045
+    assert hop.max_ms == 1.234
+    assert hop.avg_ms == pytest.approx(1.127, abs=0.001)
+    assert hop.jitter_ms == pytest.approx(0.095, abs=0.001)
+
+
+def test_linux_captures_unreachable_annotations(trace_fixture):
+    hops = parse_linux(trace_fixture("linux", "gnu_unreachable.txt"))
+    assert hops[1].annotations == ["!N"]
+    assert hops[2].annotations == ["!H"]
+    assert hops[1].probes == [8.2, 8.3, 8.25]
+
+
+def test_linux_max_hops_run_yields_trailing_dead_hops(trace_fixture):
+    hops = parse_linux(trace_fixture("linux", "gnu_maxhops.txt"))
+    assert len(hops) == 5
+    assert all(h.ip is None for h in hops[1:])
+    assert all(h.loss_pct == 100.0 for h in hops[1:])
+
+
+def test_linux_parser_tolerates_empty_and_garbage_input():
+    assert parse_linux("") == []
+    assert parse_linux("bash: traceroute: command not found\n") == []