diff --git a/src/netcheck/traceparse.py b/src/netcheck/traceparse.py new file mode 100644 index 0000000..9837195 --- /dev/null +++ b/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\d+(?:\.\d+)?)\s*ms|(?P\*)") +_UNIX_HOSTIP_RE = re.compile(r"(?P[A-Za-z0-9._-]+)\s+\((?P[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 diff --git a/tests/fixtures/traceroute/linux/gnu_basic.txt b/tests/fixtures/traceroute/linux/gnu_basic.txt new file mode 100644 index 0000000..c982c78 --- /dev/null +++ b/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 diff --git a/tests/fixtures/traceroute/linux/gnu_maxhops.txt b/tests/fixtures/traceroute/linux/gnu_maxhops.txt new file mode 100644 index 0000000..82c7d99 --- /dev/null +++ b/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 * * * diff --git a/tests/fixtures/traceroute/linux/gnu_unreachable.txt b/tests/fixtures/traceroute/linux/gnu_unreachable.txt new file mode 100644 index 0000000..bfb272d --- /dev/null +++ b/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 diff --git a/tests/test_traceparse.py b/tests/test_traceparse.py new file mode 100644 index 0000000..d104789 --- /dev/null +++ b/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") == []