mirror of
https://github.com/Horizontal42/netsleuth.git
synced 2026-11-10 15:42:10 +03:00
bc8482cc99
Stayed unprivileged on both platforms deliberately. Windows: echo_once() gained a df: bool param setting IP_FLAG_DF on the IP_OPTION_INFORMATION struct's Flags field (present but unused until now), and classify_status() learned IP_PACKET_TOO_BIG (11009) - the same unprivileged IcmpSendEcho2 API the rest of the ICMP tier already uses, no new capability needed. Unix deliberately does not hand-roll raw ICMP echo requests - unprivileged SOCK_DGRAM/IPPROTO_ICMP checksum and identifier handling varies enough across kernel versions that shipping it without a Linux/macOS box to verify on felt like the wrong risk to take. Shells out to system ping -M do -s <size> (Linux) / ping -D -s <size> (macOS/BSD) instead, the same subprocess-and-parse pattern traceroute.py's system tiers already use. Matches "frag"/"too long" loosely in the output rather than an exact phrase, since a locally-cached-PMTU failure and a genuine on-path ICMP response word themselves differently across ping implementations - same "parse by shape" principle traceparse.py already established. classify_pmtu() distinguishes a path that's just narrower (a frag-needed signal was seen - PMTUD is working, normal behind a VPN) from an actual blackhole (large packets vanish with no ICMP reply at all - the "SSH connects then hangs" symptom). path.pmtu_below_iface_mtu flags separately when the path MTU is narrower than the local NIC's own MTU, which is exactly what makes this class of problem invisible locally. Opt-in --pmtud, excluded from --full. Live-verified on Windows against real cloudflare.com/google.com - found the full 1500-byte MTU in 11 probes each, matching the expected binary-search depth, correct probe history in the JSON output. Part of plans/zesty-zooming-wilkinson.md (gitignored).
86 lines
2.8 KiB
Python
86 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import struct
|
|
|
|
import pytest
|
|
|
|
from netsleuth.probes.icmp_win import (
|
|
IP_DEST_HOST_UNREACHABLE,
|
|
IP_PACKET_TOO_BIG,
|
|
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"),
|
|
(IP_PACKET_TOO_BIG, "packet_too_big"),
|
|
(-1, "error"),
|
|
(11050, "error"),
|
|
],
|
|
)
|
|
def test_status_classification(status, expected):
|
|
assert classify_status(status) == expected
|