From 2f95a79ef5d1166d1d88219ceebde8a98f038a62 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Wed, 19 Aug 2026 11:26:38 +0000 Subject: [PATCH] Refactor cli.py imports and typing annotations Co-authored-by: Horizontal42 <75383519+Horizontal42@users.noreply.github.com> --- src/netsleuth/cli.py | 45 +++++++++++++++++++++++++++----------------- 1 file changed, 28 insertions(+), 17 deletions(-) diff --git a/src/netsleuth/cli.py b/src/netsleuth/cli.py index 7b7a8b2..77ef25c 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,9 +18,15 @@ from rich.console import Console from netsleuth import __version__, history from netsleuth.compare import diff_reports, load_report, render_diff, render_diff_brief 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.metrics import collect_metrics, render_csv, render_prometheus +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, captive_portal_findings, @@ -42,6 +48,7 @@ from netsleuth.interpret import ( tls_findings, ) from netsleuth.ip_geo import dual_stack_mismatch, gather_identity +from netsleuth.metrics import collect_metrics, render_csv, render_prometheus from netsleuth.models import ( BindTarget, CaptivePortal, @@ -53,7 +60,11 @@ from netsleuth.models import ( SpeedResult, VpnContext, ) -from netsleuth.netinfo import collect_local_net, detect_capabilities, iface_for_ip, is_tunnel_iface, primary_interface_ip +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.captive_portal import check_captive_portal from netsleuth.probes.dns_leak import collect_dns_leak @@ -96,7 +107,7 @@ class Options: ecmp: bool = False pmtud: bool = False quic: bool = False - ipv6: Optional[bool] = None + ipv6: bool | None = None dpi_target: str | None = None formats: frozenset[str] = frozenset({"md"}) interface: str | None = None @@ -246,8 +257,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, @@ -877,19 +888,19 @@ 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."), - webhook: Optional[str] = typer.Option( + webhook: str | None = typer.Option( None, "--webhook", help="With --watch, POST a JSON alert to this URL on a verdict transition (overrides watch.webhook_url).", ), - 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." ), - trend: Optional[int] = typer.Option( + trend: int | None = typer.Option( None, "--trend", help="Show a sparkline trend across the last N saved JSON reports for this network. Read-only, no probing.", @@ -917,7 +928,7 @@ def run( "--ecmp", help="Trace a handful of reference hosts multiple times to detect per-hop load balancing (ECMP).", ), - ipv6: Optional[bool] = typer.Option( + ipv6: bool | None = typer.Option( None, "--ipv6/--no-ipv6", help="Also measure latency to IPv6 reference hosts (default: auto, only on a dual-stack connection).", @@ -932,12 +943,12 @@ def run( "--quic", help="Measure QUIC/HTTP3 handshake time and check whether UDP/443 is blocked (needs the quic extra).", ), - 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=( @@ -948,7 +959,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.",