webapp: fix review findings (imaging types, NIC status, iPXE gates)
- imaging: coerce client payload and stored session fields to str/int, drop lists/dicts; tiles template guards pctype/current_stage; POST rejects non-object bodies and non-string serials. - nic_status: normalise every check-file entry; load nic.conf as-is with a [nic] fallback; split interface= lists; keep the check's binding while fresh, resolve wildcard sockets from the dnsmasq pin (never True for an excluded NIC); mark vanished NICs down; include role bridges; stale or future-dated files downgrade to warnings and drop evidence. - Overview renders the network pill in its own try and shows 'Network: unknown' when status fails. - ipxe_update: read every Authenticode signature (all WIN_CERTIFICATE entries plus nested SPC_NESTED_SIGNATURE) with openssl; shim swap and install need both Microsoft UEFI CA 2011 and 2023; install re-reads each file (snponly ipxe.efi only); staged.json schema version, older records shown stale and refused; uhci counts as native USB. - backups describe their files; risky rollbacks need a confirm flag. - mac_folders: guarded listdir, cached describe, configured TFTP root, warning for a shim without ipxe.efi. - evidence pack: years 2000..2100, 400 days inclusive. - a11y role=group on quick ranges; progress bar stays full during 'shim'. - tests/test_review_fixes.py: 29 unittest cases incl. both real shims.
This commit is contained in:
cproudlock
452
webapp/tests/test_review_fixes.py
Normal file
452
webapp/tests/test_review_fixes.py
Normal file
@@ -0,0 +1,452 @@
|
||||
"""Regression tests for the 2026-10-15 webapp review findings.
|
||||
|
||||
Run from the webapp folder:
|
||||
PYTHONPATH=.:<deps> python -m unittest discover -s tests -v
|
||||
|
||||
Every config path points into a temp folder (config.py reads env vars), so
|
||||
nothing on the host is read or written. The shim tests use the real boot
|
||||
files in the sibling pxe checkout and skip when they are not there.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tarfile
|
||||
import tempfile
|
||||
import unittest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from unittest import mock
|
||||
|
||||
T = tempfile.mkdtemp(prefix="pxe-webapp-test-")
|
||||
for _k, _sub in {"SAMBA_SHARE": "samba", "CLONEZILLA_SHARE": "cz", "BLANCCO_REPORTS": "br",
|
||||
"ENROLLMENT_SHARE": "enr", "WEB_ROOT": "www", "IMAGING_DIR": "img", "UPDATES_DIR": "upd",
|
||||
"TFTP_ROOT": "tftp", "BIOS_DIR": "bios", "UPLOAD_DIR": "upl", "SAMBA_LOG_DIR": "smblog"}.items():
|
||||
os.environ[_k] = os.path.join(T, _sub)
|
||||
os.makedirs(os.environ[_k], exist_ok=True)
|
||||
for _k in ("AUDIT_LOG", "HARDWARE_STATE", "BLANCCO_STATE", "DNSMASQ_LEASES", "APACHE_ACCESS_LOG",
|
||||
"DNSMASQ_SYSLOG", "NIC_STATUS_FILE", "NIC_CONF", "DNSMASQ_NIC_CONF", "DNSMASQ_CONF"):
|
||||
os.environ[_k] = os.path.join(T, _k.lower())
|
||||
os.makedirs(os.path.join(T, "enr", "ppkgs"), exist_ok=True)
|
||||
|
||||
WEBAPP = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
sys.path.insert(0, WEBAPP)
|
||||
os.chdir(WEBAPP)
|
||||
|
||||
import config # noqa: E402
|
||||
import app as A # noqa: E402
|
||||
from services import imaging_status, ipxe_update, nic_status # noqa: E402
|
||||
|
||||
BOOT_FILES = os.environ.get("PXE_BOOT_FILES", "/home/camp/projects/pxe/boot-files")
|
||||
DUAL_SHIM = os.path.join(BOOT_FILES, "ipxe-shimx64.efi")
|
||||
TARBALL = os.path.join(BOOT_FILES, "ipxeboot.tar.gz")
|
||||
HAVE_TOOLS = bool(shutil.which("openssl")) and bool(shutil.which("sbverify"))
|
||||
|
||||
|
||||
def _extract(member, dest):
|
||||
"""One file from the real iPXE tarball (symlinks followed), or None."""
|
||||
if not os.path.isfile(TARBALL):
|
||||
return None
|
||||
with tarfile.open(TARBALL) as tar:
|
||||
members = {m.name[2:] if m.name.startswith("./") else m.name: m for m in tar.getmembers()}
|
||||
m = ipxe_update._resolve(members, member)
|
||||
if m is None:
|
||||
return None
|
||||
with tar.extractfile(m) as src, open(dest, "wb") as out:
|
||||
shutil.copyfileobj(src, out)
|
||||
return dest
|
||||
|
||||
|
||||
FILES = os.path.join(T, "realfiles")
|
||||
os.makedirs(FILES, exist_ok=True)
|
||||
SHIM_2011 = _extract("ipxeboot/x86_64-sb/ipxe-shim.efi", os.path.join(FILES, "shim2011.efi"))
|
||||
SNPONLY = _extract("ipxeboot/x86_64-sb/snponly.efi", os.path.join(FILES, "snponly.efi"))
|
||||
NATIVE = _extract("ipxeboot/x86_64-sb/ipxe.efi", os.path.join(FILES, "ipxe-native.efi"))
|
||||
|
||||
|
||||
def _client():
|
||||
c = A.app.test_client()
|
||||
with c.session_transaction() as s:
|
||||
s["_csrf_token"] = "t"
|
||||
return c
|
||||
|
||||
|
||||
def _write(path, text):
|
||||
with open(path, "w") as f:
|
||||
f.write(text)
|
||||
|
||||
|
||||
def _iso(delta):
|
||||
return (datetime.now(timezone.utc) + delta).isoformat(timespec="seconds")
|
||||
|
||||
|
||||
# --- 1: imaging payload types ------------------------------------------------
|
||||
|
||||
class ImagingPayloadTypes(unittest.TestCase):
|
||||
def setUp(self):
|
||||
for f in os.listdir(config.IMAGING_DIR):
|
||||
os.remove(os.path.join(config.IMAGING_DIR, f))
|
||||
self.c = _client()
|
||||
|
||||
def test_list_pctype_post_does_not_500_pages(self):
|
||||
r = self.c.post("/imaging/status", json={
|
||||
"serial": "LISTPC", "pctype": ["gea-standard"], "current_stage": {"x": 1},
|
||||
"status": ["failed"], "stage_index": "3", "stage_total": [5], "mac": None,
|
||||
"log_lines": ["one", {"bad": 1}, 2], "hostname_target": 12})
|
||||
self.assertEqual(r.status_code, 200, r.get_data(as_text=True))
|
||||
state = imaging_status.get_session("LISTPC")
|
||||
self.assertNotIn("pctype", state)
|
||||
self.assertNotIn("current_stage", state)
|
||||
self.assertEqual(state["stage_index"], 3)
|
||||
self.assertNotIn("stage_total", state)
|
||||
self.assertEqual(state["hostname_target"], "12")
|
||||
self.assertEqual(state["log_tail"], ["one", "2"])
|
||||
for url in ("/", "/imaging", "/imaging/tiles"):
|
||||
self.assertEqual(self.c.get(url).status_code, 200, url)
|
||||
|
||||
def test_bad_serial_and_body_are_400(self):
|
||||
self.assertEqual(self.c.post("/imaging/status", json={"serial": ["x"]}).status_code, 400)
|
||||
self.assertEqual(self.c.post("/imaging/status", json=[1, 2]).status_code, 400)
|
||||
|
||||
def test_bad_json_on_disk_does_not_500(self):
|
||||
_write(os.path.join(config.IMAGING_DIR, "DISKBAD.json"), json.dumps({
|
||||
"serial": "DISKBAD", "pctype": ["gea-engineer"], "current_stage": ["x"],
|
||||
"stage_index": {"a": 1}, "log_tail": "not a list", "stage_history": [1, {"stage_index": "x"}],
|
||||
"last_updated": _iso(timedelta(0)), "status": "in_progress"}))
|
||||
_write(os.path.join(config.IMAGING_DIR, "LISTFILE.json"), "[1, 2, 3]")
|
||||
sessions = imaging_status.list_sessions(include_inferred=False)
|
||||
self.assertEqual([s["serial"] for s in sessions], ["DISKBAD"])
|
||||
s = sessions[0]
|
||||
self.assertNotIn("pctype", s)
|
||||
self.assertEqual(s["log_tail"], ["not a list"])
|
||||
for url in ("/", "/imaging", "/imaging/session/DISKBAD", "/imaging/DISKBAD.json"):
|
||||
self.assertEqual(self.c.get(url).status_code, 200, url)
|
||||
|
||||
def test_template_guards_unsanitised_session(self):
|
||||
with A.app.test_request_context("/"):
|
||||
html = A.render_template("_imaging_tiles.html", sessions=[{
|
||||
"serial": "RAW", "pctype": ["gea-standard"], "current_stage": {"a": 1},
|
||||
"stage_index": 1, "floor_state": "running"}])
|
||||
self.assertIn("RAW", html)
|
||||
|
||||
|
||||
# --- 4..8: network status ----------------------------------------------------
|
||||
|
||||
def _kernel(*ifaces):
|
||||
return (list(ifaces), True)
|
||||
|
||||
|
||||
def _kif(name, role, bound=None, link="up", speed=1000, addrs=None):
|
||||
return {"iface": name, "mac": "", "role": role, "link": link, "speed_mbps": speed,
|
||||
"addrs": addrs or [], "verdict": None, "evidence": [], "notes": [], "dnsmasq_bound": bound}
|
||||
|
||||
|
||||
class NicStatus(unittest.TestCase):
|
||||
def setUp(self):
|
||||
for p in (config.NIC_STATUS_FILE, config.NIC_CONF, config.DNSMASQ_NIC_CONF, config.DNSMASQ_CONF):
|
||||
if os.path.exists(p):
|
||||
os.remove(p)
|
||||
|
||||
def _check(self, interfaces, checked_at=None, guard=None):
|
||||
_write(config.NIC_STATUS_FILE, json.dumps({
|
||||
"checked_at": checked_at or _iso(timedelta(0)), "interfaces": interfaces,
|
||||
"dhcp_guard": guard or {"state": "ok", "reason": ""}}))
|
||||
|
||||
def test_malformed_file_renders(self):
|
||||
self._check([
|
||||
{"iface": "enp1", "role": "imaging", "evidence": None, "speed_mbps": "1000", "addrs": "x",
|
||||
"notes": 5, "dnsmasq_bound": "yes", "verdict": ["ge"], "link": "up"},
|
||||
"not a dict", None, 7,
|
||||
{"iface": ["x"], "role": {"a": 1}, "mac": 5},
|
||||
])
|
||||
with mock.patch.object(nic_status, "_kernel_view", return_value=([], False)):
|
||||
st = nic_status.status()
|
||||
for i in st["interfaces"]:
|
||||
self.assertIsInstance(i["evidence"], list)
|
||||
self.assertIsInstance(i["notes"], list)
|
||||
self.assertIsInstance(i["addrs"], list)
|
||||
self.assertTrue(i["speed_mbps"] is None or isinstance(i["speed_mbps"], int))
|
||||
self.assertIn(i["dnsmasq_bound"], (True, False, None))
|
||||
for k in ("role", "iface", "mac", "link"):
|
||||
self.assertIsInstance(i[k], str)
|
||||
self.assertEqual(st["interfaces"][0]["speed_mbps"], 1000)
|
||||
self.assertEqual(_client().get("/").status_code, 200)
|
||||
for junk in ("[1,2]", "{\"interfaces\": \"x\"}", "garbage", "{\"interfaces\": [], \"dhcp_guard\": 5}"):
|
||||
_write(config.NIC_STATUS_FILE, junk)
|
||||
with mock.patch.object(nic_status, "_kernel_view", return_value=([], False)):
|
||||
self.assertEqual(_client().get("/").status_code, 200, junk)
|
||||
|
||||
def test_status_raising_shows_unknown_pill(self):
|
||||
with mock.patch.object(nic_status, "status", side_effect=RuntimeError("boom")):
|
||||
r = _client().get("/")
|
||||
self.assertEqual(r.status_code, 200)
|
||||
self.assertIn("Network: unknown", r.get_data(as_text=True))
|
||||
|
||||
def test_roles_with_headered_conf(self):
|
||||
_write(config.NIC_CONF, "# managed\n[nic]\nimaging_mac = AA-BB-CC-DD-EE-01 # pxe\n"
|
||||
"ge_mac = aa:bb:cc:dd:ee:02 ; corp\n")
|
||||
self.assertEqual(nic_status._roles(), {"aa:bb:cc:dd:ee:01": "imaging", "aa:bb:cc:dd:ee:02": "ge"})
|
||||
_write(config.NIC_CONF, "imaging_iface = enp1\n") # no header: still read
|
||||
self.assertEqual(nic_status._roles(), {"enp1": "imaging"})
|
||||
|
||||
def test_roles_from_dnsmasq_pin_split_commas(self):
|
||||
_write(config.DNSMASQ_NIC_CONF, "interface=enp1,br0\nexcept-interface=enp2\n")
|
||||
self.assertEqual(nic_status._roles(), {"enp1": "imaging", "br0": "imaging", "enp2": "ge"})
|
||||
|
||||
def test_wildcard_binding_not_flagged(self):
|
||||
_write(config.DNSMASQ_NIC_CONF, "interface=enp1\nexcept-interface=enp2\nno-dhcp-interface=enp2\n")
|
||||
rules = nic_status._dnsmasq_rules()
|
||||
socks = [("0.0.0.0", None)]
|
||||
self.assertFalse(nic_status._bound("enp2", ["10.1.2.3"], socks, rules))
|
||||
self.assertTrue(nic_status._bound("enp1", ["172.16.9.1"], socks, rules))
|
||||
self.assertIsNone(nic_status._bound("enp2", ["10.1.2.3"], socks, None))
|
||||
self.assertTrue(nic_status._bound("enp2", [], [("0.0.0.0", "enp2")], rules))
|
||||
|
||||
def test_fresh_check_keeps_its_binding(self):
|
||||
self._check([{"iface": "enp1", "role": "imaging", "verdict": "isolated", "dnsmasq_bound": True},
|
||||
{"iface": "enp2", "role": "ge", "verdict": "ge", "dnsmasq_bound": False}])
|
||||
k = _kernel(_kif("enp1", "imaging", True), _kif("enp2", "ge", True))
|
||||
with mock.patch.object(nic_status, "_kernel_view", return_value=k):
|
||||
st = nic_status.status()
|
||||
self.assertEqual(st["state"], "ok", st["problems"])
|
||||
self.assertFalse(st["interfaces"][1]["dnsmasq_bound"])
|
||||
|
||||
def test_stale_file_downgrades_and_clears(self):
|
||||
guard = {"state": "blocked", "reason": "foreign DHCP server seen"}
|
||||
ifs = [{"iface": "enp1", "role": "imaging", "verdict": "ge", "evidence": ["LLDP from corp switch"]}]
|
||||
k = _kernel(_kif("enp1", "imaging", True, addrs=["172.16.9.1/24"]))
|
||||
for stamp in (_iso(-timedelta(hours=2)), _iso(timedelta(hours=1)), "not a date"):
|
||||
self._check(ifs, checked_at=stamp, guard=guard)
|
||||
with mock.patch.object(nic_status, "_kernel_view", return_value=k):
|
||||
st = nic_status.status()
|
||||
self.assertTrue(st["stale"], stamp)
|
||||
self.assertNotEqual(st["state"], "bad", (stamp, st["problems"]))
|
||||
self.assertTrue(any("last network check" in t.lower() for _, t in st["problems"]))
|
||||
self.assertEqual(st["interfaces"][0]["evidence"], [])
|
||||
self.assertIsNone(st["interfaces"][0]["verdict"])
|
||||
# Fresh, the same file is a danger.
|
||||
self._check(ifs, guard=guard)
|
||||
with mock.patch.object(nic_status, "_kernel_view", return_value=k):
|
||||
self.assertEqual(nic_status.status()["state"], "bad")
|
||||
|
||||
def test_interface_gone_from_kernel(self):
|
||||
self._check([{"iface": "enx1", "role": "imaging", "link": "up", "speed_mbps": 5000,
|
||||
"addrs": ["172.16.9.1/24"], "dnsmasq_bound": True}])
|
||||
with mock.patch.object(nic_status, "_kernel_view", return_value=_kernel(_kif("enp2", "ge"))):
|
||||
st = nic_status.status()
|
||||
i = st["interfaces"][0]
|
||||
self.assertEqual((i["link"], i["speed_mbps"], i["addrs"]), ("down", None, []))
|
||||
self.assertIn("not present (unplugged?)", i["notes"])
|
||||
|
||||
def test_kernel_view_includes_role_bridge(self):
|
||||
sysnet = os.path.join(T, "sysnet")
|
||||
shutil.rmtree(sysnet, ignore_errors=True)
|
||||
for name, mac in (("br0", "aa:bb:cc:dd:ee:01"), ("enp1", "aa:bb:cc:dd:ee:01"), ("enp2", "aa:bb:cc:dd:ee:02")):
|
||||
os.makedirs(os.path.join(sysnet, name))
|
||||
_write(os.path.join(sysnet, name, "address"), mac + "\n")
|
||||
_write(os.path.join(sysnet, name, "operstate"), "up\n")
|
||||
os.makedirs(os.path.join(sysnet, "br0", "bridge"))
|
||||
os.makedirs(os.path.join(sysnet, "br0", "brif", "enp1"))
|
||||
os.makedirs(os.path.join(sysnet, "enp1", "device"))
|
||||
os.makedirs(os.path.join(sysnet, "enp2", "device"))
|
||||
_write(config.NIC_CONF, "[nic]\nimaging_mac = aa:bb:cc:dd:ee:01\nge_mac = aa:bb:cc:dd:ee:02\n")
|
||||
with mock.patch.object(nic_status, "SYS_NET", sysnet), \
|
||||
mock.patch.object(nic_status, "_addrs", return_value={"br0": ["172.16.9.1/24"]}), \
|
||||
mock.patch.object(nic_status, "_dhcp_sockets", return_value=[("172.16.9.1", None)]):
|
||||
view, ok = nic_status._kernel_view()
|
||||
by = {i["iface"]: i for i in view}
|
||||
self.assertTrue(ok)
|
||||
self.assertEqual(by["br0"]["role"], "imaging")
|
||||
self.assertTrue(by["br0"]["dnsmasq_bound"])
|
||||
self.assertEqual(by["enp1"]["role"], "other")
|
||||
self.assertEqual(by["enp2"]["role"], "ge")
|
||||
self.assertFalse(by["enp2"]["dnsmasq_bound"])
|
||||
|
||||
|
||||
# --- 2, 3, 9, 10, 11: iPXE ---------------------------------------------------
|
||||
|
||||
@unittest.skipUnless(HAVE_TOOLS and os.path.isfile(DUAL_SHIM) and SHIM_2011 and SNPONLY and NATIVE,
|
||||
"real boot files, openssl or sbverify missing")
|
||||
class IpxeRealFiles(unittest.TestCase):
|
||||
def setUp(self):
|
||||
ipxe_update._DESCRIBE_CACHE.clear()
|
||||
for d in (os.path.join(config.UPDATES_DIR, "ipxe"), config.TFTP_ROOT):
|
||||
shutil.rmtree(d, ignore_errors=True)
|
||||
os.makedirs(config.TFTP_ROOT)
|
||||
|
||||
def test_dual_shim_accepted(self):
|
||||
self.assertEqual(ipxe_update.microsoft_cas(DUAL_SHIM), [ipxe_update.MS_CA_2011, ipxe_update.MS_CA_2023])
|
||||
d = ipxe_update.describe(DUAL_SHIM)
|
||||
self.assertTrue(d["dual"])
|
||||
self.assertEqual(ipxe_update.risks("ipxe-shim.efi", d), [])
|
||||
|
||||
def test_2011_only_shim_refused(self):
|
||||
self.assertEqual(ipxe_update.microsoft_cas(SHIM_2011), [ipxe_update.MS_CA_2011])
|
||||
d = ipxe_update.describe(SHIM_2011)
|
||||
self.assertFalse(d["dual"])
|
||||
self.assertTrue(ipxe_update.risks("ipxe-shim.efi", d))
|
||||
|
||||
def test_shim_swap_only_for_dual(self):
|
||||
folder = tempfile.mkdtemp(dir=T)
|
||||
for src, expect_swap in ((DUAL_SHIM, True), (SHIM_2011, False)):
|
||||
files = {"ipxe-shim.efi": dict(ipxe_update.describe(SHIM_2011), ok=True, error="", source="tar")}
|
||||
rel = {"tag_name": "t1", "assets": [{"name": ipxe_update.SHIM_ASSET, "browser_download_url": "u",
|
||||
"digest": "sha256:" + ipxe_update.uc.file_hash(src)}]}
|
||||
dest = os.path.join(folder, ipxe_update.SHIM_ASSET)
|
||||
shutil.copy(src, dest)
|
||||
with mock.patch.object(ipxe_update.uc, "fetch_text", return_value=json.dumps(rel)):
|
||||
ipxe_update._use_dual_signed_shim(folder, files)
|
||||
f = files["ipxe-shim.efi"]
|
||||
self.assertEqual(bool(f.get("dual")), expect_swap, f.get("note"))
|
||||
if not expect_swap:
|
||||
self.assertIn("not signed under both", f["note"])
|
||||
|
||||
def _stage_real(self, shim_src):
|
||||
tag = "vreal"
|
||||
folder = os.path.join(ipxe_update._dir(), "staged", tag)
|
||||
os.makedirs(folder, exist_ok=True)
|
||||
shutil.copy(TARBALL, os.path.join(folder, ipxe_update.TARBALL))
|
||||
shutil.copy(shim_src, os.path.join(folder, ipxe_update.SHIM_ASSET))
|
||||
info = {"tag": tag, "sha256": ipxe_update.uc.file_hash(TARBALL), "url": "unused",
|
||||
"size": os.path.getsize(TARBALL)}
|
||||
rel = {"tag_name": "s1", "assets": [{"name": ipxe_update.SHIM_ASSET, "browser_download_url": "unused",
|
||||
"digest": "sha256:" + ipxe_update.uc.file_hash(shim_src)}]}
|
||||
with mock.patch.object(ipxe_update.uc, "fetch_text", return_value=json.dumps(rel)), \
|
||||
mock.patch.object(ipxe_update.uc, "download", side_effect=AssertionError("no network")):
|
||||
return ipxe_update.stage(info)
|
||||
|
||||
def test_stage_real_tarball_dual_shim(self):
|
||||
rec = self._stage_real(DUAL_SHIM)
|
||||
self.assertEqual(rec["schema"], ipxe_update.STAGED_SCHEMA)
|
||||
self.assertTrue(rec["files"]["ipxe-shim.efi"]["ok"], rec["files"]["ipxe-shim.efi"])
|
||||
self.assertTrue(rec["files"]["ipxe-shim.efi"]["dual"])
|
||||
self.assertTrue(rec["files"]["ipxe.efi"]["ok"], rec["files"]["ipxe.efi"])
|
||||
self.assertEqual(rec["files"]["ipxe.efi"]["drivers"], "firmware")
|
||||
ipxe_update._install("vreal", ipxe_update.staged("vreal")[0], ["ipxe-shim.efi", "ipxe.efi"])
|
||||
|
||||
def test_stage_real_tarball_2011_shim_kept_not_installable(self):
|
||||
rec = self._stage_real(SHIM_2011) # the "dual" release asset is really 2011-only
|
||||
f = rec["files"]["ipxe-shim.efi"]
|
||||
self.assertFalse(f["ok"])
|
||||
self.assertIn("not dual-signed", f["error"])
|
||||
self.assertIn("dual-signed shim not used", f["note"])
|
||||
self.assertTrue(rec["files"]["ipxe.efi"]["ok"])
|
||||
|
||||
def test_drivers_of(self):
|
||||
self.assertEqual(ipxe_update.drivers_of(SNPONLY), "firmware")
|
||||
self.assertEqual(ipxe_update.drivers_of(NATIVE), "native")
|
||||
self.assertEqual(ipxe_update.drivers_of(DUAL_SHIM), "")
|
||||
|
||||
def _stage(self, tag, files, schema=ipxe_update.STAGED_SCHEMA):
|
||||
folder = os.path.join(ipxe_update._dir(), "staged", tag)
|
||||
os.makedirs(folder, exist_ok=True)
|
||||
recs = {}
|
||||
for name, (src, lie) in files.items():
|
||||
shutil.copy(src, os.path.join(folder, name))
|
||||
d = ipxe_update.describe(os.path.join(folder, name))
|
||||
d.update(lie)
|
||||
recs[name] = dict(d, ok=True, error="", source="test")
|
||||
rec = {"tag": tag, "folder": folder, "files": recs, "staged": "now"}
|
||||
if schema is not None:
|
||||
rec["schema"] = schema
|
||||
ipxe_update.uc.write_json(os.path.join(folder, "staged.json"), rec)
|
||||
return rec
|
||||
|
||||
def test_install_refuses_native_ipxe(self):
|
||||
# The record claims snponly; the file on disk is the native-USB build.
|
||||
self._stage("v9", {"ipxe-shim.efi": (DUAL_SHIM, {}), "ipxe.efi": (NATIVE, {"drivers": "firmware"})})
|
||||
rec = ipxe_update.staged("v9")[0]
|
||||
with self.assertRaisesRegex(ValueError, "drives USB itself"):
|
||||
ipxe_update._install("v9", rec, ["ipxe-shim.efi", "ipxe.efi"])
|
||||
self.assertEqual(os.listdir(config.TFTP_ROOT), [])
|
||||
|
||||
def test_install_refuses_2011_shim(self):
|
||||
self._stage("v8", {"ipxe-shim.efi": (SHIM_2011, {"dual": True}), "ipxe.efi": (SNPONLY, {})})
|
||||
rec = ipxe_update.staged("v8")[0]
|
||||
with self.assertRaisesRegex(ValueError, "not dual-signed"):
|
||||
ipxe_update._install("v8", rec, ["ipxe-shim.efi", "ipxe.efi"])
|
||||
|
||||
def test_install_good_set(self):
|
||||
self._stage("v7", {"ipxe-shim.efi": (DUAL_SHIM, {}), "ipxe.efi": (SNPONLY, {})})
|
||||
rec = ipxe_update.staged("v7")[0]
|
||||
ipxe_update._install("v7", rec, ["ipxe-shim.efi", "ipxe.efi"])
|
||||
self.assertTrue(os.path.isfile(os.path.join(config.TFTP_ROOT, "ipxe.efi")))
|
||||
|
||||
def test_old_schema_record_is_stale(self):
|
||||
self._stage("v6", {"ipxe-shim.efi": (DUAL_SHIM, {}), "ipxe.efi": (SNPONLY, {})}, schema=None)
|
||||
rec = ipxe_update.staged("v6")[0]
|
||||
self.assertTrue(rec["stale"])
|
||||
self.assertFalse(rec["files"]["ipxe.efi"]["ok"])
|
||||
with mock.patch.object(ipxe_update.uc, "job_running", return_value=False):
|
||||
with self.assertRaisesRegex(ValueError, "older version"):
|
||||
ipxe_update.install("v6", ["ipxe-shim.efi", "ipxe.efi"])
|
||||
with self.assertRaisesRegex(ValueError, "older version"):
|
||||
ipxe_update._install("v6", rec, ["ipxe.efi"])
|
||||
with mock.patch.object(ipxe_update.uc, "job_running", return_value=False):
|
||||
r = _client().get("/updates/ipxe/")
|
||||
self.assertEqual(r.status_code, 200)
|
||||
self.assertIn("Stale: download again", r.get_data(as_text=True))
|
||||
|
||||
def test_rollback_needs_confirm_for_risky_backup(self):
|
||||
folder = os.path.join(ipxe_update._dir(), "backups", "20260101-000000")
|
||||
os.makedirs(folder)
|
||||
shutil.copy(NATIVE, os.path.join(folder, "ipxe.efi"))
|
||||
shutil.copy(SHIM_2011, os.path.join(folder, "ipxe-shim.efi"))
|
||||
b = ipxe_update.backups()[0]
|
||||
self.assertEqual(len(b["warnings"]), 2)
|
||||
with mock.patch.object(ipxe_update.uc, "job_running", return_value=False):
|
||||
html = _client().get("/updates/ipxe/").get_data(as_text=True)
|
||||
self.assertIn('name="confirm"', html)
|
||||
self.assertIn("Restore anyway", html)
|
||||
r = _client().post("/updates/ipxe/rollback", data={"_csrf_token": "t", "backup": "20260101-000000"})
|
||||
self.assertEqual(r.status_code, 302)
|
||||
self.assertFalse(os.path.exists(os.path.join(config.TFTP_ROOT, "ipxe.efi")))
|
||||
with self.assertRaisesRegex(ValueError, "Tick the confirmation"):
|
||||
ipxe_update.rollback("20260101-000000")
|
||||
self.assertEqual(sorted(ipxe_update.rollback("20260101-000000", confirm=True)),
|
||||
["ipxe-shim.efi", "ipxe.efi"])
|
||||
|
||||
def test_mac_folder_shim_without_ipxe_warns(self):
|
||||
d = os.path.join(config.TFTP_ROOT, "aa-bb-cc-dd-ee-ff")
|
||||
os.makedirs(d)
|
||||
shutil.copy(DUAL_SHIM, os.path.join(d, "ipxe-shim.efi"))
|
||||
m = ipxe_update.mac_folders()[0]
|
||||
self.assertTrue(any("no ipxe.efi" in w for w in m["warnings"]))
|
||||
with mock.patch.object(ipxe_update.uc, "job_running", return_value=False):
|
||||
html = _client().get("/updates/ipxe/").get_data(as_text=True)
|
||||
self.assertIn(config.TFTP_ROOT + "/aa-bb-cc-dd-ee-ff/", html)
|
||||
self.assertNotIn("/srv/tftp/aa-bb", html)
|
||||
|
||||
def test_describe_is_cached(self):
|
||||
with mock.patch.object(ipxe_update, "signers", wraps=ipxe_update.signers) as sp:
|
||||
ipxe_update.describe(SNPONLY)
|
||||
ipxe_update.describe(SNPONLY)
|
||||
self.assertEqual(sp.call_count, 1)
|
||||
|
||||
|
||||
class NestedSignatureParse(unittest.TestCase):
|
||||
def test_finds_nested_contentinfo(self):
|
||||
inner = bytes([0x30, 0x03, 0x02, 0x01, 0x05]) # SEQUENCE { INTEGER 5 }
|
||||
outer = b"\x30\x00junk" + ipxe_update.NESTED_SIG_OID + bytes([0x31, len(inner)]) + inner + b"tail"
|
||||
self.assertEqual(ipxe_update._nested_signatures(outer), [inner])
|
||||
self.assertEqual(ipxe_update._nested_signatures(b"no oid here"), [])
|
||||
self.assertEqual(ipxe_update._nested_signatures(ipxe_update.NESTED_SIG_OID + b"\x31\x7f"), [])
|
||||
|
||||
|
||||
# --- 12: evidence pack dates -------------------------------------------------
|
||||
|
||||
class EvidenceDates(unittest.TestCase):
|
||||
def test_far_future_dates_do_not_500(self):
|
||||
r = _client().get("/reports/evidence?from=9999-12-01&to=9999-12-31")
|
||||
self.assertEqual(r.status_code, 302)
|
||||
|
||||
def test_400_days_inclusive(self):
|
||||
c = _client()
|
||||
ok = c.get("/reports/evidence?from=2025-01-01&to=2026-02-04") # 400 days with both ends
|
||||
self.assertEqual(ok.status_code, 200)
|
||||
self.assertEqual(ok.mimetype, "application/zip")
|
||||
self.assertEqual(c.get("/reports/evidence?from=2025-01-01&to=2026-02-05").status_code, 302)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user