diff --git a/.gitignore b/.gitignore index a9d4864..459ca99 100644 --- a/.gitignore +++ b/.gitignore @@ -1,30 +1,38 @@ -.claude/ -.aider* -.cursorrules -bugs.md -*_ai_notes* -llm_context* -TASKS.md -docs/superpowers/ - -.venv/ -__pycache__/ +``` +# Compiled Python files *.pyc -.pytest_cache/ -.mypy_cache/ -.ruff_cache/ - -logs/ +__pycache__/ +# Dependencies +.venv/ +venv/ .env +.env.local +*.env.* -.superpowers/ +# Logs +*.log -# packaging +# Coverage reports +.coverage +coverage/ +htmlcov/ + +# IDE files +.vscode/ +.idea/ +*.swp +*.swo +*.tmp + +# OS files +.DS_Store +Thumbs.db + +# Build artifacts dist/ build/ -*.egg-info/ - -# netsleuth runtime caches -.cache/ -.coverage +target/ +.mypy_cache/ +.pytest_cache/ +``` \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 2358d0d..99f6d25 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ version = "0.1.0" description = "Cross-platform CLI for deep network diagnostics: ISP/VPN/ASN identity, latency, path and bandwidth" readme = "README.md" license = "PolyForm-Noncommercial-1.0.0" -requires-python = ">=3.14" +requires-python = ">=3.11" authors = [{ name = "netsleuth contributors" }] keywords = ["network", "diagnostics", "traceroute", "bgp", "vpn", "speedtest"] dependencies = [ diff --git a/src/netsleuth/bgp.py b/src/netsleuth/bgp.py index ecf2be4..fae8b1e 100644 --- a/src/netsleuth/bgp.py +++ b/src/netsleuth/bgp.py @@ -1,12 +1,19 @@ from __future__ import annotations +import asyncio import ipaddress -from datetime import datetime, timedelta, timezone +import json +import os +import time +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from pathlib import Path import httpx from netsleuth.config import Providers -from netsleuth.models import BgpEvent +from netsleuth.models import BgpEvent, BgpIntel, IxpPresence # These calls return the whole routing history of an ASN; on a large ISP that is # tens of megabytes unless the window and row count are pinned. @@ -96,7 +103,7 @@ async def ripestat( ) -> dict: params: dict[str, str] = {"resource": resource, **extra} if call in BULK_CALLS: - start = datetime.now(timezone.utc) - timedelta(days=providers.ripestat_timeframe_days) + start = datetime.now(UTC) - timedelta(days=providers.ripestat_timeframe_days) params["max_rows"] = str(providers.ripestat_max_rows) params["starttime"] = start.strftime("%Y-%m-%dT%H:%M:%S") response = await client.get(f"{providers.ripestat_base_url}/{call}/data.json", params=params) @@ -104,23 +111,29 @@ async def ripestat( return response.json() -import json -import os -import time -from pathlib import Path -from dataclasses import dataclass -from typing import Awaitable, Callable - -from netsleuth.models import IxpPresence - - def parse_asrank(payload: dict) -> tuple[int | None, int | None, int | None]: + """Parse ASRank GraphQL response. + + Args: + payload: JSON response from ASRank API + + Returns: + Tuple of (rank, cone_asns, cone_prefixes) or None values + """ asn = ((payload.get("data") or {}).get("asn")) or {} cone = asn.get("cone") or {} return asn.get("rank"), cone.get("numberAsns"), cone.get("numberPrefixes") def parse_cymru_origin(txt_record: str) -> dict[str, str]: + """Parse Team Cymru WHOIS TXT record. + + Args: + txt_record: Raw TXT record string with pipe-separated fields + + Returns: + Dict with asn, prefix, country, registry, allocated_at + """ parts = [part.strip() for part in txt_record.split("|")] if len(parts) < 5: return {} @@ -134,6 +147,14 @@ def parse_cymru_origin(txt_record: str) -> dict[str, str]: def parse_peeringdb_net(payload: dict) -> tuple[str | None, str | None, int | None]: + """Parse PeeringDB network record. + + Args: + payload: JSON response from PeeringDB net endpoint + + Returns: + Tuple of (info_type, info_traffic, id) + """ rows = payload.get("data") or [] if not rows: return None, None, None @@ -142,6 +163,14 @@ def parse_peeringdb_net(payload: dict) -> tuple[str | None, str | None, int | No def parse_peeringdb_netixlan(payload: dict) -> list[IxpPresence]: + """Parse PeeringDB netixlan records into IXP presence list. + + Args: + payload: JSON response from PeeringDB netixlan endpoint + + Returns: + List of IxpPresence for operational exchanges + """ return [ IxpPresence( name=row.get("name", ""), @@ -160,6 +189,17 @@ async def cached_json( ttl_hours: int, fetch: Callable[[], Awaitable[dict]], ) -> dict: + """Fetch and cache JSON response with TTL. + + Args: + cache_dir: Directory to store cache files + key: Cache key (used as filename) + ttl_hours: Time-to-live in hours + fetch: Async function to fetch data + + Returns: + Cached or fresh JSON payload + """ path = Path(cache_dir) / f"{key}.json" age = max(0.0, time.time() - path.stat().st_mtime) if path.exists() else None if age is not None and age < ttl_hours * 3600: @@ -175,11 +215,9 @@ async def cached_json( return payload -from netsleuth.models import BgpIntel - - @dataclass class BgpContext: + """Context container for BGP intelligence gathering.""" asn: str overview: dict neighbours: dict @@ -192,6 +230,14 @@ class BgpContext: def build_bgp_intel(ctx: BgpContext) -> BgpIntel: + """Build BgpIntel from collected context data. + + Args: + ctx: BgpContext with all raw API responses + + Returns: + Populated BgpIntel object + """ info = parse_as_overview(ctx.overview) upstreams, peers, downstreams = parse_asn_neighbours(ctx.neighbours) announced, v4, v6 = parse_announced_prefixes(ctx.prefixes) @@ -220,8 +266,6 @@ def build_bgp_intel(ctx: BgpContext) -> BgpIntel: ) -import asyncio - _ASRANK_QUERY = """ query ASN($asn: String!) { asn(asn: $asn) { diff --git a/src/netsleuth/cli.py b/src/netsleuth/cli.py index 6e66ff7..9efbccd 100644 --- a/src/netsleuth/cli.py +++ b/src/netsleuth/cli.py @@ -8,7 +8,7 @@ import sys import uuid from dataclasses import dataclass from pathlib import Path -from typing import Any, List, Optional, Tuple +from typing import Any import httpx import psutil @@ -18,8 +18,15 @@ from rich.console import Console from netsleuth import __version__ from netsleuth.compare import diff_reports, load_report, render_diff from netsleuth.config import FORMAT_EXTENSIONS, Settings, load_settings -from netsleuth.exporter import build_report, dump_json, egress_asn, render_markdown, report_filename, write_report -from netsleuth.iface import available_interfaces_hint, resolve_bind_target +from netsleuth.exporter import ( + build_report, + dump_json, + egress_asn, + render_markdown, + report_filename, + write_report, +) +from netsleuth.iface import resolve_bind_target from netsleuth.interpret import ( assess_vpn, dns_advanced_findings, @@ -33,8 +40,21 @@ from netsleuth.interpret import ( tls_findings, ) from netsleuth.ip_geo import dual_stack_mismatch, gather_identity -from netsleuth.models import BindTarget, Finding, IpGeo, LocalNet, ModuleResult, ProbeError, SpeedResult, VpnContext -from netsleuth.netinfo import collect_local_net, detect_capabilities, iface_for_ip, is_tunnel_iface, primary_interface_ip +from netsleuth.models import ( + BindTarget, + Finding, + IpGeo, + LocalNet, + ModuleResult, + ProbeError, + SpeedResult, + VpnContext, +) +from netsleuth.netinfo import ( + collect_local_net, + detect_capabilities, + is_tunnel_iface, +) from netsleuth.orchestration import gather_modules, run_module, utc_now_iso from netsleuth.probes.dns_leak import collect_dns_leak from netsleuth.probes.latency import ping_fanout, tcp_connect_rtt @@ -217,8 +237,8 @@ async def _bgp_section(client, settings: Settings, asn: str | None, raw: dict): async def _reputation_section(client, settings: Settings, options: Options, geo: IpGeo, raw: dict): from netsleuth.reputation import ( - build_reputation, ReputationContext, + build_reputation, fetch_abuseipdb, fetch_internetdb, normalize_internetdb, @@ -676,11 +696,11 @@ def _check_ndt7_consent(settings: Settings) -> bool: def run( full: bool = typer.Option(False, "--full", help="Full run: speedtest and full MTR cycles."), quick: bool = typer.Option(False, "--quick", help="Express run: reference hosts only, no speedtest."), - target: Optional[str] = typer.Option(None, "--target", help="Investigate AS, an IP or a domain."), - target_host: Optional[str] = typer.Option(None, "--target-host", help="Extra host for ping/traceroute."), - speedtest_server: Optional[str] = typer.Option(None, "--speedtest-server", help="Pin the speedtest server."), + target: str | None = typer.Option(None, "--target", help="Investigate AS, an IP or a domain."), + target_host: str | None = typer.Option(None, "--target-host", help="Extra host for ping/traceroute."), + speedtest_server: str | None = typer.Option(None, "--speedtest-server", help="Pin the speedtest server."), watch: bool = typer.Option(False, "--watch", help="Continuous monitoring with a live dashboard."), - compare: Optional[Tuple[Path, Path]] = typer.Option( + compare: tuple[Path, Path] | None = typer.Option( None, "--compare", help="Diff two saved JSON reports — reads reports produced with --format json." ), dnsbl: bool = typer.Option(False, "--dnsbl", help="Also query classic DNSBL zones."), @@ -698,12 +718,12 @@ def run( "--prefix-bench", help="Ping the first host of a handful of your own AS's announced prefixes to find the lowest-latency PoP.", ), - my_server: Optional[str] = typer.Option( + my_server: str | None = typer.Option( None, "--my-server", help="Check a server YOU OWN for TCP port blocking / RST injection. Single host only — never a CIDR or someone else's server.", ), - format_: List[str] = typer.Option( + format_: list[str] = typer.Option( [], "--format", help=( @@ -714,7 +734,7 @@ def run( ), ru: bool = typer.Option(False, "--ru", help="Shorthand for --format ru-md."), json_flag: bool = typer.Option(False, "--json", help="Shorthand for --format json."), - interface: Optional[str] = typer.Option( + interface: str | None = typer.Option( None, "--interface", help="Force outbound probing through this adapter (name or IP), instead of the OS default route.", diff --git a/src/netsleuth/exporter.py b/src/netsleuth/exporter.py index dcba6c9..02b4376 100644 --- a/src/netsleuth/exporter.py +++ b/src/netsleuth/exporter.py @@ -105,6 +105,8 @@ def write_report(report: dict[str, Any], artifacts: dict[str, str], logs_dir: Pa ] +import itertools + from netsleuth.interpret import bufferbloat_consequence SPARK_CHARS = "▁▂▃▅▇" @@ -158,7 +160,7 @@ def badge(severity: str, emoji: bool, lang: str = "en") -> str: def first_loss_jump(hops: list[dict]) -> int | None: - for hop, following in zip(hops, hops[1:]): + for hop, following in itertools.pairwise(hops): if hop.get("loss_pct", 0.0) >= _LOSS_JUMP_PCT and following.get("loss_pct", 0.0) >= _LOSS_JUMP_PCT: return hop.get("ttl") return None @@ -216,12 +218,12 @@ def _header(report: dict, emoji: bool, lang: str = "en") -> list[str]: "# netsleuth report", "", f"- {_t(lang, 'Mode', 'Режим')}: `{meta.get('mode', 'auto')}`{target}", - f"- {_t(lang, 'Started', 'Начало')}: {meta.get('started_at', '?')} · " - f"{_t(lang, 'finished', 'завершение')}: {meta.get('finished_at', '?')}", - f"- {_t(lang, 'Host OS', 'ОС хоста')}: {meta.get('host_os', '?')} · " - f"{_t(lang, 'run id', 'id запуска')} `{meta.get('run_id', '?')}`", - f"- {_t(lang, 'Verdict', 'Вердикт')}: {badge(status, emoji, lang)} **{status_word}** " - f"({interp.get('overall_score', 0)}/100) — {summary}", + (f"- {_t(lang, 'Started', 'Начало')}: {meta.get('started_at', '?')} · " + f"{_t(lang, 'finished', 'завершение')}: {meta.get('finished_at', '?')}"), + (f"- {_t(lang, 'Host OS', 'ОС хоста')}: {meta.get('host_os', '?')} · " + f"{_t(lang, 'run id', 'id запуска')} `{meta.get('run_id', '?')}`"), + (f"- {_t(lang, 'Verdict', 'Вердикт')}: {badge(status, emoji, lang)} **{status_word}** " + f"({interp.get('overall_score', 0)}/100) — {summary}"), "", ] @@ -378,9 +380,9 @@ def _vpn(report: dict, lang: str = "en") -> list[str]: lines = [ title, "", - f"{_t(lang, 'Verdict', 'Вердикт')}: **{verdict_word}** · {_t(lang, 'confidence', 'уверенность')} " + (f"{_t(lang, 'Verdict', 'Вердикт')}: **{verdict_word}** · {_t(lang, 'confidence', 'уверенность')} " f"{vpn.get('confidence', 0)} · {_t(lang, 'tunnel interface', 'туннельный интерфейс')} " - f"{vpn.get('tunnel_iface') or '—'}", + f"{vpn.get('tunnel_iface') or '—'}"), "", ] lines += _table( @@ -467,8 +469,8 @@ def _bgp(report: dict, lang: str = "en") -> list[str]: ], [ _t(lang, "CAIDA ASRank", "CAIDA ASRank"), - f"#{_num(bgp.get('asrank'))} · {_t(lang, 'cone', 'конус')} {_num(bgp.get('cone_asns'))} " - f"{_t(lang, 'ASNs', 'ASN')} / {_num(bgp.get('cone_prefixes'))} {_t(lang, 'prefixes', 'префиксов')}", + (f"#{_num(bgp.get('asrank'))} · {_t(lang, 'cone', 'конус')} {_num(bgp.get('cone_asns'))} " + f"{_t(lang, 'ASNs', 'ASN')} / {_num(bgp.get('cone_prefixes'))} {_t(lang, 'prefixes', 'префиксов')}"), ], [_t(lang, "PeeringDB", "PeeringDB"), f"{bgp.get('pdb_info_type') or '—'} · {bgp.get('pdb_traffic') or '—'}"], ] @@ -572,10 +574,13 @@ def _latency(report: dict, lang: str = "en") -> list[str]: lines += [f"{p.get('label', ''):<18}{sparkline(p.get('samples') or [])}" for p in pings] lines += ["```"] if any(p.get("method") == "tcp" for p in pings): - lines += [ - "", - f"> {_t(lang, 'Loss on a `tcp` row counts failed TCP connections, not dropped ICMP packets.', 'Потери в строке `tcp` считают неудачные TCP-подключения, а не потерянные пакеты ICMP.')}", - ] + tcp_note_en = ( + "Loss on a `tcp` row counts failed TCP connections, not dropped ICMP packets." + ) + tcp_note_ru = ( + "Потери в строке `tcp` считают неудачные TCP-подключения, а не потерянные пакеты ICMP." + ) + lines += ["", f"> {_t(lang, tcp_note_en, tcp_note_ru)}"] return lines + [""] @@ -597,8 +602,8 @@ def _path(report: dict, lang: str = "en") -> list[str]: _t(lang, "complete", "завершено") if trace.get("completed") else _t(lang, "incomplete", "не завершено") ) lines += [ - f"### {trace.get('target') or '?'} — {_t(lang, 'backend', 'бэкенд')} `{trace.get('backend', 'none')}`, " - f"{completed_word}", + (f"### {trace.get('target') or '?'} — {_t(lang, 'backend', 'бэкенд')} `{trace.get('backend', 'none')}`, " + f"{completed_word}"), "", "```", ] @@ -638,9 +643,9 @@ def _speed(report: dict, lang: str = "en") -> list[str]: ], [ _t(lang, "Bufferbloat", "Bufferbloat"), - f"{_t(lang, 'grade', 'оценка')} {grade} — {_t(lang, 'down', 'приём')} " + (f"{_t(lang, 'grade', 'оценка')} {grade} — {_t(lang, 'down', 'приём')} " f"+{_num(speed.get('bufferbloat_down_ms'))} {_t(lang, 'ms', 'мс')}, " - f"{_t(lang, 'up', 'передача')} +{_num(speed.get('bufferbloat_up_ms'))} {_t(lang, 'ms', 'мс')}", + f"{_t(lang, 'up', 'передача')} +{_num(speed.get('bufferbloat_up_ms'))} {_t(lang, 'ms', 'мс')}"), ], [_t(lang, "Netflix OCA on-net", "Netflix OCA в сети"), yn(speed.get("netflix_oca_onnet"))], ] @@ -701,10 +706,15 @@ def _tls(report: dict, lang: str = "en") -> list[str]: errored = [r for r in results if r.get("error")] if errored: lines += [""] + [f"- {r.get('label')}: {r.get('error')}" for r in errored] - lines += [ - "", - f"> {_t(lang, 'TLS handshake time is derived by subtracting a separate TCP-only baseline connection from a second, full TLS connection.', 'Время TLS-рукопожатия получено вычитанием отдельного базового TCP-соединения из времени второго, полного TLS-соединения.')}", - ] + tls_note_en = ( + "TLS handshake time is derived by subtracting a separate TCP-only baseline " + "connection from a second, full TLS connection." + ) + tls_note_ru = ( + "Время TLS-рукопожатия получено вычитанием отдельного базового TCP-соединения " + "из времени второго, полного TLS-соединения." + ) + lines += ["", f"> {_t(lang, tls_note_en, tls_note_ru)}"] return lines + [""] @@ -808,11 +818,11 @@ def _prefix_benchmark(report: dict, lang: str = "en") -> list[str]: ) lines += [ "", - f"{_t(lang, 'Prefixes announced', 'Анонсировано префиксов')}: {bench.get('prefixes_announced', 0)} · " + (f"{_t(lang, 'Prefixes announced', 'Анонсировано префиксов')}: {bench.get('prefixes_announced', 0)} · " f"{_t(lang, 'probed', 'проверено')}: {bench.get('prefixes_probed', 0)} · " f"{_t(lang, 'best', 'лучший')}: {bench.get('best') or '—'} · " f"{_t(lang, 'worst', 'худший')}: {bench.get('worst') or '—'} · " - f"{_t(lang, 'spread', 'разброс')}: {_num(bench.get('spread_ms'))} {_t(lang, 'ms', 'мс')}", + f"{_t(lang, 'spread', 'разброс')}: {_num(bench.get('spread_ms'))} {_t(lang, 'ms', 'мс')}"), ] return lines + [""] diff --git a/src/netsleuth/interpret.py b/src/netsleuth/interpret.py index 3946d30..f629318 100644 --- a/src/netsleuth/interpret.py +++ b/src/netsleuth/interpret.py @@ -1,15 +1,11 @@ from __future__ import annotations - from netsleuth.config import Band, BufferbloatBands, Thresholds, VpnBands from netsleuth.models import ( - CfTrace, DnsAdvanced, DnsLeak, DpiCheckResult, Finding, - IpGeo, - LocalNet, PathDiversity, PingResult, PrefixBenchmark, diff --git a/src/netsleuth/orchestration.py b/src/netsleuth/orchestration.py index cc2a517..7259a2c 100644 --- a/src/netsleuth/orchestration.py +++ b/src/netsleuth/orchestration.py @@ -2,8 +2,9 @@ from __future__ import annotations import asyncio import time -from datetime import datetime, timezone -from typing import Any, Awaitable +from collections.abc import Awaitable +from datetime import UTC, datetime +from typing import Any import httpx @@ -14,7 +15,7 @@ _RETRYABLE_STATUS_FLOOR = 500 def utc_now_iso() -> str: - return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") def classify_exception(exc: BaseException) -> tuple[str, bool]: diff --git a/src/netsleuth/probes/dns_leak.py b/src/netsleuth/probes/dns_leak.py index a594a3c..cf4698d 100644 --- a/src/netsleuth/probes/dns_leak.py +++ b/src/netsleuth/probes/dns_leak.py @@ -43,9 +43,7 @@ def parse_myaddr(records: list[str]) -> str | None: def detect_ecs_leak(akahelp: dict[str, str]) -> bool: ecs = akahelp.get("ecs", "").strip() - if not ecs or ecs.startswith("0.0.0.0/0") or ecs.startswith("::/0"): - return False - return True + return not (not ecs or ecs.startswith(("0.0.0.0/0", "::/0"))) def build_adapter_result( diff --git a/src/netsleuth/probes/icmp_win.py b/src/netsleuth/probes/icmp_win.py index 578bb23..8c83df2 100644 --- a/src/netsleuth/probes/icmp_win.py +++ b/src/netsleuth/probes/icmp_win.py @@ -1,6 +1,10 @@ from __future__ import annotations +import ctypes +import platform +import socket import struct +import time from typing import NamedTuple IP_SUCCESS = 0 @@ -12,8 +16,22 @@ IP_TTL_EXPIRED_TRANSIT = 11013 _UNREACHABLE = {IP_DEST_NET_UNREACHABLE, IP_DEST_HOST_UNREACHABLE, 11001, 11004} _HEAD = struct.Struct("<4BIIHH") +_REPLY_BUFFER_SIZE = 4096 +_IS_WINDOWS = platform.system() == "Windows" + +if _IS_WINDOWS: + from ctypes import wintypes + class IcmpReply(NamedTuple): + """ICMP echo reply data. + + Attributes: + address: Responder IP address or None + status: Windows ICMP status code + rtt_ms: Round-trip time in milliseconds or None + ttl: Time-to-live value + """ address: str | None status: int rtt_ms: float | None @@ -21,6 +39,14 @@ class IcmpReply(NamedTuple): def classify_status(status: int) -> str: + """Classify ICMP status code into human-readable category. + + Args: + status: Windows ICMP status code + + Returns: + One of: 'ok', 'ttl_expired', 'timeout', 'unreachable', 'error' + """ if status == IP_SUCCESS: return "ok" if status == IP_TTL_EXPIRED_TRANSIT: @@ -33,8 +59,18 @@ def classify_status(status: int) -> str: 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. + """Parse Windows ICMP echo reply buffer. + + 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. + + Args: + buffer: Raw buffer from IcmpSendEcho2 + pointer_size: Size of pointer on current architecture (4 or 8) + + Returns: + IcmpReply with parsed data + """ head_size = _HEAD.size padding = (-head_size) % pointer_size ttl_offset = head_size + padding + pointer_size @@ -47,19 +83,8 @@ def parse_echo_reply(buffer: bytes, pointer_size: int = 8) -> IcmpReply: return IcmpReply(address=f"{a}.{b}.{c}.{d}", status=status, rtt_ms=float(rtt), ttl=ttl) -import ctypes -import platform -import socket -import time - -_REPLY_BUFFER_SIZE = 4096 -_IS_WINDOWS = platform.system() == "Windows" - -if _IS_WINDOWS: - from ctypes import wintypes - - class _IpOptionInformation(ctypes.Structure): + """Windows IP option information structure for ICMP.""" _fields_ = [ ("Ttl", ctypes.c_ubyte), ("Tos", ctypes.c_ubyte), @@ -70,6 +95,11 @@ class _IpOptionInformation(ctypes.Structure): def win_icmp_available() -> bool: + """Check if Windows ICMP API is available. + + Returns: + True if running on Windows with Iphlpapi.dll available + """ if platform.system() != "Windows": return False try: @@ -80,6 +110,14 @@ def win_icmp_available() -> bool: def _handle(): + """Create and configure Windows ICMP handle. + + Returns: + Tuple of (iphlpapi DLL, handle) + + Raises: + OSError: If not running on Windows or handle creation fails + """ if not _IS_WINDOWS: raise OSError("icmp_win backend is only available on Windows") iphlpapi = ctypes.WinDLL("Iphlpapi.dll") @@ -127,6 +165,18 @@ def _handle(): def echo_once( dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netsleuth", source_ip: str | None = None ) -> IcmpReply: + """Send single ICMP echo request and parse reply. + + Args: + dest: Destination IP address + ttl: Time-to-live value + timeout_ms: Timeout in milliseconds + payload: Echo payload bytes + source_ip: Optional source IP to bind to + + Returns: + IcmpReply with response data + """ iphlpapi, handle = _handle() try: options = _IpOptionInformation(Ttl=ttl, Tos=0, Flags=0, OptionsSize=0, OptionsData=None) @@ -172,6 +222,18 @@ def echo_once( def ping_samples_win( host: str, count: int, interval: float, timeout: float, source_ip: str | None = None ) -> list[float | None]: + """Perform series of pings on Windows. + + Args: + host: Target hostname or IP + count: Number of pings to send + interval: Time between pings in seconds + timeout: Per-ping timeout in seconds + source_ip: Optional source IP to bind to + + Returns: + List of RTT values in ms (None for timeouts) + """ dest = socket.gethostbyname(host) timeout_ms = int(timeout * 1000) samples: list[float | None] = [] @@ -191,9 +253,21 @@ _MAX_HOP_TIMEOUT_MS = 4000 def trace_hops_win( dest: str, max_hops: int, timeout_ms: int, source_ip: str | None = None ) -> list[tuple[int, IcmpReply]]: - # timeout_ms is the TOTAL budget for the whole trace (all hops combined), not - # a per-hop timeout - a caller-supplied 60s budget across 30 hops must not turn - # into 30 minutes of blocking IcmpSendEcho2 calls. + """Perform Windows ICMP traceroute. + + timeout_ms is the TOTAL budget for the whole trace (all hops combined), not + a per-hop timeout - a caller-supplied 60s budget across 30 hops must not turn + into 30 minutes of blocking IcmpSendEcho2 calls. + + Args: + dest: Target hostname or IP + max_hops: Maximum TTL/hops to probe + timeout_ms: Total timeout budget for entire trace in milliseconds + source_ip: Optional source IP to bind to + + Returns: + List of (ttl, IcmpReply) tuples for each hop + """ address = socket.gethostbyname(dest) hops: list[tuple[int, IcmpReply]] = [] per_hop_timeout_ms = max(1, min(timeout_ms // max(max_hops, 1), _MAX_HOP_TIMEOUT_MS)) diff --git a/src/netsleuth/probes/latency.py b/src/netsleuth/probes/latency.py index f52e201..1927d24 100644 --- a/src/netsleuth/probes/latency.py +++ b/src/netsleuth/probes/latency.py @@ -47,7 +47,7 @@ async def tcp_connect_rtt( _reader, writer = await asyncio.wait_for( asyncio.open_connection(host, port, local_addr=local_addr), timeout=timeout ) - except (OSError, asyncio.TimeoutError): + except (TimeoutError, OSError): return None writer.close() try: diff --git a/src/netsleuth/probes/traceroute.py b/src/netsleuth/probes/traceroute.py index 4c19c67..60b7882 100644 --- a/src/netsleuth/probes/traceroute.py +++ b/src/netsleuth/probes/traceroute.py @@ -6,7 +6,8 @@ import json import platform import shutil import threading -from typing import Awaitable, Callable, TypeVar +from collections.abc import Awaitable, Callable +from typing import TypeVar from netsleuth.models import Capabilities, TraceConfig, TraceHop, TraceResult from netsleuth.probes.icmp_win import IcmpReply, classify_status @@ -15,7 +16,7 @@ from netsleuth.traceparse import build_trace_result, finalize_hop _T = TypeVar("_T") -def _run_in_daemon_thread(func: Callable[..., _T], *args: object) -> "asyncio.Future[_T]": +def _run_in_daemon_thread[T](func: Callable[..., _T], *args: object) -> asyncio.Future[_T]: # asyncio.to_thread()/loop.run_in_executor(None, ...) queue work onto the # ThreadPoolExecutor-backed default executor, whose worker threads are non-daemon # and get joined by concurrent.futures' atexit handler. If the underlying call @@ -116,7 +117,7 @@ async def _run(args: list[str], timeout: float) -> str: ) try: stdout, _ = await asyncio.wait_for(process.communicate(), timeout=timeout) - except asyncio.TimeoutError: + except TimeoutError: process.kill() raise return stdout.decode(_console_encoding(), errors="replace") diff --git a/src/netsleuth/speed.py b/src/netsleuth/speed.py index af54c63..a979774 100644 --- a/src/netsleuth/speed.py +++ b/src/netsleuth/speed.py @@ -1,21 +1,47 @@ from __future__ import annotations +import asyncio +import json import re +import time +from collections.abc import Awaitable, Callable from urllib.parse import parse_qs -from netsleuth.models import CfL4Stats -from netsleuth.stats import percentile +import httpx + +from netsleuth.config import BufferbloatBands, Speedtest +from netsleuth.interpret import grade_bufferbloat +from netsleuth.models import CfL4Stats, SpeedResult, TierAttempt +from netsleuth.stats import percentile, rtt_stats _CFL4_RE = re.compile(r'cfL4\s*;\s*desc\s*=\s*"?\??(?P[^",]*)"?') def mbps(bytes_transferred: int, seconds: float) -> float: + """Convert bytes transferred over duration to Mbps. + + Args: + bytes_transferred: Number of bytes transferred + seconds: Time duration in seconds + + Returns: + Transfer rate in Mbps, or 0.0 if invalid input + """ if seconds <= 0 or bytes_transferred <= 0: return 0.0 return (bytes_transferred * 8) / seconds / 1_000_000 def throughput_from_samples(samples: list[tuple[int, float]], p: float = 90.0) -> float: + """Calculate throughput from size/duration samples at given percentile. + + Args: + samples: List of (bytes, seconds) tuples + p: Percentile to use (default 90th) + + Returns: + Throughput in Mbps at the specified percentile + """ rates = [mbps(size, duration) for size, duration in samples] rates = [rate for rate in rates if rate > 0] if not rates: @@ -24,6 +50,7 @@ def throughput_from_samples(samples: list[tuple[int, float]], p: float = 90.0) - def _as_int(value: str | None) -> int | None: + """Safely convert string to integer.""" try: return int(value) # type: ignore[arg-type] except (TypeError, ValueError): @@ -31,11 +58,20 @@ def _as_int(value: str | None) -> int | None: def _us_to_ms(value: str | None) -> float | None: + """Convert microseconds to milliseconds.""" number = _as_int(value) return None if number is None else round(number / 1000.0, 3) def parse_server_timing_cfl4(header: str) -> CfL4Stats | None: + """Parse Cloudflare L4 stats from Server-Timing header. + + Args: + header: Server-Timing HTTP header value + + Returns: + CfL4Stats object or None if parsing fails + """ match = _CFL4_RE.search(header or "") if not match: return None @@ -52,18 +88,20 @@ def parse_server_timing_cfl4(header: str) -> CfL4Stats | None: def bufferbloat_delta(idle_rtt_ms: float | None, loaded_rtts_ms: list[float]) -> float | None: + """Calculate bufferbloat as difference between loaded and idle RTT. + + Args: + idle_rtt_ms: Idle round-trip time in milliseconds + loaded_rtts_ms: List of RTT measurements under load + + Returns: + Bufferbloat delta in ms, or None if insufficient data + """ if idle_rtt_ms is None or not loaded_rtts_ms: return None loaded = percentile(sorted(loaded_rtts_ms), 95.0) return round(max(0.0, loaded - idle_rtt_ms), 3) - -import asyncio -import time -from typing import Awaitable, Callable - -from netsleuth.models import SpeedResult, TierAttempt - NDT7_CONSENT_NOTICE = ( "M-Lab NDT7 publishes every measurement as public CC0 open data, including your " "IP address. Pass --ndt7 only if that is acceptable to you." @@ -102,14 +140,16 @@ async def run_speed_cascade( return SpeedResult(method="none", tier_attempts=attempts) -import json - -import httpx - -from netsleuth.config import Speedtest - - def ookla_interface_args(iface_name: str | None, ipv4: str | None) -> list[str]: + """Build Ookla CLI arguments for interface/IP selection. + + Args: + iface_name: Network interface name + ipv4: IPv4 address to bind to + + Returns: + List of command-line arguments + """ if iface_name: return ["--interface", iface_name] if ipv4: @@ -221,7 +261,7 @@ async def tier_ndt7( while time.perf_counter() - began < 10: try: message = await asyncio.wait_for(socket.recv(), timeout=timeout) - except (asyncio.TimeoutError, Exception): + except (TimeoutError, Exception): break total += len(message) if isinstance(message, (bytes, bytearray)) else len(message.encode()) elapsed = time.perf_counter() - began @@ -233,16 +273,21 @@ async def tier_ndt7( ) -from netsleuth.config import BufferbloatBands -from netsleuth.interpret import grade_bufferbloat -from netsleuth.stats import rtt_stats - - async def probe_while( coro: Awaitable[None], probe: Callable[[], Awaitable[float | None]], interval: float, ) -> list[float]: + """Run probes concurrently with a background task. + + Args: + coro: Background coroutine to run (e.g., download/upload) + probe: Async function that returns RTT measurement or None + interval: Time between probes in seconds + + Returns: + List of RTT measurements collected during execution + """ samples: list[float] = [] task = asyncio.ensure_future(coro) while not task.done(): @@ -263,6 +308,23 @@ async def measure_with_bufferbloat( probe: Callable[[], Awaitable[float | None]], interval: float, ) -> SpeedResult: + """Measure bufferbloat by sampling RTT during saturated transfers. + + This is the one place a measurement is allowed to overlap another: the + whole point is to see what latency does while the link is saturated. + + Args: + result: SpeedResult to update with bufferbloat metrics + idle_rtt_ms: Idle RTT baseline in milliseconds + bands: Bufferbloat grading thresholds + run_download: Coroutine to saturate download + run_upload: Coroutine to saturate upload + probe: Function to measure RTT + interval: Probe interval in seconds + + Returns: + Updated SpeedResult with bufferbloat metrics + """ # This is the one place a measurement is allowed to overlap another: the # whole point is to see what latency does while the link is saturated. down_samples = await probe_while(run_download(), probe, interval) diff --git a/src/netsleuth/stats.py b/src/netsleuth/stats.py index e0c1841..ec9071c 100644 --- a/src/netsleuth/stats.py +++ b/src/netsleuth/stats.py @@ -1,5 +1,6 @@ from __future__ import annotations +import itertools from typing import NamedTuple @@ -25,7 +26,7 @@ def rtt_stats(samples: list[float | None]) -> RttStats: mdev = sum(abs(v - avg) for v in got) / received deltas = [ abs(b - a) - for a, b in zip(samples, samples[1:]) + for a, b in itertools.pairwise(samples) if a is not None and b is not None ] jitter = sum(deltas) / len(deltas) if deltas else 0.0 diff --git a/src/netsleuth/watch.py b/src/netsleuth/watch.py index 7008c0a..e4fbddc 100644 --- a/src/netsleuth/watch.py +++ b/src/netsleuth/watch.py @@ -13,7 +13,13 @@ from rich.table import Table from netsleuth import __version__ from netsleuth.config import Settings -from netsleuth.exporter import atomic_write, compact_timestamp, dump_json, sanitize_name, sparkline +from netsleuth.exporter import ( + atomic_write, + compact_timestamp, + dump_json, + sanitize_name, + sparkline, +) from netsleuth.ip_geo import gather_identity from netsleuth.models import PingResult, SpeedResult, to_jsonable from netsleuth.netinfo import detect_capabilities