feat(backend): runtime LAN-IP detection (netinfo helper) + manual override

The backend's bridge network can't see the host's NICs, so HOST_IP from
install.sh went stale silently after a DHCP change. Now:

- services/netinfo_helper.py runs as a new `netinfo` service (same backend
  image, network_mode: host): every 60s it picks the PHYSICAL LAN address
  (main-table default-route NIC if real hardware, else first real NIC with
  IPv4; never WireGuard/Tailscale/ZeroTier/bridges/veths, ignores 169.254)
  and writes it to the shared `netinfo` volume. Stdlib only. On Docker
  Desktop (linuxkit/WSL2 kernel) it reports "unsupported" instead of the
  VM's meaningless address.
- services/lan_ip.py: one resolver used by /api/system/status (lan_ip +
  lan_ip_info), the pairing QR and the cloud heartbeat's local_ip:
  override (pos_settings network.lan_ip_override) → live detection (ignored
  when older than 5 min) → HOST_IP. Flags `mismatch` when a pinned address
  is no longer on any of the machine's NICs.
- PUT /api/system/lan-ip-override (manager): set a private IPv4 or null to
  return to automatic; public/loopback/link-local/IPv6 rejected (422).
- cloud_sync._get_local_ip uses the resolver (no more socket trick that
  returned the container IP).

Tested: helper selection on a fake sysfs/route table (8 cases incl. VPN
default routes) + real ioctl/route parsing on a Linux kernel; resolver
priority/staleness/mismatch/validation (18 cases); isolated full stack:
HOST_IP fallback on Docker Desktop, override save/validate/auth, simulated
Linux detection incl. DHCP change and dead helper, heartbeat IP.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-09-28 16:13:17 +03:00
co-authored by Claude Opus 5.5
parent 034106918d
commit 76aac203d6
5 changed files with 281 additions and 18 deletions
+31 -8
View File
@@ -1,11 +1,11 @@
import asyncio
import ipaddress
import json
import os
import socket
import time
from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from sqlalchemy.orm import Session
from typing import List
@@ -19,6 +19,7 @@ from models.product import Category, Product
from models.table import Table, TableGroup
from services import printer_service
from services.cloud_sync import _sync_once, _push_menu_snapshot
from services.lan_ip import OVERRIDE_KEY, resolve_lan_ip, validate_lan_ip
from middleware.license_check import license_state
from config import settings
@@ -37,12 +38,6 @@ def health():
API_VERSION = 1
def _lan_ip() -> str | None:
"""The address phones should use. Only HOST_IP is trusted: inside Docker any
auto-detection returns the container's bridge IP, which phones can't reach."""
return os.environ.get("HOST_IP", "").strip() or None
@router.get("/identity")
def identity(db: Session = Depends(get_db)):
"""Public, unauthenticated. Lets a phone confirm which venue a server is
@@ -67,6 +62,8 @@ def system_status(db: Session = Depends(get_db), user: User = Depends(get_curren
reachable = printer_service.check_printer(p.ip_address, p.port)
printer_statuses.append({"id": p.id, "name": p.name, "reachable": reachable})
lan = resolve_lan_ip(db)
licensed = license_state.get("licensed", True)
locked = license_state.get("locked", False)
lock_pending = license_state.get("lock_pending", False)
@@ -111,11 +108,37 @@ def system_status(db: Session = Depends(get_db), user: User = Depends(get_curren
"last_sync": license_state.get("last_sync"),
"waiter_domain": license_state.get("waiter_domain"),
"site_id": settings.SITE_ID or None,
"lan_ip": _lan_ip(),
"lan_ip": lan.get("effective"),
"lan_ip_info": lan,
"printers": printer_statuses,
}
class LanIpOverride(BaseModel):
ip: str | None = None # null / empty → clear the override (back to automatic)
@router.put("/lan-ip-override")
def set_lan_ip_override(body: LanIpOverride, db: Session = Depends(get_db), user: User = Depends(require_manager)):
"""Pin the server's LAN IP (what phones and the pairing QR use). Clearing it
returns to automatic detection / HOST_IP."""
row = db.query(PosSettings).filter(PosSettings.key == OVERRIDE_KEY).first()
if body.ip and body.ip.strip():
try:
value = validate_lan_ip(body.ip)
except ValueError as e:
raise HTTPException(status_code=422, detail=str(e))
if row:
row.value = value
row.updated_by_id = user.id or None
else:
db.add(PosSettings(key=OVERRIDE_KEY, value=value, updated_by_id=user.id or None))
elif row:
db.delete(row)
db.commit()
return resolve_lan_ip(db)
@router.post("/sync-license")
async def sync_license_now(user: User = Depends(require_manager)):
"""Trigger an immediate cloud heartbeat and return the fresh license state."""
+9 -10
View File
@@ -16,7 +16,6 @@ import asyncio
import json
import logging
import os
import socket
from datetime import datetime, timedelta, timezone
from pathlib import Path
@@ -80,18 +79,18 @@ def _compute_expiry_fields(expires_at_str: str | None) -> dict:
def _get_local_ip() -> str | None:
"""Best-effort detection of the host machine's LAN IP address.
When running inside Docker the socket trick returns the container/bridge IP,
so we honour HOST_IP if it is explicitly provided via the environment."""
import os
if host_ip := os.environ.get("HOST_IP", "").strip():
return host_ip
"""The server's LAN IP as phones see it — same resolver as the manager's
pairing QR (override → live detection → HOST_IP), see services/lan_ip.py.
No socket tricks: inside Docker they return the unreachable bridge IP."""
from database import SessionLocal
from services.lan_ip import resolve_lan_ip
db = SessionLocal()
try:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
s.connect(("8.8.8.8", 80))
return s.getsockname()[0]
return resolve_lan_ip(db)["effective"]
except Exception:
return None
finally:
db.close()
async def _sync_once():
+91
View File
@@ -0,0 +1,91 @@
"""
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),
}
+134
View File
@@ -0,0 +1,134 @@
"""
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()