pxe-server/webapp/services/nic_status.py

"""Network interfaces for the Overview: link, addresses, role and verdict.

The server has two NICs: the imaging NIC (static 172.16.9.1, the only one
dnsmasq may serve DHCP/PXE on) and the GE NIC (corporate DHCP client). The
classifier pxe-nic-check (playbook/pxe-server-helpers) writes its verdicts
to config.NIC_STATUS_FILE; when that file is missing or stale this module
falls back to what the kernel says (link, speed, addresses) and which
interface dnsmasq is bound to, without a verdict.

Read-only: never changes interfaces or services.
"""
from __future__ import annotations

import configparser
import json
import os
import re
import subprocess
from datetime import datetime, timezone

import config

STALE_SECONDS = 20 * 60  # the watch timer runs every 5 min
SYS_NET = "/sys/class/net"
# cdc_ncm (the generic USB NIC driver, e.g. a WisdPI RTL8157 5G adapter on a
# kernel whose r8152 does not know the chip) keeps the link speed as a 32-bit
# bits-per-second value, so multi-gig links wrap: 5 Gb/s reads 705 Mb/s and
# 10 Gb/s reads 1410 Mb/s.
CDC_NCM_WRAPPED = {705: 5000, 1410: 10000}


def _read(path):
    try:
        with open(path) as f:
            return f.read().strip()
    except OSError:
        return ""


def _roles():
    """{iface or mac: role} from nic.conf, else from dnsmasq's NIC pin."""
    roles = {}
    cp = configparser.ConfigParser()
    try:
        cp.read_string("[nic]\n" + open(config.NIC_CONF).read())
        sec = cp["nic"]
        for role in ("imaging", "ge"):
            for key in (f"{role}_iface", f"{role}_mac"):
                if sec.get(key):
                    roles[sec.get(key).strip().lower()] = role
    except (OSError, configparser.Error):
        pass
    if not roles:
        for line in _read(config.DNSMASQ_NIC_CONF).splitlines():
            key, _, val = line.split("#", 1)[0].strip().partition("=")
            if key == "interface" and val:
                roles[val.strip()] = "imaging"
            elif key == "except-interface" and val:
                roles.setdefault(val.strip(), "ge")
    return roles


def _addrs():
    """{iface: ["a.b.c.d/nn", ...]} (IPv4 and global IPv6) from `ip -j addr`."""
    try:
        out = subprocess.run(["ip", "-j", "addr", "show"], capture_output=True, text=True, timeout=5).stdout
        data = json.loads(out or "[]")
    except (OSError, subprocess.SubprocessError, ValueError):
        return {}
    res = {}
    for link in data:
        res[link.get("ifname", "")] = [
            f"{a['local']}/{a['prefixlen']}" for a in link.get("addr_info", [])
            if a.get("family") == "inet" or (a.get("family") == "inet6" and a.get("scope") == "global")]
    return res


def _dhcp_bound():
    """Interfaces dnsmasq's DHCP socket (UDP 67) is bound to; None if unknown."""
    try:
        out = subprocess.run(["ss", "-H", "-uln", "sport", "= :67"], capture_output=True, text=True,
                             timeout=5).stdout
    except (OSError, subprocess.SubprocessError):
        return None
    bound = set(re.findall(r"%([\w.-]+):67\b", out))
    if re.search(r"(^|\s)(0\.0\.0\.0|\*|\[::\]):67\b", out):
        bound.add("*")  # bound on every interface
    return bound


def _kernel_view():
    """Interfaces as the kernel sees them (physical NICs only)."""
    roles, addrs, bound = _roles(), _addrs(), _dhcp_bound()
    out = []
    try:
        names = sorted(os.listdir(SYS_NET))
    except OSError:
        names = []
    for name in names:
        base = os.path.join(SYS_NET, name)
        if name == "lo" or not os.path.exists(os.path.join(base, "device")):
            continue  # skip loopback, bridges, veths, tunnels
        mac = _read(os.path.join(base, "address")).lower()
        oper = _read(os.path.join(base, "operstate"))
        try:
            speed = int(_read(os.path.join(base, "speed")))
            speed = speed if speed > 0 else None
        except ValueError:
            speed = None
        notes = []
        driver = os.path.basename(os.path.realpath(os.path.join(base, "device", "driver")))
        if driver == "cdc_ncm" and speed in CDC_NCM_WRAPPED:
            notes.append(f"Link {CDC_NCM_WRAPPED[speed] // 1000} Gb/s (the cdc_ncm driver reports "
                         f"{speed} Mb/s: its 32-bit speed counter wraps above 4.29 Gb/s)")
            speed = CDC_NCM_WRAPPED[speed]
        out.append({
            "iface": name, "mac": mac,
            "role": roles.get(name) or roles.get(mac) or "other",
            "link": "up" if oper == "up" else "down",
            "speed_mbps": speed if oper == "up" else None,
            "addrs": addrs.get(name, []),
            "verdict": None, "evidence": [], "notes": notes,
            "dnsmasq_bound": (None if bound is None else (name in bound or "*" in bound)),
        })
    return out


def _age_seconds(stamp):
    try:
        t = datetime.fromisoformat(str(stamp).replace("Z", "+00:00"))
        if t.tzinfo is None:
            t = t.replace(tzinfo=timezone.utc)
        return (datetime.now(timezone.utc) - t).total_seconds()
    except ValueError:
        return None


def _problems(ifaces, guard):
    """[(level, text)]: 'bad' is a DHCP-on-the-wrong-network risk."""
    out = []
    if (guard or {}).get("state") == "blocked":
        out.append(("bad", "DHCP/PXE is blocked: " + (guard.get("reason") or "the imaging NIC is not isolated")))
    imaging = [i for i in ifaces if i["role"] == "imaging"]
    for i in ifaces:
        if i["role"] == "imaging" and i.get("verdict") in ("ge", "foreign-dhcp"):
            what = "the GE network" if i["verdict"] == "ge" else "a network with another DHCP server"
            out.append(("bad", f"Imaging NIC {i['iface']} is on {what}. Do not serve DHCP/PXE here."))
        if i["role"] != "imaging" and i.get("dnsmasq_bound"):
            out.append(("bad", f"dnsmasq is answering DHCP on {i['iface']} ({i['role']} NIC), not only on the imaging NIC."))
    if not imaging:
        out.append(("warn", "No imaging NIC configured (/etc/pxe-server/nic.conf or dnsmasq's interface= pin)."))
    for i in imaging:
        if i["link"] != "up":
            out.append(("warn", f"Imaging NIC {i['iface']} has no link: no PC can network-boot."))
        elif i.get("dnsmasq_bound") is False:
            out.append(("warn", f"dnsmasq is not listening for DHCP on the imaging NIC {i['iface']}."))
    for i in ifaces:
        if i["role"] == "ge" and i["link"] != "up":
            out.append(("warn", f"GE NIC {i['iface']} has no link (no updates or remote admin)."))
    return out


def status():
    """{interfaces, source, checked_at, stale, guard, problems, state}.
    state: 'ok' | 'warn' | 'bad'."""
    data, source = None, "kernel"
    try:
        with open(config.NIC_STATUS_FILE) as f:
            data = json.load(f)
    except (OSError, ValueError):
        data = None
    stale = True
    if isinstance(data, dict) and isinstance(data.get("interfaces"), list):
        age = _age_seconds(data.get("checked_at"))
        stale = age is None or age > STALE_SECONDS
        source = "check"
    live = {i["iface"]: i for i in _kernel_view()}
    if source == "check":
        ifaces = []
        for i in data["interfaces"]:
            i = dict(i)
            k = live.get(i.get("iface"))
            if k:  # link and addresses are cheap to read now; keep the check's verdict
                i.update(link=k["link"], speed_mbps=k["speed_mbps"], addrs=k["addrs"], notes=k["notes"])
                if k["dnsmasq_bound"] is not None:
                    i["dnsmasq_bound"] = k["dnsmasq_bound"]
            if stale:
                i["verdict"] = None
            ifaces.append(i)
        guard = data.get("dhcp_guard") or {}
    else:
        ifaces, guard = list(live.values()), {}
    ifaces.sort(key=lambda i: {"imaging": 0, "ge": 1}.get(i.get("role"), 2))
    problems = _problems(ifaces, guard)
    state = "bad" if any(l == "bad" for l, _ in problems) else "warn" if problems else "ok"
    return {"interfaces": ifaces, "source": source, "stale": stale and source == "check",
            "checked_at": (data or {}).get("checked_at") if source == "check" else None,
            "guard": guard, "problems": problems, "state": state}