icmp_win: decode ICMP_ECHO_REPLY buffers for both pointer widths

src/netcheck/probes/icmp_win.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+import struct
+from typing import NamedTuple
+
+IP_SUCCESS = 0
+IP_DEST_NET_UNREACHABLE = 11002
+IP_DEST_HOST_UNREACHABLE = 11003
+IP_REQ_TIMED_OUT = 11010
+IP_TTL_EXPIRED_TRANSIT = 11013
+
+_UNREACHABLE = {IP_DEST_NET_UNREACHABLE, IP_DEST_HOST_UNREACHABLE, 11001, 11004}
+_HEAD = struct.Struct("<4BIIHH")
+
+
+class IcmpReply(NamedTuple):
+    address: str | None
+    status: int
+    rtt_ms: float | None
+    ttl: int
+
+
+def classify_status(status: int) -> str:
+    if status == IP_SUCCESS:
+        return "ok"
+    if status == IP_TTL_EXPIRED_TRANSIT:
+        return "ttl_expired"
+    if status == IP_REQ_TIMED_OUT:
+        return "timeout"
+    if status in _UNREACHABLE:
+        return "unreachable"
+    return "error"
+
+
+def parse_echo_reply(buffer: bytes, pointer_size: int = 8) -> IcmpReply:
+    # Layout after the fixed head: the Data pointer is aligned to its own width,
+    # then IP_OPTION_INFORMATION starts with Ttl as its first byte.
+    head_size = _HEAD.size
+    padding = (-head_size) % pointer_size
+    ttl_offset = head_size + padding + pointer_size
+    if len(buffer) < ttl_offset + 1:
+        return IcmpReply(address=None, status=-1, rtt_ms=None, ttl=0)
+    a, b, c, d, status, rtt, _data_size, _reserved = _HEAD.unpack_from(buffer, 0)
+    ttl = buffer[ttl_offset]
+    if status in (IP_REQ_TIMED_OUT,) or (a, b, c, d) == (0, 0, 0, 0):
+        return IcmpReply(address=None, status=status, rtt_ms=None, ttl=ttl)
+    return IcmpReply(address=f"{a}.{b}.{c}.{d}", status=status, rtt_ms=float(rtt), ttl=ttl)

tests/test_icmp_win_parse.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+import struct
+
+import pytest
+
+from netcheck.probes.icmp_win import (
+    IP_DEST_HOST_UNREACHABLE,
+    IP_REQ_TIMED_OUT,
+    IP_SUCCESS,
+    IP_TTL_EXPIRED_TRANSIT,
+    classify_status,
+    parse_echo_reply,
+)
+
+
+def make_reply(address: str, status: int, rtt: int, ttl: int, pointer_size: int = 8) -> bytes:
+    # ICMP_ECHO_REPLY: IPAddr(4) Status(4) RoundTripTime(4) DataSize(2) Reserved(2)
+    # then a pointer (Data), then IP_OPTION_INFORMATION { Ttl Tos Flags OptionsSize }
+    # followed by another pointer. Both pointers are alignment-padded to their width.
+    packed = struct.pack("<4B", *(int(o) for o in address.split(".")))
+    head = packed + struct.pack("<IIHH", status, rtt, 32, 0)
+    head += b"\x00" * (pointer_size - (len(head) % pointer_size) if len(head) % pointer_size else 0)
+    head += b"\x00" * pointer_size
+    head += struct.pack("<4B", ttl, 0, 0, 0)
+    head += b"\x00" * (pointer_size - 4 if pointer_size > 4 else 0)
+    head += b"\x00" * pointer_size
+    return head
+
+
+def test_parses_a_successful_reply_on_64_bit():
+    reply = parse_echo_reply(make_reply("1.1.1.1", IP_SUCCESS, 12, 57), pointer_size=8)
+    assert reply.address == "1.1.1.1"
+    assert reply.status == IP_SUCCESS
+    assert reply.rtt_ms == 12.0
+    assert reply.ttl == 57
+
+
+def test_parses_a_successful_reply_on_32_bit():
+    reply = parse_echo_reply(make_reply("8.8.4.4", IP_SUCCESS, 9, 120, pointer_size=4), pointer_size=4)
+    assert reply.address == "8.8.4.4"
+    assert reply.rtt_ms == 9.0
+    assert reply.ttl == 120
+
+
+def test_address_bytes_are_read_in_network_order():
+    reply = parse_echo_reply(make_reply("192.168.1.34", IP_SUCCESS, 1, 64))
+    assert reply.address == "192.168.1.34"
+
+
+def test_a_ttl_expired_reply_keeps_the_intermediate_router_address():
+    reply = parse_echo_reply(make_reply("10.64.0.1", IP_TTL_EXPIRED_TRANSIT, 8, 253))
+    assert reply.address == "10.64.0.1"
+    assert reply.status == IP_TTL_EXPIRED_TRANSIT
+    assert reply.rtt_ms == 8.0
+
+
+def test_a_timed_out_reply_has_no_address_and_no_rtt():
+    reply = parse_echo_reply(make_reply("0.0.0.0", IP_REQ_TIMED_OUT, 0, 0))
+    assert reply.address is None
+    assert reply.rtt_ms is None
+    assert reply.status == IP_REQ_TIMED_OUT
+
+
+def test_a_truncated_buffer_is_reported_as_an_error_not_an_exception():
+    reply = parse_echo_reply(b"\x01\x02")
+    assert reply.address is None
+    assert reply.rtt_ms is None
+    assert reply.status == -1
+
+
+@pytest.mark.parametrize(
+    ("status", "expected"),
+    [
+        (IP_SUCCESS, "ok"),
+        (IP_TTL_EXPIRED_TRANSIT, "ttl_expired"),
+        (IP_REQ_TIMED_OUT, "timeout"),
+        (IP_DEST_HOST_UNREACHABLE, "unreachable"),
+        (-1, "error"),
+        (11050, "error"),
+    ],
+)
+def test_status_classification(status, expected):
+    assert classify_status(status) == expected