Horizontal42/netsleuth
1
0
Fork 0
mirror of https://github.com/Horizontal42/netsleuth.git synced 2026-11-10 15:42:10 +03:00
Code Issues Projects Releases Packages Wiki Activity Actions

Refactor echo_once to accept a single IcmpRequest dataclass parameter instead of multiple arguments, improving readability and maintainability. Updated all internal callers in icmp_win.py and pmtud.py, as well as corresponding mock implementations in the test suite.

Browse source 
Co-authored-by: Horizontal42 <75383519+Horizontal42@users.noreply.github.com>
This commit is contained in:
google-labs-jules[bot] 2026-09-10 08:44:54 +00:00
parent 8c99ac78b7
commit d144d75385
3 changed files with 48 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
src/netsleuth/probes/icmp_win.py
View file

@ -1,6 +1,7 @@
from __future__ import annotations
import struct
from dataclasses import dataclass
from typing import NamedTuple
IP_SUCCESS = 0
@ -22,6 +23,16 @@ class IcmpReply(NamedTuple):
ttl: int
@dataclass
class IcmpRequest:
dest: str
ttl: int
timeout_ms: int
payload: bytes = b"netsleuth"
source_ip: str | None = None
df: bool = False
def classify_status(status: int) -> str:
if status == IP_SUCCESS:
return "ok"
@ -128,23 +139,16 @@ def _handle():
return iphlpapi, handle
def echo_once(
dest: str,
ttl: int,
timeout_ms: int,
payload: bytes = b"netsleuth",
source_ip: str | None = None,
df: bool = False,
) -> IcmpReply:
def echo_once(req: IcmpRequest) -> IcmpReply:
iphlpapi, handle = _handle()
try:
options = _IpOptionInformation(
Ttl=ttl, Tos=0, Flags=IP_FLAG_DF if df else 0, OptionsSize=0, OptionsData=None
Ttl=req.ttl, Tos=0, Flags=IP_FLAG_DF if req.df else 0, OptionsSize=0, OptionsData=None
)
buffer = ctypes.create_string_buffer(_REPLY_BUFFER_SIZE)
address = struct.unpack("<I", socket.inet_aton(dest))[0]
if source_ip:
source = struct.unpack("<I", socket.inet_aton(source_ip))[0]
address = struct.unpack("<I", socket.inet_aton(req.dest))[0]
if req.source_ip:
source = struct.unpack("<I", socket.inet_aton(req.source_ip))[0]
count = iphlpapi.IcmpSendEcho2Ex(
handle,
None,
@ -152,12 +156,12 @@ def echo_once(
None,
ctypes.c_uint32(source),
ctypes.c_uint32(address),
payload,
ctypes.c_ushort(len(payload)),
req.payload,
ctypes.c_ushort(len(req.payload)),
ctypes.byref(options),
buffer,
ctypes.c_uint32(_REPLY_BUFFER_SIZE),
ctypes.c_uint32(timeout_ms),
ctypes.c_uint32(req.timeout_ms),
)
else:
count = iphlpapi.IcmpSendEcho2(
@ -166,12 +170,12 @@ def echo_once(
None,
None,
ctypes.c_uint32(address),
payload,
ctypes.c_ushort(len(payload)),
req.payload,
ctypes.c_ushort(len(req.payload)),
ctypes.byref(options),
buffer,
ctypes.c_uint32(_REPLY_BUFFER_SIZE),
ctypes.c_uint32(timeout_ms),
ctypes.c_uint32(req.timeout_ms),
)
if count == 0:
return IcmpReply(address=None, status=IP_REQ_TIMED_OUT, rtt_ms=None, ttl=0)
@ -189,7 +193,9 @@ def ping_samples_win(
for index in range(count):
if index:
time.sleep(interval)
reply = echo_once(dest, ttl=128, timeout_ms=timeout_ms, source_ip=source_ip)
reply = echo_once(
IcmpRequest(dest=dest, ttl=128, timeout_ms=timeout_ms, source_ip=source_ip)
)
samples.append(reply.rtt_ms if classify_status(reply.status) == "ok" else None)
return samples
@ -212,7 +218,9 @@ def trace_hops_win(
for ttl in range(1, max_hops + 1):
if time.monotonic() >= deadline:
break
reply = echo_once(address, ttl=ttl, timeout_ms=per_hop_timeout_ms, source_ip=source_ip)
reply = echo_once(
IcmpRequest(dest=address, ttl=ttl, timeout_ms=per_hop_timeout_ms, source_ip=source_ip)
)
hops.append((ttl, reply))
if classify_status(reply.status) == "ok":
break

13
src/netsleuth/probes/pmtud.py
View file

@ -69,13 +69,22 @@ def supports_df() -> tuple[bool, str]:
async def _probe_windows(
resolved_ip: str, size: int, timeout: float, source_ip: str | None
) -> tuple[bool, bool]:
from netsleuth.probes.icmp_win import classify_status, echo_once
from netsleuth.probes.icmp_win import IcmpRequest, classify_status, echo_once
payload = b"\x00" * max(0, size - _IPV4_ICMP_OVERHEAD)
timeout_ms = int(timeout * 1000)
def _run():
return echo_once(resolved_ip, ttl=64, timeout_ms=timeout_ms, payload=payload, source_ip=source_ip, df=True)
return echo_once(
IcmpRequest(
dest=resolved_ip,
ttl=64,
timeout_ms=timeout_ms,
payload=payload,
source_ip=source_ip,
df=True,
)
)
reply = await asyncio.to_thread(_run)
kind = classify_status(reply.status)

18
tests/test_icmp_win_trace.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
import time
from unittest.mock import patch
from netsleuth.probes.icmp_win import IP_REQ_TIMED_OUT, IcmpReply, trace_hops_win
from netsleuth.probes.icmp_win import IP_REQ_TIMED_OUT, IcmpReply, IcmpRequest, trace_hops_win
def _fake_gethostbyname(host: str) -> str:
@ -19,9 +19,9 @@ def test_trace_hops_win_total_elapsed_is_bounded_by_the_total_budget_not_hops_ti
# to the 100ms total budget.
call_timeouts: list[int] = []
def fake_echo_once(dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netsleuth", source_ip=None) -> IcmpReply:
call_timeouts.append(timeout_ms)
time.sleep(timeout_ms / 1000.0)
def fake_echo_once(req: IcmpRequest) -> IcmpReply:
call_timeouts.append(req.timeout_ms)
time.sleep(req.timeout_ms / 1000.0)
return IcmpReply(address=None, status=IP_REQ_TIMED_OUT, rtt_ms=None, ttl=0)
with (
@ -43,8 +43,8 @@ def test_trace_hops_win_total_elapsed_is_bounded_by_the_total_budget_not_hops_ti
def test_trace_hops_win_per_hop_timeout_is_capped_even_for_a_huge_total_budget():
call_timeouts: list[int] = []
def fake_echo_once(dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netsleuth", source_ip=None) -> IcmpReply:
call_timeouts.append(timeout_ms)
def fake_echo_once(req: IcmpRequest) -> IcmpReply:
call_timeouts.append(req.timeout_ms)
return IcmpReply(address=None, status=IP_REQ_TIMED_OUT, rtt_ms=None, ttl=0)
with (
@ -62,9 +62,9 @@ def test_trace_hops_win_per_hop_timeout_is_capped_even_for_a_huge_total_budget()
def test_trace_hops_win_happy_path_hop_count_and_ok_break_are_unchanged():
from netsleuth.probes.icmp_win import IP_SUCCESS
def fake_echo_once(dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netsleuth", source_ip=None) -> IcmpReply:
if ttl < 3:
return IcmpReply(address=f"10.0.0.{ttl}", status=IP_REQ_TIMED_OUT, rtt_ms=None, ttl=0)
def fake_echo_once(req: IcmpRequest) -> IcmpReply:
if req.ttl < 3:
return IcmpReply(address=f"10.0.0.{req.ttl}", status=IP_REQ_TIMED_OUT, rtt_ms=None, ttl=0)
return IcmpReply(address="203.0.113.1", status=IP_SUCCESS, rtt_ms=5.0, ttl=64)
with (