""" 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()