diff --git a/scripts/hbc_mini.py b/scripts/hbc_mini.py index 9fa78f1..a8408a3 100755 --- a/scripts/hbc_mini.py +++ b/scripts/hbc_mini.py @@ -30,15 +30,17 @@ import re import shutil import signal import socket +import struct import subprocess import sys import time import zlib from abc import ABC, abstractmethod from collections import defaultdict +from dataclasses import asdict, dataclass from logging.handlers import SysLogHandler from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Awaitable, Callable, Dict, List, Optional, Set, Tuple # updated by scripts/bumpminor.sh __version__ = "5.4.4" @@ -637,6 +639,315 @@ class DiskMonitorPlugin(MonitorPlugin): return {"partitions": partitions} if partitions else {} +# --------------------------------------------------------------------------- +# Connectivity probe — per-protocol internet presence (IPv4 / IPv6) +# +# Tier 1 (every tier1_interval s, sends no packets): a UDP connect() to the probe address +# shows whether a route exists and which source address would be used. Tier 2 (every +# probe_interval s, sooner after a failure or a change of address or gateway): one ICMP echo, +# only when tier 1 found a route. A family is absent after no route at all or fail_limit +# consecutive failed echoes, present after one success, and unknown (flag omitted) when it +# cannot be probed. The result rides on every HTB as p4/p6 and, when network_monitor runs, as +# its `connectivity` metric. Mirrors hbd/client/connectivity.py. +# --------------------------------------------------------------------------- + +_FAMILIES = ("ipv4", "ipv6") +_FLAG_KEYS = {"ipv4": "p4", "ipv6": "p6"} +_DEFAULT_PROBES = {"ipv4": "1.1.1.1", "ipv6": "2606:4700:4700::1111"} +_BSD_PLATFORMS = ("darwin", "freebsd", "netbsd", "openbsd", "dragonfly") + + +class _ProbeUnavailable(Exception): + """The address family cannot be probed on this host (e.g. no IPv6 support).""" + + +class _PingUnavailable(Exception): + """The system ping binary could not be executed.""" + + +def _parse_proc_route_v4(text: str) -> Optional[str]: + """Default gateway from /proc/net/route (lowest metric wins).""" + best: Optional[Tuple[int, str]] = None + for line in text.splitlines()[1:]: + f = line.split() + if len(f) < 8 or f[1] != "00000000" or not int(f[3], 16) & 0x2: + continue + metric = int(f[6]) + if best is None or metric < best[0]: + # the kernel prints the raw network-order u32 in host order: use the host byte order + best = (metric, socket.inet_ntoa(int(f[2], 16).to_bytes(4, sys.byteorder))) + return best[1] if best else None + + +def _parse_proc_route_v6(text: str) -> Optional[str]: + """Default gateway from /proc/net/ipv6_route (lowest metric wins).""" + best: Optional[Tuple[int, str]] = None + for line in text.splitlines(): + f = line.split() + if len(f) < 10: + continue + dest, plen, nexthop, metric, flags, iface = f[0], f[1], f[4], f[5], f[8], f[9] + if dest != "0" * 32 or plen != "00" or nexthop == "0" * 32 or iface == "lo": + continue + fl = int(flags, 16) + if not fl & 0x2 or fl & 0x200: # need RTF_GATEWAY, skip RTF_REJECT + continue + m = int(metric, 16) + if best is None or m < best[0]: + best = (m, socket.inet_ntop(socket.AF_INET6, bytes.fromhex(nexthop))) + return best[1] if best else None + + +def _parse_route_get(text: str) -> Optional[str]: + """Gateway from `route -n get [-inet6] default` (macOS and the BSDs).""" + m = re.search(r"^\s*gateway:\s*(\S+)", text, re.MULTILINE) + return m.group(1) if m else None + + +def _parse_windows_route(text: str, family: str) -> Optional[str]: + """Default gateway from `route print -4` / `route print -6` (lowest metric wins).""" + best: Optional[Tuple[int, str]] = None + for line in text.splitlines(): + if family == "ipv4": + f = line.split() + if not (len(f) >= 5 and f[0] == "0.0.0.0" and f[1] == "0.0.0.0" and f[2] != "On-link"): + continue + cand = (int(f[4]), f[2]) + else: + m = re.match(r"^\s*\d+\s+(\d+)\s+::/0\s+(\S+)\s*$", line) + if not m or m.group(2) == "On-link": + continue + cand = (int(m.group(1)), m.group(2)) + if best is None or cand[0] < best[0]: + best = cand + return best[1] if best else None + + +async def _run_cmd(cmd: List[str], timeout: float = 5.0) -> str: + proc = await asyncio.create_subprocess_exec( + *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL) + try: + out, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout) + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + raise + return out.decode(errors="replace") + + +async def _default_gateway(family: str) -> Optional[str]: + """Default gateway for "ipv4"/"ipv6", or None on any failure (informational only).""" + v4 = family == "ipv4" + try: + if sys.platform.startswith("linux"): + with open("/proc/net/route" if v4 else "/proc/net/ipv6_route") as fh: + text = fh.read() + return _parse_proc_route_v4(text) if v4 else _parse_proc_route_v6(text) + if sys.platform.startswith(_BSD_PLATFORMS): + cmd = ["route", "-n", "get"] + ([] if v4 else ["-inet6"]) + ["default"] + return _parse_route_get(await _run_cmd(cmd)) + if sys.platform == "win32": + text = await _run_cmd(["route", "print", "-4" if v4 else "-6"]) + return _parse_windows_route(text, family) + except Exception: + return None + return None + + +def _probe_ping_command(host: str, family: str) -> List[str]: + """Command for ONE echo request with a ~2 s reply wait.""" + v4 = family == "ipv4" + if sys.platform == "win32": + return ["ping", "-n", "1", "-w", "2000", "-4" if v4 else "-6", host] + if sys.platform.startswith(_BSD_PLATFORMS): + # BSD/macOS -W is in milliseconds, and IPv6 is reached through ping6. + return ["ping", "-c", "1", "-W", "2000", host] if v4 else ["ping6", "-c", "1", host] + return ["ping", "-c", "1", "-W", "2", "-4" if v4 else "-6", host] + + +def _parse_ping_rtt(output: str) -> Optional[float]: + """Average RTT in ms from ping output, or None when there was no reply.""" + m = re.search(r"(?:rtt|round-trip)\s+min/avg/max/\S+\s*=\s*([\d.]+)/([\d.]+)/([\d.]+)", output) + if m: + return float(m.group(2)) + if "TTL=" in output: # Windows reply line (also when localized) + m = re.search(r"[=<]\s*(\d+)\s*ms", output) + return float(m.group(1)) if m else 0.0 + return None + + +async def _ping_rtt(target: str, family: str) -> Optional[float]: + """One ICMP echo; RTT in ms, or None if no reply. Raises _PingUnavailable without ping.""" + try: + proc = await asyncio.create_subprocess_exec( + *_probe_ping_command(target, family), + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) + except FileNotFoundError as e: + raise _PingUnavailable(str(e)) + try: + out, _ = await asyncio.wait_for(proc.communicate(), timeout=6) + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + return None + return _parse_ping_rtt(out.decode(errors="replace")) + + +def _local_address(family: str, target: str) -> Optional[str]: + """Source address the kernel would use to reach target, or None if there is no route. + + A UDP connect() sends nothing. Raises _ProbeUnavailable if the family is unsupported. + """ + af = socket.AF_INET if family == "ipv4" else socket.AF_INET6 + try: + sock = socket.socket(af, socket.SOCK_DGRAM) + except OSError as e: + raise _ProbeUnavailable(str(e)) + try: + sock.connect((target, 9)) + return str(sock.getsockname()[0]) + except OSError: + return None + finally: + sock.close() + + +@dataclass +class _FamilyState: + present: Optional[bool] = None # None = unknown (probe cannot run yet/at all) + addr: Optional[str] = None + gateway: Optional[str] = None + rtt: Optional[float] = None + + +class _Connectivity: + def __init__( + self, + probes: Optional[Dict[str, str]] = None, + probe_interval: float = 60.0, + retry_interval: float = 10.0, + tier1_interval: float = 5.0, + fail_limit: int = 2, + local_address_fn: Callable[[str, str], Optional[str]] = _local_address, + ping_fn: Callable[[str, str], Awaitable[Optional[float]]] = _ping_rtt, + gateway_fn: Callable[[str], Awaitable[Optional[str]]] = _default_gateway, + clock: Callable[[], float] = time.monotonic, + ): + self.probes = dict(_DEFAULT_PROBES) + self.probes.update(probes or {}) + self.probe_interval = probe_interval + self.retry_interval = retry_interval + self.tier1_interval = tier1_interval + self.fail_limit = fail_limit + self._local_address = local_address_fn + self._ping = ping_fn + self._gateway_fn = gateway_fn + self._clock = clock + # Reading the gateway is a cheap /proc read on Linux but a subprocess elsewhere, so + # elsewhere it is refreshed only when a tier-2 probe is due. + self._cheap_gateway = sys.platform.startswith("linux") + self.states: Dict[str, _FamilyState] = {f: _FamilyState() for f in _FAMILIES} + self._fails: Dict[str, int] = {f: 0 for f in _FAMILIES} + self._next_ping: Dict[str, float] = {f: 0.0 for f in _FAMILIES} + self._last_key: Dict[str, Optional[Tuple[Optional[str], Optional[str]]]] = { + f: None for f in _FAMILIES} + self._ping_disabled: Set[str] = set() + + @classmethod + def from_config(cls, cfg: Optional[Dict[str, Any]]) -> "_Connectivity": + cfg = cfg or {} + probes = {} + if cfg.get("probe_v4"): + probes["ipv4"] = str(cfg["probe_v4"]) + if cfg.get("probe_v6"): + probes["ipv6"] = str(cfg["probe_v6"]) + return cls(probes=probes, probe_interval=float(cfg.get("probe_interval", 60))) + + # ---- readers (cheap; safe to call on every heartbeat) --------------------------- + def snapshot(self) -> Dict[str, _FamilyState]: + return {f: _FamilyState(**asdict(s)) for f, s in self.states.items()} + + def flags(self) -> Dict[str, bool]: + """HTB flags for families whose presence is known.""" + return {_FLAG_KEYS[f]: s.present for f, s in self.states.items() if s.present is not None} + + def as_metric(self) -> Dict[str, Dict[str, Any]]: + return {f: asdict(s) for f, s in self.states.items()} + + # ---- probing ---------------------------------------------------------------------- + async def refresh(self) -> None: + for fam in _FAMILIES: + try: + await self._refresh_family(fam) + except Exception: + logging.getLogger("hbc.probe").exception("connectivity probe failed for %s", fam) + + async def run(self) -> None: + while _running: + await self.refresh() + await _sleep(self.tier1_interval) + + async def _gateway(self, fam: str) -> Optional[str]: + try: + return await self._gateway_fn(fam) + except Exception: + return None + + async def _refresh_family(self, fam: str) -> None: + st = self.states[fam] + try: + addr = self._local_address(fam, self.probes[fam]) + except _ProbeUnavailable: + st.present = None + return + + if addr is None: # tier 1: no route => absent immediately + self._fails[fam] = 0 + self._next_ping[fam] = 0.0 + self._last_key[fam] = None + st.present, st.addr, st.gateway, st.rtt = False, None, None, None + return + + now = self._clock() + gw = await self._gateway(fam) if self._cheap_gateway else st.gateway + key = (addr, gw) + due = key != self._last_key[fam] or now >= self._next_ping[fam] + if due and not self._cheap_gateway: + gw = await self._gateway(fam) + key = (addr, gw) + st.addr, st.gateway = addr, gw + self._last_key[fam] = key + if fam in self._ping_disabled: + st.present = None # cannot probe: unknown, never absent, while a route exists + elif due: + await self._tier2(fam, st, now) + + async def _tier2(self, fam: str, st: _FamilyState, now: float) -> None: + try: + rtt = await self._ping(self.probes[fam], fam) + except _PingUnavailable: + logging.getLogger("hbc.probe").warning("ping unavailable; %s presence stays unknown", fam) + self._ping_disabled.add(fam) + st.present = None + return + except (asyncio.TimeoutError, OSError): + rtt = None + if rtt is None: + self._fails[fam] += 1 + self._next_ping[fam] = now + self.retry_interval + if self._fails[fam] >= self.fail_limit: + st.present, st.rtt = False, None + else: + self._fails[fam] = 0 + self._next_ping[fam] = now + self.probe_interval + st.present, st.rtt = True, rtt + + +# The running probe (None when probe_enabled is false); shared by the heartbeat sender and +# the network_monitor plugin. Set in _async_main. +_probe: Optional[_Connectivity] = None + + # --------------------------------------------------------------------------- # Plugin: network_monitor (Linux /proc/net/dev) # --------------------------------------------------------------------------- diff --git a/tests/test_hbc_mini_probe.py b/tests/test_hbc_mini_probe.py new file mode 100644 index 0000000..ca9090d --- /dev/null +++ b/tests/test_hbc_mini_probe.py @@ -0,0 +1,306 @@ +"""Connectivity probe in the standalone scripts/hbc_mini.py (stdlib-only single-file client).""" +import asyncio +import importlib.util +import pathlib +import sys + +import pytest + +_PATH = pathlib.Path(__file__).resolve().parents[1] / "scripts" / "hbc_mini.py" +_SPEC = importlib.util.spec_from_file_location("hbc_mini", _PATH) +mini = importlib.util.module_from_spec(_SPEC) +sys.modules["hbc_mini"] = mini # dataclasses resolve their module by name +_SPEC.loader.exec_module(mini) + +PROC_V4 = ( + "Iface\tDestination\tGateway \tFlags\tRefCnt\tUse\tMetric\tMask\t\tMTU\tWindow\tIRTT\n" + "wlan0\t00000000\t0100A8C0\t0003\t0\t0\t600\t00000000\t0\t0\t0\n" + "eth0\t00000000\t0101A8C0\t0003\t0\t0\t100\t00000000\t0\t0\t0\n" + "wlan0\t0000A8C0\t00000000\t0001\t0\t0\t600\t00FFFFFF\t0\t0\t0\n" +) +PROC_V6 = ( + "00000000000000000000000000000000 00 00000000000000000000000000000000 00 " + "fe800000000000000000000000000001 00000400 00000001 00000000 00000003 wlan0\n" + "00000000000000000000000000000000 00 00000000000000000000000000000000 00 " + "00000000000000000000000000000000 ffffffff 00000001 00000000 00200200 lo\n" + "fe800000000000000000000000000000 40 00000000000000000000000000000000 00 " + "00000000000000000000000000000000 00000100 00000001 00000000 00000001 wlan0\n" +) +ROUTE_GET = """ route to: 0.0.0.0 +destination: 0.0.0.0 + mask: 0.0.0.0 + gateway: 192.0.2.1 + fib: 0 + interface: vtnet0 +""" +WIN_V4 = """IPv4 Route Table +Active Routes: +Network Destination Netmask Gateway Interface Metric + 0.0.0.0 0.0.0.0 192.168.1.1 192.168.1.50 25 + 0.0.0.0 0.0.0.0 10.0.0.1 10.0.0.9 50 + 127.0.0.0 255.0.0.0 On-link 127.0.0.1 331 +""" +WIN_V6 = """IPv6 Route Table +Active Routes: + If Metric Network Destination Gateway + 12 281 ::/0 fe80::1 + 1 331 ::1/128 On-link +""" +LINUX_PING = """PING 1.1.1.1 (1.1.1.1) 56(84) bytes of data. +64 bytes from 1.1.1.1: icmp_seq=1 ttl=57 time=11.2 ms + +--- 1.1.1.1 ping statistics --- +1 packets transmitted, 1 received, 0% packet loss, time 0ms +rtt min/avg/max/mdev = 11.234/11.234/11.234/0.000 ms +""" +FREEBSD_PING = """PING 1.1.1.1 (1.1.1.1): 56 data bytes +64 bytes from 1.1.1.1: icmp_seq=0 ttl=59 time=4.006 ms + +--- 1.1.1.1 ping statistics --- +1 packets transmitted, 1 packets received, 0.0% packet loss, 1 packets out of wait time +round-trip min/avg/max/stddev = 4.006/4.006/4.006/0.000 ms +""" +WIN_PING = """Pinging 1.1.1.1 with 32 bytes of data: +Reply from 1.1.1.1: bytes=32 time=12ms TTL=57 + +Ping statistics for 1.1.1.1: + Packets: Sent = 1, Received = 1, Lost = 0 (0% loss), +""" +LOST_PING = """--- 1.1.1.1 ping statistics --- +1 packets transmitted, 0 received, 100% packet loss, time 0ms +""" + + +# ---- gateway parsers --------------------------------------------------------------- + +def test_proc_v4_lowest_metric_wins(): + assert mini._parse_proc_route_v4(PROC_V4) == "192.168.1.1" # eth0, metric 100 + + +def test_proc_v4_none_without_default_route(): + assert mini._parse_proc_route_v4(PROC_V4.splitlines()[0] + "\n") is None + + +def test_proc_v6_skips_reject_and_non_default(): + assert mini._parse_proc_route_v6(PROC_V6) == "fe80::1" + + +def test_route_get_bsd_and_macos(): + assert mini._parse_route_get(ROUTE_GET) == "192.0.2.1" + assert mini._parse_route_get("route: writing to routing socket: not in table") is None + + +def test_windows_routes(): + assert mini._parse_windows_route(WIN_V4, "ipv4") == "192.168.1.1" + assert mini._parse_windows_route(WIN_V6, "ipv6") == "fe80::1" + + +def test_default_gateway_never_raises(monkeypatch): + monkeypatch.setattr(sys, "platform", "linux") + monkeypatch.setattr(mini, "_parse_proc_route_v4", lambda t: 1 / 0) + assert asyncio.run(mini._default_gateway("ipv4")) is None + + +# ---- ping --------------------------------------------------------------------------- + +@pytest.mark.parametrize("platform,family,expected", [ + ("linux", "ipv4", ["ping", "-c", "1", "-W", "2", "-4", "h"]), + ("linux", "ipv6", ["ping", "-c", "1", "-W", "2", "-6", "h"]), + ("freebsd15", "ipv4", ["ping", "-c", "1", "-W", "2000", "h"]), + ("freebsd15", "ipv6", ["ping6", "-c", "1", "h"]), + ("darwin", "ipv4", ["ping", "-c", "1", "-W", "2000", "h"]), + ("darwin", "ipv6", ["ping6", "-c", "1", "h"]), + ("win32", "ipv4", ["ping", "-n", "1", "-w", "2000", "-4", "h"]), + ("win32", "ipv6", ["ping", "-n", "1", "-w", "2000", "-6", "h"]), +]) +def test_probe_ping_command(monkeypatch, platform, family, expected): + monkeypatch.setattr(sys, "platform", platform) + assert mini._probe_ping_command("h", family) == expected + + +def test_parse_ping_rtt(): + assert mini._parse_ping_rtt(LINUX_PING) == 11.234 + assert mini._parse_ping_rtt(FREEBSD_PING) == 4.006 + assert mini._parse_ping_rtt(WIN_PING) == 12.0 + assert mini._parse_ping_rtt(LOST_PING) is None + assert mini._parse_ping_rtt("ping: connect: Network is unreachable") is None + + +# ---- probe state machine ------------------------------------------------------------ + +class Net: + """Scriptable network: addr None = no route, 'unavailable' = family unsupported.""" + + def __init__(self): + self.addr = {"ipv4": "192.0.2.5", "ipv6": "2001:db8::5"} + self.gw = {"ipv4": "192.0.2.1", "ipv6": "fe80::1"} + self.rtt = {"ipv4": 10.0, "ipv6": 20.0} # None = ping fails, 'unavailable' = no ping binary + self.pings = [] + self.now = 1000.0 + + def local(self, fam, target): + a = self.addr[fam] + if a == "unavailable": + raise mini._ProbeUnavailable("no such family") + return a + + async def ping(self, target, fam): + self.pings.append(fam) + r = self.rtt[fam] + if r == "unavailable": + raise mini._PingUnavailable("no ping") + return r + + async def gateway(self, fam): + return self.gw[fam] + + def clock(self): + return self.now + + +def make(net, **kw): + c = mini._Connectivity(local_address_fn=net.local, ping_fn=net.ping, gateway_fn=net.gateway, + clock=net.clock, **kw) + c._cheap_gateway = True + return c + + +def refresh(c): + asyncio.run(c.refresh()) + + +def test_present_with_addr_gateway_rtt(): + net = Net() + c = make(net) + refresh(c) + s = c.snapshot()["ipv4"] + assert (s.present, s.addr, s.gateway, s.rtt) == (True, "192.0.2.5", "192.0.2.1", 10.0) + assert c.flags() == {"p4": True, "p6": True} + assert c.as_metric()["ipv6"]["gateway"] == "fe80::1" + + +def test_no_route_is_immediately_absent(): + net = Net() + net.addr["ipv6"] = None + c = make(net) + refresh(c) + assert c.flags() == {"p4": True, "p6": False} + assert net.pings == ["ipv4"] + assert c.snapshot()["ipv6"].addr is None + + +def test_ping_is_debounced_and_rate_limited(): + net = Net() + c = make(net, probe_interval=60.0, retry_interval=10.0) + refresh(c) + refresh(c) + assert net.pings == ["ipv4", "ipv6"] + net.now += 60 + net.rtt["ipv6"] = None + refresh(c) + assert c.flags()["p6"] is True + net.now += 10 + refresh(c) + assert c.flags()["p6"] is False + + +def test_first_probe_failing_once_is_unknown(): + net = Net() + net.rtt["ipv4"] = None + c = make(net) + refresh(c) + assert "p4" not in c.flags() and c.flags()["p6"] is True + + +def test_recovers_on_first_success(): + net = Net() + net.rtt["ipv6"] = None + c = make(net, retry_interval=10.0) + refresh(c) + net.now += 10 + refresh(c) + assert c.flags()["p6"] is False + net.rtt["ipv6"] = 25.0 + net.now += 10 + refresh(c) + assert c.flags()["p6"] is True + + +def test_address_change_invalidates_cache(): + net = Net() + c = make(net) + refresh(c) + net.addr["ipv4"] = "198.51.100.7" + refresh(c) + assert net.pings.count("ipv4") == 2 + assert c.snapshot()["ipv4"].addr == "198.51.100.7" + + +def test_gateway_change_invalidates_cache(): + net = Net() + c = make(net) + refresh(c) + net.gw["ipv4"] = "192.0.2.254" + refresh(c) + assert net.pings.count("ipv4") == 2 + + +def test_route_loss_then_return_pings_immediately(): + net = Net() + c = make(net) + refresh(c) + net.addr["ipv6"] = None + refresh(c) + net.addr["ipv6"] = "2001:db8::5" + refresh(c) + assert net.pings.count("ipv6") == 2 and c.flags()["p6"] is True + + +def test_missing_ping_binary_leaves_presence_unknown_and_stops_retrying(): + net = Net() + net.rtt["ipv4"] = "unavailable" + c = make(net) + refresh(c) + net.now += 600 + refresh(c) + assert "p4" not in c.flags() + assert net.pings.count("ipv4") == 1 + + +def test_ping_less_family_stays_unknown_after_route_returns(): + net = Net() + net.rtt["ipv4"] = "unavailable" + c = make(net) + refresh(c) + assert "p4" not in c.flags() + net.addr["ipv4"] = None + refresh(c) + assert c.flags()["p4"] is False # no route is definitive + net.addr["ipv4"] = "192.0.2.5" + refresh(c) + assert "p4" not in c.flags() and net.pings.count("ipv4") == 1 + + +def test_unsupported_family_is_unknown(): + net = Net() + net.addr["ipv6"] = "unavailable" + c = make(net) + refresh(c) + assert "p6" not in c.flags() and c.flags()["p4"] is True + + +def test_from_config(): + c = mini._Connectivity.from_config({"probe_v4": "9.9.9.9", "probe_interval": 30}) + assert c.probes["ipv4"] == "9.9.9.9" + assert c.probes["ipv6"] == "2606:4700:4700::1111" + assert c.probe_interval == 30.0 + + +def test_local_address_on_this_host_never_raises(): + # Real UDP connect(): either an address string, None (no route) or _ProbeUnavailable. + for fam, target in (("ipv4", "1.1.1.1"), ("ipv6", "2606:4700:4700::1111")): + try: + r = mini._local_address(fam, target) + except mini._ProbeUnavailable: + continue + assert r is None or isinstance(r, str)