From b691a8253033aa5af058583761c41f923b8f898b Mon Sep 17 00:00:00 2001 From: cproudlock Date: Fri, 9 Oct 2026 09:23:52 -0400 Subject: [PATCH] Hardware coverage: flag models the server has no drivers or BIOS for WinPE now reports each machine's WMI model, manufacturer, Dell system SKU and BIOS version to /hardware/seen (winpe-status-push.ps1, best effort, wmic only). The webapp records every model seen and checks it against what the server has: - each image's Deploy/Control/HardwareDriver.json, matched exactly as PESetup's GetDriverByModel does (as reimplemented in lint-driver-catalogue.py), plus whether the pack's zip is on disk; - the WinPE BIOS manifest (substring match like check-bios.cmd). /hardware lists every model with per-image driver status and BIOS status. A model missing a Win11 pack or a BIOS entry becomes item 5 on the Technician To-Do (and the banner count) until the pack/BIOS is added or it is dismissed. Virtual machines are ignored. Also deploys the repo winpe-status-push.ps1 over the live copy, which predated the -CurrentStage/-StageIndex parameters startnet.cmd passes for the BIOS-flash stage, so that status post was failing. --- playbook/winpe-status-push.ps1 | 32 ++++ webapp/app.py | 37 +++- webapp/config.py | 3 + webapp/services/csrf.py | 1 + webapp/services/hardware_coverage.py | 272 +++++++++++++++++++++++++++ webapp/templates/base.html | 7 + webapp/templates/hardware.html | 81 ++++++++ webapp/templates/reports_todo.html | 53 ++++++ 8 files changed, 485 insertions(+), 1 deletion(-) create mode 100644 webapp/services/hardware_coverage.py create mode 100644 webapp/templates/hardware.html diff --git a/playbook/winpe-status-push.ps1 b/playbook/winpe-status-push.ps1 index 9fba978..5226d48 100644 --- a/playbook/winpe-status-push.ps1 +++ b/playbook/winpe-status-push.ps1 @@ -74,4 +74,36 @@ try { Log "ERR $($_.Exception.Message)" } +# Hardware facts for the webapp's driver/BIOS coverage check (/hardware): +# the exact WMI model PESetup matches driver packs on, the Dell system SKU +# (what Dell's catalogs key on), and the BIOS version check-bios.cmd sees. +function Get-WmicValue { + param([string]$Class, [string]$Property) + try { + $line = (& wmic $Class get $Property /value 2>$null | + Where-Object { $_ -match '=' } | Select-Object -First 1) + if ($line) { return ($line -split '=', 2)[1].Trim() } + } catch { Log "wmic $Class $Property exception: $_" } + return '' +} +$hw = @{ + serial = $serial + model = Get-WmicValue 'computersystem' 'Model' + manufacturer = Get-WmicValue 'computersystem' 'Manufacturer' + sku = Get-WmicValue 'computersystem' 'SystemSKUNumber' + bios_version = Get-WmicValue 'bios' 'SMBIOSBIOSVersion' + pctype = $PCType +} +Log "hardware model=$($hw.model) sku=$($hw.sku) bios=$($hw.bios_version)" +if ($hw.model) { + try { + $resp = Invoke-WebRequest -Uri "http://${PxeServer}:${Port}/hardware/seen" -Method POST ` + -Body ($hw | ConvertTo-Json -Compress) -ContentType 'application/json' ` + -UseBasicParsing -TimeoutSec $TimeoutSec -ErrorAction Stop + Log "hardware OK http=$($resp.StatusCode)" + } catch { + Log "hardware ERR $($_.Exception.Message)" + } +} + exit 0 diff --git a/webapp/app.py b/webapp/app.py index 7bfaa73..5f67604 100644 --- a/webapp/app.py +++ b/webapp/app.py @@ -41,6 +41,7 @@ from werkzeug.utils import secure_filename import config from services import ( blancco_evidence, + hardware_coverage, blancco_report, deploy, fs, @@ -721,6 +722,7 @@ def blancco_servicenow_mark(document_id): def blancco_todo(): """Technician to-do: the playbook steps the server cannot do itself.""" return render_template("reports_todo.html", todo=blancco_evidence.todo(), + hardware=hardware_coverage.open_gaps(), site=config.BLANCCO_SITE) @@ -744,6 +746,37 @@ def blancco_evidence_file(filename): return send_file(fpath, as_attachment=True) +@app.route("/hardware/seen", methods=["POST"]) +def hardware_seen(): + # CSRF-exempt machine-to-machine endpoint: WinPE (winpe-status-push.ps1) + # reports each machine's model, SKU and BIOS version before imaging. + p = request.get_json(silent=True) or {} + try: + hardware_coverage.record(p) + except ValueError as ex: + return jsonify({"error": str(ex)}), 400 + return jsonify({"ok": True}), 200 + + +@app.route("/hardware") +def hardware_page(): + """Every model WinPE has seen, with driver-pack and BIOS coverage.""" + return render_template("hardware.html", rows=hardware_coverage.coverage(), + images=hardware_coverage.IMAGES) + + +@app.route("/hardware/dismiss", methods=["POST"]) +def hardware_dismiss(): + model = request.form.get("model", "") + done = request.form.get("done", "1") == "1" + try: + hardware_coverage.dismiss(model, done) + except ValueError: + abort(400) + audit("HARDWARE_DISMISS", f"{model} {'dismissed' if done else 'reopened'}") + return redirect(request.form.get("next") or url_for("hardware_page")) + + @app.route("/reports/download/") def blancco_download_report(filename): filename = secure_filename(filename) @@ -1531,7 +1564,9 @@ def inject_globals(): return { "all_image_types": config.IMAGE_TYPES, "all_friendly_names": config.FRIENDLY_NAMES, - "blancco_todo": blancco_evidence.todo_counts(), + "blancco_todo": dict(blancco_evidence.todo_counts(), + hardware=hardware_coverage.gap_count(), + total=blancco_evidence.todo_counts()["total"] + hardware_coverage.gap_count()), } diff --git a/webapp/config.py b/webapp/config.py index f234d22..84b2800 100644 --- a/webapp/config.py +++ b/webapp/config.py @@ -28,6 +28,9 @@ BLANCCO_SITE = os.environ.get("BLANCCO_SITE", "West Jefferson") # Every time the webapp shows (and evidence-pack month boundaries) is in # this zone, DST included. Stored timestamps keep their own UTC offsets. DISPLAY_TZ = os.environ.get("DISPLAY_TZ", "America/New_York") +# Hardware models seen by WinPE, checked against driver packs and BIOS. +HARDWARE_STATE = os.environ.get("HARDWARE_STATE", "/var/lib/pxe-webapp/hardware-seen.json") +BIOS_DIR = os.environ.get("BIOS_DIR", "/srv/samba/winpeapps/_shared/BIOS") BLANCCO_STATE = os.environ.get("BLANCCO_STATE", "/var/lib/pxe-webapp/blancco-state.json") # Log sources used by services/imaging_log_tail.py to infer client progress diff --git a/webapp/services/csrf.py b/webapp/services/csrf.py index 280f9b7..d04ddc8 100644 --- a/webapp/services/csrf.py +++ b/webapp/services/csrf.py @@ -10,6 +10,7 @@ from flask import abort, request, session CSRF_EXEMPT_PATHS = { "/imaging/status", "/reports/cloud-status", + "/hardware/seen", } diff --git a/webapp/services/hardware_coverage.py b/webapp/services/hardware_coverage.py new file mode 100644 index 0000000..e4cdf6a --- /dev/null +++ b/webapp/services/hardware_coverage.py @@ -0,0 +1,272 @@ +"""Which hardware models the PXE server can image, and which it cannot. + +WinPE (playbook/winpe-status-push.ps1) reports each machine's WMI model, +manufacturer, Dell system SKU and BIOS version to /hardware/seen before +imaging starts. This module records the models seen and checks each one +against what the server actually has: + + * drivers: every image's Deploy/Control/HardwareDriver.json, matched the + way PESetup's GetDriverByModel matches (scripts/lint-driver-catalogue.py + reimplements it and documents why), plus whether the pack's zip is on + disk; + * BIOS: the WinPE BIOS manifest (models.txt, substring match like + check-bios.cmd) and whether the machine is already on that version. + +A model missing drivers for an image, or with no BIOS entry, becomes a +technician to-do until it is fixed (the check re-runs on every page load, +so adding the pack clears it) or dismissed. + +State: one JSON file, config.HARDWARE_STATE, written atomically: + {"models": {model: {manufacturer, skus[], serials[], bios_versions[], + pctypes[], first_seen, last_seen, count}}, + "dismissed": {model: {"ts"}}} +""" +from __future__ import annotations + +import json +import os +import tempfile +import time +from datetime import datetime, timezone +from pathlib import PurePosixPath + +import config + +# Images whose HardwareDriver.json is checked, with the Windows they deploy. +IMAGES = ( + ("gea-standard", "Win11"), + ("gea-shopfloor", "Win11 LTSC"), + ("gea-engineer", "Win10"), +) +KEEP = 10 # serials / versions remembered per model + + +# --- PESetup's matcher (see scripts/lint-driver-catalogue.py) --------------- + +def _model_family(model): + upper = model.upper() + family = "" + if "LATITUDE" in upper: + family = "Latitude" + if "OPTIPLEX" in upper: + family = "Optiplex" + if "PRECISION" in upper: + family = "Precision" + return family + + +def _entry_matches(entry, model): + family = _model_family(model) + if family and family.lower() not in str(entry.get("family") or "").lower(): + return False + # Not trimmed, case-insensitive substring: what the C# does. + return any(t.lower() in model.lower() + for t in str(entry.get("modelswminame") or "").split(",")) + + +def _zip_relpath(entry): + dest = (str(entry.get("destinationDir") or entry.get("DestinationDir") or "") + .replace("*destinationdir*\\", "").replace("*destinationdir*", "") + .replace("\\", "/").strip("/")) + name = str(entry.get("fileName") or entry.get("FileName") or "") + return str(PurePosixPath(dest) / name) if name else None + + +def _exists_ci(root, relpath): + """Path exists, matching each component case-insensitively (Dell_11/ + Optiplex vs OptiPlex has split driver trees before).""" + cur = root + for part in PurePosixPath(relpath).parts: + try: + names = os.listdir(cur) + except OSError: + return False + hit = next((n for n in names if n.lower() == part.lower()), None) + if hit is None: + return False + cur = os.path.join(cur, hit) + return os.path.isfile(cur) + + +_catalog_cache = {"at": 0.0, "value": None} + + +def _catalogs(ttl=60.0): + now = time.monotonic() + if _catalog_cache["value"] is None or now - _catalog_cache["at"] > ttl: + cats = {} + for image, _ in IMAGES: + path = os.path.join(config.SAMBA_SHARE, image, "Deploy", "Control", "HardwareDriver.json") + try: + with open(path, encoding="utf-8-sig") as f: + data = json.load(f) + cats[image] = data if isinstance(data, list) else [] + except (OSError, ValueError): + cats[image] = None # image not present on this server + _catalog_cache.update(at=now, value=cats) + return _catalog_cache["value"] + + +def driver_status(model): + """Per image: {"status": ok|no-entry|zip-missing|no-image, "file": ...}.""" + out = {} + for image, os_label in IMAGES: + entries = _catalogs().get(image) + if entries is None: + out[image] = {"status": "no-image", "os": os_label} + continue + entry = next((e for e in entries if _entry_matches(e, model)), None) + if entry is None: + out[image] = {"status": "no-entry", "os": os_label} + continue + rel = _zip_relpath(entry) + root = os.path.join(config.SAMBA_SHARE, image) + ok = bool(rel) and _exists_ci(root, rel) + out[image] = {"status": "ok" if ok else "zip-missing", "os": os_label, + "file": entry.get("fileName") or entry.get("FileName") or ""} + return out + + +def _ver(v): + parts = [] + for p in str(v or "").replace("A", "").split("."): + try: + parts.append(int(p)) + except ValueError: + parts.append(0) + return tuple(parts) + + +def bios_status(model, current_version=""): + """{"status": none|update|current|file-missing, "target", "file"}.""" + manifest = os.path.join(config.BIOS_DIR, "models.txt") + try: + with open(manifest, encoding="utf-8", errors="replace") as f: + lines = f.read().splitlines() + except OSError: + return {"status": "none", "target": "", "file": ""} + for line in lines: + line = line.strip() + if not line or line.startswith("#") or line.count("|") < 2: + continue + sub, bfile, target = (x.strip() for x in line.split("|")[:3]) + if sub and sub.lower() in model.lower(): + if not os.path.isfile(os.path.join(config.BIOS_DIR, bfile)): + return {"status": "file-missing", "target": target, "file": bfile} + if current_version and _ver(current_version) >= _ver(target): + return {"status": "current", "target": target, "file": bfile} + return {"status": "update", "target": target, "file": bfile} + return {"status": "none", "target": "", "file": ""} + + +# --- state ------------------------------------------------------------------ + +def _load(): + try: + with open(config.HARDWARE_STATE) as f: + state = json.load(f) + except (OSError, ValueError): + state = {} + state.setdefault("models", {}) + state.setdefault("dismissed", {}) + return state + + +def _save(state): + folder = os.path.dirname(config.HARDWARE_STATE) + os.makedirs(folder, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=folder, prefix=".hardware-") + with os.fdopen(fd, "w") as f: + json.dump(state, f, indent=1, sort_keys=True) + os.replace(tmp, config.HARDWARE_STATE) + + +def _now(): + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +def _push(lst, value): + if value and value not in lst: + lst.append(value) + del lst[:-KEEP] + + +def record(payload): + """Remember one machine sighting from WinPE.""" + model = " ".join(str(payload.get("model") or "").split()) + if not model: + raise ValueError("missing model") + model = model[:120] + state = _load() + rec = state["models"].setdefault(model, { + "first_seen": _now(), "skus": [], "serials": [], "bios_versions": [], + "pctypes": [], "count": 0}) + rec["manufacturer"] = str(payload.get("manufacturer") or rec.get("manufacturer") or "")[:60] + rec["last_seen"] = _now() + rec["count"] = int(rec.get("count") or 0) + 1 + _push(rec["skus"], str(payload.get("sku") or "").strip()[:40]) + _push(rec["serials"], str(payload.get("serial") or "").strip()[:64]) + _push(rec["bios_versions"], str(payload.get("bios_version") or "").strip()[:40]) + _push(rec["pctypes"], str(payload.get("pctype") or "").strip()[:60]) + rec["last_bios_version"] = str(payload.get("bios_version") or "").strip()[:40] + _save(state) + invalidate() + + +def dismiss(model, done=True): + state = _load() + if model not in state["models"]: + raise ValueError("unknown model") + if done: + state["dismissed"][model] = {"ts": _now()} + else: + state["dismissed"].pop(model, None) + _save(state) + invalidate() + + +# Virtual machines and other non-Dell hardware have no packs by design. +_SKIP = ("virtual", "vmware", "qemu", "hyper-v", "kvm") + + +def coverage(include_ok=True): + """Every model seen, newest first, with driver and BIOS status.""" + state = _load() + rows = [] + for model, rec in state["models"].items(): + if any(s in model.lower() for s in _SKIP): + continue + drivers = driver_status(model) + bios = bios_status(model, rec.get("last_bios_version", "")) + missing = [img for img, d in drivers.items() + if d["status"] in ("no-entry", "zip-missing") and img != "gea-engineer"] + gap = bool(missing) or bios["status"] in ("none", "file-missing") + row = dict(rec, model=model, drivers=drivers, bios=bios, missing=missing, + gap=gap, dismissed=model in state["dismissed"]) + if include_ok or gap: + rows.append(row) + rows.sort(key=lambda r: r.get("last_seen") or "", reverse=True) + return rows + + +def open_gaps(): + return [r for r in coverage(include_ok=False) if not r["dismissed"]] + + +_count_cache = {"at": 0.0, "value": None} + + +def gap_count(ttl=60.0): + now = time.monotonic() + if _count_cache["value"] is None or now - _count_cache["at"] > ttl: + try: + _count_cache["value"] = len(open_gaps()) + except Exception: + _count_cache["value"] = 0 + _count_cache["at"] = now + return _count_cache["value"] + + +def invalidate(): + _count_cache["value"] = None + _catalog_cache["value"] = None diff --git a/webapp/templates/base.html b/webapp/templates/base.html index abc2c90..1b05952 100644 --- a/webapp/templates/base.html +++ b/webapp/templates/base.html @@ -75,6 +75,12 @@ Blancco Reports +