0d69222f85
A 'Network' pill next to 'Services' lists each physical NIC with its role (imaging / GE network, from /etc/pxe-server/nic.conf or dnsmasq's interface= pin), link, speed, addresses, and whether dnsmasq serves DHCP/PXE on it. It reads pxe-nic-check's verdict file (/run/pxe-nic-status.json) when present, and shows a red banner when the imaging NIC is on the GE network or a network with another DHCP server, or when dnsmasq answers DHCP on a non-imaging NIC. Without the file it shows link and addresses only. Read-only.
188 lines
7.3 KiB
Python
188 lines
7.3 KiB
Python
"""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"
|
|
|
|
|
|
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
|
|
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": [],
|
|
"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"])
|
|
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}
|