Refactor cli.py imports and typing annotations

Co-authored-by: Horizontal42 <75383519+Horizontal42@users.noreply.github.com>
This commit is contained in:
google-labs-jules[bot] 2026-08-19 11:26:38 +00:00
parent 053eb2f530
commit 2f95a79ef5

View file

@ -8,7 +8,7 @@ import sys
import uuid import uuid
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import Any, List, Optional, Tuple from typing import Any
import httpx import httpx
import psutil import psutil
@ -18,9 +18,15 @@ from rich.console import Console
from netsleuth import __version__, history from netsleuth import __version__, history
from netsleuth.compare import diff_reports, load_report, render_diff, render_diff_brief from netsleuth.compare import diff_reports, load_report, render_diff, render_diff_brief
from netsleuth.config import FORMAT_EXTENSIONS, Settings, load_settings 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.exporter import (
from netsleuth.iface import available_interfaces_hint, resolve_bind_target build_report,
from netsleuth.metrics import collect_metrics, render_csv, render_prometheus dump_json,
egress_asn,
render_markdown,
report_filename,
write_report,
)
from netsleuth.iface import resolve_bind_target
from netsleuth.interpret import ( from netsleuth.interpret import (
assess_vpn, assess_vpn,
captive_portal_findings, captive_portal_findings,
@ -42,6 +48,7 @@ from netsleuth.interpret import (
tls_findings, tls_findings,
) )
from netsleuth.ip_geo import dual_stack_mismatch, gather_identity from netsleuth.ip_geo import dual_stack_mismatch, gather_identity
from netsleuth.metrics import collect_metrics, render_csv, render_prometheus
from netsleuth.models import ( from netsleuth.models import (
BindTarget, BindTarget,
CaptivePortal, CaptivePortal,
@ -53,7 +60,11 @@ from netsleuth.models import (
SpeedResult, SpeedResult,
VpnContext, 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.orchestration import gather_modules, run_module, utc_now_iso
from netsleuth.probes.captive_portal import check_captive_portal from netsleuth.probes.captive_portal import check_captive_portal
from netsleuth.probes.dns_leak import collect_dns_leak from netsleuth.probes.dns_leak import collect_dns_leak
@ -96,7 +107,7 @@ class Options:
ecmp: bool = False ecmp: bool = False
pmtud: bool = False pmtud: bool = False
quic: bool = False quic: bool = False
ipv6: Optional[bool] = None ipv6: bool | None = None
dpi_target: str | None = None dpi_target: str | None = None
formats: frozenset[str] = frozenset({"md"}) formats: frozenset[str] = frozenset({"md"})
interface: str | None = None 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): async def _reputation_section(client, settings: Settings, options: Options, geo: IpGeo, raw: dict):
from netsleuth.reputation import ( from netsleuth.reputation import (
build_reputation,
ReputationContext, ReputationContext,
build_reputation,
fetch_abuseipdb, fetch_abuseipdb,
fetch_internetdb, fetch_internetdb,
normalize_internetdb, normalize_internetdb,
@ -877,19 +888,19 @@ def _check_ndt7_consent(settings: Settings) -> bool:
def run( def run(
full: bool = typer.Option(False, "--full", help="Full run: speedtest and full MTR cycles."), 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."), quick: bool = typer.Option(False, "--quick", help="Express run: reference hosts only, no speedtest."),
target: Optional[str] = typer.Option(None, "--target", help="Investigate AS<n>, an IP or a domain."), target: str | None = typer.Option(None, "--target", help="Investigate AS<n>, an IP or a domain."),
target_host: Optional[str] = typer.Option(None, "--target-host", help="Extra host for ping/traceroute."), target_host: str | None = 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."), 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."), watch: bool = typer.Option(False, "--watch", help="Continuous monitoring with a live dashboard."),
webhook: Optional[str] = typer.Option( webhook: str | None = typer.Option(
None, None,
"--webhook", "--webhook",
help="With --watch, POST a JSON alert to this URL on a verdict transition (overrides watch.webhook_url).", 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." 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, None,
"--trend", "--trend",
help="Show a sparkline trend across the last N saved JSON reports for this network. Read-only, no probing.", 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", "--ecmp",
help="Trace a handful of reference hosts multiple times to detect per-hop load balancing (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, None,
"--ipv6/--no-ipv6", "--ipv6/--no-ipv6",
help="Also measure latency to IPv6 reference hosts (default: auto, only on a dual-stack connection).", help="Also measure latency to IPv6 reference hosts (default: auto, only on a dual-stack connection).",
@ -932,12 +943,12 @@ def run(
"--quic", "--quic",
help="Measure QUIC/HTTP3 handshake time and check whether UDP/443 is blocked (needs the quic extra).", 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, None,
"--my-server", "--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.", 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", "--format",
help=( help=(
@ -948,7 +959,7 @@ def run(
), ),
ru: bool = typer.Option(False, "--ru", help="Shorthand for --format ru-md."), ru: bool = typer.Option(False, "--ru", help="Shorthand for --format ru-md."),
json_flag: bool = typer.Option(False, "--json", help="Shorthand for --format json."), json_flag: bool = typer.Option(False, "--json", help="Shorthand for --format json."),
interface: Optional[str] = typer.Option( interface: str | None = typer.Option(
None, None,
"--interface", "--interface",
help="Force outbound probing through this adapter (name or IP), instead of the OS default route.", help="Force outbound probing through this adapter (name or IP), instead of the OS default route.",