#!/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") CARRIER = os.path.normpath(os.path.join( HERE, "..", "..", "playbook", "pxe-server-helpers", "pxe-nic-carrier")) 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"), # a bay PXE-booting on the imaging switch: iPXE EAPOL-Start and a # Windows EAP-Response must not read as a GE switch port ("eapol-client", ["--eapol-start"], "isolated", 0, "client EAPOL frames ignored"), ("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-eapol-client", ["--eapol-start"], "isolated", 0, "client EAPOL frames ignored"), ("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"), ] # Scenarios that also run --watch as if an operator restarted dnsmasq in # the middle of the window: the watch must re-check, and still stop it. RACE = {"rogue", "lldp-only", "br-lldp"} # What `systemctl show` would say: dnsmasq active since boot, or since # just now (inside the watch window). STATE_EARLY = "printf 'ActiveState=active\\nActiveEnterTimestampMonotonic=1\\n'" STATE_LATE = ("printf 'ActiveState=active\\n" "ActiveEnterTimestampMonotonic=%d\\n'" % (2 ** 62)) 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, "--state-cmd", STATE_EARLY], 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") # 4. watch racing an operator restart: re-check, still stop if name in RACE: marker2 = marker + ".race" r = run_script(["--watch", "--window", "3", "--write", status, "--stop-cmd", "touch " + marker2, "--state-cmd", STATE_LATE], conf, leases) if "re-checking" not in r.stderr: fails.append("race: watch did not re-check: %s" % r.stderr.strip()) if not os.path.exists(marker2): fails.append("race: re-check did not stop dnsmasq: %s" % r.stderr.strip()) if "could not join" in r.stderr: fails.append("multicast join failed: %s" % r.stderr.strip()) finally: stop_peer(p, spec) return fails, im def carrier_lab(conf, leases, tmp): """pxe-nic-carrier against real sysfs carrier in the namespace: pull the far end of v0's veth, plug it back, and poke it with GE NIC events. systemctl is a stub that records argv and says dnsmasq is active; pxe-nic-check --print-imaging is the real script.""" fails = [] stub = os.path.join(tmp, "carrier-stub") os.makedirs(stub, exist_ok=True) calls = os.path.join(stub, "calls") with open(os.path.join(stub, "systemctl"), "w") as f: f.write('#!/bin/sh\necho "$*" >> %s\n' '[ "$1" = is-active ] && echo active\nexit 0\n' % calls) with open(os.path.join(stub, "check"), "w") as f: f.write('#!/bin/sh\nexec %s %s --config %s --dnsmasq-nic-conf ' '/nonexistent --lease-dir %s "$@"\n' % (sys.executable, SCRIPT, conf, leases)) with open(os.path.join(stub, "logger"), "w") as f: # not the host log f.write('#!/bin/sh\necho "$*" >> %s/log\n' % stub) for n in ("systemctl", "check", "logger"): os.chmod(os.path.join(stub, n), 0o755) env = dict(os.environ, PATH=stub + ":" + os.environ.get("PATH", ""), PXE_SYSTEMCTL=os.path.join(stub, "systemctl"), PXE_NIC_CHECK=os.path.join(stub, "check"), PXE_NIC_CARRIER_RUN=os.path.join(stub, "run")) def event(iface, state): if os.path.exists(calls): os.unlink(calls) subprocess.run(["bash", CARRIER], env=dict(env, IFACE=iface, STATE=state), timeout=30, check=True) if not os.path.exists(calls): return [] with open(calls) as f: return f.read().splitlines() event("v0", "routable") # first sight: up sh("ip", "link", "set", "v1", "down") time.sleep(0.5) got = event("v0", "no-carrier") if "stop --no-block dnsmasq" not in got: fails.append("carrier loss did not stop dnsmasq: %s" % got) sh("ip", "link", "set", "v0", "down") # admin down: EINVAL got = event("v0", "off") if got: fails.append("repeat loss not idempotent: %s" % got) sh("ip", "link", "set", "v0", "up") sh("ip", "link", "set", "v1", "up") time.sleep(1.5) got = event("v0", "carrier") if "restart --no-block pxe-nic-carrier.service" not in got: fails.append("carrier gain did not hand off: %s" % got) if event("v0", "routable"): fails.append("repeat gain not idempotent") sh("ip", "link", "set", "g1", "down") time.sleep(0.5) got = event("g0", "no-carrier") sh("ip", "link", "set", "g1", "up") if got: fails.append("GE NIC event touched systemctl: %s" % got) time.sleep(1.0) return fails 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) if not selected or "carrier" in selected: t0 = time.monotonic() try: fails = carrier_lab(conf, leases, tmp) except Exception as e: fails = ["exception: %r" % e] state = "FAIL" if fails else "PASS" results.append(("carrier", state, fails)) print("%s %-16s pxe-nic-carrier on real carrier (%.0f s)" % (state, "carrier", time.monotonic() - t0), flush=True) for f in fails: print(" " + f, 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()