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

"""
The server's LAN address — the one phones use (http://<ip>), the pairing QR
encodes and the cloud heartbeat reports. One resolver, used everywhere.

Priority:
  1. override  — set once in the manager (pos_settings 'network.lan_ip_override')
  2. detected  — live, from the netinfo helper container (services/netinfo_helper.py)
  3. env       — HOST_IP from .env (written by install.sh)
  → None: the manager falls back to the address it was opened with.

When the address in use (override / env) differs from what the helper sees
right now, `mismatch` is set so the manager can warn before phones break
(typical cause: the router's DHCP handed the server a new address).
"""
import ipaddress
import json
import os
from datetime import datetime, timezone

OVERRIDE_KEY = "network.lan_ip_override"
NETINFO_FILE = os.environ.get("NETINFO_FILE", "/netinfo/host_ip.json")
DETECTION_MAX_AGE_SECONDS = 300  # helper writes every 60s; older means it stopped


def validate_lan_ip(value: str) -> str:
    """Normalise and check an override: a private, non-loopback IPv4 address."""
    try:
        addr = ipaddress.ip_address(value.strip())
    except ValueError:
        raise ValueError("Μη έγκυρη διεύθυνση IP (π.χ. 192.168.1.50)")
    if addr.version != 4 or not addr.is_private or addr.is_loopback or addr.is_link_local:
        raise ValueError("Η διεύθυνση πρέπει να είναι IPv4 τοπικού δικτύου (π.χ. 192.168.x.x ή 10.x.x.x)")
    return str(addr)


def read_detected(path: str = NETINFO_FILE, now: datetime | None = None) -> dict | None:
    """Latest helper result, or None if the helper isn't running / file is stale."""
    try:
        with open(path) as f:
            data = json.load(f)
        detected_at = datetime.fromisoformat(data["detected_at"])
    except (OSError, ValueError, KeyError, TypeError):
        return None
    now = now or datetime.now(timezone.utc)
    data["age_seconds"] = int((now - detected_at).total_seconds())
    data["stale"] = data["age_seconds"] > DETECTION_MAX_AGE_SECONDS
    return data


def resolve_lan_ip(db=None, override: str | None = None, detected: dict | None = None,
                   env: str | None = None) -> dict:
    """Effective LAN IP + where it came from. Pass db to read the override setting;
    the explicit arguments exist for tests."""
    if db is not None and override is None:
        from models.settings import PosSettings
        row = db.query(PosSettings).filter(PosSettings.key == OVERRIDE_KEY).first()
        override = row.value.strip() if row and row.value and row.value.strip() else None
    if detected is None:
        detected = read_detected()
    if env is None:
        env = os.environ.get("HOST_IP", "").strip() or None

    live = detected if detected and not detected.get("stale") and detected.get("ip") else None
    live_ip = live["ip"] if live else None
    live_all = {c["ip"] for c in (live or {}).get("candidates", [])} | ({live_ip} if live_ip else set())

    if override:
        effective, source = override, "override"
    elif live_ip:
        effective, source = live_ip, "detected"
    elif env:
        effective, source = env, "env"
    else:
        effective, source = None, None

    return {
        "effective": effective,
        "source": source,            # override | detected | env | None
        "override": override,
        "detected": live_ip,
        "detected_candidates": (live or {}).get("candidates", []),
        "detection": (
            "unsupported" if detected and detected.get("unsupported")
            else "stale" if detected and detected.get("stale")
            else "ok" if live_ip
            else "unavailable"
        ),
        "env": env,
        # Using a fixed address that this machine no longer has on any NIC
        "mismatch": bool(effective and source in ("override", "env") and live_ip and effective not in live_all),
    }