From 4c58f8add86a83efea9de50a14d26f9c9e7fd03d Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Sun, 9 Aug 2026 02:26:09 +0300 Subject: [PATCH] icmp_win: decode ICMP_ECHO_REPLY buffers for both pointer widths --- src/netcheck/probes/icmp_win.py | 47 ++++++++++++++++++ tests/test_icmp_win_parse.py | 84 +++++++++++++++++++++++++++++++++ 2 files changed, 131 insertions(+) create mode 100644 src/netcheck/probes/icmp_win.py create mode 100644 tests/test_icmp_win_parse.py diff --git a/src/netcheck/probes/icmp_win.py b/src/netcheck/probes/icmp_win.py new file mode 100644 index 0000000..f489266 --- /dev/null +++ b/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) diff --git a/tests/test_icmp_win_parse.py b/tests/test_icmp_win_parse.py new file mode 100644 index 0000000..52ebbf1 --- /dev/null +++ b/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(" 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