112 lines
3.1 KiB
Python
112 lines
3.1 KiB
Python
"""LAN IP detection and WESP_LISTEN parsing (shared by kiosk and network settings)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import socket
|
|
import time
|
|
from ipaddress import ip_address
|
|
from typing import Tuple
|
|
|
|
DEFAULT_WESP_HTTP_PORT = 80
|
|
DEFAULT_WESP_LISTEN = f"0.0.0.0:{DEFAULT_WESP_HTTP_PORT}"
|
|
|
|
_LAN_IP_CACHE: tuple[float, str] | None = None
|
|
_LAN_IP_CACHE_TTL_SEC = 15.0
|
|
_LAN_PROBE_TIMEOUT_SEC = 0.35
|
|
|
|
|
|
def host_priority(ip: str) -> int:
|
|
if ip.startswith("192.168."):
|
|
return 0
|
|
if ip.startswith("10."):
|
|
return 1
|
|
if ip.startswith("172."):
|
|
parts = ip.split(".")
|
|
if len(parts) == 4:
|
|
try:
|
|
second = int(parts[1])
|
|
if 16 <= second <= 31:
|
|
return 2
|
|
except ValueError:
|
|
return 99
|
|
return 99
|
|
|
|
|
|
def is_valid_candidate_ip(value: str) -> bool:
|
|
ip = (value or "").strip()
|
|
if not ip:
|
|
return False
|
|
try:
|
|
parsed = ip_address(ip)
|
|
except ValueError:
|
|
return False
|
|
if parsed.version != 4:
|
|
return False
|
|
if parsed.is_loopback or parsed.is_link_local:
|
|
return False
|
|
return host_priority(ip) < 99
|
|
|
|
|
|
def detect_lan_ip(*, use_cache: bool = True) -> str:
|
|
global _LAN_IP_CACHE
|
|
if use_cache and _LAN_IP_CACHE is not None:
|
|
ts, cached = _LAN_IP_CACHE
|
|
if cached and (time.time() - ts) < _LAN_IP_CACHE_TTL_SEC:
|
|
return cached
|
|
|
|
candidates: set[str] = set()
|
|
|
|
def _add_candidate(ip: str) -> None:
|
|
if is_valid_candidate_ip(ip):
|
|
candidates.add(ip)
|
|
|
|
probe_targets = [
|
|
("192.168.255.255", 1),
|
|
("10.255.255.255", 1),
|
|
("172.31.255.255", 1),
|
|
("8.8.8.8", 80),
|
|
]
|
|
for target_host, target_port in probe_targets:
|
|
try:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
|
sock.settimeout(_LAN_PROBE_TIMEOUT_SEC)
|
|
sock.connect((target_host, target_port))
|
|
_add_candidate(sock.getsockname()[0] or "")
|
|
except OSError:
|
|
continue
|
|
|
|
if not candidates:
|
|
host = socket.gethostname()
|
|
try:
|
|
for family, _, _, _, sockaddr in socket.getaddrinfo(host, None, socket.AF_INET):
|
|
if family != socket.AF_INET:
|
|
continue
|
|
ip = (sockaddr[0] or "").strip()
|
|
_add_candidate(ip)
|
|
except OSError:
|
|
pass
|
|
|
|
try:
|
|
for ip in socket.gethostbyname_ex(host)[2]:
|
|
_add_candidate(ip)
|
|
except OSError:
|
|
pass
|
|
|
|
prioritized = sorted(candidates, key=host_priority)
|
|
result = prioritized[0] if prioritized else ""
|
|
if use_cache and result:
|
|
_LAN_IP_CACHE = (time.time(), result)
|
|
return result
|
|
|
|
|
|
def parse_wesp_listen(listen: str | None = None) -> Tuple[str, int]:
|
|
raw = (listen or os.getenv("WESP_LISTEN", DEFAULT_WESP_LISTEN)).strip()
|
|
if ":" in raw:
|
|
host, port_s = raw.rsplit(":", 1)
|
|
try:
|
|
return (host or "0.0.0.0", int(port_s))
|
|
except ValueError:
|
|
pass
|
|
return ("0.0.0.0", DEFAULT_WESP_HTTP_PORT)
|