From 606659bf077791e67763a54a78b98f91479e3a02 Mon Sep 17 00:00:00 2001 From: cproudlock Date: Thu, 15 Oct 2026 14:20:24 -0400 Subject: [PATCH] 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. --- webapp/app.py | 40 ++- webapp/services/imaging_status.py | 124 +++++++- webapp/services/ipxe_update.py | 277 ++++++++++++++-- webapp/services/nic_status.py | 303 ++++++++++++++---- webapp/templates/_imaging_tiles.html | 28 +- webapp/templates/_nic_status.html | 50 +++ webapp/templates/dashboard.html | 46 +-- webapp/templates/reports.html | 2 +- webapp/templates/updates_ipxe.html | 25 +- webapp/tests/test_review_fixes.py | 452 +++++++++++++++++++++++++++ webapp/views/ipxe.py | 4 +- 11 files changed, 1185 insertions(+), 166 deletions(-) create mode 100644 webapp/templates/_nic_status.html create mode 100644 webapp/tests/test_review_fixes.py diff --git a/webapp/app.py b/webapp/app.py index e79cbba..7b6571d 100644 --- a/webapp/app.py +++ b/webapp/app.py @@ -101,6 +101,27 @@ app.register_blueprint(images_unattend_views.bp) # Unattend review/save/undo # Routes - pages # --------------------------------------------------------------------------- +def _nic_parts(): + """(pill, notice) HTML for the Overview's network status. Rendered here, + each inside a try, so a bad status file or a template slip shows + 'Network: unknown' and never takes the Overview down.""" + from markupsafe import Markup + try: + nic = nic_status.status() + except Exception as ex: # the page must render even if this breaks + app.logger.exception("network status failed") + nic = nic_status.unknown(ex) + parts = [] + for part in ("pill", "notice"): + try: + parts.append(Markup(render_template("_nic_status.html", nic=nic, part=part))) + except Exception as ex: + app.logger.exception("network status render failed") + parts.append(Markup(render_template("_nic_status.html", nic=nic_status.unknown(ex), part=part)) + if part == "pill" else Markup("")) + return parts[0], parts[1] + + @app.route("/") def dashboard(): image_list = [images.image_status(it) for it in config.IMAGE_TYPES] @@ -114,11 +135,7 @@ def dashboard(): img["os_label"], img["driver_count"] = "", 0 img["lint"] = image_pages._lint_counts(img["image_type"]) services = [system.service_status(s) for s in ("dnsmasq", "apache2", "smbd")] - try: - nic = nic_status.status() - except Exception: # the page must render even if this breaks - app.logger.exception("network status failed") - nic = None + nic_pill, nic_notice = _nic_parts() sessions, _hidden = imaging_status.hide_stale(imaging_status.list_sessions()) floor = imaging_status.floor_order(sessions) real = [x for x in floor if x["floor_state"] != "logs"] @@ -127,7 +144,8 @@ def dashboard(): "dashboard.html", images=image_list, services=services, - nic=nic, + nic_pill=nic_pill, + nic_notice=nic_notice, image_types=config.IMAGE_TYPES, friendly_names=config.FRIENDLY_NAMES, sessions=real[:6], @@ -810,7 +828,11 @@ def blancco_evidence_pack(): return redirect(url_for("blancco_reports")) if last < first: first, last = last, first - if (last - first).days > 400: + if first.year < 2000 or last.year > 2100: + # 9999-12-31 + 1 day overflows the date maths further down. + flash("Pick dates between the years 2000 and 2100.", "danger") + return redirect(url_for("blancco_reports")) + if (last - first).days + 1 > 400: # both ends count flash("An evidence pack covers at most about a year (400 days).", "danger") return redirect(url_for("blancco_reports")) out = tempfile.SpooledTemporaryFile(max_size=64 * 1024 * 1024) @@ -1113,8 +1135,12 @@ def imaging_stream(): def imaging_status_post(): # CSRF-exempt machine-to-machine endpoint; see services/csrf.py exempt list. payload = request.get_json(silent=True) or {} + if not isinstance(payload, dict): + return jsonify({"error": "body must be a JSON object"}), 400 if not payload.get("serial"): return jsonify({"error": "missing serial"}), 400 + if not isinstance(payload.get("serial"), str): + return jsonify({"error": "serial must be a string"}), 400 try: state = imaging_status.update_session(payload) except ValueError as ex: diff --git a/webapp/services/imaging_status.py b/webapp/services/imaging_status.py index 0f1e753..b341fa9 100644 --- a/webapp/services/imaging_status.py +++ b/webapp/services/imaging_status.py @@ -39,6 +39,110 @@ ACTIVE_WINDOW_HOURS = 6 _SAFE_SERIAL = re.compile(r"[^A-Za-z0-9._-]") +# Payload and stored-session field types. The pages compare and look these +# up (`pctype in {...}`, `'%]' in current_stage`), so a list or dict from a +# client or a hand-edited file must never reach them. +INT_FIELDS = ("stage_index", "stage_total") +# Written by this module only; a client cannot set them. +SERVER_FIELDS = ("log_tail", "stage_history", "started_at", "last_updated", "previous_run_at") +STR_MAX = 4096 +LOG_LINE_MAX = 2000 +_SAFE_KEY = re.compile(r"[A-Za-z0-9_]{1,64}") + + +def _scalar_str(v, limit=STR_MAX): + """str for a scalar, None for anything else (dict, list, None).""" + if isinstance(v, str): + return v[:limit] + if isinstance(v, bool): + return "true" if v else "false" + if isinstance(v, (int, float)): + return str(v) + return None + + +def _as_int(v): + if isinstance(v, bool): + return None + try: + return int(v) + except (TypeError, ValueError, OverflowError): + return None + + +def _str_lines(v): + """A str or a list of scalars as a list of str lines; [] otherwise.""" + if isinstance(v, str): + v = [v] + if not isinstance(v, list): + return [] + out = [] + for x in v: + x = _scalar_str(x, LOG_LINE_MAX) + if x is not None: + out.append(x) + return out + + +def _clean_payload(payload: dict) -> dict: + """A client payload with every value in the type the pages expect: + stage_index/stage_total int, log_lines a list of str, everything else a + str. Non-scalar values and server-owned fields are dropped.""" + out = {} + for key, value in payload.items(): + if not isinstance(key, str) or not _SAFE_KEY.fullmatch(key) or key in SERVER_FIELDS: + continue + if key in INT_FIELDS: + n = _as_int(value) + if n is not None: + out[key] = n + elif key == "log_lines": + lines = _str_lines(value) + if lines: + out[key] = lines + else: + v = _scalar_str(value) + if v is not None: + out[key] = v + return out + + +def sanitize_session(state) -> Optional[dict]: + """A stored session made safe to render: same rules as a payload, plus + log_tail (list of str) and stage_history (list of dicts). None when it is + not a JSON object or has no usable serial.""" + if not isinstance(state, dict): + return None + out = {} + for key, value in state.items(): + if not isinstance(key, str): + continue + if key in INT_FIELDS: + n = _as_int(value) + if n is not None: + out[key] = n + elif key == "log_tail": + out[key] = _str_lines(value)[-LOG_TAIL_MAX:] + elif key == "stage_history": + hist = [] + for h in value if isinstance(value, list) else []: + if isinstance(h, dict): + hist.append({"ts": _scalar_str(h.get("ts")) or "", + "stage_index": _as_int(h.get("stage_index")) or 0, + "current_stage": _scalar_str(h.get("current_stage")) or "", + "status": _scalar_str(h.get("status")) or ""}) + out[key] = hist[-STAGE_HISTORY_MAX:] + elif key == "log_lines": + continue + else: + v = _scalar_str(value) + if v is not None: + out[key] = v + if not out.get("serial"): + return None + return out + + def _ensure_dir(): os.makedirs(config.IMAGING_DIR, exist_ok=True) @@ -83,9 +187,15 @@ def update_session(payload: dict) -> dict: present overwrites the existing field. log_tail is appended to (capped). Returns the resulting full state. """ + if not isinstance(payload, dict): + raise ValueError("payload must be a JSON object") + if not isinstance(payload.get("serial") or "", str): + raise ValueError("serial must be a string") + payload = _clean_payload(payload) serial = (payload.get("serial") or "").strip() if not serial: raise ValueError("payload missing 'serial'") + payload["serial"] = serial # "inferred" is the server's own label for log-only bays; a client that # sends it would hide its row from the floor's failure-first order. if payload.get("source") not in (None, "client", "blancco"): @@ -101,8 +211,8 @@ def update_session(payload: dict) -> dict: if os.path.isfile(path): try: with open(path, "r") as f: - state = json.load(f) - except (json.JSONDecodeError, OSError): + state = sanitize_session(json.load(f)) or {} + except (ValueError, OSError): state = {} # Reimage detection: if the new payload's stage_index <= 2 (WinPE startnet @@ -216,9 +326,11 @@ def list_sessions(include_inferred: bool = True) -> list[dict]: path = os.path.join(config.IMAGING_DIR, name) try: with open(path, "r") as f: - out.append(json.load(f)) - except (json.JSONDecodeError, OSError): + state = sanitize_session(json.load(f)) + except (ValueError, OSError): continue + if state is not None: # a broken or hand-edited file must not 500 the pages + out.append(state) if include_inferred: from services import imaging_log_tail out = imaging_log_tail.merge_with_client_sessions(out) @@ -292,8 +404,8 @@ def get_session(serial: str) -> Optional[dict]: return None try: with open(path, "r") as f: - return json.load(f) - except (json.JSONDecodeError, OSError): + return sanitize_session(json.load(f)) + except (ValueError, OSError): return None diff --git a/webapp/services/ipxe_update.py b/webapp/services/ipxe_update.py index 3bbd6f0..b61e466 100644 --- a/webapp/services/ipxe_update.py +++ b/webapp/services/ipxe_update.py @@ -24,8 +24,10 @@ https://github.com/ipxe/ipxe/releases/latest/download/ipxeboot.tar.gz, with the signed x86 files in ipxeboot/x86_64-sb/. GitHub lists a sha256 digest per release asset, which is checked before anything is unpacked, and every EFI file's Authenticode chain is read with sbverify before it can be installed. -The chain check is by certificate name (sbverify --list); the cryptographic -trust comes from the GitHub digest. Do not swap in `sbverify --cert`: it +The chain check is by certificate name (sbverify --list, plus every +signature in the PE security directory, nested ones included, read with +openssl for the shim's Microsoft CAs); the cryptographic trust comes from +the GitHub digest. Do not swap in `sbverify --cert`: it prints "Signature verification OK" even for a missing or unrelated cert. Why it matters: the Broadcom build, and the shim in the iPXE tarball, are @@ -85,6 +87,12 @@ NEED_LABEL = {"microsoft": "a Microsoft UEFI CA", "ipxe": "the iPXE Secure Boot MS_CA_2011 = "Microsoft Corporation UEFI CA 2011" MS_CA_2023 = "Microsoft UEFI CA 2023" IPXE_CA = "iPXE Secure Boot CA" +MS_CAS = (MS_CA_2011, MS_CA_2023) + +# Bumped when the staged.json layout or the checks behind it change. A +# record written by an older version was verified by weaker checks and is +# shown as stale (download it again) and never installed. +STAGED_SCHEMA = 2 def _dir(): @@ -115,6 +123,111 @@ def signers(path): return pairs +# Authenticode nested signature (SPC_NESTED_SIGNATURE, 1.3.6.1.4.1.311.2.4.1), +# DER-encoded OID. A second signature can sit in a second WIN_CERTIFICATE +# entry or nested inside the first one's unsigned attributes; sbverify may +# list only the outer one, so both places are read here. +NESTED_SIG_OID = bytes.fromhex("060a2b060104018237020401") +WIN_CERT_TYPE_PKCS_SIGNED_DATA = 0x0002 + + +def _der_tlv(buf, i): + """(content start, content end) of the DER element at buf[i]; raises + ValueError when it runs past the buffer.""" + if i + 2 > len(buf): + raise ValueError("short DER") + n = buf[i + 1] + if n < 0x80: + start, length = i + 2, n + else: + k = n & 0x7F + if not 1 <= k <= 4 or i + 2 + k > len(buf): + raise ValueError("bad DER length") + start, length = i + 2 + k, int.from_bytes(buf[i + 2:i + 2 + k], "big") + if start + length > len(buf): + raise ValueError("DER runs past the buffer") + return start, start + length + + +def _pe_signatures(data): + """The PKCS#7 blobs in a PE file's security directory, [] if none.""" + try: + pe = struct.unpack_from("= 0 and depth < 4: + try: + i = pos + len(NESTED_SIG_OID) + if blob[i] == 0x31: # SET OF ContentInfo + start, end = _der_tlv(blob, i) + j = start + while j < end: + _cs, ce = _der_tlv(blob, j) + inner = blob[j:ce] + if blob[j] == 0x30: + out.append(inner) + out += _nested_signatures(inner, depth + 1) + j = ce + except (ValueError, IndexError): + pass + pos = blob.find(NESTED_SIG_OID, pos + 1) + return out + + +def _cert_issuers(blob): + """Issuer CNs of every certificate in one PKCS#7 blob (openssl).""" + r = subprocess.run(["openssl", "pkcs7", "-inform", "DER", "-print_certs", "-noout"], + input=blob, capture_output=True, timeout=30) + text = r.stdout.decode("utf-8", errors="replace") + return {m.strip() for m in re.findall(r"^issuer=.*?CN\s*=\s*([^,/\n]+)", text, re.M)} + + +def microsoft_cas(path): + """Which of the Microsoft UEFI CAs (2011, 2023) sign this file, looking + at every signature: each WIN_CERTIFICATE entry and any nested inside one. + Raises FileNotFoundError when openssl is not installed.""" + try: + with open(path, "rb") as f: + data = f.read() + except OSError: + return [] + issuers = set() + for blob in _pe_signatures(data): + for b in [blob] + _nested_signatures(blob): + issuers |= _cert_issuers(b) + return [ca for ca in MS_CAS if ca in issuers] + + +def dual_signed(info): + """True when a described file carries both Microsoft UEFI CAs.""" + return bool(info) and all(ca in (info.get("ms_cas") or []) for ca in MS_CAS) + + def signature_kind(pairs): """microsoft-2023 | microsoft-2011 | microsoft | ipxe | other | unsigned. A file signed under both Microsoft CAs counts as 2023.""" @@ -221,29 +334,74 @@ def drivers_of(path): return "" if b"iPXE" not in data or b"$Version: " in data: return "" # not iPXE, or a shim - return "native" if re.search(rb"\0xhci\0|\0ehci\0", data) else "firmware" + return "native" if re.search(rb"\0(?:xhci|ehci|uhci)\0", data) else "firmware" + + +_DESCRIBE_CACHE = {} def describe(path): + """Size, hash, version, signers and drivers of one boot file, or None. + Cached per (path, size, mtime): reading and sbverify-ing every file on + each page load is slow. Returns a copy the caller may change.""" + try: + st = os.stat(path) + except OSError: + return None if not os.path.isfile(path): return None - st = os.stat(path) + key = (os.path.abspath(path), st.st_size, st.st_mtime_ns) + hit = _DESCRIBE_CACHE.get(key) + if hit is None: + hit = _describe(path, st) + if len(_DESCRIBE_CACHE) > 256: + _DESCRIBE_CACHE.clear() + _DESCRIBE_CACHE[key] = hit + return dict(hit) + + +def _describe(path, st): efi = path.endswith(".efi") - pairs, error = [], "" + pairs, error, ms_cas = [], "", [] if efi: try: pairs = signers(path) except (OSError, subprocess.SubprocessError) as ex: error = f"sbverify failed: {ex}" - return { + kind = (signature_kind(pairs) if not error else "error") if efi else "bios" + if kind.startswith("microsoft"): + try: + ms_cas = microsoft_cas(path) + except (OSError, subprocess.SubprocessError) as ex: + error = f"openssl failed: {ex}" + kind = "error" + info = { "size": st.st_size, "modified": st.st_mtime, "sha256": uc.file_hash(path), "version": version_of(path), "sbat": sbat_generations(path) if efi else "", - "signers": pairs, "kind": (signature_kind(pairs) if not error else "error") if efi else "bios", + "signers": pairs, "kind": kind, "trusts": vendor_cert_cn(path) if efi else None, "drivers": drivers_of(path) if efi else "", - "sig_error": error, + "sig_error": error, "ms_cas": ms_cas, } + info["dual"] = dual_signed(info) + return info + + +def risks(name, info): + """Why a boot file should not go live, [] when none: an ipxe.efi that is + not the snponly (firmware drivers) build, or a shim that does not carry + both Microsoft UEFI CAs.""" + if not info: + return [] + if name == "ipxe.efi" and info.get("drivers") != "firmware": + if info.get("drivers") == "native": + return ["ipxe.efi drives USB itself (not snponly): USB keyboards stop working at the iPXE menu"] + return ["ipxe.efi is not recognised as an iPXE snponly build"] + if name in SHIMS and not info.get("dual"): + have = " and ".join(info.get("ms_cas") or []) or info.get("kind") or "no Microsoft CA" + return [f"{name} is not dual-signed (2011 + 2023): it carries {have}"] + return [] def _installed_log(): @@ -289,14 +447,25 @@ def mac_folders(): folder = os.path.join(config.TFTP_ROOT, d) if not (MAC_DIR_RE.match(d) and os.path.isdir(folder)): continue - files = [] - for f in sorted(os.listdir(folder)): + files, warnings = [], [] + try: + entries = sorted(os.listdir(folder)) + except OSError as ex: + entries = [] + warnings.append(f"cannot read the folder: {ex.strerror or ex}") + for f in entries: info = describe(os.path.join(folder, f)) if info: files.append({"name": f, "info": info}) by = {f["name"]: f["info"] for f in files} - _check_chain(next((by[n] for n in SHIMS if n in by), None), [by.get("ipxe.efi")]) - out.append({"mac": d.replace("-", ":"), "dir": d, "files": files}) + shim = next((by[n] for n in SHIMS if n in by), None) + if shim is None: + shim = next((f["info"] for f in files if (f["info"].get("version") or "").startswith("shim ")), None) + _check_chain(shim, [by.get("ipxe.efi")]) + if shim is not None and "ipxe.efi" not in by: + warnings.append("This folder has a shim but no ipxe.efi. The shim loads ipxe.efi from this folder " + "(no fallback to the root), so this client stops after the shim.") + out.append({"mac": d.replace("-", ":"), "dir": d, "files": files, "warnings": warnings}) return out @@ -530,10 +699,21 @@ def stage(info=None): if shim.get("ok") and ipxe.get("ok") and shim.get("trusts"): if shim["trusts"] not in {i for _, i in ipxe["signers"]}: ipxe.update(ok=False, error=f"not signed under {shim['trusts']}, which this shim trusts") - if ipxe.get("ok") and ipxe.get("drivers") == "native": - # The full build would kill USB keyboards on every client. - ipxe.update(ok=False, error="this build drives USB itself (not snponly); refusing it as ipxe.efi") - record = {"tag": info["tag"], "published": info.get("published"), "folder": folder, + if ipxe.get("ok") and ipxe.get("drivers") != "firmware": + # The full build would kill USB keyboards on every client; anything + # not recognised as snponly is refused too. + ipxe.update(ok=False, error=("this build drives USB itself (not snponly); refusing it as ipxe.efi" + if ipxe.get("drivers") == "native" else + "not recognised as an iPXE snponly build; refusing it as ipxe.efi")) + for n in SHIMS: + f = files.get(n) or {} + if f.get("ok") and not f.get("dual"): + # Kept on disk and shown, but never installed: a 2011-only shim + # stops booting once machines drop or revoke the 2011 CA. + have = " and ".join(f.get("ms_cas") or []) or f.get("kind") + f.update(ok=False, error=f"not dual-signed (needs {MS_CA_2011} and {MS_CA_2023}; has {have}). " + + (f.get("note") or "")) + record = {"schema": STAGED_SCHEMA, "tag": info["tag"], "published": info.get("published"), "folder": folder, "tarball_sha256": info["sha256"], "files": files, "staged": uc.now()} uc.write_json(os.path.join(folder, "staged.json"), record) return record @@ -541,10 +721,11 @@ def stage(info=None): def _use_dual_signed_shim(folder, files): """Swap the tarball's shim for the ipxe/shim release build when that one - is signed under the Microsoft UEFI CA 2023 and trusts the same iPXE CA. - Any problem leaves the tarball shim in place, with the reason noted.""" + carries both Microsoft UEFI CAs (2011 and 2023, read from every + signature, nested ones included) and trusts the same iPXE CA. Any + problem leaves the tarball shim in place, with the reason noted.""" tar_shim = files.get("ipxe-shim.efi") or {} - if not tar_shim.get("ok") or tar_shim.get("kind") == "microsoft-2023": + if not tar_shim.get("ok") or tar_shim.get("dual"): return try: rel = json.loads(uc.fetch_text(SHIM_API, timeout=20)) @@ -559,8 +740,12 @@ def _use_dual_signed_shim(folder, files): os.remove(dest) uc.download(asset["browser_download_url"], dest, expect=sha, timeout=300) d = describe(dest) - if d["kind"] != "microsoft-2023": - raise RuntimeError(f"{SHIM_ASSET} is {d['kind']}, not signed under {MS_CA_2023}") + if d["sig_error"]: + raise RuntimeError(f"{SHIM_ASSET}: {d['sig_error']}") + if not d["dual"]: + have = " and ".join(d["ms_cas"]) or d["kind"] + raise RuntimeError(f"{SHIM_ASSET} is not signed under both {MS_CA_2011} and {MS_CA_2023} " + f"(found {have})") if d.get("trusts") != tar_shim.get("trusts"): raise RuntimeError(f"{SHIM_ASSET} trusts {d.get('trusts')}, the tarball's iPXE needs " f"{tar_shim.get('trusts')}") @@ -574,7 +759,18 @@ def _use_dual_signed_shim(folder, files): if n in files: shutil.copy2(dest, os.path.join(folder, n)) files[n] = dict(d, ok=True, error="", source=f"github.com/ipxe/shim {rel.get('tag_name')} " - f"{SHIM_ASSET} (dual-signed, 2011 + 2023)") + f"{SHIM_ASSET} (dual-signed: {' + '.join(d['ms_cas'])})") + + +def _record_stale(rec): + try: + schema = int(rec.get("schema") or 0) + except (TypeError, ValueError): + schema = 0 + if schema < STAGED_SCHEMA: + return True + return any(isinstance(f, dict) and f.get("signers") is not None and "drivers" not in f + for f in rec["files"].values()) def staged(tag=None): @@ -590,6 +786,12 @@ def staged(tag=None): if tag is None or rec.get("tag") == tag: # Where it is now, not the path recorded at staging time. rec["folder"] = os.path.join(root, t) + if _record_stale(rec): + # Verified by older, weaker checks: show it, never install it. + rec["stale"] = True + for f in rec["files"].values(): + if isinstance(f, dict): + f.update(ok=False, error="verified by an older version of this page; download it again") out.append(rec) out.sort(key=lambda r: r.get("staged") or "", reverse=True) return out @@ -751,6 +953,8 @@ def install(tag, names=DEFAULT_INSTALL): rec = next(iter(staged(tag)), None) if not rec: raise ValueError(f"{tag} is not staged") + if rec.get("stale"): + raise ValueError(f"{tag} was verified by an older version of this page; download it again") names = [n for n in MANAGED if n in names] if not names: raise ValueError("nothing selected") @@ -763,13 +967,25 @@ def install(tag, names=DEFAULT_INSTALL): def _install(tag, rec, names): + if _record_stale(rec): + raise ValueError(f"{tag} was verified by an older version of this page; download it again") for n in names: f = rec["files"].get(n) or {} if not f.get("ok"): raise ValueError(f"{n}: {f.get('error') or 'not staged'}") path = os.path.join(rec["folder"], n) - if not os.path.isfile(path) or uc.file_hash(path) != f["sha256"]: + if not os.path.isfile(path) or uc.file_hash(path) != f.get("sha256"): raise ValueError(f"{n}: staged copy changed since it was verified; download again") + # Re-read the file itself: the record says what was true at staging. + need = MANAGED[n][1] + if need: + now = _describe(path, os.stat(path)) # uncached: this gates going live + if now.get("sig_error") or not str(now.get("kind") or "").startswith(need): + raise ValueError(f"{n}: expected a signature under {NEED_LABEL[need]}, found " + f"{now.get('sig_error') or now.get('kind')}") + bad = risks(n, now) + if bad: + raise ValueError(f"{n}: {bad[0]}; refusing to install it") if not any(n in names for n in SHIMS): # iPXE alone must still be one the installed shim will start. trusts = vendor_cert_cn(os.path.join(config.TFTP_ROOT, "ipxe-shim.efi")) @@ -799,14 +1015,19 @@ def backups(): meta = uc.read_json(os.path.join(folder, "backup.json"), {}) meta = meta if isinstance(meta, dict) else {} files = [f for f in MANAGED if os.path.isfile(os.path.join(folder, f))] + infos = {f: describe(os.path.join(folder, f)) for f in files} out.append({"id": b, "reason": meta.get("reason", ""), "at": meta.get("at", ""), "files": files, "absent": meta.get("absent", []), - "versions": {f: version_of(os.path.join(folder, f)) for f in files}}) + "info": infos, + "versions": {f: (infos[f] or {}).get("version", "") for f in files}, + "warnings": [w for f in files for w in risks(f, infos[f])]}) return out -def rollback(backup_id): - """Put the files of one backup back, after backing up what is there now.""" +def rollback(backup_id, confirm=False): + """Put the files of one backup back, after backing up what is there now. + A backup holding a native-USB ipxe.efi or a shim that is not dual-signed + is only restored with confirm=True.""" bid = os.path.basename(str(backup_id or "")) folder = os.path.join(_dir(), "backups", bid) if not bid or bid.startswith(".") or not os.path.isdir(folder): @@ -814,6 +1035,10 @@ def rollback(backup_id): files = [f for f in MANAGED if os.path.isfile(os.path.join(folder, f))] if not files: raise ValueError("that backup holds no boot files") + warnings = [w for f in files + for w in risks(f, _describe(os.path.join(folder, f), os.stat(os.path.join(folder, f))))] + if warnings and not confirm: + raise ValueError("; ".join(warnings) + ". Tick the confirmation to restore it anyway") with _Lock(): _backup(files, f"before rolling back to {bid}") _place_all([(os.path.join(folder, f), f, uc.file_hash(os.path.join(folder, f))) diff --git a/webapp/services/nic_status.py b/webapp/services/nic_status.py index ebc9d17..39b8043 100644 --- a/webapp/services/nic_status.py +++ b/webapp/services/nic_status.py @@ -21,6 +21,8 @@ from datetime import datetime, timezone import config STALE_SECONDS = 20 * 60 # the watch timer runs every 5 min +# A checked_at this far in the future is clock skew or a bad file: stale. +FUTURE_SKEW_SECONDS = 300 SYS_NET = "/sys/class/net" # cdc_ncm (the generic USB NIC driver, e.g. a WisdPI RTL8157 5G adapter on a # kernel whose r8152 does not know the chip) keeps the link speed as a 32-bit @@ -37,26 +39,73 @@ def _read(path): return "" -def _roles(): +def _norm_mac(v): + return str(v or "").strip().lower().replace("-", ":") + + +def _nic_conf(): + """{key: value} from nic.conf, loaded like pxe-nic-check does: as-is + (the playbook writes a [nic] header), else with a [nic] header added.""" + try: + with open(config.NIC_CONF) as f: + text = f.read() + except (OSError, UnicodeDecodeError): + return {} + 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) + out = {} + for sect in cp.sections(): + for k, v in cp.items(sect): + if v and v.strip(): + out[k] = v.strip() + return out + except configparser.Error: + return {} + + +def _dnsmasq_rules(): + """dnsmasq's NIC pin: {interface, except, no_dhcp, listen} sets, from + DNSMASQ_CONF (and what it includes) plus DNSMASQ_NIC_CONF. None when + neither could be read (the pin is then unknown).""" + from services import ipxe_update # its include-aware dnsmasq.conf reader + seen = set() + lines = ipxe_update._conf_lines(config.DNSMASQ_CONF, seen=seen) + lines += ipxe_update._conf_lines(config.DNSMASQ_NIC_CONF, seen=seen) + if not lines: + return None + 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 _src, line in lines: + key, _, val = line.partition("=") + key = key.strip() + if key in keymap: + for item in val.split(","): + if item.strip(): + rules[keymap[key]].add(item.strip()) + return rules + + +def _roles(rules=None): """{iface or mac: role} from nic.conf, else from dnsmasq's NIC pin.""" roles = {} - cp = configparser.ConfigParser() - try: - cp.read_string("[nic]\n" + open(config.NIC_CONF).read()) - sec = cp["nic"] - for role in ("imaging", "ge"): - for key in (f"{role}_iface", f"{role}_mac"): - if sec.get(key): - roles[sec.get(key).strip().lower()] = role - except (OSError, configparser.Error): - pass + conf = _nic_conf() + for role in ("imaging", "ge"): + if conf.get(f"{role}_iface"): + roles[conf[f"{role}_iface"].lower()] = role + if conf.get(f"{role}_mac"): + roles[_norm_mac(conf[f"{role}_mac"])] = role if not roles: - for line in _read(config.DNSMASQ_NIC_CONF).splitlines(): - key, _, val = line.split("#", 1)[0].strip().partition("=") - if key == "interface" and val: - roles[val.strip()] = "imaging" - elif key == "except-interface" and val: - roles.setdefault(val.strip(), "ge") + rules = _dnsmasq_rules() if rules is None else rules + for name in sorted((rules or {}).get("interface", ())): + roles[name] = "imaging" + for name in sorted((rules or {}).get("except", ())): + roles.setdefault(name, "ge") return roles @@ -68,39 +117,93 @@ def _addrs(): except (OSError, subprocess.SubprocessError, ValueError): return {} res = {} - for link in data: - res[link.get("ifname", "")] = [ - f"{a['local']}/{a['prefixlen']}" for a in link.get("addr_info", []) - if a.get("family") == "inet" or (a.get("family") == "inet6" and a.get("scope") == "global")] + for link in data if isinstance(data, list) else []: + if not isinstance(link, dict): + continue + res[str(link.get("ifname", ""))] = [ + f"{a.get('local')}/{a.get('prefixlen')}" for a in link.get("addr_info", []) or [] + if isinstance(a, dict) and a.get("local") + and (a.get("family") == "inet" or (a.get("family") == "inet6" and a.get("scope") == "global"))] return res -def _dhcp_bound(): - """Interfaces dnsmasq's DHCP socket (UDP 67) is bound to; None if unknown.""" +def _dhcp_sockets(): + """[(addr, dev)] of the UDP 67 sockets; dev is None for an unbound one. + None if ss cannot be run.""" try: - out = subprocess.run(["ss", "-H", "-uln", "sport", "= :67"], capture_output=True, text=True, - timeout=5).stdout + r = subprocess.run(["ss", "-H", "-uln", "sport", "= :67"], capture_output=True, text=True, timeout=5) except (OSError, subprocess.SubprocessError): return None - bound = set(re.findall(r"%([\w.-]+):67\b", out)) - if re.search(r"(^|\s)(0\.0\.0\.0|\*|\[::\]):67\b", out): - bound.add("*") # bound on every interface - return bound + if r.returncode != 0: + return None + out = [] + for line in r.stdout.splitlines(): + fields = line.split() + if len(fields) < 4: + continue + host = fields[3].rsplit(":", 1)[0] + host, _, dev = host.partition("%") + out.append((host.strip("[]"), dev or None)) + return out + + +def _bound(name, ips, socks, rules): + """Does dnsmasq answer DHCP on `name`? True/False, or None if unknown. + Mirrors pxe-nic-check: a wildcard socket serves only what dnsmasq's + interface pin allows, so it never marks an excluded NIC as served.""" + if socks is None: + return None + if not socks: + return False + for addr, dev in socks: + if dev == name or (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 in socks): + return False + if rules is None: + return None # wildcard socket and the pin is unknown + if name in rules["no_dhcp"] or name in rules["except"]: + return False + if name in rules["interface"] or any(ip in rules["listen"] for ip in ips): + return True + if not rules["interface"] and not rules["listen"]: + return bool(ips) + return False def _kernel_view(): - """Interfaces as the kernel sees them (physical NICs only).""" - roles, addrs, bound = _roles(), _addrs(), _dhcp_bound() - out = [] + """Interfaces as the kernel sees them: physical NICs, plus bridges that + carry a role (the imaging NIC is often a bridge over the port). + Returns (interfaces, ok) where ok is False if /sys could not be read.""" + rules = _dnsmasq_rules() + roles, addrs, socks = _roles(rules), _addrs(), _dhcp_sockets() try: names = sorted(os.listdir(SYS_NET)) except OSError: - names = [] + return [], False + macs = {n: _read(os.path.join(SYS_NET, n, "address")).lower() for n in names} + bridges, port_of = {}, {} + for name in names: + brif = os.path.join(SYS_NET, name, "brif") + if os.path.isdir(os.path.join(SYS_NET, name, "bridge")): + try: + ports = sorted(os.listdir(brif)) + except OSError: + ports = [] + role = roles.get(name) or roles.get(macs.get(name, "")) + role = role or next((roles.get(p) or roles.get(macs.get(p, "")) for p in ports + if roles.get(p) or roles.get(macs.get(p, ""))), None) + if role: + bridges[name] = role + for p in ports: + port_of[p] = name + out = [] for name in names: base = os.path.join(SYS_NET, name) - if name == "lo" or not os.path.exists(os.path.join(base, "device")): - continue # skip loopback, bridges, veths, tunnels - mac = _read(os.path.join(base, "address")).lower() + physical = os.path.exists(os.path.join(base, "device")) + if name == "lo" or not (physical or name in bridges): + continue # skip loopback, plain bridges, veths, tunnels + mac = macs.get(name, "") oper = _read(os.path.join(base, "operstate")) try: speed = int(_read(os.path.join(base, "speed"))) @@ -108,21 +211,30 @@ def _kernel_view(): except ValueError: speed = None notes = [] - driver = os.path.basename(os.path.realpath(os.path.join(base, "device", "driver"))) + driver = os.path.basename(os.path.realpath(os.path.join(base, "device", "driver"))) if physical else "" if driver == "cdc_ncm" and speed in CDC_NCM_WRAPPED: notes.append(f"Link {CDC_NCM_WRAPPED[speed] // 1000} Gb/s (the cdc_ncm driver reports " f"{speed} Mb/s: its 32-bit speed counter wraps above 4.29 Gb/s)") speed = CDC_NCM_WRAPPED[speed] + if name in bridges: + role = bridges[name] + notes.append("bridge") + elif name in port_of: + role = "other" # the bridge carries the role and the address + notes.append(f"port of bridge {port_of[name]}") + else: + role = roles.get(name) or roles.get(mac) or "other" + up = oper == "up" or (name in bridges and oper == "unknown") + ips = [a.split("/")[0] for a in addrs.get(name, [])] out.append({ - "iface": name, "mac": mac, - "role": roles.get(name) or roles.get(mac) or "other", - "link": "up" if oper == "up" else "down", - "speed_mbps": speed if oper == "up" else None, + "iface": name, "mac": mac, "role": role, + "link": "up" if up else "down", + "speed_mbps": speed if up else None, "addrs": addrs.get(name, []), "verdict": None, "evidence": [], "notes": notes, - "dnsmasq_bound": (None if bound is None else (name in bound or "*" in bound)), + "dnsmasq_bound": None if name in port_of else _bound(name, ips, socks, rules), }) - return out + return out, True def _age_seconds(stamp): @@ -131,21 +243,73 @@ def _age_seconds(stamp): if t.tzinfo is None: t = t.replace(tzinfo=timezone.utc) return (datetime.now(timezone.utc) - t).total_seconds() - except ValueError: + except (ValueError, TypeError, OverflowError): return None -def _problems(ifaces, guard): +def _str(v): + return v.strip() if isinstance(v, str) else "" + + +def _str_list(v): + if isinstance(v, str): + v = [v] + if not isinstance(v, (list, tuple)): + return [] + return [x if isinstance(x, str) else str(x) for x in v + if isinstance(x, (str, int, float)) and not isinstance(x, bool)] + + +def _speed(v): + if isinstance(v, bool): + return None + try: + n = int(v) + except (TypeError, ValueError, OverflowError): + return None + return n if 0 < n < 10_000_000 else None + + +def _norm_iface(i): + """One interface entry of the check file with every field the page uses + in the type it expects; None for something that is not an entry.""" + if not isinstance(i, dict): + return None + bound = i.get("dnsmasq_bound") + return { + "iface": _str(i.get("iface")) or "?", + "mac": _norm_mac(_str(i.get("mac"))), + "role": _str(i.get("role")) or "other", + "link": "up" if _str(i.get("link")) == "up" else "down", + "speed_mbps": _speed(i.get("speed_mbps")), + "addrs": _str_list(i.get("addrs")), + "verdict": _str(i.get("verdict")) or None, + "evidence": _str_list(i.get("evidence")), + "notes": _str_list(i.get("notes")), + "dnsmasq_bound": bound if isinstance(bound, bool) else None, + } + + +def _when(stamp): + """checked_at for a sentence: 'YYYY-MM-DD HH:MM'.""" + return str(stamp or "?").replace("T", " ")[:16] + + +def _problems(ifaces, guard, stale=False, checked_at=None): """[(level, text)]: 'bad' is a DHCP-on-the-wrong-network risk.""" out = [] if (guard or {}).get("state") == "blocked": - out.append(("bad", "DHCP/PXE is blocked: " + (guard.get("reason") or "the imaging NIC is not isolated"))) + reason = guard.get("reason") or "the imaging NIC is not isolated" + if stale: + out.append(("warn", f"The last network check ({_when(checked_at)}) said DHCP/PXE is blocked: {reason}")) + else: + out.append(("bad", "DHCP/PXE is blocked: " + reason)) imaging = [i for i in ifaces if i["role"] == "imaging"] for i in ifaces: if i["role"] == "imaging" and i.get("verdict") in ("ge", "foreign-dhcp"): what = "the GE network" if i["verdict"] == "ge" else "a network with another DHCP server" out.append(("bad", f"Imaging NIC {i['iface']} is on {what}. Do not serve DHCP/PXE here.")) - if i["role"] != "imaging" and i.get("dnsmasq_bound"): + if i["role"] != "imaging" and i.get("dnsmasq_bound") is True: out.append(("bad", f"dnsmasq is answering DHCP on {i['iface']} ({i['role']} NIC), not only on the imaging NIC.")) if not imaging: out.append(("warn", "No imaging NIC configured (/etc/pxe-server/nic.conf or dnsmasq's interface= pin).")) @@ -162,7 +326,8 @@ def _problems(ifaces, guard): def status(): """{interfaces, source, checked_at, stale, guard, problems, state}. - state: 'ok' | 'warn' | 'bad'.""" + state: 'ok' | 'warn' | 'bad'. Every interface entry is normalised (see + _norm_iface), whatever the check file holds.""" data, source = None, "kernel" try: with open(config.NIC_STATUS_FILE) as f: @@ -172,27 +337,45 @@ def status(): stale = True if isinstance(data, dict) and isinstance(data.get("interfaces"), list): age = _age_seconds(data.get("checked_at")) - stale = age is None or age > STALE_SECONDS + stale = age is None or age > STALE_SECONDS or age < -FUTURE_SKEW_SECONDS source = "check" - live = {i["iface"]: i for i in _kernel_view()} + kernel, kernel_ok = _kernel_view() + live = {i["iface"]: i for i in kernel} + checked_at = None if source == "check": + checked_at = data.get("checked_at") if isinstance(data.get("checked_at"), str) else None ifaces = [] - for i in data["interfaces"]: - i = dict(i) - k = live.get(i.get("iface")) + for raw in data["interfaces"]: + i = _norm_iface(raw) + if i is None: + continue + k = live.get(i["iface"]) if k: # link and addresses are cheap to read now; keep the check's verdict - i.update(link=k["link"], speed_mbps=k["speed_mbps"], addrs=k["addrs"], notes=k["notes"]) - if k["dnsmasq_bound"] is not None: + i.update(link=k["link"], speed_mbps=k["speed_mbps"], addrs=k["addrs"], + notes=k["notes"] + [n for n in i["notes"] if n not in k["notes"]]) + # The check knows the dnsmasq config and process; the + # kernel fallback is only used once the check is stale. + if stale or i["dnsmasq_bound"] is None: i["dnsmasq_bound"] = k["dnsmasq_bound"] + elif kernel_ok: + i.update(link="down", speed_mbps=None, addrs=[], dnsmasq_bound=False, + notes=i["notes"] + ["not present (unplugged?)"]) if stale: i["verdict"] = None + i["evidence"] = [] ifaces.append(i) - guard = data.get("dhcp_guard") or {} + g = data.get("dhcp_guard") + guard = {"state": _str(g.get("state")), "reason": _str(g.get("reason"))} if isinstance(g, dict) else {} else: - ifaces, guard = list(live.values()), {} + ifaces, guard = kernel, {} ifaces.sort(key=lambda i: {"imaging": 0, "ge": 1}.get(i.get("role"), 2)) - problems = _problems(ifaces, guard) + problems = _problems(ifaces, guard, stale and source == "check", checked_at) state = "bad" if any(l == "bad" for l, _ in problems) else "warn" if problems else "ok" return {"interfaces": ifaces, "source": source, "stale": stale and source == "check", - "checked_at": (data or {}).get("checked_at") if source == "check" else None, - "guard": guard, "problems": problems, "state": state} + "checked_at": checked_at, "guard": guard, "problems": problems, "state": state} + + +def unknown(error=""): + """What the Overview shows when status() itself failed.""" + return {"interfaces": [], "source": "error", "stale": False, "checked_at": None, "guard": {}, + "problems": [], "state": "unknown", "error": str(error)[:200]} diff --git a/webapp/templates/_imaging_tiles.html b/webapp/templates/_imaging_tiles.html index 8db257d..6f8d5c4 100644 --- a/webapp/templates/_imaging_tiles.html +++ b/webapp/templates/_imaging_tiles.html @@ -60,18 +60,22 @@ {% set track_kind = {'failed': 'fail', 'action': 'act', 'running': 'run', 'done': 'ok'} %} {% macro floor_row(s) %} + {# Sessions are sanitised on load; these guards keep a row rendering if a + non-string still slips through (a lookup or `in` on a list raises). #} + {% set pctype = s.pctype if s.pctype is string else '' %} + {% set cur_stage = s.current_stage if s.current_stage is string else '' %} {% set is_inferred = (s.source == 'inferred') %} - {% set is_blancco = (s.source == 'blancco' or s.pctype == 'blancco') %} + {% set is_blancco = (s.source == 'blancco' or pctype == 'blancco') %} {% set stage_idx = (s.stage_index or 0) | int(0) %} {% set stage_total = (s.stage_total or 0) | int(0) %} - {% set state = s.floor_state or 'running' %} + {% set state = s.floor_state if s.floor_state in state_chip else 'running' %} {% set drive_pct = None %} - {% if is_blancco and stage_idx == 4 and s.current_stage and '%]' in s.current_stage %} - {% set drive_pct = s.current_stage.split('[')[-1].split('%')[0] | int(0) %} + {% if is_blancco and stage_idx == 4 and cur_stage and '%]' in cur_stage %} + {% set drive_pct = cur_stage.split('[')[-1].split('%')[0] | int(0) %} {% endif %} {% set warn = False %} {% if is_inferred %} - {% set friendly = inferred_stage_labels.get(s.current_stage, ('Inferred from logs', s.current_stage or '')) %} + {% set friendly = inferred_stage_labels.get(cur_stage, ('Inferred from logs', cur_stage)) %} {% elif is_blancco %} {% if s.status == 'failed' %} {% set friendly = ('Erasure FAILED', 'Blancco reported a failed erasure step. Check the log tail and the report.') %} @@ -80,18 +84,18 @@ {% endif %} {% if stage_idx == 4 and s.status != 'failed' %}{% set warn = True %}{% endif %} {% else %} - {% set friendly = (desk_stage_labels if s.pctype in desk_kind else stage_labels).get(stage_idx, ('Stage ' ~ stage_idx, '')) %} - {% if stage_idx == 1 and s.current_stage and 'bios update staged' in s.current_stage|lower %} + {% set friendly = (desk_stage_labels if pctype in desk_kind else stage_labels).get(stage_idx, ('Stage ' ~ stage_idx, '')) %} + {% if stage_idx == 1 and cur_stage and 'bios update staged' in cur_stage|lower %} {% set friendly = ('Applying image - BIOS update staged', 'WinPE staged a firmware update for this model. It flashes during the reboot after the image is applied: do NOT power off then.') %} {% set warn = True %} - {% elif stage_idx == 1 and s.current_stage and 'bios' in s.current_stage|lower %} + {% elif stage_idx == 1 and cur_stage and 'bios' in cur_stage|lower %} {% set friendly = ('Updating BIOS: do not power off', 'WinPE detected a firmware update for this model. Do NOT power off until the next reboot. Imaging continues afterward.') %} {% set warn = True %} {% endif %} - {% if stage_idx == 7 and s.current_stage %} - {% set _cs = s.current_stage|lower %} + {% if stage_idx == 7 and cur_stage %} + {% set _cs = cur_stage|lower %} {% if 'ready for lockdown' in _cs or 'request lockdown' in _cs %} {% set friendly = ('Ready - request lockdown', 'Phase 1-4 all complete (Registration, Device Config, Software Deploy, Credentials). Click "ARTS request" to initiate the lockdown workflow.') %} @@ -112,11 +116,11 @@ {% if s.hostname_target and s.serial %}{% set _ = ident.append(s.hostname_target) %}{% endif %} {% if s.ip %}{% set _ = ident.append(s.ip) %}{% endif %} {% set title = s.serial or s.hostname_target or s.mac or s.ip or '(unknown bay)' %} - {% set kind = 'Blancco wipe' if is_blancco else desk_kind.get(s.pctype, s.pctype or '') %} + {% set kind = 'Blancco wipe' if is_blancco else desk_kind.get(pctype, pctype) %}
+ data-filter="{{ (s.serial or '')|lower }} {{ (s.hostname_target or '')|lower }} {{ pctype|lower }} {{ (s.machinenumber or '')|lower }} {{ (s.intune_device_id or '')|lower }} {{ friendly[0]|lower }} stage-{{ stage_idx }} {{ (s.status or 'in_progress')|lower }} {{ (s.mac or '')|lower }} {{ (s.ip or '') }} {{ s.source or 'client' }}">
{{ ui.chip(state_chip[state][0], state_chip[state][1]) }} diff --git a/webapp/templates/_nic_status.html b/webapp/templates/_nic_status.html new file mode 100644 index 0000000..5545d3f --- /dev/null +++ b/webapp/templates/_nic_status.html @@ -0,0 +1,50 @@ +{# Overview network pill and its problem notice. Rendered on its own by + the dashboard view (part = 'pill' or 'notice') inside a try, so a broken + status file can never take the Overview down. #} +{% import '_ui.html' as ui %} +{% if part == 'pill' %} + {% set nic_dot = {'ok': 'var(--ok)', 'warn': 'var(--act)', 'bad': 'var(--fail)'}.get(nic.state, 'var(--off-edge)') %} + {% set role_name = {'imaging': 'Imaging', 'ge': 'GE network', 'other': 'Other'} %} + {% set verdict_chip = {'isolated': ('Isolated', 'ok'), 'ge': ('GE network', 'neutral'), 'foreign-dhcp': ('Other DHCP server', 'err'), + 'no-link': ('No link', 'off'), 'error': ('Check failed', 'warn')} %} +
+ + + {{ {'ok': 'Network OK', 'warn': 'Network: check', 'bad': 'Network: DANGER'}.get(nic.state, 'Network: unknown') }} + + +
+ {% for i in nic.interfaces %} + {% set bad_here = (i.role == 'imaging' and i.verdict in ('ge', 'foreign-dhcp')) or (i.role != 'imaging' and i.dnsmasq_bound) %} +
+
+ + {{ role_name.get(i.role, i.role) }} + {{ i.iface }} + {% if i.verdict and i.verdict in verdict_chip %}{% set vc = verdict_chip[i.verdict] %}{{ ui.chip(vc[0], 'err' if bad_here else vc[1]) }}{% endif %} +
+
+ {{ 'Up' if i.link == 'up' else 'No link' }}{% if i.speed_mbps is number and i.speed_mbps > 0 %}, {% if i.speed_mbps >= 1000 %}{{ (i.speed_mbps // 1000) if i.speed_mbps % 1000 == 0 else (i.speed_mbps / 1000) | round(1) }} Gb/s{% else %}{{ i.speed_mbps }} Mb/s{% endif %}{% endif %} + {% if i.addrs and i.addrs is not string %}{{ i.addrs | join(', ') }}{% endif %} + {% if i.dnsmasq_bound is not none %}
DHCP/PXE {{ 'served here' if i.dnsmasq_bound else 'not served here' }}{% endif %} + {% for e in (i.evidence if i.evidence is iterable and i.evidence is not string else []) %}
{{ e }}{% endfor %}{% for n in (i.notes if i.notes is iterable and i.notes is not string else []) %}
{{ n }}{% endfor %} +
+
+ {% else %} +
{{ 'Interfaces unknown.' if nic.source == 'error' else 'No network interfaces found.' }}
+ {% endfor %} +
+ {% if nic.source == 'check' and not nic.stale %}Checked {{ nic.checked_at | local_dt }} by pxe-nic-check. + {% elif nic.source == 'check' %}The network check is stale (last {{ nic.checked_at | local_dt }}): showing link and addresses only. + {% elif nic.source == 'error' %}The network status could not be read{% if nic.error %}: {{ nic.error }}{% endif %}. See the webapp log. + {% else %}Link and addresses only; pxe-nic-check (the GE-network detector) is not installed yet.{% endif %} +
+
+
+{% elif part == 'notice' %} +{% if nic and nic.problems %} +
+ {% if nic.state == 'bad' %}Network danger. {% endif %}{% for level, text in nic.problems %}{{ text }}{% if not loop.last %} {% endif %}{% endfor %} +
+{% endif %} +{% endif %} diff --git a/webapp/templates/dashboard.html b/webapp/templates/dashboard.html index f7d5716..6c283fe 100644 --- a/webapp/templates/dashboard.html +++ b/webapp/templates/dashboard.html @@ -5,45 +5,7 @@ {% block content %} {% set services_up = services | selectattr('active') | list | length %} {% call ui.head('Overview', 'West Jefferson') %} - {% if nic %} - {% set nic_dot = {'ok': 'var(--ok)', 'warn': 'var(--act)', 'bad': 'var(--fail)'}[nic.state] %} - {% set role_name = {'imaging': 'Imaging', 'ge': 'GE network', 'other': 'Other'} %} - {% set verdict_chip = {'isolated': ('Isolated', 'ok'), 'ge': ('GE network', 'neutral'), 'foreign-dhcp': ('Other DHCP server', 'err'), - 'no-link': ('No link', 'off'), 'error': ('Check failed', 'warn')} %} -
- - - {{ 'Network OK' if nic.state == 'ok' else ('Network: check' if nic.state == 'warn' else 'Network: DANGER') }} - - -
- {% for i in nic.interfaces %} - {% set bad_here = (i.role == 'imaging' and i.verdict in ('ge', 'foreign-dhcp')) or (i.role != 'imaging' and i.dnsmasq_bound) %} -
-
- - {{ role_name.get(i.role, i.role) }} - {{ i.iface }} - {% if i.verdict and i.verdict in verdict_chip %}{% set vc = verdict_chip[i.verdict] %}{{ ui.chip(vc[0], 'err' if bad_here else vc[1]) }}{% endif %} -
-
- {{ 'Up' if i.link == 'up' else 'No link' }}{% if i.speed_mbps %}, {% if i.speed_mbps >= 1000 %}{{ (i.speed_mbps // 1000) if i.speed_mbps % 1000 == 0 else (i.speed_mbps / 1000) | round(1) }} Gb/s{% else %}{{ i.speed_mbps }} Mb/s{% endif %}{% endif %} - {% if i.addrs %}{{ i.addrs | join(', ') }}{% endif %} - {% if i.dnsmasq_bound is not none %}
DHCP/PXE {{ 'served here' if i.dnsmasq_bound else 'not served here' }}{% endif %} - {% for e in i.evidence %}
{{ e }}{% endfor %}{% for n in i.notes or [] %}
{{ n }}{% endfor %} -
-
- {% else %} -
No network interfaces found.
- {% endfor %} -
- {% if nic.source == 'check' and not nic.stale %}Checked {{ nic.checked_at | local_dt }} by pxe-nic-check. - {% elif nic.source == 'check' %}The network check is stale (last {{ nic.checked_at | local_dt }}): showing link and addresses only. - {% else %}Link and addresses only; pxe-nic-check (the GE-network detector) is not installed yet.{% endif %} -
-
-
- {% endif %} + {{ nic_pill }}
@@ -59,11 +21,7 @@ {% endcall %}
-{% if nic and nic.problems %} -
- {% if nic.state == 'bad' %}Network danger. {% endif %}{% for level, text in nic.problems %}{{ text }}{% if not loop.last %} {% endif %}{% endfor %} -
-{% endif %} +{{ nic_notice }} {% if blancco_todo.total %} {% set parts = [] %} {% if blancco_todo.servicenow %}{% set _ = parts.append(blancco_todo.servicenow ~ ' to tick in ServiceNow') %}{% endif %} diff --git a/webapp/templates/reports.html b/webapp/templates/reports.html index 57c2da2..4d04230 100644 --- a/webapp/templates/reports.html +++ b/webapp/templates/reports.html @@ -71,7 +71,7 @@
-
+
diff --git a/webapp/templates/updates_ipxe.html b/webapp/templates/updates_ipxe.html index 0273b9b..7bccba0 100644 --- a/webapp/templates/updates_ipxe.html +++ b/webapp/templates/updates_ipxe.html @@ -5,7 +5,8 @@ {# Signer as a short chip; the long explanation goes in the row body (it is a safety warning, so never hover-only). #} {% macro sig_chip(info) -%} - {% if info.kind == 'microsoft-2011' %}{{ ui.chip('MS UEFI CA 2011 only', 'warn') }} + {% if info.dual %}{{ ui.chip('MS UEFI CA 2011 + 2023', 'ok') }} + {% elif info.kind == 'microsoft-2011' %}{{ ui.chip('MS UEFI CA 2011 only', 'warn') }} {% elif info.kind == 'microsoft-2023' %}{{ ui.chip('MS UEFI CA 2023', 'ok') }} {% elif info.kind == 'microsoft' %}{{ ui.chip('Microsoft, other CA', 'neutral') }} {% elif info.kind == 'ipxe' %}{{ ui.chip('iPXE Secure Boot CA', 'ok') }} @@ -17,7 +18,8 @@ {% macro sig_text(info, chain_ok=none) -%} {% if info.kind == 'microsoft-2011' %}
Signed only under Microsoft Corporation UEFI CA 2011. That CA's signing period ended in 2026: machines that trust only the 2023 CA, or a DBX/SBAT revocation update, will refuse this file.
- {% elif info.kind == 'microsoft-2023' %}
Signed under Microsoft UEFI CA 2023.
+ {% elif info.dual %}
Dual-signed: under Microsoft Corporation UEFI CA 2011 and Microsoft UEFI CA 2023.
+ {% elif info.kind == 'microsoft-2023' %}
Signed under Microsoft UEFI CA 2023 only, not dual-signed: machines that trust only the 2011 CA will refuse it.
{% elif info.kind == 'ipxe' %}
Chains to the iPXE Secure Boot CA, which the iPXE shim trusts.
{% elif info.kind == 'bios' %}
Legacy BIOS has no Secure Boot; the file is unsigned by design.
{% elif info.kind == 'unsigned' %}
Unsigned: Secure Boot firmware will refuse it.
@@ -149,7 +151,7 @@
{{ 'Starting' if job.state == 'queued' else (job.step | capitalize) }} {{ job.tag }}... - +
{% elif job.state == 'error' %}
Download of {{ job.tag }} failed: {{ job.error }}
@@ -213,8 +215,9 @@

Per-client folders

With tftp-unique-root=mac, a client whose MAC has a folder gets every file from it, with no fallback to the root. Read-only here: installs do not update them.
{% for m in mac_folders %} -
{{ m.mac }}/srv/tftp/{{ m.dir }}/ +
{{ m.mac }}{{ tftp_root }}/{{ m.dir }}/ {{ m.files | length }} file{{ 's' if m.files | length != 1 }}
+ {% for w in m.warnings or [] %}
{{ w }}
{% endfor %} {% for f in m.files %} {% set i = f.info %} {% set chain_ok = i.chain_ok if i.chain_ok is defined else none %} @@ -229,7 +232,7 @@
{{ sig_text(i, chain_ok) }}
-
Path
/srv/tftp/{{ m.dir }}/{{ f.name }}
+
Path
{{ tftp_root }}/{{ m.dir }}/{{ f.name }}
Size
{{ "{:,}".format(i.size) }} B
Modified
{{ i.modified | timestamp_fmt }}
sha256
{{ i.sha256 }}
@@ -289,6 +292,7 @@
{{ s.tag }} verified {{ s.staged | local_dt }} + {% if s.stale %}{{ ui.chip('Stale: download again', 'warn') }}{% endif %} tarball sha256 {{ (s.tarball_sha256 or '')[:12] }} matches GitHub
{% for name, f in s.files.items() %} @@ -380,7 +384,8 @@
{% if b.reason %}
{{ b.reason }}
{% endif %} - {% for f in b.files %}
{{ f }} {{ b.versions[f] or '' }}
{% else %}
Empty backup.
{% endfor %} + {% for f in b.files %}
{{ f }} {{ b.versions[f] or '' }}{% if b.info and b.info[f] %} {{ sig_chip(b.info[f]) }}{% endif %}
{% else %}
Empty backup.
{% endfor %} + {% for w in b.warnings or [] %}
{{ w }}
{% endfor %}
{% if b.files %} @@ -392,6 +397,8 @@ onsubmit="return confirm(this.dataset.confirm);"> + {% if b.warnings %}{% endif %} {% endcall %} @@ -437,8 +444,8 @@ ls -l ipxeboot/x86_64-sb/ 
sbverify --list ipxe-shimx64.efi
 sbverify --list ipxeboot/x86_64-sb/ipxe.efi
 sbverify --list ipxeboot/x86_64-sb/snponly.efi
- The dual-signed shim shows CN=Microsoft UEFI CA 2023 (sbverify lists the first signature only; - the 2011 one is nested inside it). Fall back to the tarball's shimx64.efi + The dual-signed shim shows CN=Microsoft UEFI CA 2023 and CN=Microsoft Corporation UEFI CA 2011 + (depending on the sbverify version it lists both signatures or only the outer one; this page reads every signature, nested ones too). Fall back to the tarball's shimx64.efi (CN=Microsoft Corporation UEFI CA 2011) only if the dual-signed one is unavailable. ipxe.efi must show CN=iPXE Secure Boot Intermediate ... issued by CN=iPXE Secure Boot CA. The shim's warning: data remaining ... gaps between PE/COFF sections? @@ -484,7 +491,7 @@ install -m 0644 ipxe-shimx64.efi /srv/tftp/.ipxe-shim.efi.new & if (j.state !== 'queued' && j.state !== 'running') { location.reload(); return; } var step = j.step === 'download' ? 'Downloading' : j.step === 'verify' ? 'Verifying' : j.step === 'shim' ? 'Fetching the dual-signed shim for' : 'Starting'; if (text) text.textContent = step + ' ' + (j.tag || '') + (j.step === 'download' && j.percent != null ? ' (' + j.percent + '%)' : '') + '...'; - if (bar) bar.style.width = (j.step === 'verify' ? 100 : (j.percent || 0)) + '%'; + if (bar) bar.style.width = ((j.step === 'verify' || j.step === 'shim') ? 100 : (j.percent || 0)) + '%'; setTimeout(tick, 2000); }) .catch(function () { setTimeout(tick, 5000); }); diff --git a/webapp/tests/test_review_fixes.py b/webapp/tests/test_review_fixes.py new file mode 100644 index 0000000..333dda0 --- /dev/null +++ b/webapp/tests/test_review_fixes.py @@ -0,0 +1,452 @@ +"""Regression tests for the 2026-10-15 webapp review findings. + +Run from the webapp folder: + PYTHONPATH=.: 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() diff --git a/webapp/views/ipxe.py b/webapp/views/ipxe.py index bbde839..3194260 100644 --- a/webapp/views/ipxe.py +++ b/webapp/views/ipxe.py @@ -9,6 +9,7 @@ services/ipxe_update.py; this module is the route surface. from flask import Blueprint, flash, jsonify, redirect, render_template, request, url_for +import config from services import ipxe_update, updates_common from services.audit import audit from views import save_speed_cap @@ -33,6 +34,7 @@ def page(): installed_sha={n: (i or {}).get("sha256") for n, i in by_name.items()}, routing=ipxe_update.dnsmasq_routing(), mac_folders=ipxe_update.mac_folders(), + tftp_root=config.TFTP_ROOT.rstrip("/") or "/", latest=latest, status=status, job=ipxe_update.job_state(), @@ -99,7 +101,7 @@ def rollback(): # rollback() looks the id up by its last path part only; log that. backup_id = request.form.get("backup", "").replace("\\", "/").rsplit("/", 1)[-1] try: - files = ipxe_update.rollback(backup_id) + files = ipxe_update.rollback(backup_id, confirm=request.form.get("confirm") == "1") except Exception as ex: flash(f"Rollback failed: {ex}", "danger") return _back()