"""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 Exception: return None return None