diff --git a/hbd/client/gateway.py b/hbd/client/gateway.py new file mode 100644 index 0000000..39cd24f --- /dev/null +++ b/hbd/client/gateway.py @@ -0,0 +1,103 @@ +"""Best-effort default-gateway lookup (informational only; never affects presence).""" + +import asyncio +import re +import socket +import struct +import sys +from typing import List, Optional, Tuple + +_RTF_GATEWAY = 0x2 +_RTF_REJECT = 0x200 + + +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: + continue + _iface, dest, gw, flags = f[0], f[1], f[2], f[3] # noqa: F841 + if dest != "00000000" or not int(flags, 16) & _RTF_GATEWAY: + continue + metric = int(f[6]) + if best is None or metric < best[0]: + best = (metric, socket.inet_ntoa(struct.pack(" 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 & _RTF_GATEWAY or fl & _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_macos_route(text: str) -> Optional[str]: + """Gateway from `route -n get [-inet6] default`.""" + 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 len(f) >= 5 and f[0] == "0.0.0.0" and f[1] == "0.0.0.0" and f[2] != "On-link": + cand = (int(f[4]), f[2]) + else: + continue + 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: List[str]) -> 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=5) + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + raise + return out.decode(errors="replace") + + +async def default_gateway(family: str) -> Optional[str]: + """Return the default gateway address for "ipv4"/"ipv6", or None on any failure.""" + 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 == "darwin": + cmd = ["route", "-n", "get"] + ([] if v4 else ["-inet6"]) + ["default"] + return parse_macos_route(await _run(cmd)) + if sys.platform == "win32": + return parse_windows_route(await _run(["route", "print", "-4" if v4 else "-6"]), family) + except (OSError, asyncio.TimeoutError, ValueError): + return None + return None diff --git a/tests/test_gateway.py b/tests/test_gateway.py new file mode 100644 index 0000000..167d0a4 --- /dev/null +++ b/tests/test_gateway.py @@ -0,0 +1,64 @@ +from hbd.client import gateway + +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" +) + +MACOS = """ route to: default +destination: default + mask: default + gateway: 192.168.1.1 + interface: en0 +""" + +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 +""" + + +def test_proc_v4_picks_lowest_metric_gateway(): + assert gateway.parse_proc_route_v4(PROC_V4) == "192.168.1.1" # eth0, metric 100 + + +def test_proc_v4_none_without_default_route(): + assert gateway.parse_proc_route_v4(PROC_V4.splitlines()[0] + "\n") is None + + +def test_proc_v6_skips_reject_and_non_default(): + assert gateway.parse_proc_route_v6(PROC_V6) == "fe80::1" + + +def test_macos(): + assert gateway.parse_macos_route(MACOS) == "192.168.1.1" + assert gateway.parse_macos_route("route: writing to routing socket: not in table") is None + + +def test_windows_v4_lowest_metric(): + assert gateway.parse_windows_route(WIN_V4, "ipv4") == "192.168.1.1" + + +def test_windows_v6(): + assert gateway.parse_windows_route(WIN_V6, "ipv6") == "fe80::1"