feat(client): best-effort default gateway lookup
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Uc9V1v7iXruubpoLiSE24M
This commit is contained in:
@@ -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("<L", int(gw, 16))))
|
||||
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 & _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
|
||||
Reference in New Issue
Block a user