mirror of
https://github.com/Horizontal42/netsleuth.git
synced 2026-11-10 15:42:10 +03:00
add --interface flag to force outbound probing through a specific adapter
Every probe used to follow whatever the OS default route picked, with no way to point the tool at a specific adapter for comparison (VPN vs raw) or to confirm which one actually carries traffic on a multi-adapter machine. New iface.py resolves an adapter name or IP to a BindTarget (pure, tested): exact/case-insensitive/unique-substring name match, rejects APIPA-only, link-local-only and down adapters. Wired through every httpx provider client, the Ookla binary tier, ndt7, and the whole ping/traceroute cascade (icmplib, mtr, unix traceroute, TCP connect, scapy) plus --watch. Windows icmp_win now uses IcmpSendEcho2Ex (adds SourceAddress) only when a bind is forced, so the unforced path is unchanged. Windows tracert has no IPv4 source-address option (verified from its own usage text) so that tier is dropped when forced, noted in the report rather than silently measuring the wrong path. An unresolvable/down adapter aborts with every adapter listed; a resolvable adapter with no route to the internet degrades gracefully instead of aborting - that's a real diagnostic answer, not an error. netinfo's observation of the OS's actual default route stays untouched, so the report can show the forced adapter and the OS default side by side. Verified live: forcing Tailscale (no internet route on this box) triggered the preflight warning and degraded correctly; forcing Ethernet ran clean end-to-end including a full speedtest+MTR cycle. Also noted, not fixed: probes/traceroute.py's icmplib tier imports async_traceroute, which doesn't exist in icmplib 3.0.4 - that tier has always raised ImportError and silently fallen through. Found while verifying the icmplib source-binding wiring live.
This commit is contained in:
Andrew Pimenov
parent
51c01bd8c3
commit
fd72604240
18 changed files with 665 additions and 63 deletions
|
|
@ -13,6 +13,7 @@ src/netsleuth/
|
|||
models.py Every shared dataclass, plus to_jsonable() for serialization.
|
||||
orchestration.py run_module(): timing, timeout, exception -> ProbeError classification.
|
||||
netinfo.py Capability detection and local interface/gateway/MTU/resolver facts.
|
||||
iface.py --interface resolution: adapter name/IP -> BindTarget, pure and tested.
|
||||
ip_geo.py Six-provider identity chain, per-provider normalizers, field-wise merge.
|
||||
bgp.py RIPEstat, CAIDA ASRank, Team Cymru DNS, PeeringDB (disk-cached).
|
||||
reputation.py FireHOL netset cache + prefix lookup, Shodan InternetDB, DNSBL decode.
|
||||
|
|
@ -82,6 +83,10 @@ Two hard concurrency constraints:
|
|||
|
||||
**DNSBL responses in `127.255.255.0/24` are errors, not listings.** Spamhaus returns `127.255.255.254` for "you queried through a public resolver" — which is the case for anyone on 1.1.1.1 or 8.8.8.8 — and `127.255.255.255` for "rate limited". A naive "any `127.x.x.x` means listed" implementation red-flags most users. Both are surfaced as *result unavailable*, never as a finding.
|
||||
|
||||
**`--interface` binds where the code structurally can, drops where it structurally can't, never runs unbound and silent.** Every httpx-based provider client can bind an exact local address (`local_address=` on the transport) — cheap and solid. The Ookla binary tier passes `--interface`/`--ip` through to the external process. `icmp_win` uses `IcmpSendEcho2Ex` (which adds a `SourceAddress`) instead of `IcmpSendEcho2` only when a bind is forced, so the unforced path is byte-for-byte unchanged. The one structural dead end is Windows `tracert.exe`, which exposes a source-address override for IPv6 only, never IPv4 (`-S` is documented IPv6-only in its own usage text) — `filter_trace_tiers()` drops that tier rather than let it silently measure the OS-default path under a report claiming a forced adapter. `netinfo.collect_local_net()`'s own observation of the OS's real default route is deliberately left untouched by the flag: the report needs to show *both* the forced adapter and what the OS would have chosen on its own, and overwriting the observation would turn the VPN gateway/tunnel-interface signals into a tautology (force the VPN adapter, get "VPN confirmed" for free).
|
||||
|
||||
**An interface with no route to the internet is a diagnostic answer, not an error.** Resolving an unknown/down/addressless adapter aborts immediately (`typer.BadParameter`, every adapter listed) — that is bad *input*. But an adapter that resolves fine and turns out to carry nothing is exactly the kind of thing this tool exists to surface, so the run proceeds, every probe fails into the normal `unavailable`/`timeout` classification, and the report says so plainly rather than pretending to be a default run.
|
||||
|
||||
**Bulk RIPEstat endpoints are always bounded.** `routing-history` and friends return 10 MB+ payloads for large ISPs, so every such call carries `max_rows` and a timeframe from `config.yaml`.
|
||||
|
||||
**PeeringDB is cached to disk.** Anonymous limits are 20 requests/minute with large responses throttled to one per hour, so a response is never re-fetched within a run and persists in `.cache/` between runs for `providers.peeringdb_cache_hours`.
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ src/netsleuth/
|
|||
models.py Все общие датаклассы плюс to_jsonable() для сериализации.
|
||||
orchestration.py run_module(): тайминг, таймаут, классификация исключений в ProbeError.
|
||||
netinfo.py Определение возможностей и локальные факты: интерфейс, шлюз, MTU, резолверы.
|
||||
iface.py Резолв --interface: имя/IP адаптера -> BindTarget, чистый код, покрыт тестами.
|
||||
ip_geo.py Цепочка из шести провайдеров, нормализаторы, пополевое слияние.
|
||||
bgp.py RIPEstat, CAIDA ASRank, Team Cymru через DNS, PeeringDB (кэш на диске).
|
||||
reputation.py Кэш netset'ов FireHOL и локальный поиск, Shodan InternetDB, декодер DNSBL.
|
||||
|
|
@ -82,6 +83,10 @@ src/netsleuth/
|
|||
|
||||
**Ответы DNSBL в диапазоне `127.255.255.0/24` — это ошибки, а не листинг.** Spamhaus возвращает `127.255.255.254` на «вы спросили через публичный резолвер» — а это все, кто сидит на 1.1.1.1 или 8.8.8.8 — и `127.255.255.255` на «превышен лимит». Наивная реализация «любой `127.x.x.x` значит в списке» помечает красным большинство пользователей. Оба случая подаются как *результат недоступен*, но никогда как находка.
|
||||
|
||||
**`--interface` привязывается там, где это структурно возможно, отбрасывается там, где невозможно, и никогда не работает молча "без привязки".** Любой httpx-клиент-провайдер умеет привязаться к точному локальному адресу (`local_address=` на транспорте) — дёшево и надёжно. Уровень бинарника Ookla прокидывает `--interface`/`--ip` в сам процесс. `icmp_win` использует `IcmpSendEcho2Ex` (добавляет `SourceAddress`) вместо `IcmpSendEcho2` только когда привязка запрошена, поэтому непривязанный путь не меняется ни на байт. Единственный структурный тупик — Windows `tracert.exe`: у него есть переопределение адреса-источника только для IPv6, никогда для IPv4 (`-S` документирован как IPv6-only в его же справке) — `filter_trace_tiers()` отбрасывает этот уровень, а не даёт ему молча замерить путь по умолчанию ОС под отчётом, заявляющим о принудительном адаптере. Собственное наблюдение `netinfo.collect_local_net()` за реальным маршрутом ОС по умолчанию намеренно не трогается флагом: отчёт должен показывать *и* принудительный адаптер, *и* то, что ОС выбрала бы сама — перезапись наблюдения превратила бы сигналы шлюза/туннельного интерфейса для VPN в тавтологию (форсируешь VPN-адаптер — бесплатно получаешь "VPN подтверждён").
|
||||
|
||||
**Интерфейс без маршрута в интернет — диагностический ответ, а не ошибка.** Резолв неизвестного/выключенного/безадресного адаптера падает сразу (`typer.BadParameter`, со списком всех адаптеров) — это плохой *ввод*. А адаптер, который резолвится нормально, но не несёт трафика — ровно то, для чего существует эта утилита, поэтому прогон продолжается, каждая проба падает в обычную классификацию `unavailable`/`timeout`, и отчёт прямо это показывает, а не притворяется обычным прогоном.
|
||||
|
||||
**Массовые эндпоинты RIPEstat всегда ограничены.** `routing-history` и подобные на крупных провайдерах возвращают 10 МБ+, поэтому каждый такой вызов несёт `max_rows` и временное окно из `config.yaml`.
|
||||
|
||||
**PeeringDB кэшируется на диск.** Анонимные лимиты — 20 запросов в минуту, а крупные ответы душатся до одного в час, поэтому ответ не запрашивается повторно внутри прогона и сохраняется в `.cache/` между прогонами на `providers.peeringdb_cache_hours`.
|
||||
|
|
|
|||
|
|
@ -73,6 +73,7 @@ No configuration is required. Copy `.env.example` to `.env` only if you have opt
|
|||
| `--format md,ru-md,json,all,none` | Choose which report artifacts to write (default: `md`). Repeatable and comma-separated (`--format md,json`); any use replaces the default instead of adding to it. `all` writes all three, `none` writes nothing. Ignored with `--watch`/`--compare` |
|
||||
| `--ru` | Shorthand for `--format ru-md` |
|
||||
| `--json` | Shorthand for `--format json` |
|
||||
| `--interface <name\|ip>` | Force outbound probing (identity, BGP/reputation, latency, traceroute, speedtest) through this adapter instead of the OS default route — e.g. compare a VPN adapter against your raw connection. Accepts an adapter name or one of its IPs. Errors out immediately, listing every adapter, if the name doesn't resolve or the adapter is down; if it resolves but has no route to the internet, the run continues and reports that honestly rather than aborting. Windows' `tracert` has no IPv4 source-address option, so that one traceroute tier is skipped when forced (noted in the report) — every other backend, including the Windows ICMP API, honors it |
|
||||
|
||||
Each run writes the selected artifacts into `./logs/`: an English Markdown report, a full Russian translation of the same report (`.ru.md`, same sections, same tables, same numbers — cross-linked with the English one when both are written), and/or a JSON dump containing every raw provider response, untruncated. Default is the English Markdown report only; set `output.formats` in `config.yaml` to change the default permanently, or pass `--format` per run. `./logs/` is gitignored — reports contain your external IP, city, coordinates and ISP name.
|
||||
|
||||
|
|
|
|||
|
|
@ -73,6 +73,7 @@ netsleuth
|
|||
| `--format md,ru-md,json,all,none` | Выбрать, какие артефакты отчёта писать (по умолчанию: `md`). Повторяемый флаг и через запятую (`--format md,json`); любое использование заменяет дефолт, а не дополняет его. `all` пишет все три, `none` — ничего. Игнорируется с `--watch`/`--compare` |
|
||||
| `--ru` | Сокращение для `--format ru-md` |
|
||||
| `--json` | Сокращение для `--format json` |
|
||||
| `--interface <имя\|ip>` | Пустить все исходящие пробы (идентификация, BGP/репутация, задержки, трассировка, спидтест) через конкретный адаптер вместо маршрута ОС по умолчанию — например, сравнить VPN-адаптер с "сырым" подключением. Принимает имя адаптера или один из его IP. Сразу падает с ошибкой и списком всех адаптеров, если имя не резолвится или адаптер выключен; если адаптер есть, но маршрута в интернет нет — прогон продолжается и честно это показывает, а не падает. У Windows `tracert` нет опции адреса-источника для IPv4, поэтому этот уровень трассировки при форсированном интерфейсе пропускается (отмечено в отчёте) — все остальные бэкенды, включая Windows ICMP API, флаг соблюдают |
|
||||
|
||||
Каждый прогон пишет в `./logs/` выбранные артефакты: Markdown-отчёт на английском, его полный перевод на русский (`.ru.md` — те же разделы, те же таблицы, те же числа, со взаимными ссылками, если пишутся оба) и/или JSON-дамп со всеми сырыми ответами провайдеров без усечения. По умолчанию — только английский Markdown; чтобы поменять дефолт навсегда, задайте `output.formats` в `config.yaml`, либо используйте `--format` на конкретном прогоне. `./logs/` в gitignore — отчёты содержат ваш внешний IP, город, координаты и название провайдера.
|
||||
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ from __future__ import annotations
|
|||
import asyncio
|
||||
import os
|
||||
import platform
|
||||
import socket
|
||||
import sys
|
||||
import uuid
|
||||
from dataclasses import dataclass
|
||||
|
|
@ -10,6 +11,7 @@ from pathlib import Path
|
|||
from typing import Any, List, Optional, Tuple
|
||||
|
||||
import httpx
|
||||
import psutil
|
||||
import typer
|
||||
from rich.console import Console
|
||||
|
||||
|
|
@ -17,6 +19,7 @@ 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.interpret import (
|
||||
assess_vpn,
|
||||
gather_vpn_signals,
|
||||
|
|
@ -25,12 +28,12 @@ from netsleuth.interpret import (
|
|||
speed_findings,
|
||||
)
|
||||
from netsleuth.ip_geo import dual_stack_mismatch, gather_identity
|
||||
from netsleuth.models import Finding, IpGeo, LocalNet, ModuleResult, SpeedResult
|
||||
from netsleuth.netinfo import collect_local_net, detect_capabilities, is_tunnel_iface
|
||||
from netsleuth.models import BindTarget, Finding, IpGeo, LocalNet, ModuleResult, SpeedResult
|
||||
from netsleuth.netinfo import collect_local_net, detect_capabilities, iface_for_ip, is_tunnel_iface, primary_interface_ip
|
||||
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
|
||||
from netsleuth.probes.traceroute import traceroute
|
||||
from netsleuth.probes.traceroute import filter_trace_tiers, tier_order, traceroute
|
||||
from netsleuth.speed import NDT7_CONSENT_NOTICE
|
||||
|
||||
if sys.platform == "win32":
|
||||
|
|
@ -60,6 +63,8 @@ class Options:
|
|||
ndt7: bool = False
|
||||
tcp_trace: bool = False
|
||||
formats: frozenset[str] = frozenset({"md"})
|
||||
interface: str | None = None
|
||||
bind: BindTarget | None = None
|
||||
|
||||
|
||||
def parse_formats(
|
||||
|
|
@ -136,13 +141,15 @@ async def _identity(client: httpx.AsyncClient, settings: Settings, options: Opti
|
|||
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")
|
||||
bind_v4 = options.bind.ipv4 if options.bind else "0.0.0.0"
|
||||
warnings: list[str] = []
|
||||
if own_network:
|
||||
# The shared client has no address family pinned, so on a dual-stack host it can
|
||||
# silently connect over IPv6 (OS/happy-eyeballs preference) and hand back an IPv6
|
||||
# literal mislabeled as egress_v4. Force IPv4 here the same way egress_v6 is forced
|
||||
# below, so the two slots can never both resolve to the same protocol.
|
||||
try:
|
||||
transport = httpx.AsyncHTTPTransport(local_address="0.0.0.0")
|
||||
transport = httpx.AsyncHTTPTransport(local_address=bind_v4)
|
||||
async with httpx.AsyncClient(
|
||||
transport=transport, timeout=settings.timeouts.http_seconds
|
||||
) as v4_client:
|
||||
|
|
@ -159,9 +166,13 @@ async def _identity(client: httpx.AsyncClient, settings: Settings, options: Opti
|
|||
)
|
||||
raw.update(payloads)
|
||||
v6 = None
|
||||
if own_network:
|
||||
if own_network and options.bind and options.bind.ipv6 is None:
|
||||
warnings.append(
|
||||
f"IPv6 identity skipped: forced interface {options.bind.iface_name!r} has no global IPv6 address"
|
||||
)
|
||||
elif own_network:
|
||||
try:
|
||||
transport = httpx.AsyncHTTPTransport(local_address="::")
|
||||
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:
|
||||
|
|
@ -177,6 +188,7 @@ async def _identity(client: httpx.AsyncClient, settings: Settings, options: Opti
|
|||
"dual_stack_note": dual_stack[0] if dual_stack else None,
|
||||
"dual_stack_note_ru": dual_stack[1] if dual_stack else None,
|
||||
"_flags": flags,
|
||||
"_warnings": warnings,
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -248,6 +260,7 @@ async def _traces(hosts, caps, settings: Settings, cycles: int, options: Options
|
|||
timeout=settings.timeouts.subprocess_seconds,
|
||||
semaphore=semaphore,
|
||||
tcp_trace=options.tcp_trace,
|
||||
source_ip=options.bind.ipv4 if options.bind else None,
|
||||
)
|
||||
for _label, host in hosts
|
||||
)
|
||||
|
|
@ -269,13 +282,18 @@ async def _speed_section(client, settings: Settings, options: Options, idle_rtt_
|
|||
|
||||
cfg = settings.speedtest
|
||||
timeout = settings.timeouts.speedtest_seconds
|
||||
bind = options.bind
|
||||
builders = {
|
||||
"ookla_bin": lambda: tier_ookla(
|
||||
shutil.which("speedtest") or "speedtest", options.speedtest_server, timeout
|
||||
shutil.which("speedtest") or "speedtest",
|
||||
options.speedtest_server,
|
||||
timeout,
|
||||
iface_name=bind.iface_name if bind else None,
|
||||
ipv4=bind.ipv4 if bind else None,
|
||||
),
|
||||
"cloudflare": lambda: tier_cloudflare(client, cfg, timeout),
|
||||
"fastcom": lambda: tier_fastcom(client, cfg, timeout),
|
||||
"ndt7": lambda: tier_ndt7(client, cfg, timeout),
|
||||
"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])
|
||||
|
|
@ -314,7 +332,11 @@ async def diagnose(settings: Settings, options: Options) -> tuple[dict, list[Pat
|
|||
modules: dict[str, ModuleResult] = {}
|
||||
raw: dict[str, Any] = {}
|
||||
|
||||
client_transport = (
|
||||
httpx.AsyncHTTPTransport(local_address=options.bind.ipv4, http2=True) if options.bind else None
|
||||
)
|
||||
async with httpx.AsyncClient(
|
||||
transport=client_transport,
|
||||
timeout=timeouts.http_seconds,
|
||||
follow_redirects=True,
|
||||
http2=True,
|
||||
|
|
@ -330,9 +352,11 @@ async def diagnose(settings: Settings, options: Options) -> tuple[dict, list[Pat
|
|||
)
|
||||
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 []
|
||||
geo = bundle.get("egress_v4") or IpGeo()
|
||||
if bundle.get("dual_stack_note"):
|
||||
modules["ip_geo"].warnings.append(bundle["dual_stack_note"])
|
||||
modules["ip_geo"].warnings.extend(identity_warnings)
|
||||
asn = geo.asn if options.target_kind != "asn" else options.target_value
|
||||
|
||||
# Phase 2 — bgp || reputation || dns_leak
|
||||
|
|
@ -390,6 +414,7 @@ async def diagnose(settings: Settings, options: Options) -> tuple[dict, list[Pat
|
|||
count,
|
||||
settings.probing.ping_interval_seconds,
|
||||
settings.probing.ping_timeout_seconds,
|
||||
source_ip=options.bind.ipv4 if options.bind else None,
|
||||
),
|
||||
timeout=timeouts.subprocess_seconds,
|
||||
),
|
||||
|
|
@ -432,11 +457,25 @@ async def diagnose(settings: Settings, options: Options) -> tuple[dict, list[Pat
|
|||
"dnsbl": options.dnsbl,
|
||||
"ndt7": options.ndt7,
|
||||
"tcp_trace": options.tcp_trace,
|
||||
"interface": options.interface,
|
||||
},
|
||||
"formats": sorted(options.formats),
|
||||
"host_os": f"{platform.system()} {platform.release()}",
|
||||
"capabilities": caps,
|
||||
}
|
||||
if options.bind:
|
||||
_kept, dropped = filter_trace_tiers(
|
||||
tier_order(caps, options.tcp_trace), caps.os_name, forced_v4=bool(options.bind.ipv4)
|
||||
)
|
||||
meta["interface"] = {
|
||||
"requested": options.bind.requested,
|
||||
"resolved_iface": options.bind.iface_name,
|
||||
"bind_ipv4": options.bind.ipv4,
|
||||
"bind_ipv6": options.bind.ipv6,
|
||||
"os_default_iface": local.iface_name,
|
||||
"os_default_ipv4": local.local_ipv4,
|
||||
"dropped_backends": dropped,
|
||||
}
|
||||
report = build_report(meta, modules, findings, raw)
|
||||
asn = meta.get("target") or egress_asn(report)
|
||||
md_name = report_filename(asn, started_at, "md")
|
||||
|
|
@ -478,6 +517,11 @@ 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(
|
||||
None,
|
||||
"--interface",
|
||||
help="Force outbound probing through this adapter (name or IP), instead of the OS default route.",
|
||||
),
|
||||
) -> None:
|
||||
settings = load_settings()
|
||||
if compare:
|
||||
|
|
@ -500,6 +544,29 @@ def run(
|
|||
tcp_trace=tcp_trace,
|
||||
formats=formats,
|
||||
)
|
||||
if interface:
|
||||
addrs_by_iface = {
|
||||
name: [(a.family, a.address.split("%")[0]) for a in addrs]
|
||||
for name, addrs in psutil.net_if_addrs().items()
|
||||
}
|
||||
up_by_iface = {name: stats.isup for name, stats in psutil.net_if_stats().items()}
|
||||
bind = resolve_bind_target(interface, addrs_by_iface, up_by_iface)
|
||||
if bind.error:
|
||||
raise typer.BadParameter(bind.error, param_hint="--interface")
|
||||
options.interface = interface
|
||||
options.bind = bind
|
||||
if bind.ipv4:
|
||||
probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
try:
|
||||
probe.bind((bind.ipv4, 0))
|
||||
probe.connect(("1.1.1.1", 53))
|
||||
except OSError:
|
||||
console.print(
|
||||
f"[yellow]--interface {bind.iface_name}: no route to the internet from this "
|
||||
"address yet — continuing, this run may show every module as unreachable.[/yellow]"
|
||||
)
|
||||
finally:
|
||||
probe.close()
|
||||
if ndt7:
|
||||
consent_marker = Path(settings.output.cache_dir) / "ndt7_consent"
|
||||
if not consent_marker.exists():
|
||||
|
|
|
|||
|
|
@ -281,6 +281,22 @@ def unavailable(report: dict[str, Any], section: str, lang: str = "en") -> str |
|
|||
return f"_{label} — {status_word}: {detail}._"
|
||||
|
||||
|
||||
def _forced_interface_rows(meta: dict, lang: str = "en") -> list[list[str]]:
|
||||
forced = meta.get("interface")
|
||||
if not forced:
|
||||
return []
|
||||
rows = [
|
||||
[
|
||||
_t(lang, "Forced interface", "Принудительный интерфейс"),
|
||||
f"{forced.get('resolved_iface') or '?'} — {forced.get('bind_ipv4') or '?'} (--interface)",
|
||||
]
|
||||
]
|
||||
os_iface, os_ipv4 = forced.get("os_default_iface"), forced.get("os_default_ipv4")
|
||||
if os_iface and (os_iface, os_ipv4) != (forced.get("resolved_iface"), forced.get("bind_ipv4")):
|
||||
rows.append([_t(lang, "OS default interface", "Интерфейс ОС по умолчанию"), f"{os_iface} — {os_ipv4 or '?'}"])
|
||||
return rows
|
||||
|
||||
|
||||
def _connection(report: dict, lang: str = "en") -> list[str]:
|
||||
title = _section_title("connection", lang)
|
||||
conn_note = unavailable(report, "connection", lang)
|
||||
|
|
@ -293,8 +309,10 @@ def _connection(report: dict, lang: str = "en") -> list[str]:
|
|||
geo = geo_bundle.get("egress_v4") or {}
|
||||
v6 = geo_bundle.get("egress_v6") or {}
|
||||
cf = geo_bundle.get("cf_trace") or {}
|
||||
meta = report.get("meta") or {}
|
||||
rows = [
|
||||
[_t(lang, "Interface", "Интерфейс"), f"{local.get('iface_name') or '?'} (MTU {local.get('iface_mtu') or '?'})"],
|
||||
*_forced_interface_rows(meta, lang),
|
||||
[
|
||||
_t(lang, "Local address", "Локальный адрес"),
|
||||
f"{local.get('local_ipv4') or '?'} / {local.get('local_ipv6') or '—'}",
|
||||
|
|
@ -317,6 +335,17 @@ def _connection(report: dict, lang: str = "en") -> list[str]:
|
|||
note = geo_bundle.get("dual_stack_note_ru" if lang == "ru" else "dual_stack_note")
|
||||
if note:
|
||||
lines += [f"> {note}", ""]
|
||||
dropped = (meta.get("interface") or {}).get("dropped_backends") or []
|
||||
if dropped:
|
||||
lines += [
|
||||
"> "
|
||||
+ _t(
|
||||
lang,
|
||||
f"--interface could not be honored by {', '.join(dropped)} (no IPv4 source-address option); that tier was skipped for this run.",
|
||||
f"--interface не удалось применить к {', '.join(dropped)} (нет опции адреса-источника IPv4); этот уровень пропущен для данного запуска.",
|
||||
),
|
||||
"",
|
||||
]
|
||||
return lines
|
||||
|
||||
|
||||
|
|
|
|||
87
src/netsleuth/iface.py
Normal file
87
src/netsleuth/iface.py
Normal file
|
|
@ -0,0 +1,87 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import ipaddress
|
||||
import socket
|
||||
|
||||
from netsleuth.models import BindTarget
|
||||
from netsleuth.netinfo import iface_for_ip
|
||||
|
||||
|
||||
def usable_addresses(addrs: list[tuple[int, str]]) -> tuple[str | None, str | None]:
|
||||
v4 = v6 = None
|
||||
for family, address in addrs:
|
||||
if family == socket.AF_INET and v4 is None:
|
||||
if not ipaddress.IPv4Address(address).is_loopback and not address.startswith("169.254."):
|
||||
v4 = address
|
||||
elif family == socket.AF_INET6 and v6 is None:
|
||||
zone_free = address.split("%")[0]
|
||||
parsed = ipaddress.IPv6Address(zone_free)
|
||||
if not parsed.is_loopback and not parsed.is_link_local:
|
||||
v6 = zone_free
|
||||
return v4, v6
|
||||
|
||||
|
||||
def _match_iface_name(requested: str, names: list[str]) -> str | list[str]:
|
||||
if requested in names:
|
||||
return requested
|
||||
lowered = requested.lower()
|
||||
exact_ci = [n for n in names if n.lower() == lowered]
|
||||
if len(exact_ci) == 1:
|
||||
return exact_ci[0]
|
||||
substring = [n for n in names if lowered in n.lower()]
|
||||
if len(substring) == 1:
|
||||
return substring[0]
|
||||
return substring or exact_ci
|
||||
|
||||
|
||||
def resolve_bind_target(
|
||||
requested: str,
|
||||
addrs_by_iface: dict[str, list[tuple[int, str]]],
|
||||
up_by_iface: dict[str, bool],
|
||||
) -> BindTarget:
|
||||
try:
|
||||
ip = str(ipaddress.ip_address(requested))
|
||||
except ValueError:
|
||||
ip = None
|
||||
|
||||
if ip is not None:
|
||||
iface_name = iface_for_ip(ip, addrs_by_iface)
|
||||
if iface_name is None:
|
||||
return BindTarget(requested=requested, error=f"no adapter has the address {requested!r}")
|
||||
else:
|
||||
match = _match_iface_name(requested, list(addrs_by_iface))
|
||||
if isinstance(match, list):
|
||||
hint = available_interfaces_hint(addrs_by_iface, up_by_iface)
|
||||
if not match:
|
||||
return BindTarget(requested=requested, error=f"no adapter matches {requested!r}.\n{hint}")
|
||||
return BindTarget(
|
||||
requested=requested,
|
||||
error=f"{requested!r} matches multiple adapters: {', '.join(match)}. Be more specific.",
|
||||
)
|
||||
iface_name = match
|
||||
|
||||
is_up = up_by_iface.get(iface_name, False)
|
||||
v4, v6 = usable_addresses(addrs_by_iface.get(iface_name, []))
|
||||
if not is_up:
|
||||
return BindTarget(requested=requested, iface_name=iface_name, is_up=False, error=f"adapter {iface_name!r} is down")
|
||||
if v4 is None and v6 is None:
|
||||
return BindTarget(
|
||||
requested=requested,
|
||||
iface_name=iface_name,
|
||||
is_up=is_up,
|
||||
error=f"adapter {iface_name!r} has no usable (routable) address",
|
||||
)
|
||||
return BindTarget(requested=requested, iface_name=iface_name, ipv4=v4, ipv6=v6, is_up=is_up)
|
||||
|
||||
|
||||
def available_interfaces_hint(
|
||||
addrs_by_iface: dict[str, list[tuple[int, str]]],
|
||||
up_by_iface: dict[str, bool],
|
||||
) -> str:
|
||||
lines = ["Available adapters:"]
|
||||
for name, addrs in addrs_by_iface.items():
|
||||
v4, v6 = usable_addresses(addrs)
|
||||
state = "up" if up_by_iface.get(name, False) else "down"
|
||||
ips = ", ".join(a for a in (v4, v6) if a) or "no usable address"
|
||||
lines.append(f" {name} ({state}): {ips}")
|
||||
return "\n".join(lines)
|
||||
|
|
@ -95,6 +95,16 @@ class Capabilities:
|
|||
notes: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
@dataclass
|
||||
class BindTarget:
|
||||
requested: str
|
||||
iface_name: str | None = None
|
||||
ipv4: str | None = None
|
||||
ipv6: str | None = None
|
||||
is_up: bool = False
|
||||
error: str | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class LocalNet:
|
||||
iface_name: str | None = None
|
||||
|
|
|
|||
|
|
@ -100,6 +100,23 @@ def _handle():
|
|||
ctypes.c_uint32,
|
||||
]
|
||||
iphlpapi.IcmpSendEcho2.restype = ctypes.c_uint32
|
||||
# Ex variant inserts SourceAddress (IPAddr) before DestinationAddress; only
|
||||
# used when a caller asks to bind to a specific adapter's address.
|
||||
iphlpapi.IcmpSendEcho2Ex.argtypes = [
|
||||
wintypes.HANDLE,
|
||||
wintypes.HANDLE,
|
||||
ctypes.c_void_p,
|
||||
ctypes.c_void_p,
|
||||
ctypes.c_uint32,
|
||||
ctypes.c_uint32,
|
||||
ctypes.c_void_p,
|
||||
ctypes.c_uint16,
|
||||
ctypes.c_void_p,
|
||||
ctypes.c_void_p,
|
||||
ctypes.c_uint32,
|
||||
ctypes.c_uint32,
|
||||
]
|
||||
iphlpapi.IcmpSendEcho2Ex.restype = ctypes.c_uint32
|
||||
iphlpapi.IcmpCloseHandle.argtypes = [wintypes.HANDLE]
|
||||
handle = iphlpapi.IcmpCreateFile()
|
||||
if handle == wintypes.HANDLE(-1).value:
|
||||
|
|
@ -107,25 +124,44 @@ def _handle():
|
|||
return iphlpapi, handle
|
||||
|
||||
|
||||
def echo_once(dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netsleuth") -> IcmpReply:
|
||||
def echo_once(
|
||||
dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netsleuth", source_ip: str | None = None
|
||||
) -> IcmpReply:
|
||||
iphlpapi, handle = _handle()
|
||||
try:
|
||||
options = _IpOptionInformation(Ttl=ttl, Tos=0, Flags=0, OptionsSize=0, OptionsData=None)
|
||||
buffer = ctypes.create_string_buffer(_REPLY_BUFFER_SIZE)
|
||||
address = struct.unpack("<I", socket.inet_aton(dest))[0]
|
||||
count = iphlpapi.IcmpSendEcho2(
|
||||
handle,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
ctypes.c_uint32(address),
|
||||
payload,
|
||||
ctypes.c_ushort(len(payload)),
|
||||
ctypes.byref(options),
|
||||
buffer,
|
||||
ctypes.c_uint32(_REPLY_BUFFER_SIZE),
|
||||
ctypes.c_uint32(timeout_ms),
|
||||
)
|
||||
if source_ip:
|
||||
source = struct.unpack("<I", socket.inet_aton(source_ip))[0]
|
||||
count = iphlpapi.IcmpSendEcho2Ex(
|
||||
handle,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
ctypes.c_uint32(source),
|
||||
ctypes.c_uint32(address),
|
||||
payload,
|
||||
ctypes.c_ushort(len(payload)),
|
||||
ctypes.byref(options),
|
||||
buffer,
|
||||
ctypes.c_uint32(_REPLY_BUFFER_SIZE),
|
||||
ctypes.c_uint32(timeout_ms),
|
||||
)
|
||||
else:
|
||||
count = iphlpapi.IcmpSendEcho2(
|
||||
handle,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
ctypes.c_uint32(address),
|
||||
payload,
|
||||
ctypes.c_ushort(len(payload)),
|
||||
ctypes.byref(options),
|
||||
buffer,
|
||||
ctypes.c_uint32(_REPLY_BUFFER_SIZE),
|
||||
ctypes.c_uint32(timeout_ms),
|
||||
)
|
||||
if count == 0:
|
||||
return IcmpReply(address=None, status=IP_REQ_TIMED_OUT, rtt_ms=None, ttl=0)
|
||||
return parse_echo_reply(buffer.raw, pointer_size=ctypes.sizeof(ctypes.c_void_p))
|
||||
|
|
@ -133,14 +169,16 @@ def echo_once(dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netsleuth
|
|||
iphlpapi.IcmpCloseHandle(handle)
|
||||
|
||||
|
||||
def ping_samples_win(host: str, count: int, interval: float, timeout: float) -> list[float | None]:
|
||||
def ping_samples_win(
|
||||
host: str, count: int, interval: float, timeout: float, source_ip: str | None = None
|
||||
) -> list[float | None]:
|
||||
dest = socket.gethostbyname(host)
|
||||
timeout_ms = int(timeout * 1000)
|
||||
samples: list[float | None] = []
|
||||
for index in range(count):
|
||||
if index:
|
||||
time.sleep(interval)
|
||||
reply = echo_once(dest, ttl=128, timeout_ms=timeout_ms)
|
||||
reply = echo_once(dest, ttl=128, timeout_ms=timeout_ms, source_ip=source_ip)
|
||||
samples.append(reply.rtt_ms if classify_status(reply.status) == "ok" else None)
|
||||
return samples
|
||||
|
||||
|
|
@ -150,7 +188,9 @@ def ping_samples_win(host: str, count: int, interval: float, timeout: float) ->
|
|||
_MAX_HOP_TIMEOUT_MS = 4000
|
||||
|
||||
|
||||
def trace_hops_win(dest: str, max_hops: int, timeout_ms: int) -> list[tuple[int, IcmpReply]]:
|
||||
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.
|
||||
|
|
@ -161,7 +201,7 @@ def trace_hops_win(dest: str, max_hops: int, timeout_ms: int) -> list[tuple[int,
|
|||
for ttl in range(1, max_hops + 1):
|
||||
if time.monotonic() >= deadline:
|
||||
break
|
||||
reply = echo_once(address, ttl=ttl, timeout_ms=per_hop_timeout_ms)
|
||||
reply = echo_once(address, ttl=ttl, timeout_ms=per_hop_timeout_ms, source_ip=source_ip)
|
||||
hops.append((ttl, reply))
|
||||
if classify_status(reply.status) == "ok":
|
||||
break
|
||||
|
|
|
|||
|
|
@ -34,10 +34,15 @@ def summarize_ping(
|
|||
)
|
||||
|
||||
|
||||
async def tcp_connect_rtt(host: str, port: int = 443, timeout: float = 2.0) -> float | None:
|
||||
async def tcp_connect_rtt(
|
||||
host: str, port: int = 443, timeout: float = 2.0, source_ip: str | None = None
|
||||
) -> float | None:
|
||||
began = time.perf_counter()
|
||||
local_addr = (source_ip, 0) if source_ip else None
|
||||
try:
|
||||
_reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=timeout)
|
||||
_reader, writer = await asyncio.wait_for(
|
||||
asyncio.open_connection(host, port, local_addr=local_addr), timeout=timeout
|
||||
)
|
||||
except (OSError, asyncio.TimeoutError):
|
||||
return None
|
||||
writer.close()
|
||||
|
|
@ -55,16 +60,18 @@ def _resolve(host: str) -> str | None:
|
|||
return None
|
||||
|
||||
|
||||
async def _icmp_samples(host: str, count: int, interval: float, timeout: float, backend: str) -> list[float | None]:
|
||||
async def _icmp_samples(
|
||||
host: str, count: int, interval: float, timeout: float, backend: str, source_ip: str | None = None
|
||||
) -> list[float | None]:
|
||||
if backend == "icmp_win":
|
||||
from netsleuth.probes.icmp_win import ping_samples_win
|
||||
|
||||
return await asyncio.to_thread(ping_samples_win, host, count, interval, timeout)
|
||||
return await asyncio.to_thread(ping_samples_win, host, count, interval, timeout, source_ip)
|
||||
from icmplib import async_ping
|
||||
|
||||
privileged = backend == "icmp_raw"
|
||||
host_result = await async_ping(
|
||||
host, count=count, interval=interval, timeout=timeout, privileged=privileged
|
||||
host, count=count, interval=interval, timeout=timeout, privileged=privileged, source=source_ip
|
||||
)
|
||||
samples: list[float | None] = list(host_result.rtts)
|
||||
samples.extend([None] * (count - len(samples)))
|
||||
|
|
@ -78,11 +85,12 @@ async def ping_host(
|
|||
interval: float,
|
||||
timeout: float,
|
||||
backend: str,
|
||||
source_ip: str | None = None,
|
||||
) -> PingResult:
|
||||
resolved = _resolve(host)
|
||||
if backend in ("icmp_win", "icmp_dgram", "icmp_raw"):
|
||||
try:
|
||||
samples = await _icmp_samples(host, count, interval, timeout, backend)
|
||||
samples = await _icmp_samples(host, count, interval, timeout, backend, source_ip)
|
||||
return summarize_ping(label, host, resolved, backend, samples)
|
||||
except Exception:
|
||||
backend = "tcp"
|
||||
|
|
@ -90,7 +98,7 @@ async def ping_host(
|
|||
for index in range(count):
|
||||
if index:
|
||||
await asyncio.sleep(interval)
|
||||
samples.append(await tcp_connect_rtt(host, timeout=timeout))
|
||||
samples.append(await tcp_connect_rtt(host, timeout=timeout, source_ip=source_ip))
|
||||
return summarize_ping(label, host, resolved, "tcp", samples)
|
||||
|
||||
|
||||
|
|
@ -100,10 +108,11 @@ async def ping_fanout(
|
|||
count: int,
|
||||
interval: float,
|
||||
timeout: float,
|
||||
source_ip: str | None = None,
|
||||
) -> list[PingResult]:
|
||||
backend = choose_latency_backend(caps)
|
||||
return list(
|
||||
await asyncio.gather(
|
||||
*(ping_host(host, label, count, interval, timeout, backend) for label, host in hosts)
|
||||
*(ping_host(host, label, count, interval, timeout, backend, source_ip) for label, host in hosts)
|
||||
)
|
||||
)
|
||||
|
|
|
|||
|
|
@ -53,6 +53,41 @@ def tier_order(caps: Capabilities, tcp_trace: bool = False) -> list[str]:
|
|||
return order
|
||||
|
||||
|
||||
def filter_trace_tiers(order: list[str], os_name: str, forced_v4: bool) -> tuple[list[str], list[str]]:
|
||||
# Windows tracert exposes a source-address override for IPv6 only (-S), never
|
||||
# for IPv4, so it cannot honor a forced IPv4 bind; running it anyway would
|
||||
# silently measure the OS-default path under a report claiming otherwise.
|
||||
if not forced_v4 or os_name != "Windows" or "system_traceroute" not in order:
|
||||
return order, []
|
||||
return [t for t in order if t != "system_traceroute"], ["system_traceroute"]
|
||||
|
||||
|
||||
def trace_argv(
|
||||
backend: str,
|
||||
binary: str,
|
||||
target: str,
|
||||
max_hops: int,
|
||||
cycles: int,
|
||||
os_name: str,
|
||||
source_ip: str | None = None,
|
||||
) -> list[str]:
|
||||
if backend == "mtr_json":
|
||||
args = [binary, "--json", "--report-cycles", str(cycles), "--max-ttl", str(max_hops)]
|
||||
if source_ip:
|
||||
args += ["--address", source_ip]
|
||||
args.append(target)
|
||||
return args
|
||||
if backend == "system_traceroute":
|
||||
if os_name == "Windows":
|
||||
return [binary, "-h", str(max_hops), "-w", "2", target]
|
||||
args = [binary, "-m", str(max_hops), "-w", "2"]
|
||||
if source_ip:
|
||||
args += ["-s", source_ip]
|
||||
args.append(target)
|
||||
return args
|
||||
raise ValueError(f"unsupported backend for trace_argv: {backend!r}")
|
||||
|
||||
|
||||
async def run_cascade(tiers: list[tuple[str, Callable[[], Awaitable[TraceResult]]]]) -> TraceResult:
|
||||
for _name, tier in tiers:
|
||||
try:
|
||||
|
|
@ -100,9 +135,11 @@ def _console_encoding() -> str:
|
|||
return "utf-8"
|
||||
|
||||
|
||||
async def _tier_mtr(target: str, binary: str, cycles: int, max_hops: int, timeout: float) -> TraceResult:
|
||||
async def _tier_mtr(
|
||||
target: str, binary: str, cycles: int, max_hops: int, timeout: float, source_ip: str | None = None
|
||||
) -> TraceResult:
|
||||
text = await _run(
|
||||
[binary, "--json", "--report-cycles", str(cycles), "--max-ttl", str(max_hops), target], timeout
|
||||
trace_argv("mtr_json", binary, target, max_hops, cycles, platform.system(), source_ip), timeout
|
||||
)
|
||||
payload = json.loads(text)
|
||||
report = payload.get("report") or {}
|
||||
|
|
@ -128,10 +165,12 @@ async def _tier_mtr(target: str, binary: str, cycles: int, max_hops: int, timeou
|
|||
)
|
||||
|
||||
|
||||
async def _tier_icmp_win(target: str, max_hops: int, timeout: float) -> TraceResult:
|
||||
async def _tier_icmp_win(
|
||||
target: str, max_hops: int, timeout: float, source_ip: str | None = None
|
||||
) -> TraceResult:
|
||||
from netsleuth.probes.icmp_win import trace_hops_win
|
||||
|
||||
replies = await _run_in_daemon_thread(trace_hops_win, target, max_hops, int(timeout * 1000))
|
||||
replies = await _run_in_daemon_thread(trace_hops_win, target, max_hops, int(timeout * 1000), source_ip)
|
||||
hops = hops_from_win_replies(replies)
|
||||
return TraceResult(
|
||||
target=target,
|
||||
|
|
@ -143,10 +182,14 @@ async def _tier_icmp_win(target: str, max_hops: int, timeout: float) -> TraceRes
|
|||
)
|
||||
|
||||
|
||||
async def _tier_icmplib(target: str, max_hops: int, timeout: float, privileged: bool) -> TraceResult:
|
||||
async def _tier_icmplib(
|
||||
target: str, max_hops: int, timeout: float, privileged: bool, source_ip: str | None = None
|
||||
) -> TraceResult:
|
||||
from icmplib import async_traceroute
|
||||
|
||||
raw = await async_traceroute(target, max_hops=max_hops, timeout=timeout, privileged=privileged)
|
||||
raw = await async_traceroute(
|
||||
target, max_hops=max_hops, timeout=timeout, privileged=privileged, source=source_ip
|
||||
)
|
||||
hops = [
|
||||
finalize_hop(TraceHop(ttl=h.distance, ip=h.address, probes=list(h.rtts)))
|
||||
for h in raw
|
||||
|
|
@ -161,26 +204,29 @@ async def _tier_icmplib(target: str, max_hops: int, timeout: float, privileged:
|
|||
)
|
||||
|
||||
|
||||
async def _tier_system(target: str, binary: str, max_hops: int, timeout: float) -> TraceResult:
|
||||
async def _tier_system(
|
||||
target: str, binary: str, max_hops: int, timeout: float, source_ip: str | None = None
|
||||
) -> TraceResult:
|
||||
os_name = platform.system()
|
||||
args = (
|
||||
[binary, "-h", str(max_hops), "-w", "2", target]
|
||||
if os_name == "Windows"
|
||||
else [binary, "-m", str(max_hops), "-w", "2", target]
|
||||
)
|
||||
args = trace_argv("system_traceroute", binary, target, max_hops, 1, os_name, source_ip)
|
||||
text = await _run(args, timeout)
|
||||
return build_trace_result(text, os_name, target=target, resolved_ip=None, max_hops=max_hops)
|
||||
|
||||
|
||||
async def _tier_tcp_trace(target: str, max_hops: int, timeout: float, port: int = 443) -> TraceResult:
|
||||
async def _tier_tcp_trace(
|
||||
target: str, max_hops: int, timeout: float, port: int = 443, source_ip: str | None = None
|
||||
) -> TraceResult:
|
||||
# scapy ships in the optional `tcptrace` extra and needs Npcap (Windows) or
|
||||
# root (Unix); an ImportError here just falls through to the next tier.
|
||||
from scapy.layers.inet import IP, TCP
|
||||
from scapy.sendrecv import sr
|
||||
|
||||
def _probe() -> list[TraceHop]:
|
||||
packet = IP(dst=target, ttl=(1, max_hops))
|
||||
if source_ip:
|
||||
packet.src = source_ip
|
||||
answered, _unanswered = sr(
|
||||
IP(dst=target, ttl=(1, max_hops)) / TCP(dport=port, flags="S"),
|
||||
packet / TCP(dport=port, flags="S"),
|
||||
timeout=timeout,
|
||||
verbose=0,
|
||||
)
|
||||
|
|
@ -211,15 +257,22 @@ async def traceroute(
|
|||
timeout: float,
|
||||
semaphore: asyncio.Semaphore | None = None,
|
||||
tcp_trace: bool = False,
|
||||
source_ip: str | None = None,
|
||||
) -> TraceResult:
|
||||
builders = {
|
||||
"tcp_trace": lambda: _tier_tcp_trace(target, max_hops, timeout),
|
||||
"mtr_json": lambda: _tier_mtr(target, caps.mtr_binary or shutil.which("mtr") or "mtr", cycles, max_hops, timeout),
|
||||
"icmp_win": lambda: _tier_icmp_win(target, max_hops, timeout),
|
||||
"icmplib": lambda: _tier_icmplib(target, max_hops, timeout, privileged=not caps.icmp_dgram),
|
||||
"system_traceroute": lambda: _tier_system(target, caps.traceroute_binary or "traceroute", max_hops, timeout),
|
||||
"tcp_trace": lambda: _tier_tcp_trace(target, max_hops, timeout, source_ip=source_ip),
|
||||
"mtr_json": lambda: _tier_mtr(
|
||||
target, caps.mtr_binary or shutil.which("mtr") or "mtr", cycles, max_hops, timeout, source_ip
|
||||
),
|
||||
"icmp_win": lambda: _tier_icmp_win(target, max_hops, timeout, source_ip),
|
||||
"icmplib": lambda: _tier_icmplib(target, max_hops, timeout, privileged=not caps.icmp_dgram, source_ip=source_ip),
|
||||
"system_traceroute": lambda: _tier_system(
|
||||
target, caps.traceroute_binary or "traceroute", max_hops, timeout, source_ip
|
||||
),
|
||||
}
|
||||
tiers = [(name, builders[name]) for name in tier_order(caps, tcp_trace)]
|
||||
order = tier_order(caps, tcp_trace)
|
||||
order, _dropped = filter_trace_tiers(order, caps.os_name, forced_v4=bool(source_ip))
|
||||
tiers = [(name, builders[name]) for name in order]
|
||||
if semaphore is None:
|
||||
return await run_cascade(tiers)
|
||||
async with semaphore:
|
||||
|
|
|
|||
|
|
@ -109,8 +109,23 @@ import httpx
|
|||
from netsleuth.config import Speedtest
|
||||
|
||||
|
||||
async def tier_ookla(binary: str, server: str | None, timeout: float) -> SpeedResult:
|
||||
def ookla_interface_args(iface_name: str | None, ipv4: str | None) -> list[str]:
|
||||
if iface_name:
|
||||
return ["--interface", iface_name]
|
||||
if ipv4:
|
||||
return ["--ip", ipv4]
|
||||
return []
|
||||
|
||||
|
||||
async def tier_ookla(
|
||||
binary: str,
|
||||
server: str | None,
|
||||
timeout: float,
|
||||
iface_name: str | None = None,
|
||||
ipv4: str | None = None,
|
||||
) -> SpeedResult:
|
||||
args = [binary, "--format=json", "--accept-license", "--accept-gdpr"]
|
||||
args += ookla_interface_args(iface_name, ipv4)
|
||||
if server:
|
||||
args += ["--server-id", server] if server.isdigit() else ["--host", server]
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
|
|
@ -186,7 +201,9 @@ async def tier_fastcom(client: httpx.AsyncClient, cfg: Speedtest, timeout: float
|
|||
)
|
||||
|
||||
|
||||
async def tier_ndt7(client: httpx.AsyncClient, cfg: Speedtest, timeout: float) -> SpeedResult:
|
||||
async def tier_ndt7(
|
||||
client: httpx.AsyncClient, cfg: Speedtest, timeout: float, source_ip: str | None = None
|
||||
) -> SpeedResult:
|
||||
import websockets
|
||||
|
||||
locate = await client.get(cfg.ndt7_locate_url, timeout=timeout)
|
||||
|
|
@ -196,8 +213,11 @@ async def tier_ndt7(client: httpx.AsyncClient, cfg: Speedtest, timeout: float) -
|
|||
raise RuntimeError("no ndt7 server offered by locate.measurementlab.net")
|
||||
url = results[0]["urls"]["wss:///ndt/v7/download"]
|
||||
total = 0
|
||||
connect_kwargs = {"local_addr": (source_ip, 0)} if source_ip else {}
|
||||
began = time.perf_counter()
|
||||
async with websockets.connect(url, subprotocols=["net.measurementlab.ndt.v7"]) as socket:
|
||||
async with websockets.connect(
|
||||
url, subprotocols=["net.measurementlab.ndt.v7"], **connect_kwargs
|
||||
) as socket:
|
||||
while time.perf_counter() - began < 10:
|
||||
try:
|
||||
message = await asyncio.wait_for(socket.recv(), timeout=timeout)
|
||||
|
|
|
|||
|
|
@ -72,6 +72,8 @@ class WatchSession:
|
|||
asn: str | None = None
|
||||
interval_seconds: int = 60
|
||||
speedtest_every_n_cycles: int = 10
|
||||
interface: str | None = None
|
||||
bind_ipv4: str | None = None
|
||||
cycles: list[dict] = field(default_factory=list)
|
||||
|
||||
def add(self, summary: dict) -> None:
|
||||
|
|
@ -102,6 +104,8 @@ class WatchSession:
|
|||
"interval_seconds": self.interval_seconds,
|
||||
"speedtest_every_n_cycles": self.speedtest_every_n_cycles,
|
||||
"cycle_count": len(self.cycles),
|
||||
"interface": self.interface,
|
||||
"bind_ipv4": self.bind_ipv4,
|
||||
},
|
||||
"cycles": to_jsonable(self.cycles),
|
||||
}
|
||||
|
|
@ -140,10 +144,13 @@ async def run_watch(settings: Settings, options) -> Path:
|
|||
from netsleuth.cli import _speed_section
|
||||
|
||||
caps = detect_capabilities()
|
||||
bind = options.bind
|
||||
session = WatchSession(
|
||||
started_at=utc_now_iso(),
|
||||
interval_seconds=settings.watch.interval_seconds,
|
||||
speedtest_every_n_cycles=settings.watch.speedtest_every_n_cycles,
|
||||
interface=bind.iface_name if bind else None,
|
||||
bind_ipv4=bind.ipv4 if bind else None,
|
||||
)
|
||||
hosts = [(h.label, h.host) for h in settings.probing.reference_hosts]
|
||||
if options.extra_host:
|
||||
|
|
@ -151,7 +158,9 @@ async def run_watch(settings: Settings, options) -> Path:
|
|||
console = Console()
|
||||
console.print(f"netsleuth {__version__} · watching every {session.interval_seconds}s · Ctrl-C to stop")
|
||||
|
||||
transport = httpx.AsyncHTTPTransport(local_address=bind.ipv4) if bind else None
|
||||
async with httpx.AsyncClient(
|
||||
transport=transport,
|
||||
timeout=settings.timeouts.http_seconds,
|
||||
follow_redirects=True,
|
||||
headers={"User-Agent": f"netsleuth/{__version__}"},
|
||||
|
|
@ -173,6 +182,7 @@ async def run_watch(settings: Settings, options) -> Path:
|
|||
settings.probing.quick_ping_count,
|
||||
settings.probing.ping_interval_seconds,
|
||||
settings.probing.ping_timeout_seconds,
|
||||
source_ip=bind.ipv4 if bind else None,
|
||||
)
|
||||
speed = None
|
||||
if is_speedtest_cycle(cycle, session.speedtest_every_n_cycles) and not options.quick:
|
||||
|
|
|
|||
|
|
@ -263,7 +263,7 @@ def test_format_extensions_covers_the_three_known_formats():
|
|||
assert FORMAT_EXTENSIONS == {"md": "md", "ru-md": "ru.md", "json": "json"}
|
||||
|
||||
|
||||
from netsleuth.exporter import SPARK_CHARS, badge, first_loss_jump, render_markdown, sparkline
|
||||
from netsleuth.exporter import SPARK_CHARS, _forced_interface_rows, badge, first_loss_jump, render_markdown, sparkline
|
||||
from netsleuth.models import (
|
||||
AdapterLeakResult,
|
||||
BgpIntel,
|
||||
|
|
@ -574,6 +574,47 @@ def test_markdown_without_emoji_uses_text_badges():
|
|||
assert "🟡" not in text
|
||||
|
||||
|
||||
def test_forced_interface_rows_are_empty_without_a_forced_interface():
|
||||
assert _forced_interface_rows({}, "en") == []
|
||||
|
||||
|
||||
def test_forced_interface_rows_show_the_resolved_adapter():
|
||||
meta = {
|
||||
"interface": {
|
||||
"requested": "Ethernet",
|
||||
"resolved_iface": "Ethernet",
|
||||
"bind_ipv4": "192.168.3.72",
|
||||
"os_default_iface": "Ethernet",
|
||||
"os_default_ipv4": "192.168.3.72",
|
||||
}
|
||||
}
|
||||
rows = _forced_interface_rows(meta, "en")
|
||||
assert rows == [["Forced interface", "Ethernet — 192.168.3.72 (--interface)"]]
|
||||
|
||||
|
||||
def test_forced_interface_rows_add_the_os_default_when_it_differs():
|
||||
meta = {
|
||||
"interface": {
|
||||
"requested": "Tailscale",
|
||||
"resolved_iface": "Tailscale",
|
||||
"bind_ipv4": "100.84.46.36",
|
||||
"os_default_iface": "Ethernet",
|
||||
"os_default_ipv4": "192.168.3.72",
|
||||
}
|
||||
}
|
||||
rows = _forced_interface_rows(meta, "en")
|
||||
assert rows == [
|
||||
["Forced interface", "Tailscale — 100.84.46.36 (--interface)"],
|
||||
["OS default interface", "Ethernet — 192.168.3.72"],
|
||||
]
|
||||
|
||||
|
||||
def test_forced_interface_rows_are_translated_in_russian():
|
||||
meta = {"interface": {"requested": "Ethernet", "resolved_iface": "Ethernet", "bind_ipv4": "192.168.3.72"}}
|
||||
rows = _forced_interface_rows(meta, "ru")
|
||||
assert rows[0][0] == "Принудительный интерфейс"
|
||||
|
||||
|
||||
from netsleuth.exporter import unavailable
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@ def test_trace_hops_win_total_elapsed_is_bounded_by_the_total_budget_not_hops_ti
|
|||
# to the 100ms total budget.
|
||||
call_timeouts: list[int] = []
|
||||
|
||||
def fake_echo_once(dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netsleuth") -> IcmpReply:
|
||||
def fake_echo_once(dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netsleuth", source_ip=None) -> IcmpReply:
|
||||
call_timeouts.append(timeout_ms)
|
||||
time.sleep(timeout_ms / 1000.0)
|
||||
return IcmpReply(address=None, status=IP_REQ_TIMED_OUT, rtt_ms=None, ttl=0)
|
||||
|
|
@ -43,7 +43,7 @@ def test_trace_hops_win_total_elapsed_is_bounded_by_the_total_budget_not_hops_ti
|
|||
def test_trace_hops_win_per_hop_timeout_is_capped_even_for_a_huge_total_budget():
|
||||
call_timeouts: list[int] = []
|
||||
|
||||
def fake_echo_once(dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netsleuth") -> IcmpReply:
|
||||
def fake_echo_once(dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netsleuth", source_ip=None) -> IcmpReply:
|
||||
call_timeouts.append(timeout_ms)
|
||||
return IcmpReply(address=None, status=IP_REQ_TIMED_OUT, rtt_ms=None, ttl=0)
|
||||
|
||||
|
|
@ -62,7 +62,7 @@ def test_trace_hops_win_per_hop_timeout_is_capped_even_for_a_huge_total_budget()
|
|||
def test_trace_hops_win_happy_path_hop_count_and_ok_break_are_unchanged():
|
||||
from netsleuth.probes.icmp_win import IP_SUCCESS
|
||||
|
||||
def fake_echo_once(dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netsleuth") -> IcmpReply:
|
||||
def fake_echo_once(dest: str, ttl: int, timeout_ms: int, payload: bytes = b"netsleuth", source_ip=None) -> IcmpReply:
|
||||
if ttl < 3:
|
||||
return IcmpReply(address=f"10.0.0.{ttl}", status=IP_REQ_TIMED_OUT, rtt_ms=None, ttl=0)
|
||||
return IcmpReply(address="203.0.113.1", status=IP_SUCCESS, rtt_ms=5.0, ttl=64)
|
||||
|
|
|
|||
149
tests/test_iface.py
Normal file
149
tests/test_iface.py
Normal file
|
|
@ -0,0 +1,149 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import socket
|
||||
|
||||
from netsleuth.iface import available_interfaces_hint, resolve_bind_target, usable_addresses
|
||||
|
||||
|
||||
def addrs(**kw) -> dict[str, list[tuple[int, str]]]:
|
||||
return kw
|
||||
|
||||
|
||||
def up(**kw) -> dict[str, bool]:
|
||||
return kw
|
||||
|
||||
|
||||
def test_usable_addresses_picks_first_non_loopback_v4_and_v6():
|
||||
v4, v6 = usable_addresses(
|
||||
[
|
||||
(socket.AF_INET, "127.0.0.1"),
|
||||
(socket.AF_INET, "192.168.1.34"),
|
||||
(socket.AF_INET6, "::1"),
|
||||
(socket.AF_INET6, "fe80::1"),
|
||||
(socket.AF_INET6, "2001:db8::1"),
|
||||
]
|
||||
)
|
||||
assert v4 == "192.168.1.34"
|
||||
assert v6 == "2001:db8::1"
|
||||
|
||||
|
||||
def test_usable_addresses_rejects_apipa_v4():
|
||||
v4, v6 = usable_addresses([(socket.AF_INET, "169.254.1.2")])
|
||||
assert v4 is None
|
||||
assert v6 is None
|
||||
|
||||
|
||||
def test_usable_addresses_rejects_link_local_v6():
|
||||
v4, v6 = usable_addresses([(socket.AF_INET6, "fe80::1")])
|
||||
assert v6 is None
|
||||
|
||||
|
||||
def test_usable_addresses_accepts_link_local_only_when_nothing_else_present():
|
||||
# A link-local-only adapter (typical for an interface with no DHCP/global
|
||||
# IPv6) is a real, common case, not an error -- it just means ipv6=None.
|
||||
v4, v6 = usable_addresses([(socket.AF_INET, "192.168.1.34"), (socket.AF_INET6, "fe80::1")])
|
||||
assert v4 == "192.168.1.34"
|
||||
assert v6 is None
|
||||
|
||||
|
||||
def test_resolve_bind_target_by_exact_name():
|
||||
result = resolve_bind_target(
|
||||
"Ethernet",
|
||||
addrs(Ethernet=[(socket.AF_INET, "192.168.3.72")], VPN=[(socket.AF_INET, "10.0.0.5")]),
|
||||
up(Ethernet=True, VPN=True),
|
||||
)
|
||||
assert result.iface_name == "Ethernet"
|
||||
assert result.ipv4 == "192.168.3.72"
|
||||
assert result.error is None
|
||||
|
||||
|
||||
def test_resolve_bind_target_by_ip_finds_owning_adapter():
|
||||
result = resolve_bind_target(
|
||||
"192.168.3.72",
|
||||
addrs(Ethernet=[(socket.AF_INET, "192.168.3.72")]),
|
||||
up(Ethernet=True),
|
||||
)
|
||||
assert result.iface_name == "Ethernet"
|
||||
assert result.ipv4 == "192.168.3.72"
|
||||
|
||||
|
||||
def test_resolve_bind_target_exact_name_beats_substring_match():
|
||||
result = resolve_bind_target(
|
||||
"Ethernet",
|
||||
addrs(
|
||||
Ethernet=[(socket.AF_INET, "192.168.3.72")],
|
||||
**{"vEthernet (Default Switch)": [(socket.AF_INET, "172.20.0.1")]},
|
||||
),
|
||||
up(**{"Ethernet": True, "vEthernet (Default Switch)": True}),
|
||||
)
|
||||
assert result.iface_name == "Ethernet"
|
||||
|
||||
|
||||
def test_resolve_bind_target_case_insensitive_exact_match():
|
||||
result = resolve_bind_target(
|
||||
"ethernet",
|
||||
addrs(Ethernet=[(socket.AF_INET, "192.168.3.72")]),
|
||||
up(Ethernet=True),
|
||||
)
|
||||
assert result.iface_name == "Ethernet"
|
||||
|
||||
|
||||
def test_resolve_bind_target_unique_substring_match():
|
||||
result = resolve_bind_target(
|
||||
"radmin",
|
||||
addrs(**{"Radmin VPN": [(socket.AF_INET, "26.0.0.5")]}),
|
||||
up(**{"Radmin VPN": True}),
|
||||
)
|
||||
assert result.iface_name == "Radmin VPN"
|
||||
|
||||
|
||||
def test_resolve_bind_target_ambiguous_substring_is_an_error():
|
||||
result = resolve_bind_target(
|
||||
"eth",
|
||||
addrs(Ethernet=[(socket.AF_INET, "192.168.3.72")], **{"Ethernet 2": [(socket.AF_INET, "10.0.0.9")]}),
|
||||
up(Ethernet=True, **{"Ethernet 2": True}),
|
||||
)
|
||||
assert result.error is not None
|
||||
assert "Ethernet" in result.error and "Ethernet 2" in result.error
|
||||
|
||||
|
||||
def test_resolve_bind_target_unknown_name_is_an_error_listing_candidates():
|
||||
result = resolve_bind_target("Nope", addrs(Ethernet=[(socket.AF_INET, "192.168.3.72")]), up(Ethernet=True))
|
||||
assert result.error is not None
|
||||
assert "Ethernet" in result.error
|
||||
|
||||
|
||||
def test_resolve_bind_target_down_adapter_is_an_error():
|
||||
result = resolve_bind_target(
|
||||
"Ethernet", addrs(Ethernet=[(socket.AF_INET, "192.168.3.72")]), up(Ethernet=False)
|
||||
)
|
||||
assert result.error is not None
|
||||
assert result.is_up is False
|
||||
|
||||
|
||||
def test_resolve_bind_target_apipa_only_adapter_is_an_error():
|
||||
result = resolve_bind_target(
|
||||
"PdaNet", addrs(PdaNet=[(socket.AF_INET, "169.254.1.2")]), up(PdaNet=True)
|
||||
)
|
||||
assert result.error is not None
|
||||
|
||||
|
||||
def test_resolve_bind_target_ipv6_comes_from_the_same_matched_adapter():
|
||||
result = resolve_bind_target(
|
||||
"Ethernet",
|
||||
addrs(Ethernet=[(socket.AF_INET, "192.168.3.72"), (socket.AF_INET6, "2001:db8::1")]),
|
||||
up(Ethernet=True),
|
||||
)
|
||||
assert result.ipv4 == "192.168.3.72"
|
||||
assert result.ipv6 == "2001:db8::1"
|
||||
|
||||
|
||||
def test_available_interfaces_hint_lists_every_adapter_with_its_state():
|
||||
hint = available_interfaces_hint(
|
||||
addrs(Ethernet=[(socket.AF_INET, "192.168.3.72")], PdaNet=[(socket.AF_INET, "169.254.1.2")]),
|
||||
up(Ethernet=True, PdaNet=False),
|
||||
)
|
||||
assert "Ethernet" in hint
|
||||
assert "192.168.3.72" in hint
|
||||
assert "PdaNet" in hint
|
||||
assert "down" in hint.lower()
|
||||
|
|
@ -2,7 +2,7 @@ from __future__ import annotations
|
|||
|
||||
import pytest
|
||||
|
||||
from netsleuth.speed import bufferbloat_delta, mbps, parse_server_timing_cfl4, throughput_from_samples
|
||||
from netsleuth.speed import bufferbloat_delta, mbps, ookla_interface_args, parse_server_timing_cfl4, throughput_from_samples
|
||||
|
||||
|
||||
def test_mbps_converts_bytes_and_seconds():
|
||||
|
|
@ -81,6 +81,18 @@ def test_bufferbloat_delta_never_goes_negative():
|
|||
assert bufferbloat_delta(50.0, [10.0, 12.0, 11.0]) == 0.0
|
||||
|
||||
|
||||
def test_ookla_interface_args_prefers_interface_name_over_ip():
|
||||
assert ookla_interface_args(iface_name="Ethernet", ipv4="192.168.3.72") == ["--interface", "Ethernet"]
|
||||
|
||||
|
||||
def test_ookla_interface_args_falls_back_to_ip_without_a_name():
|
||||
assert ookla_interface_args(iface_name=None, ipv4="192.168.3.72") == ["--ip", "192.168.3.72"]
|
||||
|
||||
|
||||
def test_ookla_interface_args_is_empty_without_a_forced_bind():
|
||||
assert ookla_interface_args(iface_name=None, ipv4=None) == []
|
||||
|
||||
|
||||
def test_bufferbloat_delta_needs_both_a_baseline_and_samples():
|
||||
assert bufferbloat_delta(None, [10.0]) is None
|
||||
assert bufferbloat_delta(12.0, []) is None
|
||||
|
|
|
|||
|
|
@ -7,7 +7,14 @@ import pytest
|
|||
|
||||
from netsleuth.models import Capabilities, TraceHop, TraceResult
|
||||
from netsleuth.probes.icmp_win import IP_REQ_TIMED_OUT, IP_SUCCESS, IP_TTL_EXPIRED_TRANSIT, IcmpReply
|
||||
from netsleuth.probes.traceroute import _run_in_daemon_thread, hops_from_win_replies, run_cascade, tier_order
|
||||
from netsleuth.probes.traceroute import (
|
||||
_run_in_daemon_thread,
|
||||
filter_trace_tiers,
|
||||
hops_from_win_replies,
|
||||
run_cascade,
|
||||
tier_order,
|
||||
trace_argv,
|
||||
)
|
||||
|
||||
|
||||
def caps(**kw) -> Capabilities:
|
||||
|
|
@ -204,3 +211,59 @@ def test_run_in_daemon_thread_never_creates_a_non_daemon_worker_thread():
|
|||
|
||||
assert seen
|
||||
assert all(t.daemon for t in seen)
|
||||
|
||||
|
||||
def test_filter_trace_tiers_drops_windows_tracert_when_a_v4_bind_is_forced():
|
||||
kept, dropped = filter_trace_tiers(["icmp_win", "system_traceroute"], "Windows", forced_v4=True)
|
||||
assert kept == ["icmp_win"]
|
||||
assert dropped == ["system_traceroute"]
|
||||
|
||||
|
||||
def test_filter_trace_tiers_keeps_everything_when_nothing_is_forced():
|
||||
order = ["icmp_win", "system_traceroute"]
|
||||
kept, dropped = filter_trace_tiers(order, "Windows", forced_v4=False)
|
||||
assert kept == order
|
||||
assert dropped == []
|
||||
|
||||
|
||||
def test_filter_trace_tiers_leaves_unix_traceroute_alone_even_when_forced():
|
||||
order = ["mtr_json", "icmplib", "system_traceroute"]
|
||||
kept, dropped = filter_trace_tiers(order, "Linux", forced_v4=True)
|
||||
assert kept == order
|
||||
assert dropped == []
|
||||
|
||||
|
||||
def test_filter_trace_tiers_is_a_noop_when_system_traceroute_is_absent():
|
||||
kept, dropped = filter_trace_tiers(["icmp_win"], "Windows", forced_v4=True)
|
||||
assert kept == ["icmp_win"]
|
||||
assert dropped == []
|
||||
|
||||
|
||||
def test_trace_argv_mtr_adds_address_flag_when_source_given():
|
||||
args = trace_argv("mtr_json", "/usr/bin/mtr", "1.1.1.1", max_hops=30, cycles=10, os_name="Linux", source_ip="192.168.1.34")
|
||||
assert args == [
|
||||
"/usr/bin/mtr", "--json", "--report-cycles", "10", "--max-ttl", "30",
|
||||
"--address", "192.168.1.34", "1.1.1.1",
|
||||
]
|
||||
|
||||
|
||||
def test_trace_argv_mtr_omits_address_flag_without_a_source():
|
||||
args = trace_argv("mtr_json", "/usr/bin/mtr", "1.1.1.1", max_hops=30, cycles=10, os_name="Linux")
|
||||
assert "--address" not in args
|
||||
|
||||
|
||||
def test_trace_argv_unix_traceroute_adds_source_flag():
|
||||
args = trace_argv(
|
||||
"system_traceroute", "/usr/bin/traceroute", "1.1.1.1", max_hops=30, cycles=1,
|
||||
os_name="Linux", source_ip="192.168.1.34",
|
||||
)
|
||||
assert args == ["/usr/bin/traceroute", "-m", "30", "-w", "2", "-s", "192.168.1.34", "1.1.1.1"]
|
||||
|
||||
|
||||
def test_trace_argv_windows_tracert_never_gets_a_source_flag():
|
||||
args = trace_argv(
|
||||
"system_traceroute", "tracert", "1.1.1.1", max_hops=30, cycles=1,
|
||||
os_name="Windows", source_ip="192.168.1.34",
|
||||
)
|
||||
assert args == ["tracert", "-h", "30", "-w", "2", "1.1.1.1"]
|
||||
assert "-s" not in args and "-S" not in args
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue