ge-aerospace/pxe-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
9af1c6d89a6e9a8cdb90c037fcb0aeb09f0dbfa2
pxe-server/webapp/tests/test_review_fixes.py
cproudlock 606659bf07 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.
2026-10-15 14:20:24 -04:00

453 lines
23 KiB
Python
Raw Blame History

"""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 View Git Blame Copy Permalink