andreasandClaude Sonnet 5
ad19d2596a
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Uc9V1v7iXruubpoLiSE24M
104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
"""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 Exception:
|
|
return None
|
|
return None
|