xenia-pos-local/local_backend/services/netinfo_helper.py

"""
Host LAN-IP detector — runs as its own container with `network_mode: host`.

The backend lives in a Docker bridge network and can only see its container
address, never the machine's real network cards. This helper shares the host's
network namespace, finds the server's address on the PHYSICAL network
(Ethernet/WiFi — never a WireGuard/Tailscale/ZeroTier tunnel, Docker bridge or
veth) and writes it to a small JSON file on a volume the backend reads
(services/lan_ip.py). It refreshes every minute, so a DHCP change shows up
without anyone re-running install.sh.

Selection (same rules as install.sh's detect_host_ip):
  1. the interface of the main-table default route, if it is real hardware
     (real NICs have /sys/class/net/<if>/device; tunnels and bridges don't);
  2. otherwise the first real-hardware interface that has an IPv4 address.

On Docker Desktop (Windows/Mac dev machines) "the host network" is Docker's own
Linux VM, whose address means nothing to phones — the helper detects that and
reports no IP instead of a wrong one.

Standalone on purpose: standard library only, no app imports, no database.
Run: python -m services.netinfo_helper
"""
import fcntl
import json
import os
import platform
import socket
import struct
import time
from datetime import datetime, timezone

OUT_FILE = os.environ.get("NETINFO_FILE", "/netinfo/host_ip.json")
INTERVAL_SECONDS = int(os.environ.get("NETINFO_INTERVAL", "60"))
SIOCGIFADDR = 0x8915


def ipv4_of(ifname: str) -> str | None:
    """Primary IPv4 address of an interface, or None if it has none."""
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
        try:
            packed = fcntl.ioctl(s.fileno(), SIOCGIFADDR, struct.pack("256s", ifname[:15].encode()))
        except OSError:
            return None
    return socket.inet_ntoa(packed[20:24])


def default_route_iface(route_file: str = "/proc/net/route") -> str | None:
    """Interface of the main routing table's default route (lowest metric)."""
    best = None
    try:
        with open(route_file) as f:
            next(f)  # header
            for line in f:
                cols = line.split()
                if len(cols) < 7:
                    continue
                iface, dest, flags, metric = cols[0], cols[1], int(cols[3], 16), int(cols[6])
                if dest == "00000000" and flags & 0x1:  # default route, RTF_UP
                    if best is None or metric < best[1]:
                        best = (iface, metric)
    except OSError:
        return None
    return best[0] if best else None


def detect(sys_net: str = "/sys/class/net", route_file: str = "/proc/net/route", ipv4=ipv4_of) -> dict:
    """Pick the physical LAN address. Pure apart from the injected lookups (testable)."""
    try:
        names = sorted(os.listdir(sys_net))
    except OSError:
        names = []
    candidates = []
    for name in names:
        if not os.path.exists(os.path.join(sys_net, name, "device")):
            continue  # tunnel, bridge, veth, loopback
        addr = ipv4(name)
        if addr and not addr.startswith("169.254."):
            candidates.append({"iface": name, "ip": addr})

    default_if = default_route_iface(route_file)
    chosen = next((c for c in candidates if c["iface"] == default_if), None)
    if chosen is None and candidates:
        chosen = candidates[0]
    return {
        "ip": chosen["ip"] if chosen else None,
        "iface": chosen["iface"] if chosen else None,
        "candidates": candidates,
        "default_iface": default_if,
    }


def docker_desktop() -> bool:
    """Docker Desktop's VM kernel identifies itself; its 'host' network isn't the LAN."""
    release = platform.release().lower()
    return "linuxkit" in release or "microsoft" in release


def write_atomic(path: str, data: dict) -> None:
    os.makedirs(os.path.dirname(path), exist_ok=True)
    tmp = f"{path}.tmp"
    with open(tmp, "w") as f:
        json.dump(data, f)
    os.replace(tmp, path)


def run_once() -> dict:
    if docker_desktop():
        result = {"ip": None, "iface": None, "candidates": [], "default_iface": None,
                  "unsupported": "docker-desktop"}
    else:
        result = detect()
    result["detected_at"] = datetime.now(timezone.utc).isoformat()
    write_atomic(OUT_FILE, result)
    return result


def main() -> None:
    last_ip = object()
    while True:
        try:
            result = run_once()
            if result["ip"] != last_ip:
                print(f"netinfo: LAN IP {result['ip']} via {result['iface']} "
                      f"(candidates: {result['candidates']}{', ' + result['unsupported'] if result.get('unsupported') else ''})",
                      flush=True)
                last_ip = result["ip"]
        except Exception as e:  # never die — the backend treats a stale file as "unknown"
            print(f"netinfo: detection failed: {e}", flush=True)
        time.sleep(INTERVAL_SECONDS)


if __name__ == "__main__":
    main()