1052 lines
37 KiB
Python
Executable File
1052 lines
37 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
# Copyright (c) 2026 Petr Balvín <opensource@petrbalvin.org> (https://petrbalvin.org)
|
||
# SPDX-License-Identifier: MIT
|
||
|
||
"""Network diagnostics — latency, DNS, MTU, packet loss, dual-stack, and port checks.
|
||
|
||
No external dependencies — leverages system tools (ping, ip, ss, ifconfig,
|
||
netstat, sockstat) with fallback to pure stdlib where possible.
|
||
|
||
Supports Linux and FreeBSD (13+ recommended; on pre-13 FreeBSD the legacy
|
||
ping6 binary is used for IPv6 instead of the unified ``ping -6``).
|
||
|
||
Usage:
|
||
network-diag.py # full diagnostic
|
||
network-diag.py --target cloudflare.com # target a specific host
|
||
network-diag.py --protocol v4 # IPv4 only
|
||
network-diag.py --protocol v6 # IPv6 only
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import ipaddress
|
||
import os
|
||
import re
|
||
import shutil
|
||
import socket
|
||
import statistics
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from concurrent.futures import Future, ThreadPoolExecutor, as_completed
|
||
|
||
BOLD = "\033[1m"
|
||
RED = "\033[31m"
|
||
GREEN = "\033[32m"
|
||
YELLOW = "\033[33m"
|
||
CYAN = "\033[36m"
|
||
DIM = "\033[2m"
|
||
RESET = "\033[0m"
|
||
VERSION = "2.1.1"
|
||
|
||
# Well-known hosts for connectivity tests
|
||
CONNECTIVITY_HOSTS = [
|
||
("cloudflare.com", 443),
|
||
("google.com", 443),
|
||
]
|
||
|
||
# Ping statistics differ between iputils (Linux) and FreeBSD:
|
||
# Linux: "5 packets transmitted, 5 received, 0% packet loss, time 4003ms"
|
||
# FreeBSD: "5 packets transmitted, 5 packets received, 0.0% packet loss"
|
||
_PING_TIME_RE = re.compile(r"time=(\d+(?:\.\d+)?)")
|
||
_PING_RECEIVED_RE = re.compile(r"(\d+)\s+(?:packets?\s+)?received")
|
||
|
||
|
||
def detect_os() -> str:
|
||
"""Return 'linux' or 'freebsd' based on uname."""
|
||
system = os.uname().sysname.lower()
|
||
if system == "freebsd":
|
||
return "freebsd"
|
||
return "linux" # default
|
||
|
||
|
||
def _status(msg: str) -> None:
|
||
"""Print progress message to stderr — stays on same line until ``_status_done``."""
|
||
print(f" {msg}...", end="", flush=True, file=sys.stderr)
|
||
|
||
|
||
def _status_done(msg: str = "done") -> None:
|
||
"""Complete the current progress line on stderr."""
|
||
print(f" {msg}", file=sys.stderr)
|
||
|
||
|
||
def run(cmd: list[str], timeout: int = 15) -> tuple[str, str] | None:
|
||
"""Execute a command and return (stdout, stderr), or None if not installed.
|
||
|
||
Sets LC_ALL=C (and LANG=C) for reliable English-language output parsing
|
||
and decodes with ``errors="replace"`` so unexpected bytes never raise.
|
||
Returns ("", "") on timeout and None only when the binary is missing, so
|
||
callers can distinguish "tool unavailable" from "no output".
|
||
"""
|
||
try:
|
||
env = {**os.environ, "LANG": "C", "LC_ALL": "C"}
|
||
result = subprocess.run(
|
||
cmd,
|
||
capture_output=True,
|
||
text=True,
|
||
errors="replace",
|
||
timeout=timeout,
|
||
env=env,
|
||
)
|
||
return result.stdout, result.stderr
|
||
except subprocess.TimeoutExpired:
|
||
return "", ""
|
||
except FileNotFoundError:
|
||
return None
|
||
|
||
|
||
def _parse_ping_output(output: str) -> tuple[list[float], int | None]:
|
||
"""Parse ping output for per-reply latencies and the received counter.
|
||
|
||
Returns (latencies in ms, received count). The received count is None
|
||
when no statistics line was recognised; callers should then fall back
|
||
to len(latencies).
|
||
"""
|
||
latencies = [float(m.group(1)) for m in _PING_TIME_RE.finditer(output)]
|
||
received: int | None = None
|
||
for line in output.split("\n"):
|
||
if "transmitted" in line:
|
||
match = _PING_RECEIVED_RE.search(line)
|
||
if match:
|
||
received = int(match.group(1))
|
||
break
|
||
return latencies, received
|
||
|
||
|
||
def _ping_result_to_dict(
|
||
host: str,
|
||
latencies: list[float],
|
||
received: int,
|
||
count: int,
|
||
protocol: str,
|
||
) -> dict[str, object]:
|
||
"""Build a standardised ping result dictionary from parsed data.
|
||
|
||
``loss_pct`` is derived from the authoritative received counter, not
|
||
from the number of parsed ``time=`` lines.
|
||
"""
|
||
loss_pct = round((count - received) / count * 100, 1) if count > 0 else 100.0
|
||
result: dict[str, object] = {
|
||
"host": host,
|
||
"protocol": protocol,
|
||
"sent": count,
|
||
"received": received,
|
||
"loss_pct": loss_pct,
|
||
"min_ms": 0.0,
|
||
"avg_ms": 0.0,
|
||
"max_ms": 0.0,
|
||
"stddev_ms": 0.0,
|
||
}
|
||
if latencies:
|
||
result["min_ms"] = round(min(latencies), 1)
|
||
result["avg_ms"] = round(statistics.mean(latencies), 1)
|
||
result["max_ms"] = round(max(latencies), 1)
|
||
if len(latencies) > 1:
|
||
result["stddev_ms"] = round(statistics.stdev(latencies), 1)
|
||
return result
|
||
|
||
|
||
def _freebsd_major() -> int:
|
||
"""Parse the FreeBSD major version from uname release (99 if unsure)."""
|
||
try:
|
||
return int(os.uname().release.split("-")[0].split(".")[0])
|
||
except (ValueError, IndexError):
|
||
return 99 # assume a modern, unified ping when unparseable
|
||
|
||
|
||
def _ping_base_cmd(os_type: str, family: str) -> list[str] | None:
|
||
"""Return the ping command prefix for *family*, or None if unavailable.
|
||
|
||
FreeBSD 13+ ships a unified ping understanding -4/-6; earlier releases
|
||
need the legacy ping6 binary for IPv6.
|
||
"""
|
||
if os_type == "freebsd" and _freebsd_major() < 13:
|
||
if family == "v6":
|
||
return ["ping6"] if shutil.which("ping6") else None
|
||
return ["ping"]
|
||
return ["ping", "-6" if family == "v6" else "-4"]
|
||
|
||
|
||
def _ping_wait_args(os_type: str, base_cmd: list[str], timeout: int) -> list[str]:
|
||
"""Per-reply wait flag for the given ping binary.
|
||
|
||
Linux iputils ``-W`` takes seconds; FreeBSD ping ``-W`` takes
|
||
milliseconds; legacy FreeBSD ping6 uses ``-x`` (milliseconds) instead.
|
||
"""
|
||
if os_type == "freebsd":
|
||
flag = "-x" if base_cmd[0] == "ping6" else "-W"
|
||
return [flag, str(timeout * 1000)]
|
||
return ["-W", str(timeout)]
|
||
|
||
|
||
def ping_host(
|
||
host: str,
|
||
os_type: str,
|
||
count: int = 5,
|
||
timeout: int = 5,
|
||
protocol: str = "v4",
|
||
) -> dict[str, object]:
|
||
"""Ping a host over one address family (``"v4"`` or ``"v6"``).
|
||
|
||
Uses the OS-appropriate binary and flags (see ``_ping_base_cmd``).
|
||
When no suitable ping binary exists, the result carries a ``"note"``
|
||
key explaining the gap instead of silently reporting zero replies.
|
||
"""
|
||
base = _ping_base_cmd(os_type, protocol)
|
||
if base is None:
|
||
result = _ping_result_to_dict(host, [], 0, count, protocol)
|
||
result["note"] = "no ping binary with IPv6 support found"
|
||
return result
|
||
cmd = [*base, "-c", str(count), *_ping_wait_args(os_type, base, timeout), "--", host]
|
||
# The subprocess timeout must also cover ping's own DNS resolution time.
|
||
res = run(cmd, timeout=count + timeout + 10)
|
||
if res is None:
|
||
result = _ping_result_to_dict(host, [], 0, count, protocol)
|
||
result["note"] = f"'{base[0]}' not found"
|
||
return result
|
||
latencies, parsed_received = _parse_ping_output(res[0])
|
||
received = parsed_received if parsed_received is not None else len(latencies)
|
||
return _ping_result_to_dict(host, latencies, received, count, protocol)
|
||
|
||
|
||
def check_interface_info(os_type: str = "linux") -> dict[str, object]:
|
||
"""Collect network interface addresses and default routes."""
|
||
info: dict[str, object] = {
|
||
"interfaces": [],
|
||
"default_route_v4": "",
|
||
"default_route_v6": "",
|
||
"error": "",
|
||
}
|
||
if os_type == "freebsd":
|
||
return _check_interface_info_freebsd(info)
|
||
return _check_interface_info_linux(info)
|
||
|
||
|
||
def _check_interface_info_linux(info: dict[str, object]) -> dict[str, object]:
|
||
"""Collect interface info on Linux using iproute2."""
|
||
res = run(["ip", "-br", "addr", "show"])
|
||
if res is None:
|
||
info["error"] = "'ip' not found"
|
||
return info
|
||
for line in res[0].split("\n"):
|
||
parts = line.split()
|
||
if len(parts) >= 2 and parts[0] != "lo":
|
||
info["interfaces"].append(
|
||
{
|
||
"name": parts[0],
|
||
"state": parts[1],
|
||
"ips": parts[2:],
|
||
}
|
||
)
|
||
|
||
route4 = run(["ip", "-4", "route", "show", "default"])
|
||
if route4 and route4[0].strip():
|
||
parts = route4[0].strip().split()
|
||
for i, p in enumerate(parts):
|
||
if p == "via" and i + 1 < len(parts):
|
||
info["default_route_v4"] = parts[i + 1]
|
||
break
|
||
|
||
route6 = run(["ip", "-6", "route", "show", "default"])
|
||
if route6 and route6[0].strip():
|
||
parts = route6[0].strip().split()
|
||
for i, p in enumerate(parts):
|
||
if p == "via" and i + 1 < len(parts):
|
||
info["default_route_v6"] = parts[i + 1]
|
||
break
|
||
return info
|
||
|
||
|
||
def _check_interface_info_freebsd(info: dict[str, object]) -> dict[str, object]:
|
||
"""Collect interface info on FreeBSD using ifconfig and netstat."""
|
||
iflist = run(["ifconfig", "-l"])
|
||
if iflist is None:
|
||
info["error"] = "'ifconfig' not found"
|
||
return info
|
||
ifaces = iflist[0].strip().split()
|
||
if not ifaces:
|
||
return info
|
||
|
||
for ifname in ifaces:
|
||
if ifname == "lo0":
|
||
continue
|
||
iface = run(["ifconfig", ifname])
|
||
if iface is None:
|
||
continue
|
||
lines = iface[0].split("\n")
|
||
if not lines or not lines[0]:
|
||
continue
|
||
|
||
# Parse flags line: name: flags=...<UP,...>
|
||
state = "DOWN"
|
||
if "<UP," in lines[0] or "<UP>" in lines[0]:
|
||
state = "UP"
|
||
|
||
ips: list[str] = []
|
||
for line in lines[1:]:
|
||
stripped = line.strip()
|
||
if stripped.startswith("inet "):
|
||
# inet 192.168.1.100 netmask ...
|
||
addr = stripped.split()[1]
|
||
if addr != "127.0.0.1":
|
||
ips.append(addr)
|
||
elif stripped.startswith("inet6 "):
|
||
addr = stripped.split()[1].split("%")[0] # strip %scope suffix
|
||
# Skip link-local (fe80:) and loopback (::1)
|
||
if addr.startswith("fe80:") or addr == "::1":
|
||
continue
|
||
ips.append(addr)
|
||
|
||
info["interfaces"].append({"name": ifname, "state": state, "ips": ips})
|
||
|
||
# Default routes via netstat
|
||
route = run(["netstat", "-rn"])
|
||
if route is None:
|
||
return info
|
||
for line in route[0].split("\n"):
|
||
if not line.startswith("default"):
|
||
continue
|
||
parts = line.split()
|
||
if len(parts) < 2:
|
||
continue
|
||
gateway = parts[1].split("%")[0] # strip %scope from link-local gateways
|
||
# Determine v4 vs v6 by checking for colons
|
||
if ":" in gateway:
|
||
if not info["default_route_v6"]:
|
||
info["default_route_v6"] = gateway
|
||
elif not info["default_route_v4"]:
|
||
info["default_route_v4"] = gateway
|
||
|
||
return info
|
||
|
||
|
||
def check_connectivity() -> dict[str, object]:
|
||
"""TCP handshake probes to well-known hosts over both address families.
|
||
|
||
This is the single place performing TCP probes; the per-family
|
||
reachability sections reuse these results (see ``_reachability_report``).
|
||
"""
|
||
info: dict[str, object] = {"v4": {}, "v6": {}}
|
||
for host, port in CONNECTIVITY_HOSTS:
|
||
info["v6"][host] = _tcp_probe(host, port, socket.AF_INET6) # IPv6 first
|
||
info["v4"][host] = _tcp_probe(host, port, socket.AF_INET)
|
||
return info
|
||
|
||
|
||
def _tcp_probe(host: str, port: int, family: int) -> dict[str, object]:
|
||
"""Attempt a TCP connection and return timing info.
|
||
|
||
Iterates all resolved candidates (in getaddrinfo order) and passes the
|
||
full sockaddr tuple to connect(), preserving the IPv6 scope id.
|
||
"""
|
||
try:
|
||
addrs = socket.getaddrinfo(host, port, family=family, type=socket.SOCK_STREAM)
|
||
except OSError:
|
||
return {"reachable": False}
|
||
for af, socktype, proto, _canonname, sockaddr in addrs:
|
||
try:
|
||
with socket.socket(af, socktype, proto) as sock:
|
||
sock.settimeout(5)
|
||
start = time.perf_counter()
|
||
sock.connect(sockaddr)
|
||
elapsed = round((time.perf_counter() - start) * 1000, 1)
|
||
return {"reachable": True, "ip": str(sockaddr[0]), "ms": elapsed}
|
||
except OSError:
|
||
continue
|
||
return {"reachable": False}
|
||
|
||
|
||
def check_dns(protocol: str = "auto") -> dict[str, object]:
|
||
"""Test DNS resolution latency for multiple domains.
|
||
|
||
Tests both A (IPv4) and AAAA (IPv6) resolution when ``protocol`` is
|
||
``"auto"``. Respects ``--protocol v4`` / ``--protocol v6`` to restrict
|
||
to a single address family.
|
||
"""
|
||
info: dict[str, object] = {
|
||
"hostname": socket.gethostname(),
|
||
"fqdn": "",
|
||
"resolvers": [],
|
||
"v4": {},
|
||
"v6": {},
|
||
}
|
||
|
||
try:
|
||
info["fqdn"] = socket.getfqdn()
|
||
except OSError:
|
||
pass
|
||
|
||
domains = [host for host, _port in CONNECTIVITY_HOSTS]
|
||
|
||
# IPv6 first
|
||
if protocol in ("auto", "v6"):
|
||
for domain in domains:
|
||
info["v6"][domain] = _dns_resolve(domain, socket.AF_INET6)
|
||
if protocol in ("auto", "v4"):
|
||
for domain in domains:
|
||
info["v4"][domain] = _dns_resolve(domain, socket.AF_INET)
|
||
|
||
try:
|
||
with open("/etc/resolv.conf") as f:
|
||
for line in f:
|
||
parts = line.split()
|
||
if line.startswith("nameserver") and len(parts) >= 2:
|
||
info["resolvers"].append(parts[1])
|
||
except OSError:
|
||
pass
|
||
|
||
return info
|
||
|
||
|
||
def _dns_resolve(domain: str, family: int) -> list[dict[str, object]]:
|
||
"""Resolve *domain* three times and return latency samples."""
|
||
results: list[dict[str, object]] = []
|
||
for _ in range(3):
|
||
try:
|
||
start = time.perf_counter()
|
||
addr = socket.getaddrinfo(domain, None, family=family)
|
||
elapsed = (time.perf_counter() - start) * 1000
|
||
results.append(
|
||
{
|
||
"ip": addr[0][4][0] if addr else "?",
|
||
"ms": round(elapsed, 1),
|
||
}
|
||
)
|
||
except OSError:
|
||
results.append({"ip": "N/A", "ms": -1})
|
||
return results
|
||
|
||
|
||
def _family_addresses(
|
||
iface_info: dict[str, object], family: str
|
||
) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]:
|
||
"""Usable addresses of *family* from collected interface data.
|
||
|
||
Excludes loopback, link-local, unspecified and multicast addresses.
|
||
Globally-scoped addresses (per the IANA registries in ``ipaddress``)
|
||
come first, private/ULA ones after.
|
||
"""
|
||
version = 6 if family == "v6" else 4
|
||
global_addrs: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = []
|
||
other_addrs: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = []
|
||
interfaces = iface_info.get("interfaces") if isinstance(iface_info, dict) else None
|
||
for ifc in interfaces or []:
|
||
for raw in ifc.get("ips", []):
|
||
try:
|
||
ip = ipaddress.ip_interface(raw).ip
|
||
except ValueError:
|
||
continue
|
||
if ip.version != version:
|
||
continue
|
||
if ip.is_loopback or ip.is_link_local or ip.is_unspecified or ip.is_multicast:
|
||
continue
|
||
(global_addrs if ip.is_global else other_addrs).append(ip)
|
||
return global_addrs + other_addrs
|
||
|
||
|
||
def _reachability_report(
|
||
family: str, iface_info: dict[str, object], probes: dict[str, object]
|
||
) -> dict[str, object]:
|
||
"""Build a per-family reachability section from interface data + probes.
|
||
|
||
``addr`` is the best address found on any interface (global scope
|
||
preferred); ``is_global`` distinguishes it from private/ULA space.
|
||
"""
|
||
addresses = _family_addresses(iface_info, family)
|
||
best = addresses[0] if addresses else None
|
||
return {
|
||
"available": best is not None,
|
||
"addr": str(best) if best is not None else "",
|
||
"is_global": bool(best is not None and best.is_global),
|
||
"test_sites": probes,
|
||
}
|
||
|
||
|
||
def _resolve_probe_target(target: str, protocol: str) -> tuple[str | None, str | None]:
|
||
"""Resolve *target* to (family, numeric address), preferring IPv6.
|
||
|
||
Returns (None, None) when no address of the requested family exists.
|
||
Resolving here (instead of letting ping decide) makes the address
|
||
family deterministic so MTU header arithmetic stays correct.
|
||
"""
|
||
families = {"v4": socket.AF_INET, "v6": socket.AF_INET6}
|
||
order = ["v6", "v4"] if protocol == "auto" else [protocol]
|
||
for family in order:
|
||
try:
|
||
infos = socket.getaddrinfo(target, None, family=families[family])
|
||
except OSError:
|
||
continue
|
||
if infos:
|
||
return family, str(infos[0][4][0])
|
||
return None, None
|
||
|
||
|
||
def check_mtu(
|
||
target: str = "cloudflare.com",
|
||
protocol: str = "auto",
|
||
os_type: str = "linux",
|
||
) -> dict[str, object]:
|
||
"""Path MTU discovery using system ping with fragmentation disabled.
|
||
|
||
Success is defined POSITIVELY (a parsed reply); failure patterns are
|
||
matched in BOTH stdout and stderr, because iputils prints the local
|
||
"message too long" error to stderr while the PING banner goes to
|
||
stdout. Header sizes are per family: 28 bytes for IPv4 (20 IP + 8
|
||
ICMP), 48 for IPv6 (40 IPv6 + 8 ICMPv6). On Linux uses ``ping -M
|
||
do``; on FreeBSD ``ping -D``.
|
||
"""
|
||
info: dict[str, object] = {"path_mtu": 0, "target": target, "method": ""}
|
||
|
||
family, addr = _resolve_probe_target(target, protocol)
|
||
if family is None or addr is None:
|
||
info["method"] = "target did not resolve"
|
||
return info
|
||
header = 48 if family == "v6" else 28
|
||
|
||
base = _ping_base_cmd(os_type, family)
|
||
if base is None:
|
||
info["method"] = "no ping binary with IPv6 support found"
|
||
return info
|
||
df_args = ["-D"] if os_type == "freebsd" else ["-M", "do"]
|
||
wait_args = _ping_wait_args(os_type, base, 2)
|
||
|
||
for size in (1500, 1400, 1300, 1280, 1200, 1000):
|
||
payload = size - header
|
||
cmd = [*base, *df_args, "-c", "1", *wait_args, "-s", str(payload), "--", addr]
|
||
res = run(cmd, timeout=12) # generous: covers the wait plus extra slack
|
||
if res is None:
|
||
info["method"] = f"'{base[0]}' not found"
|
||
return info
|
||
out, err = res
|
||
received = _parse_ping_output(out)[1]
|
||
if received and received > 0:
|
||
info["path_mtu"] = size
|
||
info["method"] = f"ping DF probe ({size}B OK)"
|
||
break
|
||
combined = (out + err).lower()
|
||
if "frag needed" in combined or "message too long" in combined:
|
||
continue # definitively too big — try a smaller size
|
||
# timeout, black-holed ICMP errors, or unreachable — try smaller anyway
|
||
else:
|
||
info["method"] = "ping DF probe (no probe size succeeded)"
|
||
|
||
return info
|
||
|
||
|
||
def _parse_ss_output(stdout: str) -> list[dict[str, str]]:
|
||
"""Parse ``ss -tlnp`` output into a list of {addr, process, bucket} dicts."""
|
||
entries: list[dict[str, str]] = []
|
||
for line in stdout.split("\n")[1:]:
|
||
if "LISTEN" not in line:
|
||
continue
|
||
parts = line.split()
|
||
if len(parts) < 5:
|
||
continue
|
||
# ss -tlnp columns: State Recv-Q Send-Q LocalAddress:Port PeerAddress:Port Process
|
||
# 0 1 2 3 4 5+
|
||
local = parts[3]
|
||
# IPv6 addresses in ss output use bracket notation: [::]:port
|
||
bucket = "tcp6" if local.startswith("[") else "tcp4"
|
||
process = parts[-1] if len(parts) > 5 else ""
|
||
entries.append({"addr": local, "process": process, "bucket": bucket})
|
||
return entries
|
||
|
||
|
||
def check_listening_ports(os_type: str = "linux") -> dict[str, object]:
|
||
"""Show locally listening TCP ports, split by address family.
|
||
|
||
On Linux, uses ``ss -tlnp`` for the port listing. IPv6 entries are
|
||
detected by the bracket notation (``[::]:port``) in the local address
|
||
column. When process info is missing (ports owned by other users),
|
||
re-runs the command via ``sudo -n`` for non-interactive privilege
|
||
escalation. Entries that remain unresolvable are shown as
|
||
``(no permission)``.
|
||
|
||
On FreeBSD, uses ``sockstat -l4 -l6``; dual-stack sockets (``tcp46``)
|
||
are reported in their own bucket.
|
||
"""
|
||
if os_type == "freebsd":
|
||
return _check_listening_ports_freebsd()
|
||
return _check_listening_ports_linux()
|
||
|
||
|
||
def _check_listening_ports_linux() -> dict[str, object]:
|
||
"""Listening ports on Linux via ss."""
|
||
info: dict[str, object] = {"tcp4": [], "tcp6": [], "tcp46": [], "error": ""}
|
||
res = run(["ss", "-tlnp"], timeout=5)
|
||
if res is None:
|
||
info["error"] = "'ss' not found"
|
||
return info
|
||
entries = _parse_ss_output(res[0])
|
||
|
||
# If any process fields came back empty, try sudo for the full picture.
|
||
if any(e["process"] == "" for e in entries):
|
||
sudo_res = run(["sudo", "-n", "ss", "-tlnp"], timeout=5)
|
||
if sudo_res and sudo_res[0]:
|
||
sudo_entries = _parse_ss_output(sudo_res[0])
|
||
# Merge: overwrite entries where the sudo output gave us a process name.
|
||
sudo_map = {e["addr"]: e for e in sudo_entries}
|
||
for e in entries:
|
||
if e["process"] == "" and e["addr"] in sudo_map:
|
||
e["process"] = sudo_map[e["addr"]]["process"]
|
||
|
||
for entry in entries:
|
||
process = entry["process"] if entry["process"] else "(no permission)"
|
||
info[entry["bucket"]].append({"addr": entry["addr"], "process": process})
|
||
|
||
return info
|
||
|
||
|
||
def _check_listening_ports_freebsd() -> dict[str, object]:
|
||
"""Listening ports on FreeBSD via sockstat (tcp46 = dual-stack)."""
|
||
info: dict[str, object] = {"tcp4": [], "tcp6": [], "tcp46": [], "error": ""}
|
||
res = run(["sockstat", "-l4", "-l6"], timeout=5)
|
||
if res is None:
|
||
info["error"] = "'sockstat' not found"
|
||
return info
|
||
for line in res[0].split("\n")[1:]: # skip header
|
||
parts = line.split()
|
||
if len(parts) < 6:
|
||
continue
|
||
proto = parts[4] # tcp4, tcp6, or tcp46 (dual-stack)
|
||
if proto not in ("tcp4", "tcp6", "tcp46"):
|
||
continue
|
||
info[proto].append({"addr": parts[5], "process": parts[1]})
|
||
return info
|
||
|
||
|
||
def compute_grade(data: dict[str, object]) -> dict[str, object]:
|
||
"""Compute an A–F summary grade based on key metrics.
|
||
|
||
Scoring factors (each 0–20 points, total 100):
|
||
- Packet loss (lower is better)
|
||
- Latency (lower is better)
|
||
- DNS resolution speed (lower is better)
|
||
- IPv4 reachability (presence of address + test-site reachability)
|
||
- IPv6 / dual-stack support (presence of address + test-site reachability)
|
||
|
||
Letter mapping: A ≥ 90, B ≥ 80, C ≥ 70, D ≥ 60, F < 60.
|
||
"""
|
||
score = 0
|
||
breakdown: dict[str, int] = {}
|
||
|
||
# --- Packet loss (20 pts) ---
|
||
ping = data.get("ping", {})
|
||
loss = float(ping.get("loss_pct", 100))
|
||
if loss <= 0:
|
||
loss_score = 20
|
||
elif loss <= 5:
|
||
loss_score = 15
|
||
elif loss <= 10:
|
||
loss_score = 10
|
||
elif loss <= 30:
|
||
loss_score = 5
|
||
else:
|
||
loss_score = 0
|
||
breakdown["packet_loss"] = loss_score
|
||
score += loss_score
|
||
|
||
# --- Latency (20 pts) ---
|
||
avg_ms = float(ping.get("avg_ms", 0))
|
||
if avg_ms <= 0:
|
||
latency_score = 0
|
||
elif avg_ms < 30:
|
||
latency_score = 20
|
||
elif avg_ms < 80:
|
||
latency_score = 15
|
||
elif avg_ms < 150:
|
||
latency_score = 10
|
||
elif avg_ms < 300:
|
||
latency_score = 5
|
||
else:
|
||
latency_score = 0
|
||
breakdown["latency"] = latency_score
|
||
score += latency_score
|
||
|
||
# --- DNS speed (20 pts) ---
|
||
dns = data.get("dns", {})
|
||
dns_samples = 0
|
||
dns_total = 0.0
|
||
for family in ("v6", "v4"):
|
||
for domain_results in dns.get(family, {}).values():
|
||
for r in domain_results:
|
||
if r["ms"] > 0:
|
||
dns_samples += 1
|
||
dns_total += r["ms"]
|
||
dns_score = 0
|
||
if dns_samples > 0:
|
||
dns_avg = dns_total / dns_samples
|
||
if dns_avg < 20:
|
||
dns_score = 20
|
||
elif dns_avg < 50:
|
||
dns_score = 15
|
||
elif dns_avg < 100:
|
||
dns_score = 10
|
||
elif dns_avg < 200:
|
||
dns_score = 5
|
||
breakdown["dns_speed"] = dns_score
|
||
score += dns_score
|
||
|
||
# --- IPv4 reachability (20 pts) ---
|
||
ipv4 = data.get("ipv4", {})
|
||
v4_score = 0
|
||
if ipv4.get("available"):
|
||
v4_score += 10
|
||
# bonus for each reachable test site
|
||
sites = ipv4.get("test_sites", {})
|
||
reachable_count = sum(1 for s in sites.values() if s.get("reachable"))
|
||
v4_score += min(reachable_count * 5, 10)
|
||
breakdown["ipv4_reachability"] = v4_score
|
||
score += v4_score
|
||
|
||
# --- IPv6 / dual-stack (20 pts) ---
|
||
ipv6 = data.get("ipv6", {})
|
||
v6_score = 0
|
||
if ipv6.get("available"):
|
||
v6_score += 10
|
||
sites = ipv6.get("test_sites", {})
|
||
reachable_count = sum(1 for s in sites.values() if s.get("reachable"))
|
||
v6_score += min(reachable_count * 5, 10)
|
||
breakdown["ipv6_dualstack"] = v6_score
|
||
score += v6_score
|
||
|
||
# Letter mapping
|
||
if score >= 90:
|
||
letter = "A"
|
||
elif score >= 80:
|
||
letter = "B"
|
||
elif score >= 70:
|
||
letter = "C"
|
||
elif score >= 60:
|
||
letter = "D"
|
||
else:
|
||
letter = "F"
|
||
|
||
grade_color = GREEN if letter in ("A", "B") else YELLOW if letter in ("C", "D") else RED
|
||
|
||
return {
|
||
"grade": letter,
|
||
"score": score,
|
||
"max_score": 100,
|
||
"colour": grade_color,
|
||
"breakdown": breakdown,
|
||
}
|
||
|
||
|
||
def _ms_colour(ms: float) -> str:
|
||
"""Return an ANSI colour code for a given latency value in ms."""
|
||
if ms <= 0:
|
||
return DIM
|
||
if ms < 50:
|
||
return GREEN
|
||
if ms < 200:
|
||
return YELLOW
|
||
return RED
|
||
|
||
|
||
def _print_reachability(family: str, info: dict[str, object] | None) -> None:
|
||
"""Pretty-print one per-family reachability section."""
|
||
if not info:
|
||
return
|
||
label = "IPv6" if family == "v6" else "IPv4"
|
||
if not info.get("available"):
|
||
print(f" {YELLOW}○{RESET} No {label} address detected on any interface")
|
||
return
|
||
scope = "Global" if info.get("is_global") else "Private"
|
||
print(f" {GREEN}✓{RESET} {scope} {label} address: {info['addr']}")
|
||
suffix = " over IPv6" if family == "v6" else ""
|
||
for site, result in info.get("test_sites", {}).items():
|
||
if result.get("reachable"):
|
||
print(f" {GREEN}✓{RESET} {site:<20s} {result['ms']}ms ({result['ip']})")
|
||
else:
|
||
print(f" {RED}✗{RESET} {site:<20s} unreachable{suffix}")
|
||
|
||
|
||
def print_report(data: dict[str, object]) -> None:
|
||
"""Render the human-readable diagnostic report to stdout."""
|
||
grade_info = data.get("grade", {})
|
||
grade_letter = str(grade_info.get("grade", "?"))
|
||
grade_colour = str(grade_info.get("colour", RESET))
|
||
|
||
# --- Header ---
|
||
os_label = str(data.get("os", "Linux")).capitalize()
|
||
print(f"\n{BOLD}{'═' * 60}{RESET}")
|
||
print(f"{BOLD} Network Diagnostics v{VERSION} — {os_label}{RESET}")
|
||
print(f"{BOLD}{'═' * 60}{RESET}")
|
||
print(f" Hostname: {data['hostname']}")
|
||
print(f" Protocol: {data.get('protocol', 'auto')}")
|
||
score = grade_info.get("score", "?")
|
||
max_score = grade_info.get("max_score", 100)
|
||
print(f" Summary Grade: {grade_colour}{BOLD}{grade_letter}{RESET} ({score}/{max_score})")
|
||
|
||
# --- Interfaces ---
|
||
print(f"\n{BOLD}── Interfaces ──{RESET}")
|
||
iface = data.get("interfaces", {})
|
||
if iface.get("error"):
|
||
print(f" {DIM}unavailable: {iface['error']}{RESET}")
|
||
for ifc in iface.get("interfaces", []):
|
||
state_color = GREEN if ifc["state"] == "UP" else RED
|
||
ips = ", ".join(ifc["ips"])
|
||
print(f" {ifc['name']:<8s} {state_color}{ifc['state']:<5s}{RESET} {ips}")
|
||
if iface.get("default_route_v6"):
|
||
print(f" Default IPv6 route → {iface['default_route_v6']}")
|
||
if iface.get("default_route_v4"):
|
||
print(f" Default IPv4 route → {iface['default_route_v4']}")
|
||
|
||
# --- Internet connectivity ---
|
||
print(f"\n{BOLD}── Internet Connectivity ──{RESET}")
|
||
conn = data.get("connectivity", {})
|
||
if conn.get("error"):
|
||
print(f" {DIM}unavailable: {conn['error']}{RESET}")
|
||
elif conn:
|
||
for family_label, fam_key in [("IPv6", "v6"), ("IPv4", "v4")]:
|
||
fam_data = conn.get(fam_key, {})
|
||
status_parts = []
|
||
for host in fam_data:
|
||
result = fam_data[host]
|
||
if result.get("reachable"):
|
||
ms = result.get("ms", 0)
|
||
colour = _ms_colour(ms)
|
||
status_parts.append(f"{GREEN}✓{RESET} {host} {colour}{ms:.0f}ms{RESET}")
|
||
else:
|
||
status_parts.append(f"{RED}✗{RESET} {host}")
|
||
hosts_line = ", ".join(status_parts) if status_parts else f"{DIM}(none){RESET}"
|
||
print(f" {family_label}: {hosts_line}")
|
||
|
||
# --- Latency (ICMP ping) ---
|
||
print(f"\n{BOLD}── Latency (ICMP ping) ──{RESET}")
|
||
_print_ping_block("IPv6", data.get("ping_v6"))
|
||
_print_ping_block("IPv4", data.get("ping_v4"))
|
||
|
||
# --- DNS ---
|
||
print(f"\n{BOLD}── DNS Resolution ──{RESET}")
|
||
dns = data.get("dns", {})
|
||
if dns:
|
||
print(f" FQDN: {dns.get('fqdn', '?')}")
|
||
if dns.get("resolvers"):
|
||
print(f" Resolvers: {', '.join(dns['resolvers'])}")
|
||
for family_label, fam_key in [("AAAA (IPv6)", "v6"), ("A (IPv4)", "v4")]:
|
||
domains = dns.get(fam_key, {})
|
||
if not domains:
|
||
continue
|
||
print(f" {BOLD}{family_label}:{RESET}")
|
||
for domain, results in domains.items():
|
||
valid = [r["ms"] for r in results if r["ms"] > 0]
|
||
avg = statistics.mean(valid) if valid else 0
|
||
colour = _ms_colour(avg)
|
||
ips = [r["ip"] for r in results if r["ip"] != "N/A"]
|
||
ip_str = ips[0] if ips else "N/A"
|
||
print(f" {domain:<25s} → {colour}{ip_str:<18s}{RESET} {avg:.0f}ms")
|
||
|
||
# --- IPv6 reachability ---
|
||
print(f"\n{BOLD}── IPv6 Reachability ──{RESET}")
|
||
_print_reachability("v6", data.get("ipv6"))
|
||
|
||
# --- IPv4 reachability ---
|
||
print(f"\n{BOLD}── IPv4 Reachability ──{RESET}")
|
||
_print_reachability("v4", data.get("ipv4"))
|
||
|
||
# --- Path MTU ---
|
||
print(f"\n{BOLD}── Path MTU ──{RESET}")
|
||
mtu = data.get("mtu", {})
|
||
if mtu:
|
||
if mtu.get("path_mtu"):
|
||
colour = YELLOW if mtu["path_mtu"] < 1500 else GREEN
|
||
print(f" {colour}{mtu['path_mtu']}B{RESET} ({mtu.get('method', '')})")
|
||
elif mtu.get("method"):
|
||
print(f" {DIM}{mtu['method']}{RESET}")
|
||
elif mtu.get("error"):
|
||
print(f" {DIM}unavailable: {mtu['error']}{RESET}")
|
||
|
||
# --- Listening ports ---
|
||
print(f"\n{BOLD}── Listening Ports ──{RESET}")
|
||
ports = data.get("listening", {})
|
||
if ports:
|
||
if ports.get("error"):
|
||
print(f" {DIM}unavailable: {ports['error']}{RESET}")
|
||
else:
|
||
total = sum(len(v) for v in ports.values() if isinstance(v, list))
|
||
if total > 0:
|
||
for version in ["tcp6", "tcp4", "tcp46"]:
|
||
for entry in ports.get(version, []):
|
||
print(f" {version} {entry['addr']:<22s} {DIM}{entry['process']}{RESET}")
|
||
else:
|
||
print(" (none)")
|
||
|
||
# --- Grade breakdown ---
|
||
print(f"\n{BOLD}── Grade Breakdown ──{RESET}")
|
||
bd = grade_info.get("breakdown", {})
|
||
labels = {
|
||
"packet_loss": "Packet loss",
|
||
"latency": "Latency",
|
||
"dns_speed": "DNS speed",
|
||
"ipv6_dualstack": "IPv6 / dual-stack",
|
||
"ipv4_reachability": "IPv4 reachability",
|
||
}
|
||
for key, label in labels.items():
|
||
pts = bd.get(key, 0)
|
||
bar = "█" * (pts // 2) + "░" * (10 - pts // 2)
|
||
print(f" {label:<20s} {bar} {pts}/20")
|
||
|
||
# --- Footer ---
|
||
print(f"\n{BOLD}{'═' * 60}{RESET}")
|
||
print(
|
||
f" {BOLD}Overall Grade: {grade_colour}{grade_letter}{RESET} "
|
||
f"({grade_info.get('score', '?')}/{grade_info.get('max_score', 100)})"
|
||
)
|
||
print(f"{BOLD}{'═' * 60}{RESET}\n")
|
||
|
||
|
||
def _print_ping_block(label: str, ping_info: dict[str, object] | None) -> None:
|
||
"""Pretty-print a single ping result block for a given protocol."""
|
||
if not isinstance(ping_info, dict):
|
||
return
|
||
if ping_info.get("received", 0) == 0:
|
||
note = ping_info.get("note") or ping_info.get("error")
|
||
suffix = f" ({note})" if note else ""
|
||
print(f" {label}: {RED}no response{suffix}{RESET}")
|
||
return
|
||
|
||
loss = float(ping_info.get("loss_pct", 0))
|
||
loss_color = RED if loss > 10 else YELLOW if loss > 0 else GREEN
|
||
avg = float(ping_info.get("avg_ms", 0))
|
||
latency_color = _ms_colour(avg)
|
||
print(
|
||
f" {label}: {ping_info.get('sent')}/{ping_info.get('received')} received"
|
||
f" Loss: {loss_color}{loss}%{RESET}"
|
||
f" Avg: {latency_color}{avg:.1f}ms{RESET}"
|
||
f" Min/Max: {ping_info.get('min_ms', 0):.1f}/{ping_info.get('max_ms', 0):.1f}ms"
|
||
f" σ: {ping_info.get('stddev_ms', 0):.1f}ms"
|
||
)
|
||
|
||
|
||
def main() -> None:
|
||
"""Parse arguments, collect diagnostics, and print the report."""
|
||
parser = argparse.ArgumentParser(
|
||
description="Network diagnostics — latency, DNS, MTU, dual-stack, ports",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
)
|
||
parser.add_argument(
|
||
"--target",
|
||
default="cloudflare.com",
|
||
help="Target host for tests (default: cloudflare.com)",
|
||
)
|
||
parser.add_argument(
|
||
"--protocol",
|
||
choices=["auto", "v4", "v6"],
|
||
default="auto",
|
||
help="Address family: auto (dual-stack), v4 (IPv4 only), v6 (IPv6 only)",
|
||
)
|
||
parser.add_argument(
|
||
"--version",
|
||
action="version",
|
||
version=f"%(prog)s {VERSION}",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
target: str = args.target
|
||
protocol: str = args.protocol
|
||
|
||
os_type = detect_os()
|
||
|
||
data: dict[str, object] = {
|
||
"hostname": socket.gethostname(),
|
||
"version": VERSION,
|
||
"os": os_type,
|
||
"protocol": protocol,
|
||
"interfaces": {},
|
||
"connectivity": {},
|
||
"ping": {},
|
||
"ping_v4": None,
|
||
"ping_v6": None,
|
||
"dns": {},
|
||
"ipv4": {},
|
||
"ipv6": {},
|
||
"mtu": {},
|
||
"listening": {},
|
||
"grade": {},
|
||
}
|
||
|
||
# Phase 1: interface info (instant, sequential)
|
||
_status("Checking interfaces")
|
||
data["interfaces"] = check_interface_info(os_type)
|
||
iface_error = data["interfaces"].get("error")
|
||
_status_done(f"warning: {iface_error}" if iface_error else "done")
|
||
|
||
# Phase 2: all other checks run in parallel
|
||
with ThreadPoolExecutor(max_workers=6) as executor:
|
||
future_map: dict[Future[object], tuple[str, str]] = {}
|
||
|
||
# Connectivity (TCP handshakes; results shared with reachability)
|
||
fut: Future[object] = executor.submit(check_connectivity)
|
||
future_map[fut] = ("connectivity", "internet connectivity")
|
||
|
||
# Ping(s) — honour protocol preference, IPv6 first
|
||
if protocol in ("auto", "v6"):
|
||
fut = executor.submit(ping_host, target, os_type, protocol="v6")
|
||
future_map[fut] = ("ping_v6", "ping (IPv6)")
|
||
if protocol in ("auto", "v4"):
|
||
fut = executor.submit(ping_host, target, os_type, protocol="v4")
|
||
future_map[fut] = ("ping_v4", "ping (IPv4)")
|
||
|
||
fut = executor.submit(check_dns, protocol)
|
||
future_map[fut] = ("dns", "DNS resolution")
|
||
fut = executor.submit(check_mtu, target, protocol, os_type)
|
||
future_map[fut] = ("mtu", "path MTU")
|
||
fut = executor.submit(check_listening_ports, os_type)
|
||
future_map[fut] = ("listening", "listening ports")
|
||
|
||
# Collect results as they complete, showing progress on stderr
|
||
print(f" Running {len(future_map)} checks in parallel...", file=sys.stderr)
|
||
for future in as_completed(future_map):
|
||
key, label = future_map[future]
|
||
try:
|
||
data[key] = future.result()
|
||
except Exception as exc:
|
||
data[key] = {"error": str(exc)}
|
||
print(f" {label}... failed ({exc})", file=sys.stderr)
|
||
continue
|
||
note = data[key].get("note") if isinstance(data[key], dict) else None
|
||
if note:
|
||
print(f" {label}... warning ({note})", file=sys.stderr)
|
||
else:
|
||
print(f" {label}... done", file=sys.stderr)
|
||
|
||
# Derive per-family reachability from interface data + shared probes
|
||
conn = data.get("connectivity")
|
||
probes_v4 = conn.get("v4", {}) if isinstance(conn, dict) else {}
|
||
probes_v6 = conn.get("v6", {}) if isinstance(conn, dict) else {}
|
||
data["ipv6"] = _reachability_report("v6", data["interfaces"], probes_v6)
|
||
data["ipv4"] = _reachability_report("v4", data["interfaces"], probes_v4)
|
||
|
||
# Post-process ping results: pick the best for the dual-stack summary
|
||
if protocol == "v4":
|
||
data["ping"] = data.get("ping_v4", {})
|
||
elif protocol == "v6":
|
||
data["ping"] = data.get("ping_v6", {})
|
||
else:
|
||
# Dual-stack: prefer IPv6 when it answered (IPv6-first)
|
||
ping_v4 = data.get("ping_v4")
|
||
ping_v6 = data.get("ping_v6")
|
||
v6_recv = int(ping_v6.get("received", 0)) if isinstance(ping_v6, dict) else 0
|
||
if v6_recv > 0 and isinstance(ping_v6, dict):
|
||
data["ping"] = ping_v6
|
||
elif isinstance(ping_v4, dict):
|
||
data["ping"] = ping_v4
|
||
else:
|
||
data["ping"] = {}
|
||
|
||
# Compute summary grade
|
||
data["grade"] = compute_grade(data)
|
||
|
||
print_report(data)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except KeyboardInterrupt:
|
||
print("\nInterrupted.", file=sys.stderr)
|
||
sys.exit(130)
|