04db6d6605
Everything West Jefferson specific moved out to a separate repo (pxe-site-wj): NTLARS/UDC/heat-treat backups keyed by WJ machine numbers, CMM and Wax/Trace bay lists, WJ OpenText profiles, site-config.json, the WJ preinstall list, and the WJ shopfloor PC types (Collections, NoCollections, Common, Keyence, CMM, Genspect, HeatTreat, Wax/Trace, Display, PartMarker) with GE-Enforce, machine numbers and the desktop layout. The part-marker bench tools moved to their own repo. Dead scripts removed: build-proxmox-iso, deploy-bios, sync_hardware_models, mirror-from-gold, pull-bios, startnet-template.cmd. Shopfloor is one generic type, gea-shopfloor-standard, with today's timing: - PPKG at first logon (unattend Order 10, unchanged). - Install-AppList.ps1 runs C:\PreInstall\preinstall.json during setup. - A new SYSTEM task, GE Shopfloor Post-PPKG, runs postppkg.json once after the post-PPKG reboot. It also clears AutoLogonCount after the ShopFloor handover (the Users-level monitor can be denied that HKLM write) and removes the Intune Sync task when done. - Both lists ship empty. A site adds entries and installers. - WinPE choice 3 goes straight to the PPKG menu. Staging, fetch and heal carry only the generic payload. The webapp's shopfloor sub-menu editor is gone, and the imaging tiles name the new stages. - The playbook deploys the empty lists only if missing. It now deploys the PowerShell 7 MSI the shopfloor unattend installs, which it never did before. Blancco site settings move into the webapp (Updates > Blancco releases): business location, licensing Wi-Fi SSID and password. They are rendered into the served preferences.xml and applied to every staged release, and are seeded from site-vars (site_name, blancco_wifi_ssid, blancco_wifi_password) only while none exist. The repo XML carries no site values; the plaintext Wi-Fi password is gone from it. New site_name and site_timezone settings: report and evidence-pack site name, server clock and webapp time zone. No more West Jefferson defaults in webapp/config.py or the dashboard title. Sync-BPRT.ps1 gets the server address filled in at download instead of WJ's GE address. USB on Windows: - scripts/Build-PxeUsb.ps1 runs the same checks as build-usb.sh. It writes one FAT32 CIDATA partition (ISO files plus patched grub.cfg plus payload) with diskpart and reads every file back by SHA-256. - Tested with Windows PowerShell 5.1 in the win11 VM, and the layout installs and first-boots in the pxetest VM. - scripts/stage-usb-share.sh stages the build folder on a share. - site-content/ is optional for both builders. KB docs/kb/How_To_Deploy_PXE_Server_New_Site.html: new-site deployment, Windows only, with a site-vars.yml guide. Checks: webapp tests pass (5 suites); the unattend lint and playbook syntax checks pass; every .ps1 parses; there are no WJ hosts, IPs, names or PC types in playbook/, webapp/, scripts/ or autoinstall/.
1423 lines
52 KiB
Python
Executable File
1423 lines
52 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
pxe-nic-check - classify this PXE server's network interfaces and keep
|
|
dnsmasq from ever serving DHCP/PXE on the GE corporate network.
|
|
|
|
Each PXE server has two NICs:
|
|
|
|
imaging static 172.16.9.1/24 on the isolated imaging switch. dnsmasq
|
|
serves DHCP/TFTP here and ONLY here.
|
|
ge DHCP client on the GE corporate network (updates, webapp).
|
|
|
|
If the imaging NIC ever lands on a network with another DHCP server or a
|
|
managed/corporate switch (cable swap, uplinked switch, second PXE server),
|
|
dnsmasq must not serve. This script finds out which network each NIC is
|
|
really on, without trusting cable labels:
|
|
|
|
passive an AF_PACKET raw socket listens for N seconds for LLDP and CDP
|
|
switch announcements, 802.1X EAPOL (port authentication), STP
|
|
BPDUs, DHCP OFFER/ACK sent by other servers, and counts the
|
|
distinct ARP sender MACs. Sends nothing.
|
|
probe (imaging NIC) one DHCPDISCOVER broadcast out of the interface,
|
|
offers collected for 6 s (dnsmasq ARP/ICMP-pings an address
|
|
for ~3 s before it offers it). Needs no IP config. An offer is "own"
|
|
when its server id is one of this interface's addresses AND it
|
|
came from one of this host's MACs; a box that answers with our
|
|
address from a foreign MAC is a second PXE server and is foreign.
|
|
lease (DHCP-client NICs) the systemd-networkd lease in
|
|
/run/systemd/netif/leases/<ifindex> is read instead of probing.
|
|
|
|
Verdicts per interface:
|
|
|
|
no-link no carrier
|
|
isolated no foreign DHCP server, no LLDP/CDP/EAPOL seen
|
|
ge foreign DHCP with a corp domain (opt 15 / 119), or any
|
|
LLDP, CDP or authenticator EAPOL seen (unmanaged switches
|
|
send none). EAPOL from clients (iPXE and Windows send
|
|
EAPOL-Start and EAP-Response) is ignored.
|
|
foreign-dhcp a non-own DHCP server without a corp domain
|
|
error the check itself failed
|
|
|
|
Healthy states: the ge NIC should be "ge"; the imaging NIC should be
|
|
"isolated" (or "no-link"). STP is reported but does not change a verdict.
|
|
|
|
Modes:
|
|
|
|
(default) check, print a summary (or --json), optionally --write.
|
|
--guard dnsmasq ExecStartPre. Imaging NIC, 5 s window + probe (~6 s).
|
|
exit 0 = safe to serve, exit 1 = DO NOT serve (ge or
|
|
foreign-dhcp). Internal errors and no link exit 0 with a loud
|
|
log line: fail-open on bugs, fail-closed only on detection.
|
|
--print-imaging
|
|
the imaging NIC (and its bridge ports), for pxe-nic-carrier.
|
|
--watch systemd timer. 60 s window + probe, writes the status file,
|
|
and on a bad imaging verdict runs "systemctl stop dnsmasq"
|
|
and logs CRIT. It never starts dnsmasq. If dnsmasq was
|
|
(re)started inside the window (its guard passed after part
|
|
of the evidence was heard), it re-checks for ~6 s first and
|
|
stops dnsmasq only if that still says blocked.
|
|
|
|
Config: /etc/pxe-server/nic.conf (INI, [nic] section optional)
|
|
|
|
imaging_iface = br-pxe # or imaging_mac = aa:bb:cc:dd:ee:ff
|
|
ge_iface = enp1s0 # or ge_mac = ...
|
|
corp_domain_suffixes = geaerospace.net, ge.com
|
|
own_ip = 172.16.9.1
|
|
allowed_switches = # LLDP/CDP names or chassis MACs of a
|
|
# managed IMAGING switch to ignore
|
|
|
|
Without imaging_iface/imaging_mac the imaging NIC is the interface= line of
|
|
/etc/dnsmasq.d/pxe-nic-only.conf, else the interface holding own_ip. A MAC
|
|
that belongs to a bridge port resolves to the bridge. Without ge_iface/ge_mac
|
|
the ge NIC is the first other interface holding a networkd DHCP lease.
|
|
|
|
Status file (--write, 0644, atomic): /run/pxe-nic-status.json, read by the
|
|
webapp. Shape: see build_status().
|
|
|
|
Operator tests (floor idle) are in docs/pxe-nic-guard.md.
|
|
|
|
Python 3 stdlib only. Runs as root (raw sockets); the parsers do not.
|
|
"""
|
|
|
|
import argparse
|
|
import configparser
|
|
import datetime
|
|
import glob
|
|
import json
|
|
import os
|
|
import random
|
|
import select
|
|
import socket
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
CONFIG_PATH = "/etc/pxe-server/nic.conf"
|
|
DNSMASQ_NIC_CONF = "/etc/dnsmasq.d/pxe-nic-only.conf"
|
|
DNSMASQ_CONF_GLOBS = ["/etc/dnsmasq.conf", "/etc/dnsmasq.d/*.conf"]
|
|
LEASE_DIR = "/run/systemd/netif/leases"
|
|
SYS_NET = "/sys/class/net"
|
|
STATUS_PATH = "/run/pxe-nic-status.json"
|
|
DEFAULT_SUFFIXES = "geaerospace.net, ge.com"
|
|
DEFAULT_OWN_IP = "172.16.9.1"
|
|
STOP_CMD = "systemctl stop dnsmasq"
|
|
STATE_CMD = ("systemctl show -p ActiveState "
|
|
"-p ActiveEnterTimestampMonotonic dnsmasq")
|
|
|
|
ETH_P_ALL = 0x0003
|
|
PACKET_OUTGOING = 4
|
|
ETH_LLDP = 0x88CC
|
|
ETH_EAPOL = 0x888E
|
|
ETH_ARP = 0x0806
|
|
ETH_IP = 0x0800
|
|
VLAN_TPIDS = (0x8100, 0x88A8)
|
|
MAC_CDP = bytes.fromhex("01000ccccccc")
|
|
MAC_STP = bytes.fromhex("0180c2000000")
|
|
MAC_PVST = bytes.fromhex("01000ccccccd")
|
|
DHCP_MAGIC = b"\x63\x82\x53\x63"
|
|
|
|
PROG = "pxe-nic-check"
|
|
|
|
|
|
# ---------------------------------------------------------------- logging
|
|
|
|
def log(msg, level="info"):
|
|
"""stderr; under systemd the <N> prefix sets the journal priority."""
|
|
prio = {"crit": 2, "err": 3, "warning": 4, "notice": 5, "info": 6}[level]
|
|
prefix = "<%d>" % prio if os.environ.get("JOURNAL_STREAM") else ""
|
|
tag = "" if level == "info" else level.upper() + ": "
|
|
sys.stderr.write("%s%s: %s%s\n" % (prefix, PROG, tag, msg))
|
|
sys.stderr.flush()
|
|
|
|
|
|
# ---------------------------------------------------------------- helpers
|
|
|
|
def mac_str(b):
|
|
return ":".join("%02x" % x for x in b)
|
|
|
|
|
|
def norm_mac(s):
|
|
return (s or "").strip().lower().replace("-", ":")
|
|
|
|
|
|
def ip_str(b):
|
|
return socket.inet_ntoa(b)
|
|
|
|
|
|
def read_file(path, default=""):
|
|
try:
|
|
with open(path) as f:
|
|
return f.read().strip()
|
|
except OSError:
|
|
return default
|
|
|
|
|
|
def parse_suffixes(text):
|
|
out = []
|
|
for part in (text or "").replace(";", ",").split(","):
|
|
part = part.strip().strip(".").lower()
|
|
if part:
|
|
out.append(part)
|
|
return out
|
|
|
|
|
|
def domain_matches(domain, suffixes):
|
|
"""Whole-label suffix match: site.geaerospace.net and geaerospace.net
|
|
match geaerospace.net; notgeaerospace.net does not."""
|
|
d = (domain or "").strip().strip(".").lower()
|
|
if not d:
|
|
return False
|
|
for s in suffixes:
|
|
if d == s or d.endswith("." + s):
|
|
return True
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------- frame parsing
|
|
|
|
def split_ether(frame):
|
|
"""(dst, src, ethertype_or_len, payload). Skips 802.1Q/QinQ tags;
|
|
returns the innermost VLAN id as well."""
|
|
if len(frame) < 14:
|
|
return None
|
|
dst, src = frame[0:6], frame[6:12]
|
|
etype = struct.unpack("!H", frame[12:14])[0]
|
|
off = 14
|
|
vlan = None
|
|
while etype in VLAN_TPIDS and len(frame) >= off + 4:
|
|
tci, etype = struct.unpack("!HH", frame[off:off + 4])
|
|
vlan = tci & 0x0FFF
|
|
off += 4
|
|
return dst, src, etype, frame[off:], vlan
|
|
|
|
|
|
def _lldp_id(subtype, value, chassis):
|
|
# chassis subtype 4 / port subtype 3 = MAC address; 5 / 4 = network addr
|
|
mac_sub = 4 if chassis else 3
|
|
net_sub = 5 if chassis else 4
|
|
if subtype == mac_sub and len(value) == 6:
|
|
return mac_str(value)
|
|
if subtype == net_sub and len(value) >= 5 and value[0] == 1:
|
|
return ip_str(value[1:5])
|
|
return value.decode("utf-8", "replace").strip("\x00 ")
|
|
|
|
|
|
def parse_lldp(payload):
|
|
"""LLDPDU -> dict(chassis, port, system, port_desc, vlan) or None."""
|
|
out = {"chassis": None, "port": None, "system": None,
|
|
"port_desc": None, "vlan": None}
|
|
off = 0
|
|
while off + 2 <= len(payload):
|
|
hdr = struct.unpack("!H", payload[off:off + 2])[0]
|
|
ttype, tlen = hdr >> 9, hdr & 0x1FF
|
|
val = payload[off + 2:off + 2 + tlen]
|
|
off += 2 + tlen
|
|
if len(val) < tlen:
|
|
return None
|
|
if ttype == 0:
|
|
break
|
|
if ttype == 1 and tlen >= 2:
|
|
out["chassis"] = _lldp_id(val[0], val[1:], True)
|
|
elif ttype == 2 and tlen >= 2:
|
|
out["port"] = _lldp_id(val[0], val[1:], False)
|
|
elif ttype == 4:
|
|
out["port_desc"] = val.decode("utf-8", "replace").strip("\x00 ")
|
|
elif ttype == 5:
|
|
out["system"] = val.decode("utf-8", "replace").strip("\x00 ")
|
|
elif ttype == 127 and tlen >= 6 and val[0:3] == b"\x00\x80\xc2" \
|
|
and val[3] == 1:
|
|
out["vlan"] = struct.unpack("!H", val[4:6])[0]
|
|
if out["chassis"] is None and out["port"] is None:
|
|
return None
|
|
return out
|
|
|
|
|
|
def parse_cdp(payload):
|
|
"""802.3 LLC/SNAP payload -> dict(device, port, vlan) or None."""
|
|
if len(payload) < 12 or payload[0:3] != b"\xaa\xaa\x03" \
|
|
or payload[3:6] != b"\x00\x00\x0c" \
|
|
or payload[6:8] != b"\x20\x00":
|
|
return None
|
|
body = payload[8:]
|
|
off = 4 # version, ttl, checksum
|
|
out = {"device": None, "port": None, "vlan": None}
|
|
while off + 4 <= len(body):
|
|
ttype, tlen = struct.unpack("!HH", body[off:off + 4])
|
|
if tlen < 4:
|
|
break
|
|
val = body[off + 4:off + tlen]
|
|
off += tlen
|
|
if ttype == 0x0001:
|
|
out["device"] = val.decode("utf-8", "replace").strip("\x00 ")
|
|
elif ttype == 0x0003:
|
|
out["port"] = val.decode("utf-8", "replace").strip("\x00 ")
|
|
elif ttype == 0x000A and len(val) >= 2:
|
|
out["vlan"] = struct.unpack("!H", val[0:2])[0]
|
|
if out["device"] is None and out["port"] is None:
|
|
return None
|
|
return out
|
|
|
|
|
|
def _dns_names(data):
|
|
"""RFC 1035 names with RFC 3397 compression (DHCP option 119)."""
|
|
names = []
|
|
i = 0
|
|
while i < len(data):
|
|
labels = []
|
|
pos = i
|
|
jumped = False
|
|
hops = 0
|
|
while pos < len(data):
|
|
n = data[pos]
|
|
if n == 0:
|
|
pos += 1
|
|
break
|
|
if n & 0xC0 == 0xC0:
|
|
if pos + 1 >= len(data):
|
|
return names
|
|
ptr = ((n & 0x3F) << 8) | data[pos + 1]
|
|
if not jumped:
|
|
i = pos + 2
|
|
jumped = True
|
|
hops += 1
|
|
if hops > 32 or ptr >= len(data):
|
|
return names
|
|
pos = ptr
|
|
continue
|
|
labels.append(data[pos + 1:pos + 1 + n].decode("ascii", "replace"))
|
|
pos += 1 + n
|
|
if not jumped:
|
|
i = pos
|
|
if labels:
|
|
names.append(".".join(labels))
|
|
return names
|
|
|
|
|
|
def parse_dhcp_options(opts):
|
|
"""Option bytes -> {code: bytes}; repeated codes are concatenated
|
|
(RFC 3396)."""
|
|
out = {}
|
|
i = 0
|
|
while i < len(opts):
|
|
code = opts[i]
|
|
if code == 255:
|
|
break
|
|
if code == 0:
|
|
i += 1
|
|
continue
|
|
if i + 1 >= len(opts):
|
|
break
|
|
ln = opts[i + 1]
|
|
out[code] = out.get(code, b"") + opts[i + 2:i + 2 + ln]
|
|
i += 2 + ln
|
|
return out
|
|
|
|
|
|
def parse_bootp(udp_payload):
|
|
"""BOOTP/DHCP message -> dict or None."""
|
|
p = udp_payload
|
|
if len(p) < 240 or p[236:240] != DHCP_MAGIC:
|
|
return None
|
|
opts = parse_dhcp_options(p[240:])
|
|
mtype = opts.get(53, b"\x00")[0] if opts.get(53) else 0
|
|
sid = opts.get(54)
|
|
domain = opts.get(15, b"").decode("ascii", "replace").strip("\x00 .") \
|
|
or None
|
|
search = _dns_names(opts[119]) if 119 in opts else []
|
|
return {
|
|
"op": p[0],
|
|
"xid": struct.unpack("!I", p[4:8])[0],
|
|
"yiaddr": ip_str(p[16:20]),
|
|
"chaddr": mac_str(p[28:34]),
|
|
"msgtype": mtype,
|
|
"server_id": ip_str(sid[0:4]) if sid and len(sid) >= 4 else None,
|
|
"domain": domain,
|
|
"search": search,
|
|
}
|
|
|
|
|
|
EAPOL_TYPE_EAP = 0
|
|
EAP_AUTHENTICATOR_CODES = (1, 3, 4) # Request, Success, Failure
|
|
|
|
|
|
def eapol_from_authenticator(payload):
|
|
"""True only for EAPOL that an 802.1X authenticator (a switch port)
|
|
sends: EAPOL type 0 (EAP-Packet) carrying EAP code 1 Request, 3 Success
|
|
or 4 Failure. Supplicants send EAPOL-Start (type 1), Logoff (type 2)
|
|
and EAP Responses (code 2): iPXE sends three EAPOL-Starts every time
|
|
it opens a NIC, and Windows dot3svc sends Starts and Responses, so a
|
|
bay PXE-booting next to us must not read as a GE switch port."""
|
|
if len(payload) < 5:
|
|
return False
|
|
return payload[1] == EAPOL_TYPE_EAP \
|
|
and payload[4] in EAP_AUTHENTICATOR_CODES
|
|
|
|
|
|
def parse_dhcp_frame(etype, payload):
|
|
"""IPv4/UDP 67->68 server message -> bootp dict (+ src ip) or None."""
|
|
if etype != ETH_IP or len(payload) < 20:
|
|
return None
|
|
ihl = (payload[0] & 0x0F) * 4
|
|
if payload[0] >> 4 != 4 or payload[9] != 17 or len(payload) < ihl + 8:
|
|
return None
|
|
sport, dport = struct.unpack("!HH", payload[ihl:ihl + 4])
|
|
if sport != 67 or dport not in (68, 67):
|
|
return None
|
|
msg = parse_bootp(payload[ihl + 8:])
|
|
if not msg or msg["op"] != 2:
|
|
return None
|
|
msg["src_ip"] = ip_str(payload[12:16])
|
|
return msg
|
|
|
|
|
|
class Observations:
|
|
"""Everything seen on one interface (bridge + its ports)."""
|
|
|
|
def __init__(self):
|
|
self.lldp = {} # (chassis, port) -> dict
|
|
self.cdp = {} # (device, port) -> dict
|
|
self.eapol = False # EAPOL from an authenticator (a switch)
|
|
self.eapol_client = 0 # supplicant EAPOL frames, ignored
|
|
self.stp = False
|
|
self.arp_macs = set()
|
|
self.dhcp = {} # (server, src_mac) -> dict
|
|
self.frames = 0
|
|
|
|
def feed(self, frame, own_macs, probe_xid=None):
|
|
"""Classify one received (not outgoing) frame."""
|
|
parts = split_ether(frame)
|
|
if not parts:
|
|
return
|
|
dst, src, etype, payload, _vlan = parts
|
|
self.frames += 1
|
|
if etype == ETH_LLDP:
|
|
info = parse_lldp(payload)
|
|
if info:
|
|
self.lldp[(info["chassis"], info["port"])] = info
|
|
return
|
|
if etype == ETH_EAPOL:
|
|
if eapol_from_authenticator(payload):
|
|
self.eapol = True
|
|
else:
|
|
self.eapol_client += 1
|
|
return
|
|
if etype <= 1500: # 802.3 length field: LLC frames
|
|
if dst == MAC_CDP:
|
|
info = parse_cdp(payload)
|
|
if info:
|
|
self.cdp[(info["device"], info["port"])] = info
|
|
elif dst in (MAC_STP, MAC_PVST) or payload[0:2] == b"\x42\x42":
|
|
self.stp = True
|
|
return
|
|
if dst == MAC_PVST:
|
|
self.stp = True
|
|
return
|
|
if etype == ETH_ARP and len(payload) >= 14:
|
|
sha = mac_str(payload[8:14])
|
|
if sha not in own_macs and sha != "00:00:00:00:00:00":
|
|
self.arp_macs.add(sha)
|
|
return
|
|
msg = parse_dhcp_frame(etype, payload)
|
|
if msg and msg["msgtype"] in (2, 5): # OFFER, ACK
|
|
self.add_dhcp(msg, mac_str(src),
|
|
"probe" if msg["xid"] == probe_xid else "passive")
|
|
|
|
def add_dhcp(self, msg, src_mac, how):
|
|
server = msg["server_id"] or msg["src_ip"]
|
|
key = (server, src_mac)
|
|
if key in self.dhcp:
|
|
return
|
|
self.dhcp[key] = {
|
|
"server": server, "src_mac": src_mac,
|
|
"domain": msg["domain"], "search": msg["search"],
|
|
"yiaddr": msg["yiaddr"], "how": how,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------- DHCP probe
|
|
|
|
def _ip_checksum(hdr):
|
|
s = 0
|
|
for i in range(0, len(hdr), 2):
|
|
s += (hdr[i] << 8) + hdr[i + 1]
|
|
while s >> 16:
|
|
s = (s & 0xFFFF) + (s >> 16)
|
|
return (~s) & 0xFFFF
|
|
|
|
|
|
def build_discover(mac_bytes, xid):
|
|
"""Ethernet/IPv4/UDP DHCPDISCOVER, broadcast flag set."""
|
|
bootp = struct.pack("!BBBBIHH4s4s4s4s16s64s128s",
|
|
1, 1, 6, 0, xid, 0, 0x8000,
|
|
b"\0" * 4, b"\0" * 4, b"\0" * 4, b"\0" * 4,
|
|
mac_bytes + b"\0" * 10, b"\0" * 64, b"\0" * 128)
|
|
opts = DHCP_MAGIC + bytes([53, 1, 1]) \
|
|
+ bytes([55, 6, 1, 3, 6, 15, 54, 119]) \
|
|
+ bytes([60, len(PROG)]) + PROG.encode() + b"\xff"
|
|
bootp += opts
|
|
if len(bootp) < 300:
|
|
bootp += b"\0" * (300 - len(bootp))
|
|
udp = struct.pack("!HHHH", 68, 67, 8 + len(bootp), 0) + bootp
|
|
ip = struct.pack("!BBHHHBBH4s4s", 0x45, 0x10, 20 + len(udp),
|
|
random.randint(0, 0xFFFF), 0, 64, 17, 0,
|
|
b"\0" * 4, b"\xff" * 4)
|
|
ip = ip[:10] + struct.pack("!H", _ip_checksum(ip)) + ip[12:]
|
|
eth = b"\xff" * 6 + mac_bytes + struct.pack("!H", ETH_IP)
|
|
return eth + ip + udp
|
|
|
|
|
|
# ---------------------------------------------------------------- interfaces
|
|
|
|
def iface_mac(name):
|
|
return norm_mac(read_file(os.path.join(SYS_NET, name, "address")))
|
|
|
|
|
|
def all_ifaces():
|
|
try:
|
|
return sorted(os.listdir(SYS_NET))
|
|
except OSError:
|
|
return []
|
|
|
|
|
|
def is_bridge(name):
|
|
return os.path.isdir(os.path.join(SYS_NET, name, "bridge"))
|
|
|
|
|
|
def bridge_ports(name):
|
|
try:
|
|
return sorted(os.listdir(os.path.join(SYS_NET, name, "brif")))
|
|
except OSError:
|
|
return []
|
|
|
|
|
|
def bridge_master(name):
|
|
link = os.path.join(SYS_NET, name, "master")
|
|
if os.path.islink(link):
|
|
return os.path.basename(os.readlink(link))
|
|
return None
|
|
|
|
|
|
def is_physical(name):
|
|
return os.path.exists(os.path.join(SYS_NET, name, "device"))
|
|
|
|
|
|
def link_info(name):
|
|
carrier = read_file(os.path.join(SYS_NET, name, "carrier"), "0")
|
|
oper = read_file(os.path.join(SYS_NET, name, "operstate"), "down")
|
|
up = carrier == "1" and oper in ("up", "unknown")
|
|
speed = None
|
|
if up:
|
|
try:
|
|
v = int(read_file(os.path.join(SYS_NET, name, "speed"), "-1"))
|
|
speed = v if v > 0 else None
|
|
except ValueError:
|
|
speed = None
|
|
return up, speed
|
|
|
|
|
|
def iface_addrs():
|
|
"""{iface: ["a.b.c.d/nn", ...]} IPv4 only, via iproute2."""
|
|
out = {}
|
|
try:
|
|
res = subprocess.run(["ip", "-j", "-4", "addr", "show"],
|
|
capture_output=True, text=True, timeout=5)
|
|
for entry in json.loads(res.stdout or "[]"):
|
|
name = entry.get("ifname")
|
|
for a in entry.get("addr_info", []):
|
|
if a.get("family") == "inet":
|
|
out.setdefault(name, []).append(
|
|
"%s/%s" % (a["local"], a["prefixlen"]))
|
|
except (OSError, ValueError, subprocess.SubprocessError):
|
|
pass
|
|
return out
|
|
|
|
|
|
def parse_lease(text):
|
|
"""systemd-networkd lease file -> dict of KEY=VALUE."""
|
|
out = {}
|
|
for line in text.splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
k, v = line.split("=", 1)
|
|
out[k.strip()] = v.strip()
|
|
return out
|
|
|
|
|
|
def read_lease(name, lease_dir=LEASE_DIR):
|
|
idx = read_file(os.path.join(SYS_NET, name, "ifindex"))
|
|
if not idx:
|
|
return None
|
|
path = os.path.join(lease_dir, idx)
|
|
if not os.path.exists(path):
|
|
return None
|
|
return parse_lease(read_file(path))
|
|
|
|
|
|
# ---------------------------------------------------------------- config
|
|
|
|
def load_config(path=CONFIG_PATH, dnsmasq_nic_conf=DNSMASQ_NIC_CONF):
|
|
cfg = {"imaging_iface": "", "imaging_mac": "", "ge_iface": "",
|
|
"ge_mac": "", "corp_domain_suffixes": DEFAULT_SUFFIXES,
|
|
"own_ip": DEFAULT_OWN_IP, "allowed_switches": "",
|
|
"dnsmasq_ifaces": []}
|
|
text = read_file(path)
|
|
if text:
|
|
cp = configparser.ConfigParser(inline_comment_prefixes=("#", ";"))
|
|
try:
|
|
try:
|
|
cp.read_string(text)
|
|
except configparser.MissingSectionHeaderError:
|
|
cp = configparser.ConfigParser(
|
|
inline_comment_prefixes=("#", ";"))
|
|
cp.read_string("[nic]\n" + text)
|
|
for sect in cp.sections():
|
|
for k, v in cp.items(sect):
|
|
if k in cfg and v.strip():
|
|
cfg[k] = v.strip()
|
|
except configparser.Error as e:
|
|
log("bad config %s: %s (using defaults)" % (path, e), "warning")
|
|
cfg["imaging_mac"] = norm_mac(cfg["imaging_mac"])
|
|
cfg["ge_mac"] = norm_mac(cfg["ge_mac"])
|
|
cfg["suffixes"] = parse_suffixes(cfg["corp_domain_suffixes"])
|
|
cfg["allowed"] = allowed_list(cfg["allowed_switches"])
|
|
for line in read_file(dnsmasq_nic_conf).splitlines():
|
|
line = line.split("#", 1)[0].strip()
|
|
if line.startswith("interface="):
|
|
for name in line.split("=", 1)[1].split(","):
|
|
if name.strip():
|
|
cfg["dnsmasq_ifaces"].append(name.strip())
|
|
return cfg
|
|
|
|
|
|
def find_by_mac(mac, ifaces):
|
|
if not mac:
|
|
return None
|
|
hits = [i for i in ifaces if iface_mac(i) == mac]
|
|
# a bridge often clones a port's MAC; the port's master owns the IP
|
|
for i in hits:
|
|
master = bridge_master(i)
|
|
if master:
|
|
return master
|
|
return hits[0] if hits else None
|
|
|
|
|
|
def resolve_roles(cfg, ifaces, addrs, lease_dir=LEASE_DIR):
|
|
"""-> (imaging, ge, notes). Either may be None."""
|
|
notes = []
|
|
imaging = None
|
|
if cfg["imaging_iface"] and cfg["imaging_iface"] in ifaces:
|
|
imaging = cfg["imaging_iface"]
|
|
elif cfg["imaging_mac"]:
|
|
imaging = find_by_mac(cfg["imaging_mac"], ifaces)
|
|
if not imaging:
|
|
notes.append("imaging_mac %s not found" % cfg["imaging_mac"])
|
|
if not imaging:
|
|
for name in cfg["dnsmasq_ifaces"]:
|
|
if name in ifaces:
|
|
imaging = name
|
|
notes.append("imaging NIC from %s" % DNSMASQ_NIC_CONF)
|
|
break
|
|
if not imaging:
|
|
for name, alist in sorted(addrs.items()):
|
|
if any(a.split("/")[0] == cfg["own_ip"] for a in alist):
|
|
imaging = name
|
|
notes.append("imaging NIC = holder of %s" % cfg["own_ip"])
|
|
break
|
|
ge = None
|
|
if cfg["ge_iface"] and cfg["ge_iface"] in ifaces:
|
|
ge = cfg["ge_iface"]
|
|
elif cfg["ge_mac"]:
|
|
ge = find_by_mac(cfg["ge_mac"], ifaces)
|
|
if not ge:
|
|
notes.append("ge_mac %s not found" % cfg["ge_mac"])
|
|
if not ge:
|
|
for name in ifaces:
|
|
if name != imaging and name != "lo" and \
|
|
read_lease(name, lease_dir) is not None:
|
|
ge = name
|
|
notes.append("ge NIC = holder of a DHCP lease")
|
|
break
|
|
if ge == imaging:
|
|
ge = None
|
|
return imaging, ge, notes
|
|
|
|
|
|
# ---------------------------------------------------------------- dnsmasq binding
|
|
|
|
def dnsmasq_conf_rules(globs=DNSMASQ_CONF_GLOBS):
|
|
rules = {"interface": set(), "except": set(), "no_dhcp": set(),
|
|
"listen": set()}
|
|
keymap = {"interface": "interface", "except-interface": "except",
|
|
"no-dhcp-interface": "no_dhcp", "listen-address": "listen"}
|
|
for g in globs:
|
|
for path in sorted(glob.glob(g)):
|
|
for line in read_file(path).splitlines():
|
|
line = line.split("#", 1)[0].strip()
|
|
if "=" not in line:
|
|
continue
|
|
k, v = line.split("=", 1)
|
|
if k.strip() in keymap:
|
|
for item in v.split(","):
|
|
if item.strip():
|
|
rules[keymap[k.strip()]].add(item.strip())
|
|
return rules
|
|
|
|
|
|
def udp67_sockets():
|
|
"""[(addr, dev, proc)] or None if ss is unavailable."""
|
|
try:
|
|
res = subprocess.run(["ss", "-H", "-uanp", "sport = :67"],
|
|
capture_output=True, text=True, timeout=5)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return None
|
|
if res.returncode != 0:
|
|
return None
|
|
out = []
|
|
for line in res.stdout.splitlines():
|
|
cols = line.split()
|
|
if len(cols) < 4:
|
|
continue
|
|
local = cols[3]
|
|
hostpart = local.rsplit(":", 1)[0]
|
|
dev = None
|
|
if "%" in hostpart:
|
|
hostpart, dev = hostpart.split("%", 1)
|
|
proc = None
|
|
if "users:((" in line:
|
|
proc = line.split('users:(("', 1)[1].split('"', 1)[0]
|
|
out.append((hostpart.strip("[]"), dev, proc))
|
|
return out
|
|
|
|
|
|
def dnsmasq_bound(iface, addrs, socks, rules):
|
|
if socks is None:
|
|
return None
|
|
mine = [s for s in socks if s[2] is None or "dnsmasq" in s[2]]
|
|
if not mine:
|
|
return False
|
|
ips = [a.split("/")[0] for a in addrs]
|
|
for addr, dev, _proc in mine:
|
|
if dev == iface:
|
|
return True
|
|
if dev is None and addr in ips:
|
|
return True
|
|
if not any(dev is None and addr in ("0.0.0.0", "*", "")
|
|
for addr, dev, _p in mine):
|
|
return False
|
|
# wildcard socket: dnsmasq filters per interface by its config
|
|
if iface in rules["no_dhcp"] or iface in rules["except"]:
|
|
return False
|
|
if iface in rules["interface"]:
|
|
return True
|
|
if any(ip in rules["listen"] for ip in ips):
|
|
return True
|
|
if not rules["interface"] and not rules["listen"]:
|
|
return bool(ips)
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------- BPF filter
|
|
|
|
# The imaging NIC moves gigabits of SMB while a watch runs. A classic BPF
|
|
# program in the kernel passes only the frames this script parses, so the
|
|
# Python loop never sees bulk traffic. Ops are (code, k, jt, jf) with jt/jf
|
|
# as labels (None = next instruction).
|
|
_BPF = {"ldh": 0x28, "ldb": 0x30, "jeq": 0x15, "jset": 0x45, "ret": 0x06,
|
|
"ldxmsh": 0xB1, "ldhind": 0x48}
|
|
BPF_ACCEPT = 0x40000
|
|
|
|
BPF_PROGRAM = [
|
|
("ldh", 12), # ethertype / 802.3 length
|
|
("jeq", ETH_LLDP, "ok"),
|
|
("jeq", ETH_EAPOL, "ok"),
|
|
("jeq", ETH_ARP, "ok"),
|
|
("jeq", 0x8100, "ok"), # tagged: let Python unwrap it
|
|
("jeq", 0x88A8, "ok"),
|
|
("jeq", ETH_IP, "ip"),
|
|
("ldh", 0), # destination MAC prefix
|
|
("jeq", 0x0180, "ok"), # 01:80:c2 STP, LLDP, EAPOL
|
|
("jeq", 0x0100, None, "drop"),
|
|
("ldh", 2),
|
|
("jeq", 0x0CCC, "ok", "drop"), # 01:00:0c:cc:cc:cc/cd CDP, PVST
|
|
("label", "ip"),
|
|
("ldb", 23), # IP protocol
|
|
("jeq", 17, None, "drop"), # UDP only
|
|
("ldh", 20),
|
|
("jset", 0x1FFF, "drop"), # not a later fragment
|
|
("ldxmsh", 14), # X = IP header length
|
|
("ldhind", 14), # UDP source port
|
|
("jeq", 67, "ok", "drop"), # from a DHCP server
|
|
("label", "ok"),
|
|
("ret", BPF_ACCEPT),
|
|
("label", "drop"),
|
|
("ret", 0),
|
|
]
|
|
|
|
|
|
def assemble_bpf(program=BPF_PROGRAM):
|
|
"""-> list of (code, jt, jf, k) with relative jumps resolved."""
|
|
labels = {}
|
|
insns = []
|
|
for op in program:
|
|
if op[0] == "label":
|
|
labels[op[1]] = len(insns)
|
|
else:
|
|
insns.append(op)
|
|
out = []
|
|
for i, op in enumerate(insns):
|
|
name, k = op[0], op[1]
|
|
jt = jf = 0
|
|
if name in ("jeq", "jset"):
|
|
t = op[2] if len(op) > 2 else None
|
|
f = op[3] if len(op) > 3 else None
|
|
jt = labels[t] - i - 1 if t else 0
|
|
jf = labels[f] - i - 1 if f else 0
|
|
if not (0 <= jt < 256 and 0 <= jf < 256):
|
|
raise ValueError("BPF jump out of range at %d" % i)
|
|
out.append((_BPF[name], jt, jf, k))
|
|
return out
|
|
|
|
|
|
def attach_bpf(sock):
|
|
import ctypes
|
|
insns = assemble_bpf()
|
|
raw = b"".join(struct.pack("HBBI", c, jt, jf, k)
|
|
for c, jt, jf, k in insns)
|
|
buf = ctypes.create_string_buffer(raw, len(raw))
|
|
fprog = struct.pack("HL", len(insns), ctypes.addressof(buf))
|
|
sock.setsockopt(socket.SOL_SOCKET, 26, fprog) # SO_ATTACH_FILTER
|
|
|
|
|
|
# ---------------------------------------------------------------- capture
|
|
|
|
SOL_PACKET = 263
|
|
PACKET_ADD_MEMBERSHIP = 1
|
|
PACKET_MR_MULTICAST = 0
|
|
|
|
# Link-local groups the NIC's hardware filter drops unless someone joins
|
|
# them: STP, EAPOL (PAE), LLDP, CDP/VTP/DTP, PVST+. Without them a plain
|
|
# (non-promiscuous) NIC on a CDP-only Cisco port hears nothing.
|
|
CAPTURE_GROUPS = ["01:80:c2:00:00:00", "01:80:c2:00:00:03",
|
|
"01:80:c2:00:00:0e", "01:00:0c:cc:cc:cc",
|
|
"01:00:0c:cc:cc:cd"]
|
|
|
|
|
|
def packet_mreq(ifindex, mac):
|
|
"""struct packet_mreq {int mr_ifindex; unsigned short mr_type,
|
|
mr_alen; unsigned char mr_address[8];} for PACKET_MR_MULTICAST."""
|
|
addr = bytes.fromhex(norm_mac(mac).replace(":", ""))
|
|
if len(addr) != 6:
|
|
raise ValueError("bad MAC %r" % mac)
|
|
return struct.pack("iHH8s", ifindex, PACKET_MR_MULTICAST, len(addr),
|
|
addr)
|
|
|
|
|
|
def join_groups(sock, ifname, groups=CAPTURE_GROUPS):
|
|
"""Join each group on this socket's interface. The kernel drops the
|
|
memberships when the socket closes. Returns the groups that failed."""
|
|
failed = []
|
|
try:
|
|
ifindex = socket.if_nametoindex(ifname)
|
|
except OSError:
|
|
return list(groups)
|
|
for g in groups:
|
|
try:
|
|
sock.setsockopt(SOL_PACKET, PACKET_ADD_MEMBERSHIP,
|
|
packet_mreq(ifindex, g))
|
|
except (OSError, ValueError):
|
|
failed.append(g)
|
|
return failed
|
|
|
|
|
|
def capture(targets, window, probe_ifaces, own_macs, probe_window=6.0):
|
|
"""targets: {report_iface: [capture ifaces]}. Listens on all of them
|
|
for `window` s (at least probe_window after the probe). Returns
|
|
({report_iface: Observations}, {report_iface: error or None},
|
|
{report_iface: probe_xid or None})."""
|
|
obs = {name: Observations() for name in targets}
|
|
errors = {name: None for name in targets}
|
|
xids = {name: None for name in targets}
|
|
socks = {}
|
|
for name, cap_list in targets.items():
|
|
for cap in cap_list:
|
|
try:
|
|
# protocol 0 = receive nothing until the filter is on
|
|
s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, 0)
|
|
try:
|
|
attach_bpf(s)
|
|
except (OSError, ValueError) as e:
|
|
log("no BPF filter on %s (%s), parsing every frame"
|
|
% (cap, e), "warning")
|
|
s.bind((cap, ETH_P_ALL))
|
|
bad = join_groups(s, cap)
|
|
if bad:
|
|
log("could not join %s on %s; a quiet switch may be "
|
|
"missed" % (", ".join(bad), cap), "warning")
|
|
s.setblocking(False)
|
|
socks[s] = name
|
|
except OSError as e:
|
|
if cap == name:
|
|
errors[name] = "capture on %s failed: %s" % (cap, e)
|
|
start = time.monotonic()
|
|
for name in probe_ifaces:
|
|
if errors.get(name) or name not in targets:
|
|
continue
|
|
try:
|
|
mac = bytes.fromhex(iface_mac(name).replace(":", ""))
|
|
xid = random.randint(1, 0xFFFFFFFF)
|
|
s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW)
|
|
s.bind((name, 0))
|
|
s.send(build_discover(mac, xid))
|
|
s.close()
|
|
xids[name] = xid
|
|
except (OSError, ValueError) as e:
|
|
errors[name] = "DHCP probe on %s failed: %s" % (name, e)
|
|
deadline = start + max(window, probe_window if any(xids.values()) else 0)
|
|
try:
|
|
while socks:
|
|
left = deadline - time.monotonic()
|
|
if left <= 0:
|
|
break
|
|
ready, _, _ = select.select(list(socks), [], [], min(left, 1.0))
|
|
for s in ready:
|
|
name = socks[s]
|
|
while True:
|
|
try:
|
|
frame, addr = s.recvfrom(65535)
|
|
except BlockingIOError:
|
|
break
|
|
except OSError:
|
|
break
|
|
if len(addr) > 2 and addr[2] == PACKET_OUTGOING:
|
|
continue
|
|
try:
|
|
obs[name].feed(frame, own_macs, xids[name])
|
|
except Exception as e: # never let one frame kill a run
|
|
log("frame parse error on %s: %s" % (name, e),
|
|
"warning")
|
|
finally:
|
|
for s in socks:
|
|
s.close()
|
|
return obs, errors, xids
|
|
|
|
|
|
# ---------------------------------------------------------------- verdicts
|
|
|
|
def allowed_list(text):
|
|
"""Names match case-insensitively; MACs in any separator style."""
|
|
out = set()
|
|
for x in parse_suffixes(text):
|
|
out.add(x.lower())
|
|
out.add(norm_mac(x))
|
|
return out
|
|
|
|
|
|
def _allowed(names, allowed):
|
|
return any(n and (n.lower() in allowed or norm_mac(n) in allowed)
|
|
for n in names)
|
|
|
|
|
|
def classify(link_up, error, obs, dhcp_servers, suffixes, window,
|
|
probed, lease_used, allowed=()):
|
|
"""-> (verdict, evidence). dhcp_servers: list of dicts with keys
|
|
server, domain, search, own, src_mac, yiaddr, how. `allowed` lists
|
|
LLDP/CDP system names or chassis ids of a managed imaging switch that
|
|
should not count (EAPOL always counts)."""
|
|
evidence = []
|
|
if error:
|
|
return "error", [error]
|
|
if not link_up:
|
|
return "no-link", ["no carrier"]
|
|
|
|
switch = False
|
|
for info in obs.lldp.values() if obs else []:
|
|
txt = "LLDP from %s port %s" % (info["system"] or info["chassis"],
|
|
info["port"])
|
|
if info.get("port_desc"):
|
|
txt += " (%s)" % info["port_desc"]
|
|
if info.get("vlan") is not None:
|
|
txt += " vlan %s" % info["vlan"]
|
|
if _allowed([info["system"], info["chassis"]], allowed):
|
|
txt += " (allowed imaging switch, ignored)"
|
|
else:
|
|
switch = True
|
|
evidence.append(txt)
|
|
for info in obs.cdp.values() if obs else []:
|
|
txt = "CDP from %s port %s" % (info["device"], info["port"])
|
|
if info.get("vlan") is not None:
|
|
txt += " vlan %s" % info["vlan"]
|
|
if _allowed([info["device"]], allowed):
|
|
txt += " (allowed imaging switch, ignored)"
|
|
else:
|
|
switch = True
|
|
evidence.append(txt)
|
|
if obs and obs.eapol:
|
|
evidence.append("802.1X EAPOL seen (port authentication)")
|
|
if obs and obs.eapol_client:
|
|
evidence.append("%d client EAPOL frames ignored (supplicants such "
|
|
"as iPXE EAPOL-Start, informational)"
|
|
% obs.eapol_client)
|
|
if obs and obs.stp:
|
|
evidence.append("STP BPDUs seen (informational)")
|
|
|
|
corp = False
|
|
foreign = False
|
|
for d in dhcp_servers:
|
|
doms = [x for x in [d.get("domain")] + list(d.get("search") or [])
|
|
if x]
|
|
is_corp = any(domain_matches(x, suffixes) for x in doms)
|
|
if d.get("own"):
|
|
evidence.append("own DHCP server %s answered (ignored)"
|
|
% d["server"])
|
|
continue
|
|
txt = "DHCP from %s" % d["server"]
|
|
txt += ", domain %s" % ", ".join(doms) if doms else ", no domain"
|
|
if d.get("yiaddr") and d["yiaddr"] != "0.0.0.0":
|
|
txt += ", offered %s" % d["yiaddr"]
|
|
if d.get("how"):
|
|
txt += " (%s)" % d["how"]
|
|
if d.get("claims_own"):
|
|
txt += ", CLAIMS OUR ADDRESS from %s (second PXE server?)" \
|
|
% d.get("src_mac")
|
|
evidence.append(txt)
|
|
foreign = True
|
|
corp = corp or is_corp
|
|
|
|
switch = switch or bool(obs is not None and obs.eapol)
|
|
if switch or corp:
|
|
verdict = "ge"
|
|
elif foreign:
|
|
verdict = "foreign-dhcp"
|
|
else:
|
|
verdict = "isolated"
|
|
quiet = []
|
|
what = "no foreign DHCP server" if any(
|
|
d.get("own") for d in dhcp_servers) else "no DHCP server"
|
|
quiet.append(what if (probed or lease_used)
|
|
else what + " seen (not probed)")
|
|
quiet.append("no switch announcements in %d s" % window)
|
|
evidence.append(", ".join(quiet))
|
|
if obs is not None:
|
|
evidence.append("%d ARP hosts seen" % len(obs.arp_macs))
|
|
return verdict, evidence
|
|
|
|
|
|
def dhcp_list(obs, own_ips, local_macs):
|
|
out = []
|
|
for d in (obs.dhcp.values() if obs else []):
|
|
claims_own = d["server"] in own_ips
|
|
own = claims_own and d["src_mac"] in local_macs
|
|
e = dict(d)
|
|
e["own"] = own
|
|
e["claims_own"] = claims_own and not own
|
|
out.append(e)
|
|
return out
|
|
|
|
|
|
def lease_servers(lease):
|
|
if not lease:
|
|
return []
|
|
server = lease.get("SERVER_ADDRESS") or lease.get("ROUTER") or "?"
|
|
search = [x for x in lease.get("DOMAIN_SEARCH_LIST", "").split() if x]
|
|
return [{"server": server, "domain": lease.get("DOMAINNAME") or None,
|
|
"search": search, "own": False, "src_mac": None,
|
|
"yiaddr": lease.get("ADDRESS"), "how": "networkd lease",
|
|
"claims_own": False}]
|
|
|
|
|
|
def guard_state(imaging_entry):
|
|
if imaging_entry is None:
|
|
return "unknown", "imaging NIC could not be determined"
|
|
v = imaging_entry["verdict"]
|
|
name = imaging_entry["iface"]
|
|
if v in ("ge", "foreign-dhcp"):
|
|
why = "; ".join(e for e in imaging_entry["evidence"]
|
|
if not e.endswith("ARP hosts seen"))
|
|
return "blocked", "imaging NIC %s verdict %s: %s" % (name, v, why)
|
|
if v == "isolated":
|
|
return "allowed", "imaging NIC %s is isolated" % name
|
|
if v == "no-link":
|
|
return "allowed", "imaging NIC %s has no link" % name
|
|
return "unknown", "imaging NIC %s check failed: %s" % (
|
|
name, "; ".join(imaging_entry["evidence"]))
|
|
|
|
|
|
def build_status(entries, duration, guard):
|
|
"""The webapp contract. Field names are fixed."""
|
|
return {
|
|
"checked_at": datetime.datetime.now().astimezone().isoformat(
|
|
timespec="seconds"),
|
|
"duration_s": int(round(duration)),
|
|
"dhcp_guard": {"state": guard[0], "reason": guard[1]},
|
|
"interfaces": entries,
|
|
}
|
|
|
|
|
|
def write_status(path, status):
|
|
d = os.path.dirname(os.path.abspath(path)) or "."
|
|
fd, tmp = tempfile.mkstemp(prefix=".pxe-nic-status.", dir=d)
|
|
try:
|
|
with os.fdopen(fd, "w") as f:
|
|
json.dump(status, f, indent=2)
|
|
f.write("\n")
|
|
os.chmod(tmp, 0o644)
|
|
os.replace(tmp, path)
|
|
except BaseException:
|
|
try:
|
|
os.unlink(tmp)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
|
|
|
|
# ---------------------------------------------------------------- main run
|
|
|
|
def run_check(args, cfg):
|
|
ifaces = all_ifaces()
|
|
addrs = iface_addrs()
|
|
imaging, ge, notes = resolve_roles(cfg, ifaces, addrs, args.lease_dir)
|
|
for n in notes:
|
|
log(n)
|
|
|
|
if args.iface:
|
|
report = [i for i in args.iface if i in ifaces]
|
|
for i in args.iface:
|
|
if i not in ifaces:
|
|
log("no such interface: %s" % i, "warning")
|
|
elif args.mode == "guard":
|
|
# guard: imaging NIC only, plus the GE NIC when its networkd lease
|
|
# (or lack of link) says something without capturing on it
|
|
report = [imaging] if imaging else []
|
|
if ge and (read_lease(ge, args.lease_dir) is not None
|
|
or not link_info(ge)[0]):
|
|
report.append(ge)
|
|
else:
|
|
report = [x for x in (imaging, ge) if x]
|
|
for i in ifaces:
|
|
if i not in report and is_physical(i) and not bridge_master(i):
|
|
report.append(i)
|
|
|
|
local_macs = set(iface_mac(i) for i in ifaces) - {"00:00:00:00:00:00"}
|
|
links = {i: link_info(i) for i in report}
|
|
leases = {i: read_lease(i, args.lease_dir) for i in report}
|
|
|
|
targets = {}
|
|
for i in report:
|
|
if not links[i][0]:
|
|
continue
|
|
if args.mode == "guard" and i != imaging:
|
|
continue # guard: GE NIC from its lease only, no capture
|
|
cap = [i]
|
|
if is_bridge(i):
|
|
# link-local frames (LLDP, EAPOL, STP) stop at the bridge port
|
|
cap += bridge_ports(i)
|
|
targets[i] = cap
|
|
|
|
# Probe the imaging NIC (or every --iface given). A NIC holding a
|
|
# networkd lease is a DHCP client: its lease says who the server is,
|
|
# so it is never probed.
|
|
probe_ifaces = []
|
|
if args.probe:
|
|
probe_ifaces = [i for i in targets
|
|
if (i == imaging or args.iface)
|
|
and leases.get(i) is None]
|
|
|
|
t0 = time.monotonic()
|
|
obs, errors, xids = {}, {}, {}
|
|
if targets:
|
|
obs, errors, xids = capture(targets, args.window, probe_ifaces,
|
|
local_macs, args.probe_window)
|
|
duration = time.monotonic() - t0
|
|
|
|
socks = udp67_sockets()
|
|
rules = dnsmasq_conf_rules()
|
|
entries = []
|
|
imaging_entry = None
|
|
for i in report:
|
|
up, speed = links[i]
|
|
role = "imaging" if i == imaging else "ge" if i == ge else "other"
|
|
o = obs.get(i)
|
|
own_ips = set(a.split("/")[0] for a in addrs.get(i, []))
|
|
servers = dhcp_list(o, own_ips, local_macs)
|
|
lease_used = False
|
|
if leases.get(i) is not None:
|
|
servers += lease_servers(leases[i])
|
|
lease_used = True
|
|
window = int(round(duration)) if i in targets else 0
|
|
verdict, evidence = classify(
|
|
up, errors.get(i), o if i in targets else None, servers,
|
|
cfg["suffixes"], window, bool(xids.get(i)), lease_used,
|
|
cfg["allowed"])
|
|
if i not in targets and lease_used:
|
|
evidence.append("from the networkd lease only (guard mode)")
|
|
if xids.get(i) and not any(s["how"] == "probe" for s in servers):
|
|
evidence.append("DHCP probe: no offers in %g s"
|
|
% args.probe_window)
|
|
if role == "imaging" and verdict not in ("isolated", "no-link"):
|
|
evidence.append("UNHEALTHY: imaging NIC must be isolated")
|
|
if role == "ge" and verdict not in ("ge", "no-link"):
|
|
evidence.append("note: GE NIC does not look like the GE network")
|
|
seen = set()
|
|
dhcp_out = []
|
|
for s in servers:
|
|
key = (s["server"], s["domain"], s["own"])
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
dom = s["domain"] or (s["search"][0] if s.get("search") else None)
|
|
dhcp_out.append({"server": s["server"], "domain": dom,
|
|
"own": bool(s["own"])})
|
|
entry = {
|
|
"iface": i,
|
|
"mac": iface_mac(i),
|
|
"role": role,
|
|
"link": "up" if up else "down",
|
|
"speed_mbps": speed,
|
|
"addrs": addrs.get(i, []),
|
|
"verdict": verdict,
|
|
"evidence": evidence,
|
|
"dhcp_servers": dhcp_out,
|
|
"switch": {
|
|
"lldp": [{"system": x["system"] or x["chassis"],
|
|
"port": x["port"], "vlan": x["vlan"]}
|
|
for x in (o.lldp.values() if o else [])],
|
|
"cdp": [{"device": x["device"], "port": x["port"]}
|
|
for x in (o.cdp.values() if o else [])],
|
|
"eapol": bool(o and o.eapol),
|
|
"stp": bool(o and o.stp),
|
|
},
|
|
"arp_hosts": len(o.arp_macs) if o else 0,
|
|
"dnsmasq_bound": dnsmasq_bound(i, addrs.get(i, []), socks, rules),
|
|
}
|
|
entries.append(entry)
|
|
if role == "imaging":
|
|
imaging_entry = entry
|
|
guard = guard_state(imaging_entry)
|
|
return build_status(entries, duration, guard), imaging_entry
|
|
|
|
|
|
def summary(status):
|
|
lines = ["checked %s (%d s), dhcp_guard %s: %s" % (
|
|
status["checked_at"], status["duration_s"],
|
|
status["dhcp_guard"]["state"], status["dhcp_guard"]["reason"])]
|
|
for e in status["interfaces"]:
|
|
lines.append(" %-16s %-8s %-4s %-13s %s" % (
|
|
e["iface"], e["role"], e["link"], e["verdict"],
|
|
", ".join(e["addrs"]) or "-"))
|
|
for ev in e["evidence"]:
|
|
lines.append(" " + ev)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def parse_unit_state(text):
|
|
"""`systemctl show -p ActiveState -p ActiveEnterTimestampMonotonic`
|
|
output -> (active_state or None, active_enter_us or None)."""
|
|
props = {}
|
|
for line in (text or "").splitlines():
|
|
if "=" in line:
|
|
k, v = line.split("=", 1)
|
|
props[k.strip()] = v.strip()
|
|
state = props.get("ActiveState") or None
|
|
try:
|
|
since = int(props.get("ActiveEnterTimestampMonotonic", ""))
|
|
except ValueError:
|
|
since = None
|
|
return state, since
|
|
|
|
|
|
def dnsmasq_state(cmd):
|
|
try:
|
|
res = subprocess.run(cmd, shell=True, capture_output=True,
|
|
text=True, timeout=10)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return None, None
|
|
if res.returncode != 0:
|
|
return None, None
|
|
return parse_unit_state(res.stdout)
|
|
|
|
|
|
def watch_decision(state, since_us, window_start_us):
|
|
"""What --watch does with a 'blocked' verdict, given dnsmasq's
|
|
ActiveState / ActiveEnterTimestampMonotonic and the CLOCK_MONOTONIC
|
|
microsecond the watch's capture began. -> "stop", "skip" (already
|
|
stopped) or "recheck" (dnsmasq was (re)started inside the window, so
|
|
its own guard passed after part of our evidence was heard; confirm
|
|
with a fresh short check before stopping it)."""
|
|
if state is None:
|
|
return "stop" # cannot tell: stopping is the safe side
|
|
if state in ("inactive", "failed", "deactivating"):
|
|
return "skip"
|
|
if state in ("activating", "reloading"):
|
|
return "recheck" # its guard is running right now
|
|
if since_us and since_us > window_start_us:
|
|
return "recheck"
|
|
return "stop"
|
|
|
|
|
|
def stop_dnsmasq(cmd):
|
|
try:
|
|
rc = subprocess.run(cmd, shell=True, timeout=60).returncode
|
|
if rc != 0:
|
|
log("'%s' exited %d" % (cmd, rc), "crit")
|
|
except (OSError, subprocess.SubprocessError) as e:
|
|
log("could not stop dnsmasq: %s" % e, "crit")
|
|
|
|
|
|
def recheck(args, cfg):
|
|
"""A guard-style check (imaging NIC, short window + probe), written to
|
|
the status file. -> (state, reason). An internal error counts as
|
|
blocked: the 60 s watch already saw the bad evidence."""
|
|
sub = argparse.Namespace(**vars(args))
|
|
sub.mode = "guard"
|
|
sub.window = 5.0
|
|
sub.probe = True
|
|
try:
|
|
status, _ = run_check(sub, cfg)
|
|
except Exception as e:
|
|
return "blocked", "re-check failed (%s); keeping the watch verdict" \
|
|
% e
|
|
if args.write:
|
|
try:
|
|
write_status(args.write, status)
|
|
except OSError as e:
|
|
log("could not write %s: %s" % (args.write, e), "err")
|
|
return status["dhcp_guard"]["state"], status["dhcp_guard"]["reason"]
|
|
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser(
|
|
prog=PROG,
|
|
description="Classify the PXE server's NICs (imaging vs GE "
|
|
"corporate) and keep dnsmasq off the corporate network.",
|
|
epilog="Operator tests: docs/pxe-nic-guard.md. Exit codes: --guard "
|
|
"0 = safe to serve, 1 = DO NOT serve; other modes 0, or 2 on "
|
|
"an internal error.")
|
|
mode = ap.add_mutually_exclusive_group()
|
|
mode.add_argument("--guard", dest="mode", action="store_const",
|
|
const="guard", help="dnsmasq ExecStartPre check")
|
|
mode.add_argument("--watch", dest="mode", action="store_const",
|
|
const="watch", help="timer mode: stop dnsmasq on a "
|
|
"bad imaging verdict")
|
|
mode.add_argument("--print-imaging", dest="mode", action="store_const",
|
|
const="print-imaging", help="print the imaging NIC "
|
|
"and, for a bridge, its ports (one per line; nothing "
|
|
"if unknown), then exit. Used by pxe-nic-carrier.")
|
|
ap.add_argument("--json", action="store_true",
|
|
help="print the status JSON")
|
|
ap.add_argument("--write", metavar="PATH", nargs="?", const=STATUS_PATH,
|
|
help="write the status JSON atomically (default %s)"
|
|
% STATUS_PATH)
|
|
ap.add_argument("--window", type=float, default=None,
|
|
help="passive capture seconds (guard 5, otherwise 60)")
|
|
ap.add_argument("--probe", dest="probe", action="store_true",
|
|
default=None, help="send one DHCPDISCOVER on the "
|
|
"imaging NIC (default on in --guard/--watch)")
|
|
ap.add_argument("--no-probe", dest="probe", action="store_false")
|
|
# 6 s, not 3: dnsmasq (the likeliest rogue, a second PXE server) pings
|
|
# the address it is about to offer and only answers ~3 s later.
|
|
ap.add_argument("--probe-window", type=float, default=6.0,
|
|
help="seconds to collect DHCP offers (default 6)")
|
|
ap.add_argument("--iface", action="append",
|
|
help="check only this interface (repeatable)")
|
|
ap.add_argument("--config", default=CONFIG_PATH)
|
|
ap.add_argument("--dnsmasq-nic-conf", default=DNSMASQ_NIC_CONF)
|
|
ap.add_argument("--lease-dir", default=LEASE_DIR)
|
|
ap.add_argument("--stop-cmd", default=STOP_CMD,
|
|
help="command --watch runs to stop dnsmasq")
|
|
ap.add_argument("--state-cmd", default=STATE_CMD,
|
|
help="command --watch runs to read dnsmasq's "
|
|
"ActiveState and ActiveEnterTimestampMonotonic")
|
|
args = ap.parse_args(argv)
|
|
args.mode = args.mode or "check"
|
|
if args.window is None:
|
|
args.window = 5.0 if args.mode == "guard" else 60.0
|
|
if args.probe is None:
|
|
args.probe = args.mode in ("guard", "watch")
|
|
if args.mode in ("guard", "watch") and not args.write:
|
|
args.write = STATUS_PATH
|
|
|
|
if args.mode == "print-imaging":
|
|
try:
|
|
cfg = load_config(args.config, args.dnsmasq_nic_conf)
|
|
ifaces = all_ifaces()
|
|
imaging, _ge, _notes = resolve_roles(cfg, ifaces, iface_addrs(),
|
|
args.lease_dir)
|
|
except Exception as e:
|
|
log("could not resolve the imaging NIC: %s" % e, "err")
|
|
return 2
|
|
if imaging:
|
|
print("\n".join([imaging] + (bridge_ports(imaging)
|
|
if is_bridge(imaging) else [])))
|
|
return 0
|
|
|
|
# CLOCK_MONOTONIC, the clock of ActiveEnterTimestampMonotonic
|
|
window_start_us = int(time.clock_gettime(time.CLOCK_MONOTONIC) * 1e6)
|
|
try:
|
|
cfg = load_config(args.config, args.dnsmasq_nic_conf)
|
|
status, imaging = run_check(args, cfg)
|
|
except Exception as e: # fail open on bugs, but loudly
|
|
import traceback
|
|
log("INTERNAL ERROR, check did not complete: %s" % e, "crit")
|
|
traceback.print_exc()
|
|
if args.mode == "guard":
|
|
log("guard FAILING OPEN: dnsmasq allowed to start unchecked",
|
|
"crit")
|
|
return 0
|
|
return 2
|
|
|
|
if args.write:
|
|
try:
|
|
write_status(args.write, status)
|
|
except OSError as e:
|
|
log("could not write %s: %s" % (args.write, e), "err")
|
|
if args.json:
|
|
print(json.dumps(status, indent=2))
|
|
elif args.mode == "check":
|
|
print(summary(status))
|
|
|
|
state, reason = status["dhcp_guard"]["state"], \
|
|
status["dhcp_guard"]["reason"]
|
|
if args.mode == "guard":
|
|
if state == "blocked":
|
|
log("REFUSING to start dnsmasq: " + reason, "crit")
|
|
return 1
|
|
if state == "unknown" or (imaging and imaging["verdict"] == "no-link"):
|
|
log("guard could not verify the imaging NIC, allowing dnsmasq: "
|
|
+ reason, "err")
|
|
else:
|
|
log("ok: " + reason)
|
|
return 0
|
|
if args.mode == "watch":
|
|
if state == "blocked":
|
|
dstate, since = dnsmasq_state(args.state_cmd)
|
|
what = watch_decision(dstate, since, window_start_us)
|
|
if what == "recheck":
|
|
log("dnsmasq was (re)started during this watch window "
|
|
"(its guard passed); re-checking before stopping it: "
|
|
+ reason, "warning")
|
|
state, reason = recheck(args, cfg)
|
|
if state == "blocked":
|
|
what = "stop"
|
|
else:
|
|
log("re-check says %s, leaving dnsmasq running: %s"
|
|
% (state, reason), "notice")
|
|
if what == "skip":
|
|
log("imaging NIC is NOT isolated, dnsmasq already %s: %s"
|
|
% (dstate, reason), "crit")
|
|
elif what == "stop":
|
|
log("imaging NIC is NOT isolated, stopping dnsmasq: "
|
|
+ reason, "crit")
|
|
stop_dnsmasq(args.stop_cmd)
|
|
elif state == "unknown":
|
|
log("watch could not verify the imaging NIC: " + reason, "err")
|
|
else:
|
|
log("ok: " + reason)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|