593 lines
23 KiB
Python
593 lines
23 KiB
Python
"""
|
|
Find the LAN address this machine is actually reachable on, live.
|
|
|
|
The dashboard publishes `http://<ip>:8770`, so picking the wrong address means
|
|
handing someone a link that cannot work. That is easy to get wrong: this very
|
|
machine has a *disconnected* Ethernet adapter still holding 192.168.123.222, and
|
|
a local HTTP fetch to that address SUCCEEDS - Windows routes traffic to any of
|
|
its own addresses through loopback, skipping the adapter entirely. It even
|
|
answers faster than the real one. There is therefore no connect, bind or fetch
|
|
test that can tell a reachable address from a dead one; the decision has to come
|
|
from adapter metadata.
|
|
|
|
The rule below was chosen by running candidate rules against 23 synthetic
|
|
adapter tables (VPNs of three different shapes, Hyper-V bridges, mobile hotspot,
|
|
docked Ethernet, duplicate-IP detection, campus public addressing, renamed
|
|
adapters). Two obvious-looking rules broke 9 times each; this one breaks once,
|
|
on a case that is genuinely undecidable (two Wi-Fi radios, nothing in the
|
|
adapter table says which one the phone is associated to).
|
|
|
|
Deliberate non-decisions, each of which was tried and rejected:
|
|
|
|
* No `is_private` test. It over-rejects campus and CGNAT (100.64/10) addresses
|
|
that phones reach perfectly well, and under-rejects Hyper-V's 172.x.
|
|
* No adapter *name* matching. "Wi-Fi" is user-renameable and every ipconfig
|
|
label is translated on non-English Windows.
|
|
* Interface metric is the LAST tiebreak, never a gate. Corporate VPNs
|
|
deliberately set a low metric, and on this machine four *disconnected*
|
|
adapters sit at metric 25 while the one working NIC is at 30.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import json
|
|
import logging
|
|
import os
|
|
import platform
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
|
|
log = logging.getLogger("x2.nic")
|
|
|
|
# Re-read this often at most. At ~2 ms a detection the cost is irrelevant; the
|
|
# TTL exists so two requests in one page load cannot disagree mid-roam, and so a
|
|
# stale address never outlives a single human retry.
|
|
TTL_SECONDS = 2.0
|
|
|
|
IF_TYPE_ETHERNET = 6
|
|
IF_TYPE_LOOPBACK = 24
|
|
IF_TYPE_WIFI = 71
|
|
IF_TYPE_TUNNEL = 131
|
|
|
|
OPER_STATUS_UP = 1
|
|
DAD_STATE_PREFERRED = 4
|
|
PREFIX_ORIGIN_WELLKNOWN = 2
|
|
PREFIX_ORIGIN_DHCP = 3
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Ranking - platform independent, operating on normalised records
|
|
# --------------------------------------------------------------------------
|
|
|
|
def _class_of(record: dict) -> int:
|
|
"""Higher is a better candidate to hand to a phone."""
|
|
hardware = record.get("hardware")
|
|
iftype = record.get("iftype")
|
|
if hardware and iftype == IF_TYPE_WIFI:
|
|
return 3 # a real Wi-Fi radio
|
|
if hardware and iftype == IF_TYPE_ETHERNET:
|
|
return 2 # a real wired NIC
|
|
if iftype == IF_TYPE_WIFI:
|
|
return 1 # hotspot / Wi-Fi Direct - still a radio
|
|
return 0 # bridge or unknown
|
|
|
|
|
|
def _rejected(record: dict) -> str | None:
|
|
"""Why this address must not be published, or None if it is usable."""
|
|
iftype = record.get("iftype")
|
|
if iftype in (IF_TYPE_LOOPBACK, IF_TYPE_TUNNEL):
|
|
return "loopback or tunnel"
|
|
if record.get("oper_status") != OPER_STATUS_UP:
|
|
return "adapter is down"
|
|
if record.get("connected") is False:
|
|
return "no link"
|
|
if record.get("dad_state") not in (None, DAD_STATE_PREFERRED):
|
|
# Deprecated = the stale address on a disconnected NIC.
|
|
# Tentative = still running duplicate-address detection.
|
|
# Duplicate = the stack itself refuses to send from it.
|
|
return "address not preferred"
|
|
if record.get("prefix_origin") == PREFIX_ORIGIN_WELLKNOWN:
|
|
return "self-assigned (APIPA)"
|
|
|
|
try:
|
|
ip = ipaddress.IPv4Address(record["ip"])
|
|
except (ipaddress.AddressValueError, KeyError):
|
|
return "not an IPv4 address"
|
|
if ip.is_loopback or ip.is_link_local:
|
|
return "loopback or link-local"
|
|
|
|
# Virtual adapters (VPN NDIS miniports, Hyper-V, WSL, docker) are not a path
|
|
# to a phone. Two exceptions: any Wi-Fi-class radio, which covers the mobile
|
|
# hotspot case; and a bridge that has taken over a live physical NIC's
|
|
# address, which is what a Hyper-V *external* switch does.
|
|
if not record.get("hardware", True):
|
|
is_radio = iftype == IF_TYPE_WIFI
|
|
is_external_bridge = (
|
|
record.get("physical_link_present")
|
|
and record.get("has_gateway")
|
|
and record.get("prefix_origin") == PREFIX_ORIGIN_DHCP
|
|
)
|
|
if not (is_radio or is_external_bridge):
|
|
return "virtual adapter"
|
|
return None
|
|
|
|
|
|
def _sort_key(record: dict):
|
|
try:
|
|
numeric = int(ipaddress.IPv4Address(record["ip"]))
|
|
except Exception:
|
|
numeric = 0
|
|
return (
|
|
1 if record.get("has_gateway") else 0,
|
|
_class_of(record),
|
|
1 if record.get("prefix_origin") == PREFIX_ORIGIN_DHCP else 0,
|
|
-(record.get("metric") or 0),
|
|
-numeric,
|
|
)
|
|
|
|
|
|
def rank(records: list[dict]) -> tuple[list[dict], list[dict]]:
|
|
"""Split candidates into (usable, best first) and (rejected, with reasons)."""
|
|
usable, rejected = [], []
|
|
for record in records:
|
|
why = _rejected(record)
|
|
if why:
|
|
rejected.append({**record, "rejected": why})
|
|
else:
|
|
usable.append(record)
|
|
usable.sort(key=_sort_key, reverse=True)
|
|
return usable, rejected
|
|
|
|
|
|
def _reason_for_nothing(rejected: list[dict]) -> str:
|
|
"""Explain an empty result honestly instead of guessing an address."""
|
|
if not rejected:
|
|
return "no_network_interface"
|
|
kinds = {r.get("rejected") for r in rejected}
|
|
wifi = [r for r in rejected if r.get("iftype") == IF_TYPE_WIFI and r.get("hardware")]
|
|
if wifi and all(r.get("oper_status") != OPER_STATUS_UP or r.get("connected") is False
|
|
for r in wifi):
|
|
return "wifi_down"
|
|
if "address not preferred" in kinds or "self-assigned (APIPA)" in kinds:
|
|
return "awaiting_dhcp"
|
|
if all(not r.get("has_gateway") for r in rejected):
|
|
return "no_router_on_this_network"
|
|
return "no_usable_address"
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Windows collector - GetAdaptersAddresses + GetIfTable2 via ctypes
|
|
# --------------------------------------------------------------------------
|
|
|
|
def _collect_windows() -> list[dict]:
|
|
import ctypes
|
|
from ctypes import wintypes
|
|
|
|
iphlpapi = ctypes.WinDLL("iphlpapi")
|
|
|
|
class SOCKET_ADDRESS(ctypes.Structure):
|
|
_fields_ = [("lpSockaddr", ctypes.c_void_p), ("iSockaddrLength", ctypes.c_int)]
|
|
|
|
class IP_ADAPTER_UNICAST_ADDRESS(ctypes.Structure):
|
|
pass
|
|
|
|
IP_ADAPTER_UNICAST_ADDRESS._fields_ = [
|
|
("Length", wintypes.ULONG),
|
|
("Flags", wintypes.DWORD),
|
|
("Next", ctypes.POINTER(IP_ADAPTER_UNICAST_ADDRESS)),
|
|
("Address", SOCKET_ADDRESS),
|
|
("PrefixOrigin", ctypes.c_int),
|
|
("SuffixOrigin", ctypes.c_int),
|
|
("DadState", ctypes.c_int),
|
|
("ValidLifetime", wintypes.ULONG),
|
|
("PreferredLifetime", wintypes.ULONG),
|
|
("LeaseLifetime", wintypes.ULONG),
|
|
("OnLinkPrefixLength", ctypes.c_ubyte),
|
|
]
|
|
|
|
class IP_ADAPTER_GATEWAY_ADDRESS(ctypes.Structure):
|
|
pass
|
|
|
|
IP_ADAPTER_GATEWAY_ADDRESS._fields_ = [
|
|
("Length", wintypes.ULONG),
|
|
("Reserved", wintypes.DWORD),
|
|
("Next", ctypes.POINTER(IP_ADAPTER_GATEWAY_ADDRESS)),
|
|
("Address", SOCKET_ADDRESS),
|
|
]
|
|
|
|
class IP_ADAPTER_ADDRESSES(ctypes.Structure):
|
|
pass
|
|
|
|
IP_ADAPTER_ADDRESSES._fields_ = [
|
|
("Length", wintypes.ULONG),
|
|
("IfIndex", wintypes.DWORD),
|
|
("Next", ctypes.POINTER(IP_ADAPTER_ADDRESSES)),
|
|
("AdapterName", ctypes.c_char_p),
|
|
("FirstUnicastAddress", ctypes.POINTER(IP_ADAPTER_UNICAST_ADDRESS)),
|
|
("FirstAnycastAddress", ctypes.c_void_p),
|
|
("FirstMulticastAddress", ctypes.c_void_p),
|
|
("FirstDnsServerAddress", ctypes.c_void_p),
|
|
("DnsSuffix", ctypes.c_wchar_p),
|
|
("Description", ctypes.c_wchar_p),
|
|
("FriendlyName", ctypes.c_wchar_p),
|
|
("PhysicalAddress", ctypes.c_ubyte * 8),
|
|
("PhysicalAddressLength", wintypes.ULONG),
|
|
("Flags", wintypes.ULONG),
|
|
("Mtu", wintypes.ULONG),
|
|
("IfType", wintypes.ULONG),
|
|
("OperStatus", ctypes.c_int),
|
|
("Ipv6IfIndex", wintypes.DWORD),
|
|
("ZoneIndices", wintypes.ULONG * 16),
|
|
("FirstPrefix", ctypes.c_void_p),
|
|
("TransmitLinkSpeed", ctypes.c_uint64),
|
|
("ReceiveLinkSpeed", ctypes.c_uint64),
|
|
("FirstWinsServerAddress", ctypes.c_void_p),
|
|
("FirstGatewayAddress", ctypes.POINTER(IP_ADAPTER_GATEWAY_ADDRESS)),
|
|
("Ipv4Metric", wintypes.ULONG),
|
|
("Ipv6Metric", wintypes.ULONG),
|
|
("Luid", ctypes.c_uint64),
|
|
("Dhcpv4Server", SOCKET_ADDRESS),
|
|
("CompartmentId", wintypes.DWORD),
|
|
("NetworkGuid", ctypes.c_ubyte * 16),
|
|
("ConnectionType", ctypes.c_int),
|
|
("TunnelType", ctypes.c_int),
|
|
]
|
|
|
|
AF_INET = 2
|
|
GAA_FLAG_INCLUDE_GATEWAYS = 0x0080
|
|
GAA_FLAG_SKIP_ANYCAST = 0x0002
|
|
GAA_FLAG_SKIP_MULTICAST = 0x0004
|
|
GAA_FLAG_SKIP_DNS_SERVER = 0x0008
|
|
flags = (GAA_FLAG_INCLUDE_GATEWAYS | GAA_FLAG_SKIP_ANYCAST
|
|
| GAA_FLAG_SKIP_MULTICAST | GAA_FLAG_SKIP_DNS_SERVER)
|
|
|
|
def sockaddr_to_ip(sa: SOCKET_ADDRESS) -> str | None:
|
|
if not sa.lpSockaddr:
|
|
return None
|
|
raw = ctypes.string_at(sa.lpSockaddr, 8)
|
|
if int.from_bytes(raw[0:2], "little") != AF_INET:
|
|
return None
|
|
return ".".join(str(b) for b in raw[4:8])
|
|
|
|
# The buffer is allocated per call; sharing one across threads corrupts it.
|
|
size = wintypes.ULONG(15000)
|
|
buffer = ctypes.create_string_buffer(size.value)
|
|
result = iphlpapi.GetAdaptersAddresses(
|
|
AF_INET, flags, None, ctypes.byref(buffer), ctypes.byref(size))
|
|
if result == 111: # ERROR_BUFFER_OVERFLOW
|
|
buffer = ctypes.create_string_buffer(size.value)
|
|
result = iphlpapi.GetAdaptersAddresses(
|
|
AF_INET, flags, None, ctypes.byref(buffer), ctypes.byref(size))
|
|
if result != 0:
|
|
raise OSError(f"GetAdaptersAddresses failed: {result}")
|
|
|
|
# First pass: collect IfType per interface so the GetIfTable2 parse below can
|
|
# be validated against a second, independent source.
|
|
known_types: dict[int, int] = {}
|
|
node = ctypes.cast(buffer, ctypes.POINTER(IP_ADAPTER_ADDRESSES))
|
|
while node:
|
|
known_types[int(node.contents.IfIndex)] = int(node.contents.IfType)
|
|
node = node.contents.Next
|
|
|
|
hardware, connected, physical_live = _windows_interface_flags(known_types)
|
|
|
|
records: list[dict] = []
|
|
node = ctypes.cast(buffer, ctypes.POINTER(IP_ADAPTER_ADDRESSES))
|
|
while node:
|
|
adapter = node.contents
|
|
gateways = []
|
|
gw = adapter.FirstGatewayAddress
|
|
while gw:
|
|
ip = sockaddr_to_ip(gw.contents.Address)
|
|
if ip:
|
|
gateways.append(ip)
|
|
gw = gw.contents.Next
|
|
|
|
unicast = adapter.FirstUnicastAddress
|
|
while unicast:
|
|
entry = unicast.contents
|
|
ip = sockaddr_to_ip(entry.Address)
|
|
if ip:
|
|
index = int(adapter.IfIndex)
|
|
records.append({
|
|
"ip": ip,
|
|
"prefix_length": int(entry.OnLinkPrefixLength),
|
|
"adapter": adapter.FriendlyName or "",
|
|
"description": adapter.Description or "",
|
|
"if_index": index,
|
|
"iftype": int(adapter.IfType),
|
|
"oper_status": int(adapter.OperStatus),
|
|
"dad_state": int(entry.DadState),
|
|
"prefix_origin": int(entry.PrefixOrigin),
|
|
"metric": int(adapter.Ipv4Metric),
|
|
"has_gateway": bool(gateways),
|
|
"gateway": gateways[0] if gateways else None,
|
|
"hardware": hardware.get(index),
|
|
"connected": connected.get(index),
|
|
"physical_link_present": physical_live,
|
|
})
|
|
unicast = entry.Next
|
|
node = adapter.Next
|
|
|
|
return records
|
|
|
|
|
|
def _windows_interface_flags(known_types: dict | None = None) -> tuple[dict, dict, bool]:
|
|
"""
|
|
Per-interface HardwareInterface / MediaConnectState from GetIfTable2.
|
|
|
|
These two gates cannot be derived from GetAdaptersAddresses, and they are
|
|
what separates a real NIC from a VPN miniport or a Hyper-V bridge without
|
|
matching English adapter names.
|
|
|
|
MIB_IF_ROW2 must be modelled COMPLETELY, trailing statistics counters and
|
|
all: the rows are read as a contiguous array, so a struct even one field
|
|
short gives the wrong stride and every row after the first is garbage -
|
|
silently, with no error from the API. `known_types` therefore carries the
|
|
IfType values already read from GetAdaptersAddresses so the parse can be
|
|
checked against them; on disagreement we discard the whole table and let the
|
|
ranking fall back to `hardware: None` (unknown), which is treated as "do not
|
|
reject", never as "virtual".
|
|
"""
|
|
import ctypes
|
|
from ctypes import wintypes
|
|
|
|
try:
|
|
iphlpapi = ctypes.WinDLL("iphlpapi")
|
|
|
|
class MIB_IF_ROW2(ctypes.Structure):
|
|
_fields_ = [
|
|
("InterfaceLuid", ctypes.c_uint64),
|
|
("InterfaceIndex", wintypes.DWORD),
|
|
("InterfaceGuid", ctypes.c_ubyte * 16),
|
|
("Alias", ctypes.c_wchar * 257),
|
|
("Description", ctypes.c_wchar * 257),
|
|
("PhysicalAddressLength", wintypes.ULONG),
|
|
("PhysicalAddress", ctypes.c_ubyte * 32),
|
|
("PermanentPhysicalAddress", ctypes.c_ubyte * 32),
|
|
("Mtu", wintypes.ULONG),
|
|
("Type", wintypes.ULONG),
|
|
("TunnelType", ctypes.c_int),
|
|
("MediaType", ctypes.c_int),
|
|
("PhysicalMediumType", ctypes.c_int),
|
|
("AccessType", ctypes.c_int),
|
|
("DirectionType", ctypes.c_int),
|
|
("InterfaceAndOperStatusFlags", ctypes.c_ubyte),
|
|
("OperStatus", ctypes.c_int),
|
|
("AdminStatus", ctypes.c_int),
|
|
("MediaConnectState", ctypes.c_int),
|
|
("NetworkGuid", ctypes.c_ubyte * 16),
|
|
("ConnectionType", ctypes.c_int),
|
|
("TransmitLinkSpeed", ctypes.c_uint64),
|
|
("ReceiveLinkSpeed", ctypes.c_uint64),
|
|
# Everything below is unused, but its size decides the stride.
|
|
("InOctets", ctypes.c_uint64),
|
|
("InUcastPkts", ctypes.c_uint64),
|
|
("InNUcastPkts", ctypes.c_uint64),
|
|
("InDiscards", ctypes.c_uint64),
|
|
("InErrors", ctypes.c_uint64),
|
|
("InUnknownProtos", ctypes.c_uint64),
|
|
("InUcastOctets", ctypes.c_uint64),
|
|
("InMulticastOctets", ctypes.c_uint64),
|
|
("InBroadcastOctets", ctypes.c_uint64),
|
|
("OutOctets", ctypes.c_uint64),
|
|
("OutUcastPkts", ctypes.c_uint64),
|
|
("OutNUcastPkts", ctypes.c_uint64),
|
|
("OutDiscards", ctypes.c_uint64),
|
|
("OutErrors", ctypes.c_uint64),
|
|
("OutUcastOctets", ctypes.c_uint64),
|
|
("OutMulticastOctets", ctypes.c_uint64),
|
|
("OutBroadcastOctets", ctypes.c_uint64),
|
|
("OutQLen", ctypes.c_uint64),
|
|
]
|
|
|
|
class MIB_IF_TABLE2(ctypes.Structure):
|
|
# NumEntries then the rows - there is no reserved field here. Adding
|
|
# one shifts Table by 8 bytes and every row parses as garbage.
|
|
_fields_ = [("NumEntries", wintypes.ULONG),
|
|
("Table", MIB_IF_ROW2 * 1)]
|
|
|
|
iphlpapi.GetIfTable2.argtypes = [ctypes.POINTER(ctypes.POINTER(MIB_IF_TABLE2))]
|
|
iphlpapi.GetIfTable2.restype = ctypes.c_ulong
|
|
|
|
table_ptr = ctypes.POINTER(MIB_IF_TABLE2)()
|
|
status = iphlpapi.GetIfTable2(ctypes.byref(table_ptr))
|
|
if status != 0 or not table_ptr:
|
|
log.debug("GetIfTable2 returned %s", status)
|
|
return {}, {}, False
|
|
|
|
try:
|
|
count = int(table_ptr.contents.NumEntries)
|
|
if not 0 < count < 4096:
|
|
log.debug("GetIfTable2 gave an implausible row count: %s", count)
|
|
return {}, {}, False
|
|
|
|
rows = ctypes.cast(
|
|
ctypes.byref(table_ptr.contents.Table),
|
|
ctypes.POINTER(MIB_IF_ROW2 * count)).contents
|
|
|
|
hardware, connected, types = {}, {}, {}
|
|
physical_live = False
|
|
for row in rows:
|
|
index = int(row.InterfaceIndex)
|
|
if not 0 < index < 10_000_000:
|
|
log.debug("GetIfTable2 row has a bogus index (%s) - bad layout", index)
|
|
return {}, {}, False
|
|
media = int(row.MediaConnectState)
|
|
is_hardware = bool(row.InterfaceAndOperStatusFlags & 0x01)
|
|
is_connected = media == 1
|
|
hardware[index] = is_hardware
|
|
connected[index] = is_connected if media in (1, 2) else None
|
|
types[index] = int(row.Type)
|
|
if is_hardware and is_connected and int(row.Type) in (
|
|
IF_TYPE_ETHERNET, IF_TYPE_WIFI):
|
|
physical_live = True
|
|
|
|
# Cross-check against the types GetAdaptersAddresses already gave
|
|
# us. If the two disagree, this parse is not trustworthy.
|
|
for index, iftype in (known_types or {}).items():
|
|
if index in types and types[index] != iftype:
|
|
log.warning("GetIfTable2 disagrees with GetAdaptersAddresses on if%s "
|
|
"(%s vs %s) - ignoring hardware flags",
|
|
index, types[index], iftype)
|
|
return {}, {}, False
|
|
|
|
return hardware, connected, physical_live
|
|
finally:
|
|
iphlpapi.FreeMibTable(table_ptr)
|
|
except Exception as exc:
|
|
log.debug("GetIfTable2 unavailable (%s); ranking without hardware flags", exc)
|
|
return {}, {}, False
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Linux collector - `ip -j addr` enriched from sysfs
|
|
# --------------------------------------------------------------------------
|
|
|
|
def _sysfs(interface: str, leaf: str) -> str | None:
|
|
try:
|
|
with open(f"/sys/class/net/{interface}/{leaf}", "r") as handle:
|
|
return handle.read().strip()
|
|
except OSError:
|
|
return None
|
|
|
|
|
|
def _linux_gateways() -> set[str]:
|
|
"""Interfaces owning a default route, from /proc/net/route."""
|
|
owners: set[str] = set()
|
|
try:
|
|
with open("/proc/net/route", "r") as handle:
|
|
next(handle, None)
|
|
for line in handle:
|
|
parts = line.split()
|
|
if len(parts) > 2 and parts[1] == "00000000":
|
|
owners.add(parts[0])
|
|
except OSError:
|
|
pass
|
|
return owners
|
|
|
|
|
|
def _collect_linux() -> list[dict]:
|
|
try:
|
|
raw = subprocess.run(["ip", "-j", "-4", "addr", "show"],
|
|
capture_output=True, text=True, timeout=5).stdout
|
|
interfaces = json.loads(raw or "[]")
|
|
except (OSError, subprocess.SubprocessError, ValueError):
|
|
return []
|
|
|
|
gateways = _linux_gateways()
|
|
physical_live = False
|
|
records: list[dict] = []
|
|
|
|
for iface in interfaces:
|
|
name = iface.get("ifname", "")
|
|
# The `device` symlink exists only when a real PCI/USB/platform device
|
|
# backs the interface - absent for docker0, veth*, tun*, br-*, lo. It is
|
|
# the structural equivalent of Windows' HardwareInterface.
|
|
is_hardware = os.path.exists(f"/sys/class/net/{name}/device")
|
|
is_wireless = os.path.isdir(f"/sys/class/net/{name}/phy80211") or \
|
|
os.path.isdir(f"/sys/class/net/{name}/wireless")
|
|
carrier = _sysfs(name, "carrier")
|
|
operstate = _sysfs(name, "operstate")
|
|
up = (operstate == "up") or ("UP" in (iface.get("flags") or []))
|
|
|
|
iftype = IF_TYPE_WIFI if is_wireless else (
|
|
IF_TYPE_LOOPBACK if name == "lo" else IF_TYPE_ETHERNET)
|
|
if is_hardware and carrier == "1" and iftype in (IF_TYPE_ETHERNET, IF_TYPE_WIFI):
|
|
physical_live = True
|
|
|
|
for addr in iface.get("addr_info", []):
|
|
if addr.get("family") != "inet":
|
|
continue
|
|
records.append({
|
|
"ip": addr.get("local"),
|
|
"prefix_length": addr.get("prefixlen"),
|
|
"adapter": name,
|
|
"description": name,
|
|
"if_index": iface.get("ifindex"),
|
|
"iftype": iftype,
|
|
"oper_status": OPER_STATUS_UP if up else 2,
|
|
# `scope: global` is the practical stand-in for Preferred; it
|
|
# also drops link-local for free.
|
|
"dad_state": DAD_STATE_PREFERRED if addr.get("scope") == "global" else 1,
|
|
"prefix_origin": PREFIX_ORIGIN_DHCP if addr.get("dynamic") else 1,
|
|
"metric": 0,
|
|
"has_gateway": name in gateways,
|
|
"gateway": None,
|
|
"hardware": is_hardware,
|
|
"connected": None if carrier is None else carrier == "1",
|
|
"physical_link_present": physical_live,
|
|
})
|
|
|
|
for record in records:
|
|
record["physical_link_present"] = physical_live
|
|
return records
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Public interface
|
|
# --------------------------------------------------------------------------
|
|
|
|
_lock = threading.Lock()
|
|
_cache: dict | None = None
|
|
_cache_at = 0.0
|
|
|
|
|
|
def _collect() -> list[dict]:
|
|
if platform.system() == "Windows":
|
|
return _collect_windows()
|
|
return _collect_linux()
|
|
|
|
|
|
def detect(force: bool = False) -> dict:
|
|
"""
|
|
The current LAN address, re-read live.
|
|
|
|
Returns {primary, alternates, rejected, reason, checked_at}. `primary` is
|
|
None when there is genuinely nothing usable - the caller must show that
|
|
honestly rather than print a stale or invented URL.
|
|
"""
|
|
global _cache, _cache_at
|
|
|
|
with _lock:
|
|
# monotonic, never time.time(): the wall clock jumps on NTP sync and on
|
|
# resume from sleep, which would freeze or expire the cache wrongly.
|
|
now = time.monotonic()
|
|
if not force and _cache is not None and (now - _cache_at) < TTL_SECONDS:
|
|
return _cache
|
|
|
|
try:
|
|
records = _collect()
|
|
except Exception as exc:
|
|
log.warning("address detection failed: %s", exc)
|
|
records = []
|
|
|
|
usable, rejected = rank(records)
|
|
result = {
|
|
"primary": usable[0] if usable else None,
|
|
"alternates": usable[1:],
|
|
"rejected": rejected,
|
|
"reason": None if usable else _reason_for_nothing(rejected),
|
|
"checked_at": time.time(),
|
|
}
|
|
_cache, _cache_at = result, now
|
|
return result
|
|
|
|
|
|
def primary_address() -> str | None:
|
|
"""Just the address, for callers that do not care why."""
|
|
found = detect()
|
|
return found["primary"]["ip"] if found["primary"] else None
|
|
|
|
|
|
REASON_TEXT = {
|
|
"wifi_down": "Wi-Fi is off or not connected",
|
|
"awaiting_dhcp": "connected, still waiting for an address",
|
|
"no_router_on_this_network": "connected, but this network has no router",
|
|
"no_physical_adapter": "no network adapter found",
|
|
"no_network_interface": "no network interface found",
|
|
"no_usable_address": "no usable network address",
|
|
}
|