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 traceroute function to use TracerouteOptions dataclass

Browse source 
Introduced a new `TracerouteOptions` dataclass in `src/netsleuth/probes/traceroute.py` to encapsulate the configuration parameters for the `traceroute` function. Updated the function signature and implementation to use the new dataclass, improving readability and maintainability. Also updated all call sites in `src/netsleuth/cli.py` to correctly instantiate and pass the `TracerouteOptions` object. Validated the changes by running tests and linting.

Co-authored-by: Horizontal42 <75383519+Horizontal42@users.noreply.github.com>
This commit is contained in:
google-labs-jules[bot] 2026-10-07 03:57:28 +00:00
parent 8c99ac78b7
commit f7599ea941
2 changed files with 41 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
src/netsleuth/cli.py
View file

@ -71,7 +71,7 @@ from netsleuth.probes.dns_leak import collect_dns_leak
from netsleuth.probes.hop_asn import enrich_hops from netsleuth.probes.hop_asn import enrich_hops
from netsleuth.probes.latency import ping_fanout, tcp_connect_rtt from netsleuth.probes.latency import ping_fanout, tcp_connect_rtt
from netsleuth.probes.nat64 import detect_nat64 from netsleuth.probes.nat64 import detect_nat64
from netsleuth.probes.traceroute import filter_trace_tiers, tier_order, traceroute from netsleuth.probes.traceroute import TracerouteOptions, filter_trace_tiers, tier_order, traceroute
from netsleuth.speed import NDT7_CONSENT_NOTICE from netsleuth.speed import NDT7_CONSENT_NOTICE
if sys.platform == "win32": if sys.platform == "win32":
@ -476,13 +476,15 @@ async def _ecmp_section(caps, settings: Settings, options: Options, cycles: int)
*( *(
traceroute( traceroute(
host, host,
caps, TracerouteOptions(
max_hops=settings.probing.max_hops, caps=caps,
cycles=cycles, max_hops=settings.probing.max_hops,
timeout=settings.timeouts.subprocess_seconds, cycles=cycles,
semaphore=semaphore, timeout=settings.timeouts.subprocess_seconds,
tcp_trace=options.tcp_trace, semaphore=semaphore,
source_ip=options.bind.ipv4 if options.bind else None, tcp_trace=options.tcp_trace,
source_ip=options.bind.ipv4 if options.bind else None,
)
) )
for _ in range(cfg.runs) for _ in range(cfg.runs)
) )
@ -498,13 +500,15 @@ async def _traces(hosts, caps, settings: Settings, cycles: int, options: Options
*( *(
traceroute( traceroute(
host, host,
caps, TracerouteOptions(
max_hops=settings.probing.max_hops, caps=caps,
cycles=cycles, max_hops=settings.probing.max_hops,
timeout=settings.timeouts.subprocess_seconds, cycles=cycles,
semaphore=semaphore, timeout=settings.timeouts.subprocess_seconds,
tcp_trace=options.tcp_trace, semaphore=semaphore,
source_ip=options.bind.ipv4 if options.bind else None, tcp_trace=options.tcp_trace,
source_ip=options.bind.ipv4 if options.bind else None,
)
) )
for _label, host in hosts for _label, host in hosts
) )

38
src/netsleuth/probes/traceroute.py
View file

@ -7,6 +7,7 @@ import platform
import shutil import shutil
import threading import threading
from collections.abc import Awaitable, Callable from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import TypeVar from typing import TypeVar
from netsleuth.models import Capabilities, TraceConfig, TraceHop, TraceResult from netsleuth.models import Capabilities, TraceConfig, TraceHop, TraceResult
@ -16,6 +17,17 @@ from netsleuth.traceparse import build_trace_result, finalize_hop
_T = TypeVar("_T") _T = TypeVar("_T")
@dataclass(frozen=True)
class TracerouteOptions:
caps: Capabilities
max_hops: int
cycles: int
timeout: float
semaphore: asyncio.Semaphore | None = None
tcp_trace: bool = False
source_ip: str | None = None
def _run_in_daemon_thread(func: Callable[..., _T], *args: object) -> asyncio.Future[_T]: def _run_in_daemon_thread(func: Callable[..., _T], *args: object) -> asyncio.Future[_T]:
# asyncio.to_thread()/loop.run_in_executor(None, ...) queue work onto the # asyncio.to_thread()/loop.run_in_executor(None, ...) queue work onto the
# ThreadPoolExecutor-backed default executor, whose worker threads are non-daemon # ThreadPoolExecutor-backed default executor, whose worker threads are non-daemon
@ -256,29 +268,23 @@ async def _tier_tcp_trace(
async def traceroute( async def traceroute(
target: str, target: str,
caps: Capabilities, options: TracerouteOptions,
max_hops: int,
cycles: int,
timeout: float,
semaphore: asyncio.Semaphore | None = None,
tcp_trace: bool = False,
source_ip: str | None = None,
) -> TraceResult: ) -> TraceResult:
builders = { builders = {
"tcp_trace": lambda: _tier_tcp_trace(target, max_hops, timeout, source_ip=source_ip), "tcp_trace": lambda: _tier_tcp_trace(target, options.max_hops, options.timeout, source_ip=options.source_ip),
"mtr_json": lambda: _tier_mtr( "mtr_json": lambda: _tier_mtr(
target, caps.mtr_binary or shutil.which("mtr") or "mtr", cycles, max_hops, timeout, source_ip target, options.caps.mtr_binary or shutil.which("mtr") or "mtr", options.cycles, options.max_hops, options.timeout, options.source_ip
), ),
"icmp_win": lambda: _tier_icmp_win(target, max_hops, timeout, source_ip), "icmp_win": lambda: _tier_icmp_win(target, options.max_hops, options.timeout, options.source_ip),
"icmplib": lambda: _tier_icmplib(target, max_hops, timeout, source_ip=source_ip), "icmplib": lambda: _tier_icmplib(target, options.max_hops, options.timeout, source_ip=options.source_ip),
"system_traceroute": lambda: _tier_system( "system_traceroute": lambda: _tier_system(
target, caps.traceroute_binary or "traceroute", max_hops, timeout, source_ip target, options.caps.traceroute_binary or "traceroute", options.max_hops, options.timeout, options.source_ip
), ),
} }
order = tier_order(caps, tcp_trace) order = tier_order(options.caps, options.tcp_trace)
order, _dropped = filter_trace_tiers(order, caps.os_name, forced_v4=bool(source_ip)) order, _dropped = filter_trace_tiers(order, options.caps.os_name, forced_v4=bool(options.source_ip))
tiers = [(name, builders[name]) for name in order] tiers = [(name, builders[name]) for name in order]
if semaphore is None: if options.semaphore is None:
return await run_cascade(tiers) return await run_cascade(tiers)
async with semaphore: async with options.semaphore:
return await run_cascade(tiers) return await run_cascade(tiers)