diff --git a/src/netsleuth/cli.py b/src/netsleuth/cli.py index 77ef25c..650e6a7 100644 --- a/src/netsleuth/cli.py +++ b/src/netsleuth/cli.py @@ -81,7 +81,7 @@ if sys.platform == "win32": for _stream in (sys.stdout, sys.stderr): try: _stream.reconfigure(encoding="utf-8", errors="replace") - except (AttributeError, ValueError): + except AttributeError, ValueError: pass app = typer.Typer(add_completion=False, help="Deep network diagnostics.") @@ -185,7 +185,9 @@ def _dedupe(findings: list[Finding]) -> list[Finding]: return out -async def _identity(client: httpx.AsyncClient, settings: Settings, options: Options, raw: dict) -> dict: +async def _identity( + client: httpx.AsyncClient, settings: Settings, options: Options, raw: dict +) -> dict: token = settings.ipinfo_token.get_secret_value() if settings.ipinfo_token else None lookup = options.target_value if options.target_kind in ("ip", "domain") else None own_network = options.target_kind in (None, "asn") @@ -204,7 +206,7 @@ async def _identity(client: httpx.AsyncClient, settings: Settings, options: Opti merged, cf, flags, payloads = await gather_identity( v4_client, settings.providers, ip=lookup, ipinfo_token=token ) - except (httpx.HTTPError, OSError): + except httpx.HTTPError, OSError: merged, cf, flags, payloads = await gather_identity( client, settings.providers, ip=lookup, ipinfo_token=token ) @@ -220,13 +222,17 @@ async def _identity(client: httpx.AsyncClient, settings: Settings, options: Opti ) elif own_network: try: - transport = httpx.AsyncHTTPTransport(local_address=options.bind.ipv6 if options.bind else "::") + transport = httpx.AsyncHTTPTransport( + local_address=options.bind.ipv6 if options.bind else "::" + ) async with httpx.AsyncClient( transport=transport, timeout=settings.timeouts.http_seconds ) as v6_client: - v6_merged, _cf6, _flags6, _raw6 = await gather_identity(v6_client, settings.providers) + v6_merged, _cf6, _flags6, _raw6 = await gather_identity( + v6_client, settings.providers + ) v6 = v6_merged if v6_merged.ip else None - except (httpx.HTTPError, OSError): + except httpx.HTTPError, OSError: v6 = None dual_stack = dual_stack_mismatch(merged, v6) nat64_prefix = await detect_nat64(settings.timeouts.dns_seconds) @@ -244,18 +250,32 @@ async def _identity(client: httpx.AsyncClient, settings: Settings, options: Opti async def _bgp_section(client, settings: Settings, asn: str | None, raw: dict): if not asn: - return ModuleResult(name="bgp", status="skipped", warnings=["no ASN was resolved for this target"]) + return ModuleResult( + name="bgp", + status="skipped", + warnings=["no ASN was resolved for this target"], + ) from netsleuth.bgp import collect_bgp - key = settings.peeringdb_api_key.get_secret_value() if settings.peeringdb_api_key else None + key = ( + settings.peeringdb_api_key.get_secret_value() + if settings.peeringdb_api_key + else None + ) intel, payloads = await collect_bgp( - client, settings.providers, Path(settings.output.cache_dir), asn, peeringdb_key=key + client, + settings.providers, + Path(settings.output.cache_dir), + asn, + peeringdb_key=key, ) raw.update(payloads) return ModuleResult(name="bgp", status="ok", data=intel) -async def _reputation_section(client, settings: Settings, options: Options, geo: IpGeo, raw: dict): +async def _reputation_section( + client, settings: Settings, options: Options, geo: IpGeo, raw: dict +): from netsleuth.reputation import ( ReputationContext, build_reputation, @@ -267,18 +287,27 @@ async def _reputation_section(client, settings: Settings, options: Options, geo: ) if not geo.ip: - return ModuleResult(name="reputation", status="skipped", warnings=["no egress IP to check"]) + return ModuleResult( + name="reputation", status="skipped", warnings=["no egress IP to check"] + ) warnings: list[str] = [] - index = await refresh_netsets(client, settings.providers, Path(settings.output.cache_dir)) + index = await refresh_netsets( + client, settings.providers, Path(settings.output.cache_dir) + ) payload = await fetch_internetdb(client, settings.providers, geo.ip) raw["internetdb"] = payload outcomes = None if options.dnsbl and geo.ip_version == 4: - outcomes = await query_dnsbl(geo.ip, settings.dnsbl.zones, settings.timeouts.dns_seconds) + outcomes = await query_dnsbl( + geo.ip, settings.dnsbl.zones, settings.timeouts.dns_seconds + ) score = reports = None if settings.abuseipdb_api_key: abuse = await fetch_abuseipdb( - client, settings.providers, geo.ip, settings.abuseipdb_api_key.get_secret_value() + client, + settings.providers, + geo.ip, + settings.abuseipdb_api_key.get_secret_value(), ) raw["abuseipdb"] = abuse data = abuse.get("data") or {} @@ -296,13 +325,18 @@ async def _reputation_section(client, settings: Settings, options: Options, geo: ) ) return ModuleResult( - name="reputation", status="partial" if warnings else "ok", data=reputation, warnings=warnings + name="reputation", + status="partial" if warnings else "ok", + data=reputation, + warnings=warnings, ) async def _tls_section(settings: Settings, options: Options) -> ModuleResult: if not options.tls: - return ModuleResult(name="tls", status="skipped", warnings=["tls probe skipped: not requested"]) + return ModuleResult( + name="tls", status="skipped", warnings=["tls probe skipped: not requested"] + ) from netsleuth.probes.tls_rtt import tls_fanout cfg = settings.tls @@ -319,24 +353,37 @@ async def _tls_section(settings: Settings, options: Options) -> ModuleResult: return ModuleResult(name="tls", status="ok", data=results) -async def _dns_advanced_section(client, settings: Settings, options: Options) -> ModuleResult: +async def _dns_advanced_section( + client, settings: Settings, options: Options +) -> ModuleResult: if not options.dns_advanced: return ModuleResult( - name="dns_advanced", status="skipped", warnings=["dns advanced probe skipped: not requested"] + name="dns_advanced", + status="skipped", + warnings=["dns advanced probe skipped: not requested"], ) from netsleuth.probes.dns_advanced import collect_dns_advanced cfg = settings.dns_advanced adv = await collect_dns_advanced( - client, cfg.control_names, cfg.doh_endpoints, cfg.bogus_resolver_ip, cfg.bogus_probe_name, settings.timeouts.dns_seconds + client, + cfg.control_names, + cfg.doh_endpoints, + cfg.bogus_resolver_ip, + cfg.bogus_probe_name, + settings.timeouts.dns_seconds, ) return ModuleResult(name="dns_advanced", status="ok", data=adv) -async def _path_diversity_section(client, settings: Settings, options: Options, client_country: str | None) -> ModuleResult: +async def _path_diversity_section( + client, settings: Settings, options: Options, client_country: str | None +) -> ModuleResult: if not options.path_diversity: return ModuleResult( - name="path_diversity", status="skipped", warnings=["path diversity probe skipped: not requested"] + name="path_diversity", + status="skipped", + warnings=["path diversity probe skipped: not requested"], ) from netsleuth.probes.bgp_path import collect_path_diversity @@ -353,14 +400,20 @@ async def _path_diversity_section(client, settings: Settings, options: Options, return ModuleResult(name="path_diversity", status="ok", data=pd) -async def _prefix_bench_section(caps, settings: Settings, options: Options, bgp) -> ModuleResult: +async def _prefix_bench_section( + caps, settings: Settings, options: Options, bgp +) -> ModuleResult: if not options.prefix_bench: return ModuleResult( - name="prefix_benchmark", status="skipped", warnings=["prefix benchmark skipped: not requested"] + name="prefix_benchmark", + status="skipped", + warnings=["prefix benchmark skipped: not requested"], ) if not bgp or not bgp.announced_prefixes: return ModuleResult( - name="prefix_benchmark", status="skipped", warnings=["prefix benchmark skipped: no announced prefixes"] + name="prefix_benchmark", + status="skipped", + warnings=["prefix benchmark skipped: no announced prefixes"], ) from netsleuth.probes.prefix_benchmark import benchmark_prefixes @@ -384,7 +437,9 @@ async def _prefix_bench_section(caps, settings: Settings, options: Options, bgp) async def _dpi_section(settings: Settings, options: Options) -> ModuleResult: if not options.dpi_target: return ModuleResult( - name="dpi_check", status="skipped", warnings=["dpi check skipped: no explicit --my-server target"] + name="dpi_check", + status="skipped", + warnings=["dpi check skipped: no explicit --my-server target"], ) from netsleuth.probes.dpi_check import check_dpi @@ -397,7 +452,13 @@ async def _dpi_section(settings: Settings, options: Options) -> ModuleResult: return ModuleResult( name="dpi_check", status="failed", - errors=[ProbeError(source="dpi_check", kind="unavailable", message=f"could not resolve {target}: {exc}")], + errors=[ + ProbeError( + source="dpi_check", + kind="unavailable", + message=f"could not resolve {target}: {exc}", + ) + ], ) cfg = settings.dpi_check console.print( @@ -418,14 +479,20 @@ async def _dpi_section(settings: Settings, options: Options) -> ModuleResult: async def _quic_section(settings: Settings, options: Options) -> ModuleResult: if not options.quic: - return ModuleResult(name="quic", status="skipped", warnings=["quic probe skipped: not requested"]) + return ModuleResult( + name="quic", + status="skipped", + warnings=["quic probe skipped: not requested"], + ) try: import aioquic # noqa: F401 except ImportError: return ModuleResult( name="quic", status="skipped", - warnings=["quic probe skipped: aioquic not installed (uv sync --extra quic)"], + warnings=[ + "quic probe skipped: aioquic not installed (uv sync --extra quic)" + ], ) from netsleuth.probes.quic_rtt import quic_fanout @@ -434,16 +501,26 @@ async def _quic_section(settings: Settings, options: Options) -> ModuleResult: # aioquic's connect() takes only local_port, never a source address, so a # forced adapter cannot be honored here -- say so rather than silently # measuring the OS-default path under a report claiming a forced interface. - warnings.append("quic probe ran over the OS default route: aioquic has no source-address bind option") + warnings.append( + "quic probe ran over the OS default route: aioquic has no source-address bind option" + ) cfg = settings.quic targets = [(t.label, t.host) for t in cfg.targets] - results = await quic_fanout(targets, port=cfg.port, timeout=cfg.timeout_seconds, concurrency=cfg.concurrency) + results = await quic_fanout( + targets, port=cfg.port, timeout=cfg.timeout_seconds, concurrency=cfg.concurrency + ) return ModuleResult(name="quic", status="ok", data=results, warnings=warnings) -async def _pmtud_section(settings: Settings, options: Options, iface_mtu: int | None) -> ModuleResult: +async def _pmtud_section( + settings: Settings, options: Options, iface_mtu: int | None +) -> ModuleResult: if not options.pmtud: - return ModuleResult(name="pmtud", status="skipped", warnings=["pmtud probe skipped: not requested"]) + return ModuleResult( + name="pmtud", + status="skipped", + warnings=["pmtud probe skipped: not requested"], + ) from netsleuth.probes.pmtud import probe_pmtu cfg = settings.pmtud @@ -462,13 +539,21 @@ async def _pmtud_section(settings: Settings, options: Options, iface_mtu: int | return ModuleResult(name="pmtud", status="ok", data=results) -async def _ecmp_section(caps, settings: Settings, options: Options, cycles: int) -> ModuleResult: +async def _ecmp_section( + caps, settings: Settings, options: Options, cycles: int +) -> ModuleResult: if not options.ecmp: - return ModuleResult(name="ecmp", status="skipped", warnings=["ecmp probe skipped: not requested"]) + return ModuleResult( + name="ecmp", + status="skipped", + warnings=["ecmp probe skipped: not requested"], + ) from netsleuth.probes.ecmp import detect_ecmp cfg = settings.ecmp - targets = [(h.label, h.host) for h in settings.probing.reference_hosts[: cfg.max_targets]] + targets = [ + (h.label, h.host) for h in settings.probing.reference_hosts[: cfg.max_targets] + ] semaphore = asyncio.Semaphore(settings.probing.trace_concurrency) reports = [] for _label, host in targets: @@ -512,7 +597,9 @@ async def _traces(hosts, caps, settings: Settings, cycles: int, options: Options ) -async def _speed_section(client, settings: Settings, options: Options, idle_rtt_ms: float | None): +async def _speed_section( + client, settings: Settings, options: Options, idle_rtt_ms: float | None +): import shutil from netsleuth.speed import ( @@ -537,25 +624,35 @@ async def _speed_section(client, settings: Settings, options: Options, idle_rtt_ ), "cloudflare": lambda: tier_cloudflare(client, cfg, timeout), "fastcom": lambda: tier_fastcom(client, cfg, timeout), - "ndt7": lambda: tier_ndt7(client, cfg, timeout, source_ip=bind.ipv4 if bind else None), + "ndt7": lambda: tier_ndt7( + client, cfg, timeout, source_ip=bind.ipv4 if bind else None + ), } enabled = list(cfg.enabled_tiers) + (["ndt7"] if options.ndt7 else []) - result = await run_speed_cascade([(name, builders[name]) for name in enabled if name in builders]) + result = await run_speed_cascade( + [(name, builders[name]) for name in enabled if name in builders] + ) if result.method == "none": return ModuleResult(name="speed", status="failed", data=result) async def _saturate_down() -> None: await client.get( - f"{cfg.cloudflare_base_url}/__down", params={"bytes": cfg.download_sizes_bytes[-1]}, timeout=timeout + f"{cfg.cloudflare_base_url}/__down", + params={"bytes": cfg.download_sizes_bytes[-1]}, + timeout=timeout, ) async def _saturate_up() -> None: await client.post( - f"{cfg.cloudflare_base_url}/__up", content=b"\x00" * cfg.upload_sizes_bytes[-1], timeout=timeout + f"{cfg.cloudflare_base_url}/__up", + content=b"\x00" * cfg.upload_sizes_bytes[-1], + timeout=timeout, ) async def _probe() -> float | None: - return await tcp_connect_rtt("1.1.1.1", timeout=settings.probing.ping_timeout_seconds) + return await tcp_connect_rtt( + "1.1.1.1", timeout=settings.probing.ping_timeout_seconds + ) result = await measure_with_bufferbloat( result, @@ -577,7 +674,9 @@ async def diagnose(settings: Settings, options: Options) -> tuple[dict, list[Pat raw: dict[str, Any] = {} client_transport = ( - httpx.AsyncHTTPTransport(local_address=options.bind.ipv4, http2=True) if options.bind else None + httpx.AsyncHTTPTransport(local_address=options.bind.ipv4, http2=True) + if options.bind + else None ) async with httpx.AsyncClient( transport=client_transport, @@ -594,26 +693,37 @@ async def diagnose(settings: Settings, options: Options) -> tuple[dict, list[Pat modules["captive_portal"] = await run_module( "captive_portal", check_captive_portal( - client, cp_cfg.check_urls, expected_status=cp_cfg.expected_status, timeout=timeouts.http_seconds + client, + cp_cfg.check_urls, + expected_status=cp_cfg.expected_status, + timeout=timeouts.http_seconds, ), timeout=timeouts.module_seconds, ) else: modules["captive_portal"] = ModuleResult( - name="captive_portal", status="skipped", warnings=["captive portal check disabled in config"] + name="captive_portal", + status="skipped", + warnings=["captive portal check disabled in config"], ) # Phase 1 — local facts and identity. Blocking: everything below needs the ASN. modules["connection"] = await run_module( - "connection", asyncio.to_thread(collect_local_net), timeout=timeouts.module_seconds + "connection", + asyncio.to_thread(collect_local_net), + timeout=timeouts.module_seconds, ) local = modules["connection"].data or LocalNet() modules["ip_geo"] = await run_module( - "ip_geo", _identity(client, settings, options, raw), timeout=timeouts.module_seconds + "ip_geo", + _identity(client, settings, options, raw), + timeout=timeouts.module_seconds, ) bundle = modules["ip_geo"].data or {} flags = bundle.pop("_flags", {}) if isinstance(bundle, dict) else {} - identity_warnings = bundle.pop("_warnings", []) if isinstance(bundle, dict) else [] + identity_warnings = ( + bundle.pop("_warnings", []) if isinstance(bundle, dict) else [] + ) geo = bundle.get("egress_v4") or IpGeo() if bundle.get("dual_stack_note"): modules["ip_geo"].warnings.append(bundle["dual_stack_note"]) @@ -621,17 +731,38 @@ async def diagnose(settings: Settings, options: Options) -> tuple[dict, list[Pat asn = geo.asn if options.target_kind != "asn" else options.target_value # Phase 2 — bgp || reputation || dns_leak || dns_advanced || path_diversity - bgp_result, rep_result, dns_result, dns_advanced_result, path_diversity_result = await gather_modules( - run_module("bgp", _bgp_section(client, settings, asn, raw), timeout=timeouts.module_seconds), + ( + bgp_result, + rep_result, + dns_result, + dns_advanced_result, + path_diversity_result, + ) = await gather_modules( run_module( - "reputation", _reputation_section(client, settings, options, geo, raw), timeout=timeouts.module_seconds + "bgp", + _bgp_section(client, settings, asn, raw), + timeout=timeouts.module_seconds, + ), + run_module( + "reputation", + _reputation_section(client, settings, options, geo, raw), + timeout=timeouts.module_seconds, ), run_module( "dns_leak", - collect_dns_leak(local, asn, settings.providers.cymru_origin_zone, timeouts.dns_seconds), + collect_dns_leak( + local, + asn, + settings.providers.cymru_origin_zone, + timeouts.dns_seconds, + ), + timeout=timeouts.module_seconds, + ), + run_module( + "dns_advanced", + _dns_advanced_section(client, settings, options), timeout=timeouts.module_seconds, ), - run_module("dns_advanced", _dns_advanced_section(client, settings, options), timeout=timeouts.module_seconds), run_module( "path_diversity", _path_diversity_section(client, settings, options, geo.country_code), @@ -639,7 +770,10 @@ async def diagnose(settings: Settings, options: Options) -> tuple[dict, list[Pat ), ) modules["bgp"], modules["reputation"] = bgp_result, rep_result - modules["dns_advanced"], modules["path_diversity"] = dns_advanced_result, path_diversity_result + modules["dns_advanced"], modules["path_diversity"] = ( + dns_advanced_result, + path_diversity_result, + ) ctx = VpnContext( local=local, geo=geo, @@ -656,7 +790,9 @@ async def diagnose(settings: Settings, options: Options) -> tuple[dict, list[Pat data=assess_vpn( signals, settings.thresholds.vpn_confidence, - tunnel_iface=local.iface_name if is_tunnel_iface(local.iface_name or "") else None, + tunnel_iface=local.iface_name + if is_tunnel_iface(local.iface_name or "") + else None, dns_leak=dns_result.data, ), errors=dns_result.errors, @@ -672,12 +808,31 @@ async def diagnose(settings: Settings, options: Options) -> tuple[dict, list[Pat hosts.append(("target-host", options.extra_host)) if options.target_kind in ("ip", "domain"): hosts.append(("target", options.target_value)) - count = settings.probing.quick_ping_count if options.quick else settings.probing.ping_count - cycles = settings.probing.quick_mtr_cycles if options.quick else settings.probing.mtr_cycles - ipv6_policy = {"auto": local.is_dual_stack, "on": True, "off": False}[settings.probing.ipv6] + count = ( + settings.probing.quick_ping_count + if options.quick + else settings.probing.ping_count + ) + cycles = ( + settings.probing.quick_mtr_cycles + if options.quick + else settings.probing.mtr_cycles + ) + ipv6_policy = {"auto": local.is_dual_stack, "on": True, "off": False}[ + settings.probing.ipv6 + ] ipv6_enabled = ipv6_policy if options.ipv6 is None else options.ipv6 - v6_hosts = [(f"{h.label}-v6", h.host) for h in settings.probing.reference_hosts_v6] if ipv6_enabled else [] - modules["latency"], modules["path"], modules["tls"], v6_result = await gather_modules( + v6_hosts = ( + [(f"{h.label}-v6", h.host) for h in settings.probing.reference_hosts_v6] + if ipv6_enabled + else [] + ) + ( + modules["latency"], + modules["path"], + modules["tls"], + v6_result, + ) = await gather_modules( run_module( "latency", ping_fanout( @@ -690,8 +845,14 @@ async def diagnose(settings: Settings, options: Options) -> tuple[dict, list[Pat ), timeout=timeouts.subprocess_seconds, ), - run_module("path", _traces(hosts, caps, settings, cycles, options), timeout=timeouts.subprocess_seconds), - run_module("tls", _tls_section(settings, options), timeout=timeouts.module_seconds), + run_module( + "path", + _traces(hosts, caps, settings, cycles, options), + timeout=timeouts.subprocess_seconds, + ), + run_module( + "tls", _tls_section(settings, options), timeout=timeouts.module_seconds + ), run_module( "latency_v6", ping_fanout( @@ -707,7 +868,9 @@ async def diagnose(settings: Settings, options: Options) -> tuple[dict, list[Pat ) # v6 rows are tagged onto the same Latency section via a "