diff --git a/docs/pxe-nic-guard.md b/docs/pxe-nic-guard.md new file mode 100644 index 0000000..5570aa6 --- /dev/null +++ b/docs/pxe-nic-guard.md @@ -0,0 +1,158 @@ +# PXE NIC guard + +The PXE server's DHCP/PXE (dnsmasq) must never serve on the GE corporate +network. Each server has two NICs: + +| Role | Network | Address | +|---------|-----------------------------|----------------------------------| +| imaging | isolated imaging switch | static 172.16.9.1/24, dnsmasq | +| ge | GE corporate network | DHCP client (systemd-networkd) | + +`/etc/dnsmasq.d/pxe-nic-only.conf` pins dnsmasq to the imaging NIC by name. +That stops dnsmasq answering on the GE NIC, but not on the imaging NIC when +that NIC itself ends up on the wrong network: a cable swap, an imaging +switch uplinked into a GE port, or a second PXE server on the same switch. +The NIC guard catches those. + +## What is installed + +| File | Purpose | +|-----------------------------------------------------------|---------------------------------| +| `/usr/local/sbin/pxe-nic-check` | the classifier | +| `/etc/pxe-server/nic.conf` | which NIC is which | +| `/etc/systemd/system/dnsmasq.service.d/nic-guard.conf` | guard before every dnsmasq start| +| `/etc/systemd/system/pxe-nic-check.{service,timer}` | watch every 5 min | +| `/run/pxe-nic-status.json` | last result, read by the webapp | + +Sources: `playbook/pxe-server-helpers/`, installed by `pxe_server_setup.yml` +(vars `pxe_imaging_mac`, `pxe_ge_mac`, `pxe_corp_domain_suffixes`, +`pxe_nic_allowed_switches`). + +## How it decides + +It listens on each NIC (raw socket, sends nothing) for: + +- LLDP and CDP: managed switches announce themselves. GE switches do; the + unmanaged imaging switch does not. +- 802.1X EAPOL: GE ports ask for port authentication. +- DHCP OFFER/ACK from any other server. +- STP and the count of distinct ARP hosts: reported, never decisive. + +On the imaging NIC it also sends one DHCPDISCOVER and collects offers for +6 s. On the GE NIC it reads the networkd lease instead (no traffic). + +| Verdict | Meaning | +|----------------|------------------------------------------------------------| +| `isolated` | no foreign DHCP, no LLDP/CDP/EAPOL | +| `ge` | LLDP, CDP or EAPOL seen, or a DHCP server whose domain ends in a corp suffix (`geaerospace.net`, `ge.com`; whole labels, so `notgeaerospace.net` does not count) | +| `foreign-dhcp` | another DHCP server, no corp domain | +| `no-link` | no carrier | +| `error` | the check itself failed | + +Healthy: imaging NIC `isolated` (or `no-link`), GE NIC `ge`. + +Our own dnsmasq never counts. An offer whose server id is our own address +but comes from a MAC that is not ours is a second PXE server and DOES count. + +If the imaging switch is itself a managed switch that sends LLDP/CDP, put +its LLDP system name, CDP device id or chassis MAC in `allowed_switches` +(`pxe_nic_allowed_switches`), or the guard will block dnsmasq. 802.1X is +never allowed. + +## Modes + +- `pxe-nic-check` - check every NIC (60 s), print a summary. `--json`, + `--write`, `--probe` (imaging DHCP probe), `--iface NAME`, `--window S`. +- `pxe-nic-check --guard` - dnsmasq `ExecStartPre`. Imaging NIC only, + about 6 s. Exit 1 = do not serve: dnsmasq does not start. A bug or no + link exits 0 (fail open) with an error line in the journal. +- `pxe-nic-check --watch` - the timer. On a bad imaging verdict it runs + `systemctl stop dnsmasq` and logs CRIT. It never starts dnsmasq. + +Every start or restart of dnsmasq runs the guard, so `systemctl restart +dnsmasq` and the `@reboot` cron restart take about 6 s longer than before. + +## Reading it + + sudo pxe-nic-check # full summary, 60 s + cat /run/pxe-nic-status.json # last guard/watch result + journalctl -u dnsmasq -b | grep pxe-nic-check + journalctl -u pxe-nic-check -b + +## dnsmasq will not start + +1. `journalctl -u dnsmasq -b | grep pxe-nic-check` shows the evidence, e.g. + `REFUSING to start dnsmasq: imaging NIC br-pxe verdict ge: LLDP from + WJ-IDF2 port Gi1/0/14`. +2. Check the imaging cabling: the imaging NIC must go to the imaging + switch only, and that switch must not be uplinked anywhere. +3. `sudo pxe-nic-check` until the imaging NIC reads `isolated`. +4. `sudo systemctl start dnsmasq` (runs the guard again). + +## Floor tests (run with the floor idle) + +Both tests stop imaging for a few minutes. Have console or GE-side SSH +access to the server; do not rely on the imaging network for the session. + +### (a) Rogue DHCP on the imaging switch: expect the alarm + +1. Confirm the baseline: `sudo pxe-nic-check --probe` shows the imaging + NIC `isolated` and `dnsmasq_bound` true. +2. Plug a device that runs a DHCP server into a free port of the IMAGING + switch (a spare home router with DHCP on, LAN port only, nothing in its + WAN port; or a laptop running dnsmasq). +3. Run the watch now instead of waiting up to 5 min: + `sudo systemctl start pxe-nic-check.service` +4. Expect: + - `journalctl -u pxe-nic-check -n 20` has a CRIT line `imaging NIC is + NOT isolated, stopping dnsmasq: ... verdict foreign-dhcp: DHCP from + 192.168.x.1, ...` (or `ge` if the router sends LLDP). + - `systemctl is-active dnsmasq` is `inactive`. + - `/run/pxe-nic-status.json` has `"dhcp_guard": {"state": "blocked", ...}`. +5. Prove the guard too: `sudo systemctl start dnsmasq` must FAIL, and + `journalctl -u dnsmasq -n 20` shows `REFUSING to start dnsmasq`. +6. Unplug the rogue device. `sudo pxe-nic-check --probe` shows `isolated`. + `sudo systemctl start dnsmasq` starts. PXE-boot one client to confirm. + +### (b) Cable swap into a GE port: expect the guard to refuse dnsmasq + +dnsmasq must be stopped BEFORE the cable moves, so it never serves on GE. + +1. `sudo systemctl stop dnsmasq`; `systemctl is-active dnsmasq` must say + `inactive`. Do not reboot during the test (the `@reboot` cron restarts + dnsmasq; the guard would still block it, but do not test that way). +2. Move the IMAGING NIC's cable from the imaging switch into a live GE + network port. Wait for link: `ip -br link` shows the imaging NIC `UP`. +3. Run the guard by hand first: + `sudo pxe-nic-check --guard; echo "exit $?"` + Expect `exit 1` and a `REFUSING to start dnsmasq` line naming LLDP/CDP, + 802.1X, or `DHCP from 10.x.x.x, domain ....geaerospace.net`. + This sends one DHCPDISCOVER on the GE port, the same as any PC being + plugged in. To send nothing, use `--no-probe --window 60` instead and + rely on LLDP/CDP/802.1X (switches announce every 30-60 s). + **If it exits 0, STOP: do not start dnsmasq. Move the cable back and + send the output of `sudo pxe-nic-check --probe --json`.** +4. Only after step 3 exited 1: `sudo systemctl start dnsmasq`. Expect + `Job for dnsmasq.service failed`; `systemctl is-active dnsmasq` says + `failed`; `journalctl -u dnsmasq -n 20` shows `REFUSING`. +5. Move the cable back to the imaging switch. `sudo pxe-nic-check --probe` + shows the imaging NIC `isolated`. `sudo systemctl start dnsmasq` + starts. PXE-boot one client to confirm. + +## Status file + +`/run/pxe-nic-status.json`, mode 0644, replaced atomically: + + {"checked_at": "2026-10-14T16:47:09-04:00", "duration_s": 60, + "dhcp_guard": {"state": "allowed"|"blocked"|"unknown", "reason": "..."}, + "interfaces": [{"iface": "br-pxe", "mac": "aa:bb:..", "role": "imaging"|"ge"|"other", + "link": "up"|"down", "speed_mbps": 5000, "addrs": ["172.16.9.1/24"], + "verdict": "isolated"|"ge"|"foreign-dhcp"|"no-link"|"error", "evidence": ["..."], + "dhcp_servers": [{"server": "10.1.2.3", "domain": "x.geaerospace.net", "own": false}], + "switch": {"lldp": [{"system": "...", "port": "...", "vlan": null}], + "cdp": [{"device": "...", "port": "..."}], "eapol": false, "stp": false}, + "arp_hosts": 3, "dnsmasq_bound": true}]} + +`dnsmasq_bound` is null when it cannot be told (no `ss`). A bridge +(br-pxe) is reported as one interface; its ports are listened on too, +because LLDP and 802.1X frames stop at the port and never reach the bridge. diff --git a/playbook/pxe-server-helpers/dnsmasq.service.d/nic-guard.conf b/playbook/pxe-server-helpers/dnsmasq.service.d/nic-guard.conf new file mode 100644 index 0000000..e35a4e0 --- /dev/null +++ b/playbook/pxe-server-helpers/dnsmasq.service.d/nic-guard.conf @@ -0,0 +1,18 @@ +# dnsmasq drop-in: refuse to start DHCP/PXE unless the imaging NIC is on +# the isolated imaging network. +# +# Installed as /etc/systemd/system/dnsmasq.service.d/nic-guard.conf by +# pxe_server_setup.yml. pxe-nic-check --guard listens on the imaging NIC +# for 5 s and sends one DHCPDISCOVER (~6 s total), then exits: +# 0 safe to serve (isolated, no link, or the check itself broke - it +# fails open on bugs and says so loudly in the journal) +# 1 DO NOT serve: a foreign DHCP server or a managed/corporate switch +# (LLDP, CDP, 802.1X) is on the imaging NIC. dnsmasq does not start. +# +# Why did dnsmasq not start? journalctl -u dnsmasq -b | grep pxe-nic-check +# Full picture: pxe-nic-check (or the webapp status page) +# Guard does not run on "systemctl reload"; every start/restart runs it, +# including the @reboot cron restart. + +[Service] +ExecStartPre=/usr/local/sbin/pxe-nic-check --guard diff --git a/playbook/pxe-server-helpers/pxe-nic-check b/playbook/pxe-server-helpers/pxe-nic-check new file mode 100755 index 0000000..964901e --- /dev/null +++ b/playbook/pxe-server-helpers/pxe-nic-check @@ -0,0 +1,1237 @@ +#!/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/ 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 EAPOL seen (unmanaged switches send none) + 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. + --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. + +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" + +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 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: wjs.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, + } + + +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 + 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: + self.eapol = True + 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 + +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)) + 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.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 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") + 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") + 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 + + 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": + log("imaging NIC is NOT isolated, stopping dnsmasq: " + reason, + "crit") + try: + rc = subprocess.run(args.stop_cmd, shell=True, + timeout=60).returncode + if rc != 0: + log("'%s' exited %d" % (args.stop_cmd, rc), "crit") + except (OSError, subprocess.SubprocessError) as e: + log("could not stop dnsmasq: %s" % e, "crit") + 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()) diff --git a/playbook/pxe-server-helpers/pxe-nic-check.service b/playbook/pxe-server-helpers/pxe-nic-check.service new file mode 100644 index 0000000..dc22d7b --- /dev/null +++ b/playbook/pxe-server-helpers/pxe-nic-check.service @@ -0,0 +1,16 @@ +# Periodic imaging-NIC watch (pxe-nic-check.timer runs this every 5 min). +# Listens 60 s on each NIC, probes the imaging NIC for foreign DHCP, writes +# /run/pxe-nic-status.json for the webapp, and STOPS dnsmasq if the imaging +# NIC turns out to be on the GE network or next to another DHCP server. +# It never starts dnsmasq: after fixing the cabling, start it by hand +# (systemctl start dnsmasq), which re-runs the guard. + +[Unit] +Description=PXE NIC watch: keep dnsmasq off the GE corporate network +After=network.target + +[Service] +Type=oneshot +ExecStart=/usr/local/sbin/pxe-nic-check --watch +TimeoutStartSec=180 +Nice=5 diff --git a/playbook/pxe-server-helpers/pxe-nic-check.timer b/playbook/pxe-server-helpers/pxe-nic-check.timer new file mode 100644 index 0000000..a7f142d --- /dev/null +++ b/playbook/pxe-server-helpers/pxe-nic-check.timer @@ -0,0 +1,11 @@ +[Unit] +Description=PXE NIC watch every 5 minutes + +[Timer] +OnBootSec=2min +OnActiveSec=1min +OnUnitActiveSec=5min +AccuracySec=15s + +[Install] +WantedBy=timers.target diff --git a/playbook/pxe_server_setup.yml b/playbook/pxe_server_setup.yml index b8a879e..c55df98 100644 --- a/playbook/pxe_server_setup.yml +++ b/playbook/pxe_server_setup.yml @@ -55,6 +55,20 @@ # _media/ links the base image's entries, so also look in the # image roots (on a rebuild the _media views may not exist yet). zscaler_ca_share_glob: "{{ samba_share }}/_media/*/Deploy/Applications/ZscalerCommercialCertificate-2048-SHA256.crt {{ samba_share }}/gea-*/Deploy/Applications/ZscalerCommercialCertificate-2048-SHA256.crt" + # NIC guard (pxe-nic-check): dnsmasq must never serve DHCP/PXE on the GE + # corporate network. Pin the two NICs by MAC (ip -br link); left blank, + # the imaging NIC is the interface= of /etc/dnsmasq.d/pxe-nic-only.conf + # or the holder of 172.16.9.1, and the GE NIC is the one with a DHCP + # lease. A MAC of a bridge port resolves to the bridge (br-pxe). + pxe_imaging_mac: "" + pxe_ge_mac: "" + # A foreign DHCP server handing out one of these domains (whole labels: + # wjs.geaerospace.net matches, notgeaerospace.net does not) = GE network. + pxe_corp_domain_suffixes: "geaerospace.net, ge.com" + # Only if the IMAGING switch is itself managed and sends LLDP/CDP: its + # LLDP system name / CDP device id / chassis MAC, comma separated, so + # it does not count as a corporate switch. 802.1X always counts. + pxe_nic_allowed_switches: "" image_types: - gea-standard - gea-engineer @@ -211,6 +225,77 @@ # firmware / iPXE requests. script-on-renewal + # --- NIC guard: keep dnsmasq off the GE corporate network --- + # pxe-nic-check classifies each NIC from what it hears (LLDP/CDP/802.1X + # from managed switches, foreign DHCP offers and their domain). The + # dnsmasq drop-in runs it as ExecStartPre (--guard: dnsmasq refuses to + # start if the imaging NIC is on the GE network or next to another DHCP + # server); the timer re-checks every 5 min (--watch: stops dnsmasq) and + # writes /run/pxe-nic-status.json for the webapp. Operator notes and + # the floor tests: docs/pxe-nic-guard.md. + - name: "NIC guard: deploy pxe-nic-check" + copy: + src: "{{ usb_mount }}/pxe-server-helpers/pxe-nic-check" + dest: /usr/local/sbin/pxe-nic-check + mode: '0755' + + - name: "NIC guard: create /etc/pxe-server" + file: + path: /etc/pxe-server + state: directory + mode: '0755' + + - name: "NIC guard: write /etc/pxe-server/nic.conf" + copy: + dest: /etc/pxe-server/nic.conf + mode: '0644' + content: | + # Managed by pxe_server_setup.yml (vars pxe_imaging_mac, pxe_ge_mac, + # pxe_corp_domain_suffixes). Read by /usr/local/sbin/pxe-nic-check. + [nic] + {% if pxe_imaging_mac %} + imaging_mac = {{ pxe_imaging_mac | lower }} + {% else %} + # imaging_mac unset: falls back to interface= in + # /etc/dnsmasq.d/pxe-nic-only.conf, then the holder of own_ip + {% endif %} + {% if pxe_ge_mac %} + ge_mac = {{ pxe_ge_mac | lower }} + {% else %} + # ge_mac unset: falls back to the interface holding a DHCP lease + {% endif %} + corp_domain_suffixes = {{ pxe_corp_domain_suffixes }} + own_ip = 172.16.9.1 + allowed_switches = {{ pxe_nic_allowed_switches }} + + - name: "NIC guard: create dnsmasq drop-in directory" + file: + path: /etc/systemd/system/dnsmasq.service.d + state: directory + mode: '0755' + + - name: "NIC guard: install dnsmasq ExecStartPre drop-in" + copy: + src: "{{ usb_mount }}/pxe-server-helpers/dnsmasq.service.d/nic-guard.conf" + dest: /etc/systemd/system/dnsmasq.service.d/nic-guard.conf + mode: '0644' + + - name: "NIC guard: install watch service and timer" + copy: + src: "{{ usb_mount }}/pxe-server-helpers/{{ item }}" + dest: "/etc/systemd/system/{{ item }}" + mode: '0644' + loop: + - pxe-nic-check.service + - pxe-nic-check.timer + + - name: "NIC guard: enable watch timer" + systemd: + name: pxe-nic-check.timer + state: started + enabled: yes + daemon_reload: yes + - name: "Create TFTP directory" file: path: "{{ tftp_dir }}" diff --git a/tests/pxe-nic-check/README.md b/tests/pxe-nic-check/README.md new file mode 100644 index 0000000..8d020c8 --- /dev/null +++ b/tests/pxe-nic-check/README.md @@ -0,0 +1,50 @@ +# pxe-nic-check tests + +Tests for `playbook/pxe-server-helpers/pxe-nic-check`. Neither needs root +or touches a real network. + +## Unit tests + + python3 -m unittest discover -s tests/pxe-nic-check -v + +Synthesized LLDP, CDP, EAPOL, STP and ARP frames and DHCPOFFER/ACK packets +(`frames.py`) fed to the parsers; the verdict table; own-server and +cloned-PXE-server handling; networkd lease parsing; config fallbacks; the +dnsmasq_bound logic; the status JSON shape and atomic 0644 write. + +## Namespace lab + + python3 tests/pxe-nic-check/lab.py # all scenarios, ~5 min + python3 tests/pxe-nic-check/lab.py rogue br-lldp # some + +Re-runs itself under `unshare --user --map-root-user --net --mount`: a +private network namespace with veth pairs, so raw sockets work without sudo +and every frame stays inside it. Each scenario runs the real script in +check (`--json --probe`), `--guard` and `--watch` modes (watch with +`--stop-cmd "touch "` instead of `systemctl stop dnsmasq`) and +asserts the imaging verdict, the guard exit code, whether watch "stopped +dnsmasq", the evidence text and the status JSON shape. + +Peers: `lab_peer.py` (a raw-socket DHCP responder and LLDP/EAPOL sender; +refuses to run unless `PXE_NIC_LAB=1`, which only the lab sets) and real +`dnsmasq` if installed (started with `--user= --group=`, since a user +namespace cannot drop privileges; scenarios are skipped without it). + +| Scenario | Peer | Expect | Guard | +|-----------------|-----------------------------------------------|---------------|-------| +| isolated | none | isolated | 0 | +| rogue | DHCP, no domain | foreign-dhcp | 1 | +| fake-ge | DHCP domain test.geaerospace.net + LLDP | ge | 1 | +| not-corp-domain | DHCP domain notgeaerospace.net | foreign-dhcp | 1 | +| lldp-only | LLDP | ge | 1 | +| eapol-only | 802.1X EAPOL | ge | 1 | +| own-server | DHCP with server id 172.16.9.1 from a local MAC | isolated | 0 | +| cloned-pxe | DHCP with server id 172.16.9.1, foreign MAC | foreign-dhcp | 1 | +| no-link | peer end down | no-link | 0 | +| dnsmasq-corp | real dnsmasq, domain lab.geaerospace.net | ge | 1 | +| dnsmasq-rogue | real dnsmasq, no domain | foreign-dhcp | 1 | +| own-dnsmasq | real dnsmasq serving the imaging NIC itself | isolated, dnsmasq_bound true | 0 | +| br-* | same, imaging NIC is a bridge (like br-pxe), named in nic.conf by a port's MAC | | | + +The GE NIC (g0) is checked from a fake networkd lease with domain +test.geaerospace.net and must read `ge` in every scenario. diff --git a/tests/pxe-nic-check/frames.py b/tests/pxe-nic-check/frames.py new file mode 100644 index 0000000..8d3dab7 --- /dev/null +++ b/tests/pxe-nic-check/frames.py @@ -0,0 +1,166 @@ +""" +frames.py - synthesize the Ethernet frames pxe-nic-check looks for. + +Shared by the unit tests (fed straight to the parsers) and the namespace +lab (sent on a veth by lab_peer.py). Plain struct packing, no scapy. +""" + +import os +import struct +import socket +import importlib.machinery +import importlib.util + +HERE = os.path.dirname(os.path.abspath(__file__)) +SCRIPT = os.path.join(HERE, "..", "..", "playbook", "pxe-server-helpers", + "pxe-nic-check") + + +def load_script(): + """Import the extensionless script as a module (no __pycache__ next + to it: that directory is copied to the server).""" + import sys + sys.dont_write_bytecode = True + loader = importlib.machinery.SourceFileLoader("pxe_nic_check", SCRIPT) + spec = importlib.util.spec_from_loader("pxe_nic_check", loader) + mod = importlib.util.module_from_spec(spec) + loader.exec_module(mod) + return mod + + +def mac(s): + return bytes.fromhex(s.replace(":", "")) + + +def eth(dst, src, etype, payload): + return mac(dst) + mac(src) + struct.pack("!H", etype) + payload + + +def _tlv(t, value): + return struct.pack("!H", (t << 9) | len(value)) + value + + +def lldp(src, system="WJ-IDF2", port="Gi1/0/14", vlan=None, + chassis_mac="00:11:22:33:44:55"): + body = _tlv(1, b"\x04" + mac(chassis_mac)) # chassis: MAC + body += _tlv(2, b"\x05" + port.encode()) # port: ifName + body += _tlv(3, struct.pack("!H", 120)) # TTL + if system is not None: + body += _tlv(5, system.encode()) + if vlan is not None: + body += _tlv(127, b"\x00\x80\xc2\x01" + struct.pack("!H", vlan)) + body += _tlv(0, b"") + return eth("01:80:c2:00:00:0e", src, 0x88CC, body) + + +def cdp(src, device="WJ-CORE1.wjs.geaerospace.net", port="GigabitEthernet1/0/3", + vlan=None): + def t(code, val): + return struct.pack("!HH", code, 4 + len(val)) + val + tlvs = t(1, device.encode()) + t(3, port.encode()) + if vlan is not None: + tlvs += t(0x0A, struct.pack("!H", vlan)) + body = bytes([2, 180]) + b"\x00\x00" + tlvs + llc = b"\xaa\xaa\x03" + b"\x00\x00\x0c" + b"\x20\x00" + body + return mac("01:00:0c:cc:cc:cc") + mac(src) + struct.pack("!H", len(llc)) \ + + llc + + +def eapol(src): + # EAPOL v2, type 0 (EAP packet), EAP Request/Identity + eap = bytes([1, 1]) + struct.pack("!H", 5) + bytes([1]) + body = bytes([2, 0]) + struct.pack("!H", len(eap)) + eap + return eth("01:80:c2:00:00:03", src, 0x888E, body) + + +def stp(src): + bpdu = b"\x00\x00\x00\x00\x00" + b"\x80\x00" + mac(src) + b"\x00" * 23 + llc = b"\x42\x42\x03" + bpdu + return mac("01:80:c2:00:00:00") + mac(src) + \ + struct.pack("!H", len(llc)) + llc + + +def arp(src, sender_ip="172.16.9.50", target_ip="172.16.9.1"): + body = struct.pack("!HHBBH", 1, 0x0800, 6, 4, 1) + mac(src) + \ + socket.inet_aton(sender_ip) + b"\x00" * 6 + socket.inet_aton(target_ip) + return eth("ff:ff:ff:ff:ff:ff", src, 0x0806, body) + + +def _dns_encode(names): + out = b"" + for n in names: + for label in n.split("."): + out += bytes([len(label)]) + label.encode() + out += b"\x00" + return out + + +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 dhcp_reply(src, server_ip, xid, chaddr, yiaddr="10.99.0.50", + domain=None, search=None, msgtype=2, server_id=None, + search_raw=None): + """DHCPOFFER (msgtype 2) or ACK (5), broadcast.""" + sid = server_ip if server_id is None else server_id + bootp = struct.pack("!BBBBIHH", 2, 1, 6, 0, xid, 0, 0x8000) + bootp += b"\x00" * 4 + socket.inet_aton(yiaddr) + \ + socket.inet_aton(server_ip) + b"\x00" * 4 + bootp += mac(chaddr) + b"\x00" * 10 + b"\x00" * 64 + b"\x00" * 128 + opts = b"\x63\x82\x53\x63" + bytes([53, 1, msgtype]) + if sid: + opts += bytes([54, 4]) + socket.inet_aton(sid) + opts += bytes([1, 4]) + socket.inet_aton("255.255.255.0") + opts += bytes([51, 4]) + struct.pack("!I", 3600) + if domain: + opts += bytes([15, len(domain)]) + domain.encode() + if search or search_raw: + raw = search_raw if search_raw is not None else _dns_encode(search) + opts += bytes([119, len(raw)]) + raw + opts += b"\xff" + bootp += opts + if len(bootp) < 300: + bootp += b"\x00" * (300 - len(bootp)) + udp = struct.pack("!HHHH", 67, 68, 8 + len(bootp), 0) + bootp + ip = struct.pack("!BBHHHBBH4s4s", 0x45, 0, 20 + len(udp), 1, 0, 64, 17, + 0, socket.inet_aton(server_ip), + socket.inet_aton("255.255.255.255")) + ip = ip[:10] + struct.pack("!H", _ip_checksum(ip)) + ip[12:] + return eth("ff:ff:ff:ff:ff:ff", src, 0x0800, ip + udp) + + +def parse_discover(frame): + """(xid, chaddr) if frame is a DHCPDISCOVER, else None.""" + if len(frame) < 14 + 20 + 8 + 240 or frame[12:14] != b"\x08\x00": + return None + ip = frame[14:] + ihl = (ip[0] & 0x0F) * 4 + if ip[9] != 17: + return None + sport, dport = struct.unpack("!HH", ip[ihl:ihl + 4]) + if (sport, dport) != (68, 67): + return None + b = ip[ihl + 8:] + if b[0] != 1 or b[236:240] != b"\x63\x82\x53\x63": + return None + opts = b[240:] + i = 0 + mtype = None + while i < len(opts) and opts[i] != 255: + if opts[i] == 0: + i += 1 + continue + if opts[i] == 53: + mtype = opts[i + 2] + i += 2 + opts[i + 1] + if mtype != 1: + return None + xid = struct.unpack("!I", b[4:8])[0] + chaddr = ":".join("%02x" % x for x in b[28:34]) + return xid, chaddr diff --git a/tests/pxe-nic-check/lab.py b/tests/pxe-nic-check/lab.py new file mode 100755 index 0000000..a57bbd0 --- /dev/null +++ b/tests/pxe-nic-check/lab.py @@ -0,0 +1,313 @@ +#!/usr/bin/env python3 +""" +lab.py - run the real pxe-nic-check against fake networks, no root needed. + +Re-executes itself inside a private user + network + mount namespace +(unshare --user --map-root-user --net --mount), so every frame stays on +veth pairs that exist only in that namespace. Nothing touches a real NIC. + +Topology inside the namespace: + + v0 (172.16.9.1/24, "imaging NIC") <-veth-> v1 (lab_peer.py / dnsmasq) + g0 (10.200.0.7/24, "GE NIC") <-veth-> g1 (quiet), plus a fake + networkd lease for g0 + br0 (172.16.9.1/24, bridge) port b0 <-veth-> b1 (lab_peer.py), the + br-pxe layout of the live box + +Each scenario starts a peer on v1, runs the script in check (--json), +--guard and --watch modes, and asserts the verdict, the guard exit code, +whether --watch called its stop command, and the status JSON shape. + + python3 tests/pxe-nic-check/lab.py # all scenarios + python3 tests/pxe-nic-check/lab.py rogue # one scenario +""" + +import json +import os +import shutil +import subprocess +import sys +import tempfile +import time + +HERE = os.path.dirname(os.path.abspath(__file__)) +SCRIPT = os.path.normpath(os.path.join( + HERE, "..", "..", "playbook", "pxe-server-helpers", "pxe-nic-check")) +PEER = os.path.join(HERE, "lab_peer.py") + +ENTRY_KEYS = ["iface", "mac", "role", "link", "speed_mbps", "addrs", + "verdict", "evidence", "dhcp_servers", "switch", "arp_hosts", + "dnsmasq_bound"] + +# name, peer (None = no peer, "down" = cable pulled, "dnsmasq:" = +# real dnsmasq on v1 as a foreign server, "own-dnsmasq" = real dnsmasq on +# v0 as our own server, else lab_peer.py args), expected imaging verdict, +# expected guard rc, evidence substring, expected dnsmasq_bound on v0 +SCENARIOS = [ + ("isolated", None, "isolated", 0, "no DHCP server, no switch", False), + ("rogue", ["--dhcp", "10.99.0.1"], "foreign-dhcp", 1, + "DHCP from 10.99.0.1, no domain"), + ("fake-ge", ["--dhcp", "10.134.48.1", "--domain", "test.geaerospace.net", + "--lldp", "WJ-IDF2"], "ge", 1, + "DHCP from 10.134.48.1, domain test.geaerospace.net"), + ("not-corp-domain", ["--dhcp", "10.99.0.1", "--domain", + "notgeaerospace.net"], "foreign-dhcp", 1, + "domain notgeaerospace.net"), + ("lldp-only", ["--lldp", "WJ-IDF2"], "ge", 1, + "LLDP from WJ-IDF2 port Gi1/0/14"), + ("eapol-only", ["--eapol"], "ge", 1, "EAPOL"), + ("own-server", ["--dhcp", "172.16.9.1"], "isolated", 0, + "own DHCP server 172.16.9.1 answered (ignored)"), + ("cloned-pxe", ["--dhcp", "172.16.9.1", "--mac", "02:cc:00:00:00:01"], + "foreign-dhcp", 1, "CLAIMS OUR ADDRESS"), + ("no-link", "down", "no-link", 0, "no carrier"), + ("dnsmasq-corp", "dnsmasq:lab.geaerospace.net", "ge", 1, + "domain lab.geaerospace.net"), + ("dnsmasq-rogue", "dnsmasq:", "foreign-dhcp", 1, "no domain"), + # our own dnsmasq serving v0 never hears our own probe and must not + # trip the guard; dnsmasq_bound must say it is serving v0 + ("own-dnsmasq", "own-dnsmasq", "isolated", 0, "no DHCP server", True), +] + +# Same idea with the imaging NIC as a bridge (br0, port b0 <-veth-> b1), +# like br-pxe on the live server, and nic.conf naming it by a PORT's MAC. +# LLDP/EAPOL stop at the bridge port and never reach br0, so these prove +# the script also listens on the bridge's ports. +BRIDGE_SCENARIOS = [ + ("br-isolated", None, "isolated", 0, "no DHCP server, no switch"), + ("br-lldp", ["--lldp", "WJ-IDF2"], "ge", 1, "LLDP from WJ-IDF2"), + ("br-eapol", ["--eapol"], "ge", 1, "EAPOL"), + ("br-rogue", ["--dhcp", "10.99.0.1"], "foreign-dhcp", 1, + "DHCP from 10.99.0.1, no domain"), + ("br-fake-ge", ["--dhcp", "10.134.48.1", "--domain", + "test.geaerospace.net"], "ge", 1, + "domain test.geaerospace.net"), +] + + +def sh(*cmd): + subprocess.run(cmd, check=True) + + +def setup(tmp): + sh("mount", "-t", "sysfs", "sysfs", "/sys") + sh("ip", "link", "set", "lo", "up") + for a, b in (("v0", "v1"), ("g0", "g1")): + sh("ip", "link", "add", a, "type", "veth", "peer", "name", b) + sh("ip", "addr", "add", "172.16.9.1/24", "dev", "v0") + sh("ip", "addr", "add", "10.200.0.7/24", "dev", "g0") + sh("ip", "link", "add", "b0", "type", "veth", "peer", "name", "b1") + sh("ip", "link", "add", "br0", "type", "bridge", "stp_state", "0") + sh("ip", "link", "set", "b0", "master", "br0") + sh("ip", "addr", "add", "172.16.9.1/24", "dev", "br0") + for i in ("v0", "v1", "g0", "g1", "b0", "b1", "br0"): + sh("ip", "link", "set", i, "up") + leases = os.path.join(tmp, "leases") + os.mkdir(leases) + with open("/sys/class/net/g0/ifindex") as f: + idx = f.read().strip() + with open(os.path.join(leases, idx), "w") as f: + f.write("ADDRESS=10.200.0.7\nSERVER_ADDRESS=10.200.0.1\n" + "DOMAINNAME=test.geaerospace.net\n") + conf = os.path.join(tmp, "nic.conf") + with open(conf, "w") as f: + f.write("[nic]\nimaging_iface = v0\nge_iface = g0\n") + with open("/sys/class/net/b0/address") as f: + b0_mac = f.read().strip() + conf_br = os.path.join(tmp, "nic-bridge.conf") + with open(conf_br, "w") as f: + f.write("# lab\nimaging_mac = %s\nge_mac = %s\n" % ( + b0_mac.upper(), open("/sys/class/net/g0/address").read().strip())) + time.sleep(1.5) # let carrier settle + return conf, conf_br, leases + + +def start_peer(spec, tmp, peer_if="v1"): + env = dict(os.environ, PXE_NIC_LAB="1") + if spec is None: + return None + if spec == "down": + sh("ip", "link", "set", peer_if, "down") + time.sleep(0.5) + return None + if isinstance(spec, str) and (spec == "own-dnsmasq" + or spec.startswith("dnsmasq:")): + if not shutil.which("dnsmasq"): + raise Skip("dnsmasq not installed") + if spec == "own-dnsmasq": + domain, dev, rng = "", "v0", "172.16.9.10,172.16.9.100,1h" + else: + domain, dev, rng = spec.split(":", 1)[1], "v1", \ + "10.99.0.10,10.99.0.20,1h" + sh("ip", "addr", "replace", "10.99.0.1/24", "dev", "v1") + # --user= --group= : skip the privilege drop, which a user + # namespace with setgroups denied cannot do + cmd = ["dnsmasq", "-k", "-C", "/dev/null", "--port=0", + "--user=", "--group=", "--interface=" + dev, + "--bind-interfaces", "--dhcp-range=" + rng, + "--dhcp-leasefile=" + os.path.join(tmp, "dm.leases"), + "--pid-file=" + os.path.join(tmp, "dm.pid"), + "--log-facility=-", "--log-dhcp"] + if domain: + cmd.append("--dhcp-option=15," + domain) + log = open(os.path.join(tmp, "dnsmasq.log"), "w") + p = subprocess.Popen(cmd, stdout=log, stderr=subprocess.STDOUT) + time.sleep(1.0) + if p.poll() is not None: + stop_peer(None, spec) + with open(os.path.join(tmp, "dnsmasq.log")) as f: + raise Skip("dnsmasq would not start in the userns: " + + f.read().strip()[-300:]) + return p + p = subprocess.Popen([sys.executable, PEER, "--iface", peer_if] + spec, + stdout=subprocess.PIPE, text=True, env=env) + line = p.stdout.readline() + if line.strip() != "ready": + raise RuntimeError("peer did not start: %r" % line) + return p + + +def stop_peer(p, spec): + if p is not None: + p.terminate() + try: + p.wait(5) + except subprocess.TimeoutExpired: + p.kill() + if spec == "down": + sh("ip", "link", "set", "v1", "up") + time.sleep(1.0) + if isinstance(spec, str) and spec.startswith("dnsmasq:"): + subprocess.run(["ip", "addr", "del", "10.99.0.1/24", "dev", "v1"], + stderr=subprocess.DEVNULL) + + +class Skip(Exception): + pass + + +def run_script(args, conf, leases): + cmd = [sys.executable, SCRIPT, "--config", conf, "--dnsmasq-nic-conf", + "/nonexistent", "--lease-dir", leases] + args + return subprocess.run(cmd, capture_output=True, text=True, timeout=120) + + +def check_shape(st): + assert sorted(st) == ["checked_at", "dhcp_guard", "duration_s", + "interfaces"], sorted(st) + assert isinstance(st["duration_s"], int) + assert st["dhcp_guard"]["state"] in ("allowed", "blocked", "unknown") + assert sorted(st["dhcp_guard"]) == ["reason", "state"] + for e in st["interfaces"]: + assert sorted(e) == sorted(ENTRY_KEYS), sorted(e) + assert sorted(e["switch"]) == ["cdp", "eapol", "lldp", "stp"] + for d in e["dhcp_servers"]: + assert sorted(d) == ["domain", "own", "server"], d + for x in e["switch"]["lldp"]: + assert sorted(x) == ["port", "system", "vlan"], x + for x in e["switch"]["cdp"]: + assert sorted(x) == ["device", "port"], x + + +def scenario(name, spec, want, want_rc, needle, bound, conf, leases, tmp, + im_if="v0", peer_if="v1"): + p = start_peer(spec, tmp, peer_if) + fails = [] + try: + status = os.path.join(tmp, name + ".json") + # 1. check mode with probe, short window + r = run_script(["--json", "--probe", "--window", "3"], conf, leases) + st = json.loads(r.stdout) + check_shape(st) + by = {e["iface"]: e for e in st["interfaces"]} + im = by[im_if] + if im["role"] != "imaging" or im["verdict"] != want: + fails.append("check: imaging verdict %s (want %s) %s" + % (im["verdict"], want, im["evidence"])) + if not any(needle in ev for ev in im["evidence"]): + fails.append("check: evidence lacks %r: %s" + % (needle, im["evidence"])) + if bound is not None and im["dnsmasq_bound"] != bound: + fails.append("check: v0 dnsmasq_bound %s (want %s)" + % (im["dnsmasq_bound"], bound)) + ge = by.get("g0") + if not ge or ge["role"] != "ge" or ge["verdict"] != "ge": + fails.append("check: g0 not ge: %s" % ge) + # 2. guard + r = run_script(["--guard", "--window", "2", "--write", status], + conf, leases) + if "no BPF filter" in r.stderr: + fails.append("BPF filter did not attach: %s" % r.stderr.strip()) + if r.returncode != want_rc: + fails.append("guard rc %d (want %d): %s" + % (r.returncode, want_rc, r.stderr.strip())) + with open(status) as f: + gst = json.load(f) + check_shape(gst) + if os.stat(status).st_mode & 0o777 != 0o644: + fails.append("status file not 0644") + want_state = "blocked" if want_rc else "allowed" + if gst["dhcp_guard"]["state"] != want_state: + fails.append("guard state %s" % gst["dhcp_guard"]) + # 3. watch, with a marker file standing in for systemctl stop + marker = os.path.join(tmp, name + ".stopped") + r = run_script(["--watch", "--window", "3", "--write", status, + "--stop-cmd", "touch " + marker], conf, leases) + stopped = os.path.exists(marker) + if stopped != bool(want_rc): + fails.append("watch stop called=%s (want %s): %s" + % (stopped, bool(want_rc), r.stderr.strip())) + if want_rc and "<2>" not in r.stderr and "CRIT" not in r.stderr: + fails.append("watch did not log CRIT") + finally: + stop_peer(p, spec) + return fails, im + + +def inside(selected): + tmp = tempfile.mkdtemp(prefix="pxe-nic-lab-") + conf, conf_br, leases = setup(tmp) + results = [] + runs = [(sc, conf, "v0", "v1") for sc in SCENARIOS] + \ + [(sc, conf_br, "br0", "b1") for sc in BRIDGE_SCENARIOS] + for sc, sconf, im_if, peer_if in runs: + name, spec, want, rc, needle = sc[:5] + bound = sc[5] if len(sc) > 5 else None + if selected and name not in selected: + continue + t0 = time.monotonic() + try: + fails, im = scenario(name, spec, want, rc, needle, bound, sconf, + leases, tmp, im_if, peer_if) + except Skip as e: + results.append((name, "SKIP", str(e))) + print("SKIP %-16s %s" % (name, e), flush=True) + continue + except Exception as e: # report and carry on + fails, im = ["exception: %r" % e], None + state = "FAIL" if fails else "PASS" + results.append((name, state, fails)) + print("%s %-16s %-13s guard_rc=%d (%.0f s)" % ( + state, name, want, rc, time.monotonic() - t0), flush=True) + for f in fails: + print(" " + f, flush=True) + if im and not fails: + print(" " + " | ".join(im["evidence"]), flush=True) + n_pass = sum(1 for r in results if r[1] == "PASS") + n_fail = sum(1 for r in results if r[1] == "FAIL") + n_skip = sum(1 for r in results if r[1] == "SKIP") + print("\nlab: %d passed, %d failed, %d skipped" % (n_pass, n_fail, n_skip)) + return 1 if n_fail else 0 + + +def main(): + args = sys.argv[1:] + if args[:1] == ["--inside"]: + sys.exit(inside(args[1:])) + cmd = ["unshare", "--user", "--map-root-user", "--net", "--mount", + sys.executable, os.path.abspath(__file__), "--inside"] + args + sys.exit(subprocess.run(cmd).returncode) + + +if __name__ == "__main__": + main() diff --git a/tests/pxe-nic-check/lab_peer.py b/tests/pxe-nic-check/lab_peer.py new file mode 100755 index 0000000..842009f --- /dev/null +++ b/tests/pxe-nic-check/lab_peer.py @@ -0,0 +1,81 @@ +#!/usr/bin/env python3 +""" +lab_peer.py - the "other end of the cable" for the namespace lab. + +Runs on one end of a veth pair inside a private network namespace and +plays a switch and/or a DHCP server with raw AF_PACKET sockets: + + --dhcp IP answer every DHCPDISCOVER with a DHCPOFFER from IP + --server-id IP option 54 to put in the offer (default = --dhcp IP) + --domain NAME option 15 in the offer + --lldp SYSNAME send an LLDP frame every second + --eapol send an EAPOL Request/Identity every second + --mac MAC source MAC for the frames (default: the iface's MAC) + +Never run this outside the lab netns: it is a rogue DHCP server. +""" + +import argparse +import os +import select +import socket +import sys +import time + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import frames as F # noqa: E402 + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--iface", required=True) + ap.add_argument("--dhcp") + ap.add_argument("--server-id") + ap.add_argument("--domain") + ap.add_argument("--lldp") + ap.add_argument("--eapol", action="store_true") + ap.add_argument("--mac") + ap.add_argument("--duration", type=float, default=120) + args = ap.parse_args() + + if os.environ.get("PXE_NIC_LAB") != "1": + sys.exit("lab_peer.py: refusing to run outside the lab (PXE_NIC_LAB)") + + with open("/sys/class/net/%s/address" % args.iface) as f: + mac = args.mac or f.read().strip() + s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(3)) + s.bind((args.iface, 0)) + end = time.monotonic() + args.duration + next_beacon = 0.0 + answered = set() + sys.stdout.write("ready\n") + sys.stdout.flush() + while time.monotonic() < end: + now = time.monotonic() + if now >= next_beacon: + if args.lldp: + s.send(F.lldp(mac, system=args.lldp, port="Gi1/0/14", + vlan=309)) + if args.eapol: + s.send(F.eapol(mac)) + next_beacon = now + 1.0 + r, _, _ = select.select([s], [], [], 0.2) + if not r: + continue + frame, addr = s.recvfrom(65535) + if addr[2] == 4 or not args.dhcp: # our own outgoing frames + continue + d = F.parse_discover(frame) + if not d or d[0] in answered: + continue + answered.add(d[0]) + xid, chaddr = d + s.send(F.dhcp_reply(mac, args.dhcp, xid, chaddr, + yiaddr="10.99.0.50", domain=args.domain, + server_id=args.server_id)) + sys.stdout.write("offered xid %08x to %s\n" % (xid, chaddr)) + sys.stdout.flush() + + +if __name__ == "__main__": + main() diff --git a/tests/pxe-nic-check/test_pxe_nic_check.py b/tests/pxe-nic-check/test_pxe_nic_check.py new file mode 100644 index 0000000..fc57a4b --- /dev/null +++ b/tests/pxe-nic-check/test_pxe_nic_check.py @@ -0,0 +1,499 @@ +""" +Unit tests for playbook/pxe-server-helpers/pxe-nic-check. No root, no +network: frames are synthesized by frames.py and fed to the parsers. + + python3 -m unittest discover -s tests/pxe-nic-check -v +""" + +import json +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import frames as F # noqa: E402 + +M = F.load_script() + +SW = "00:11:22:33:44:55" # "switch" MAC +ME = "02:00:00:00:00:01" # this host +SUFFIXES = M.parse_suffixes(M.DEFAULT_SUFFIXES) + + +def fed(*frs, own=(ME,), xid=None): + o = M.Observations() + for fr in frs: + o.feed(fr, set(own), xid) + return o + + +class DomainMatch(unittest.TestCase): + def test_table(self): + cases = { + "wjs.geaerospace.net": True, + "xyz.geaerospace.net": True, + "geaerospace.net": True, + "GEAEROSPACE.NET.": True, + "notgeaerospace.net": False, + "geaerospace.net.evil.com": False, + "rd.ds.ge.com": True, + "ge.com": True, + "huge.com": False, + "": False, + None: False, + } + for dom, want in cases.items(): + self.assertEqual(M.domain_matches(dom, SUFFIXES), want, dom) + + def test_suffix_parsing(self): + self.assertEqual(M.parse_suffixes(" geaerospace.net, .ge.com ;x.org,"), + ["geaerospace.net", "ge.com", "x.org"]) + + +class Frames(unittest.TestCase): + def test_lldp(self): + o = fed(F.lldp(SW, system="WJ-IDF2", port="Gi1/0/14", vlan=309)) + (info,) = o.lldp.values() + self.assertEqual(info["system"], "WJ-IDF2") + self.assertEqual(info["port"], "Gi1/0/14") + self.assertEqual(info["chassis"], SW) + self.assertEqual(info["vlan"], 309) + + def test_lldp_no_sysname_no_vlan(self): + o = fed(F.lldp(SW, system=None, port="7")) + (info,) = o.lldp.values() + self.assertIsNone(info["system"]) + self.assertIsNone(info["vlan"]) + + def test_lldp_vlan_tagged_frame(self): + fr = F.lldp(SW) + tagged = fr[:12] + b"\x81\x00\x01\x35" + fr[12:] + self.assertEqual(len(fed(tagged).lldp), 1) + + def test_cdp(self): + o = fed(F.cdp(SW, device="WJ-CORE1", port="Gi1/0/3", vlan=12)) + (info,) = o.cdp.values() + self.assertEqual((info["device"], info["port"], info["vlan"]), + ("WJ-CORE1", "Gi1/0/3", 12)) + self.assertFalse(o.stp) + + def test_eapol(self): + o = fed(F.eapol(SW)) + self.assertTrue(o.eapol) + + def test_stp(self): + o = fed(F.stp(SW)) + self.assertTrue(o.stp) + self.assertFalse(o.lldp or o.cdp or o.eapol) + + def test_arp_counts_distinct_and_skips_own(self): + o = fed(F.arp("02:aa:00:00:00:01"), F.arp("02:aa:00:00:00:01"), + F.arp("02:aa:00:00:00:02"), F.arp(ME)) + self.assertEqual(len(o.arp_macs), 2) + + def test_truncated_garbage_does_not_raise(self): + for junk in (b"", b"\x00" * 13, F.lldp(SW)[:20], F.cdp(SW)[:30], + F.dhcp_reply(SW, "10.1.1.1", 1, ME)[:100]): + fed(junk) + + +class Dhcp(unittest.TestCase): + def offer(self, **kw): + return F.dhcp_reply(SW, "10.134.48.1", 42, ME, **kw) + + def test_offer_domains(self): + for dom in ("wjs.geaerospace.net", "xyz.geaerospace.net", + "notgeaerospace.net", "rd.ds.ge.com", None): + o = fed(self.offer(domain=dom), xid=42) + (d,) = o.dhcp.values() + self.assertEqual(d["domain"], dom) + self.assertEqual(d["server"], "10.134.48.1") + self.assertEqual(d["yiaddr"], "10.99.0.50") + self.assertEqual(d["how"], "probe") + + def test_passive_ack_is_recorded(self): + o = fed(self.offer(msgtype=5), xid=7) + (d,) = o.dhcp.values() + self.assertEqual(d["how"], "passive") + + def test_option_119_with_compression(self): + # "a.geaerospace.net" then "b" + pointer to "geaerospace.net" (off 2) + raw = F._dns_encode(["a.geaerospace.net"]) + raw += b"\x01b\xc0\x02" + o = fed(self.offer(search_raw=raw)) + (d,) = o.dhcp.values() + self.assertEqual(d["search"], ["a.geaerospace.net", "b.geaerospace.net"]) + + def test_option_119_plain(self): + o = fed(self.offer(search=["rd.ds.ge.com", "x.example"])) + (d,) = o.dhcp.values() + self.assertEqual(d["search"], ["rd.ds.ge.com", "x.example"]) + + def test_missing_server_id_uses_source_ip(self): + o = fed(self.offer(server_id="")) + (d,) = o.dhcp.values() + self.assertEqual(d["server"], "10.134.48.1") + + def test_discover_roundtrip(self): + fr = M.build_discover(F.mac(ME), 0xDEADBEEF) + self.assertEqual(F.parse_discover(fr), (0xDEADBEEF, ME)) + # IPv4 header checksum verifies to zero + self.assertEqual(M._ip_checksum(fr[14:34]), 0) + # our own discover is not mistaken for a server + self.assertEqual(fed(fr).dhcp, {}) + + +def run_bpf(insns, pkt): + """Minimal classic-BPF interpreter for the opcodes pxe-nic-check uses. + Out-of-bounds loads reject, as in the kernel.""" + import struct + A = X = 0 + pc = 0 + while True: + code, jt, jf, k = insns[pc] + pc += 1 + if code == 0x28: # ldh [k] + if k + 2 > len(pkt): + return 0 + A = struct.unpack("!H", pkt[k:k + 2])[0] + elif code == 0x30: # ldb [k] + if k + 1 > len(pkt): + return 0 + A = pkt[k] + elif code == 0x48: # ldh [x + k] + if X + k + 2 > len(pkt): + return 0 + A = struct.unpack("!H", pkt[X + k:X + k + 2])[0] + elif code == 0xB1: # ldxb 4*([k]&0xf) + if k + 1 > len(pkt): + return 0 + X = 4 * (pkt[k] & 0x0F) + elif code == 0x15: # jeq #k + pc += jt if A == k else jf + elif code == 0x45: # jset #k + pc += jt if A & k else jf + elif code == 0x06: # ret #k + return k + else: + raise AssertionError("opcode %#x" % code) + + +class Bpf(unittest.TestCase): + def test_filter_passes_what_we_parse_and_drops_bulk(self): + insns = M.assemble_bpf() + keep = { + "lldp": F.lldp(SW), "cdp": F.cdp(SW), "eapol": F.eapol(SW), + "stp": F.stp(SW), "arp": F.arp(SW), + "offer": F.dhcp_reply(SW, "10.1.1.1", 1, ME), + "tagged lldp": F.lldp(SW)[:12] + b"\x81\x00\x00\x05" + + F.lldp(SW)[12:], + } + for name, fr in keep.items(): + self.assertTrue(run_bpf(insns, fr), name) + # TCP/445 SMB, our own DISCOVER (sport 68), IPv6, short junk + import struct + ip = bytearray(F.dhcp_reply(SW, "10.1.1.1", 1, ME)) + smb = bytes(ip[:23]) + b"\x06" + bytes(ip[24:34]) + \ + struct.pack("!HH", 445, 50000) + bytes(ip[38:]) + drop = { + "smb": smb, + "discover": M.build_discover(F.mac(ME), 5), + "ipv6": F.eth("33:33:00:00:00:01", SW, 0x86DD, b"\x60" + + b"\x00" * 60), + "junk": b"\x00" * 10, + } + for name, fr in drop.items(): + self.assertFalse(run_bpf(insns, fr), name) + + +class OwnServer(unittest.TestCase): + def test_own_needs_own_ip_and_local_mac(self): + o = M.Observations() + o.feed(F.dhcp_reply(ME, "172.16.9.1", 1, "02:bb:00:00:00:09", + yiaddr="172.16.9.20"), set(), 1) + lst = M.dhcp_list(o, {"172.16.9.1"}, {ME}) + self.assertTrue(lst[0]["own"]) + v, ev = M.classify(True, None, o, lst, SUFFIXES, 60, True, False) + self.assertEqual(v, "isolated") + self.assertTrue(any("own DHCP server 172.16.9.1" in e for e in ev)) + + def test_cloned_pxe_server_is_foreign(self): + o = M.Observations() + o.feed(F.dhcp_reply("02:cc:00:00:00:01", "172.16.9.1", 1, ME, + yiaddr="172.16.9.20"), set(), 1) + lst = M.dhcp_list(o, {"172.16.9.1"}, {ME}) + self.assertFalse(lst[0]["own"]) + v, ev = M.classify(True, None, o, lst, SUFFIXES, 60, True, False) + self.assertEqual(v, "foreign-dhcp") + self.assertTrue(any("CLAIMS OUR ADDRESS" in e for e in ev)) + + +class Verdicts(unittest.TestCase): + def run_case(self, frs, link=True, error=None, own_ips=()): + o = fed(*frs) if frs is not None else None + lst = M.dhcp_list(o, set(own_ips), {ME}) + return M.classify(link, error, o, lst, SUFFIXES, 60, True, False) + + def test_table(self): + def offer(dom): + return F.dhcp_reply(SW, "10.134.48.1", 1, ME, domain=dom) + table = [ + ("quiet", [], "isolated"), + ("arp only", [F.arp("02:aa:00:00:00:05")], "isolated"), + ("stp only", [F.stp(SW)], "isolated"), + ("lldp", [F.lldp(SW)], "ge"), + ("cdp", [F.cdp(SW)], "ge"), + ("eapol", [F.eapol(SW)], "ge"), + ("dhcp wjs", [offer("wjs.geaerospace.net")], "ge"), + ("dhcp xyz", [offer("xyz.geaerospace.net")], "ge"), + ("dhcp rd.ds.ge.com", [offer("rd.ds.ge.com")], "ge"), + ("dhcp notgeaerospace", [offer("notgeaerospace.net")], + "foreign-dhcp"), + ("dhcp no domain", [offer(None)], "foreign-dhcp"), + ("rogue + lldp", [offer(None), F.lldp(SW)], "ge"), + ("search list only", + [F.dhcp_reply(SW, "10.1.1.1", 1, ME, + search=["wjs.geaerospace.net"])], "ge"), + ] + for name, frs, want in table: + v, ev = self.run_case(frs) + self.assertEqual(v, want, "%s: %s" % (name, ev)) + + def test_allowed_imaging_switch(self): + o = fed(F.lldp(SW, system="IMG-SW1"), F.cdp(SW, device="IMG-SW1")) + v, ev = M.classify(True, None, o, [], SUFFIXES, 60, True, False, + M.allowed_list("IMG-SW1")) + self.assertEqual(v, "isolated", ev) + self.assertTrue(any("allowed imaging switch" in e for e in ev)) + # by chassis MAC, any case/separator + o = fed(F.lldp(SW, system=None)) + cfg_allowed = M.allowed_list("00-11-22-33-44-55") + v, _ = M.classify(True, None, o, [], SUFFIXES, 60, True, False, + cfg_allowed) + self.assertEqual(v, "isolated") + # another switch, or EAPOL, still counts + o = fed(F.lldp(SW, system="IMG-SW1"), + F.lldp("02:00:00:00:00:99", system="WJ-IDF2", + chassis_mac="02:00:00:00:00:99")) + v, _ = M.classify(True, None, o, [], SUFFIXES, 60, True, False, + M.allowed_list("IMG-SW1")) + self.assertEqual(v, "ge") + o = fed(F.lldp(SW, system="IMG-SW1"), F.eapol(SW)) + v, _ = M.classify(True, None, o, [], SUFFIXES, 60, True, False, + M.allowed_list("IMG-SW1")) + self.assertEqual(v, "ge") + + def test_no_link_and_error(self): + self.assertEqual(self.run_case([], link=False)[0], "no-link") + self.assertEqual(self.run_case([], error="boom")[0], "error") + + def test_evidence_strings(self): + _, ev = self.run_case([F.dhcp_reply(SW, "10.134.48.1", 1, ME, + domain="wjs.geaerospace.net")]) + self.assertTrue(any(e.startswith( + "DHCP from 10.134.48.1, domain wjs.geaerospace.net") for e in ev)) + _, ev = self.run_case([F.lldp(SW, system="WJ-IDF2", port="Gi1/0/14")]) + self.assertIn("LLDP from WJ-IDF2 port Gi1/0/14", ev) + _, ev = self.run_case([]) + self.assertIn("no DHCP server, no switch announcements in 60 s", ev) + + def test_guard_state(self): + e = {"iface": "br-pxe", "verdict": "foreign-dhcp", + "evidence": ["DHCP from 10.0.0.1, no domain", "3 ARP hosts seen"]} + st, why = M.guard_state(e) + self.assertEqual(st, "blocked") + self.assertNotIn("ARP", why) + for v, want in (("ge", "blocked"), ("isolated", "allowed"), + ("no-link", "allowed"), ("error", "unknown")): + e["verdict"] = v + self.assertEqual(M.guard_state(e)[0], want, v) + self.assertEqual(M.guard_state(None)[0], "unknown") + + +class Lease(unittest.TestCase): + SAMPLE = """# This is private data. Do not parse. +ADDRESS=10.134.49.77 +NETMASK=255.255.252.0 +ROUTER=10.134.48.1 +SERVER_ADDRESS=10.134.48.10 +NEXT_SERVER=0.0.0.0 +DNS=10.134.48.11 10.134.48.12 +DOMAINNAME=wjs.geaerospace.net +DOMAIN_SEARCH_LIST=wjs.geaerospace.net rd.ds.ge.com +LIFETIME=86400 +""" + + def test_parse(self): + d = M.parse_lease(self.SAMPLE) + self.assertEqual(d["SERVER_ADDRESS"], "10.134.48.10") + self.assertEqual(d["DOMAINNAME"], "wjs.geaerospace.net") + lst = M.lease_servers(d) + self.assertEqual(lst[0]["server"], "10.134.48.10") + self.assertEqual(lst[0]["search"], + ["wjs.geaerospace.net", "rd.ds.ge.com"]) + v, ev = M.classify(True, None, None, lst, SUFFIXES, 0, False, True) + self.assertEqual(v, "ge") + self.assertTrue(ev[0].startswith( + "DHCP from 10.134.48.10, domain wjs.geaerospace.net")) + + def test_lease_without_domain(self): + lst = M.lease_servers(M.parse_lease("ADDRESS=192.168.1.5\n" + "SERVER_ADDRESS=192.168.1.1\n")) + v, _ = M.classify(True, None, None, lst, SUFFIXES, 0, False, True) + self.assertEqual(v, "foreign-dhcp") + + def test_lease_falls_back_to_router(self): + lst = M.lease_servers({"ROUTER": "10.0.0.1"}) + self.assertEqual(lst[0]["server"], "10.0.0.1") + + def test_read_lease_missing(self): + self.assertIsNone(M.read_lease("no-such-iface-xyz", "/nonexistent")) + + +class Config(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp() + + def path(self, name, text): + p = os.path.join(self.tmp, name) + with open(p, "w") as f: + f.write(text) + return p + + def test_defaults(self): + cfg = M.load_config(os.path.join(self.tmp, "missing"), + os.path.join(self.tmp, "missing2")) + self.assertEqual(cfg["own_ip"], "172.16.9.1") + self.assertEqual(cfg["suffixes"], ["geaerospace.net", "ge.com"]) + self.assertEqual(cfg["dnsmasq_ifaces"], []) + + def test_sectionless_ini_and_mac_normalising(self): + p = self.path("nic.conf", "imaging_mac = AA-BB-CC-00-11-22\n" + "ge_iface = enp1s0 # corp\n" + "corp_domain_suffixes = example.org\n") + cfg = M.load_config(p, "/nonexistent") + self.assertEqual(cfg["imaging_mac"], "aa:bb:cc:00:11:22") + self.assertEqual(cfg["ge_iface"], "enp1s0") + self.assertEqual(cfg["suffixes"], ["example.org"]) + + def test_allowed_switches(self): + p = self.path("nic.conf", "allowed_switches = IMG-SW1, 00-11-22-33-44-55\n") + cfg = M.load_config(p, "/nonexistent") + self.assertIn("img-sw1", cfg["allowed"]) + self.assertIn("00:11:22:33:44:55", cfg["allowed"]) + + def test_section_ini(self): + p = self.path("nic.conf", "# managed by ansible\n[nic]\n" + "imaging_iface = br-pxe\n" + "own_ip = 172.16.9.2\n") + cfg = M.load_config(p, "/nonexistent") + self.assertEqual(cfg["imaging_iface"], "br-pxe") + self.assertEqual(cfg["own_ip"], "172.16.9.2") + + def test_empty_values_keep_defaults(self): + p = self.path("nic.conf", "corp_domain_suffixes =\nimaging_mac =\n") + cfg = M.load_config(p, "/nonexistent") + self.assertEqual(cfg["suffixes"], ["geaerospace.net", "ge.com"]) + + def test_dnsmasq_interface_fallback(self): + d = self.path("pxe-nic-only.conf", + "# pin dnsmasq\ninterface=br-pxe\n" + "except-interface=enp1s0\nno-dhcp-interface=enp1s0\n") + cfg = M.load_config(os.path.join(self.tmp, "missing"), d) + self.assertEqual(cfg["dnsmasq_ifaces"], ["br-pxe"]) + im, ge, notes = M.resolve_roles(cfg, ["lo", "br-pxe", "enp1s0"], {}, + "/nonexistent") + self.assertEqual(im, "br-pxe") + + def test_own_ip_fallback(self): + cfg = M.load_config("/nonexistent", "/nonexistent") + im, ge, _ = M.resolve_roles( + cfg, ["lo", "eno1", "enx1"], + {"eno1": ["10.134.49.7/22"], "enx1": ["172.16.9.1/24"]}, + "/nonexistent") + self.assertEqual(im, "enx1") + self.assertIsNone(ge) + + def test_configured_iface_wins(self): + cfg = M.load_config("/nonexistent", "/nonexistent") + cfg["imaging_iface"] = "eno1" + cfg["ge_iface"] = "enx1" + im, ge, _ = M.resolve_roles( + cfg, ["eno1", "enx1"], {"enx1": ["172.16.9.1/24"]}, "/nonexistent") + self.assertEqual((im, ge), ("eno1", "enx1")) + + +class DnsmasqBound(unittest.TestCase): + RULES = {"interface": {"br-pxe"}, "except": {"enp1s0"}, + "no_dhcp": {"enp1s0"}, "listen": {"172.16.9.1"}} + + def test_unknown_and_not_running(self): + self.assertIsNone(M.dnsmasq_bound("x", [], None, self.RULES)) + self.assertFalse(M.dnsmasq_bound("x", [], [], self.RULES)) + self.assertFalse(M.dnsmasq_bound( + "x", [], [("0.0.0.0", None, "dhclient")], self.RULES)) + + def test_device_bound(self): + socks = [("0.0.0.0", "br-pxe", "dnsmasq")] + self.assertTrue(M.dnsmasq_bound("br-pxe", [], socks, self.RULES)) + self.assertFalse(M.dnsmasq_bound("enp1s0", [], socks, self.RULES)) + + def test_wildcard_uses_config(self): + socks = [("0.0.0.0", None, "dnsmasq")] + self.assertTrue(M.dnsmasq_bound("br-pxe", ["172.16.9.1/24"], socks, + self.RULES)) + self.assertFalse(M.dnsmasq_bound("enp1s0", ["10.134.49.7/22"], socks, + self.RULES)) + open_rules = {"interface": set(), "except": set(), "no_dhcp": set(), + "listen": set()} + self.assertTrue(M.dnsmasq_bound("enp1s0", ["10.134.49.7/22"], socks, + open_rules)) + + +class StatusShape(unittest.TestCase): + KEYS = ["iface", "mac", "role", "link", "speed_mbps", "addrs", "verdict", + "evidence", "dhcp_servers", "switch", "arp_hosts", + "dnsmasq_bound"] + + def test_shape_and_atomic_write(self): + entry = {k: None for k in self.KEYS} + st = M.build_status([entry], 5.4, ("allowed", "ok")) + self.assertEqual(sorted(st), ["checked_at", "dhcp_guard", + "duration_s", "interfaces"]) + self.assertEqual(st["duration_s"], 5) + self.assertEqual(st["dhcp_guard"], {"state": "allowed", + "reason": "ok"}) + # ISO 8601 with a UTC offset + self.assertRegex(st["checked_at"], + r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d[+-]\d\d:\d\d$") + d = tempfile.mkdtemp() + p = os.path.join(d, "s.json") + M.write_status(p, st) + self.assertEqual(os.stat(p).st_mode & 0o777, 0o644) + with open(p) as f: + self.assertEqual(json.load(f), st) + self.assertEqual(os.listdir(d), ["s.json"]) + + +class CommandLine(unittest.TestCase): + """main() without root: capture fails, the guard must fail open.""" + + def test_guard_fails_open_when_nothing_resolves(self): + d = tempfile.mkdtemp() + out = os.path.join(d, "st.json") + rc = M.main(["--guard", "--config", "/nonexistent", + "--dnsmasq-nic-conf", "/nonexistent", + "--iface", "lo", "--window", "0.1", "--no-probe", + "--write", out]) + self.assertEqual(rc, 0) + with open(out) as f: + st = json.load(f) + self.assertEqual(st["dhcp_guard"]["state"], "unknown") + (e,) = st["interfaces"] + self.assertEqual(sorted(e), sorted(StatusShape.KEYS)) + self.assertEqual(sorted(e["switch"]), ["cdp", "eapol", "lldp", "stp"]) + + +if __name__ == "__main__": + unittest.main()