ge-aerospace/pxe-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Webapp: Updates pages, image page tabs with review-before-save, BPRT sync

Browse Source 
Updates group:
- Dell drivers and BIOS: check Dell's catalogs against all three driver maps
  and models.txt, download with a shared speed cap, repack, preview, apply
  with backups and rollback.
- iPXE boot files: check GitHub for the signed Secure Boot set, verify the
  sha256 and each file's signer, install atomically, roll back. Includes the
  by-hand procedure.
- Blancco: upload a release, stage it, test-boot it from the menu, activate,
  roll back.
- BPRT sync: validate a batch uploaded from the MediaCreator Lite cache
  (scripts/windows/Sync-BPRT.ps1 or a zip), show the diff, swap with the site
  overlay files kept.

Images: one sidebar entry per image with Overview, Drivers, Unattend and OS and
Packages tabs. Every save shows a diff first, keeps the file's format, makes a
backup and can be undone. Driver lint runs inline.

Enrollment: upload standard/engineer MCL packages, one per tenant/purpose,
with archive and restore. Hardware coverage counts engineer as Win11.
This commit is contained in:
cproudlock
2026-10-13 16:58:55 -04:00
parent ddf0a2fdf2
commit 397063f987
39 changed files with 15109 additions and 1106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1459
webapp/services/blancco_update.py Normal file
View File

File diff suppressed because it is too large Load Diff

1467
webapp/services/bprt_sync.py Normal file
View File

File diff suppressed because it is too large Load Diff

478
webapp/services/control_files.py Normal file
View File

@@ -0,0 +1,478 @@
"""Safe edits of an image's control files: preview, backup, atomic write, undo.
Every save on the image pages (Drivers, Unattend, OS and Packages) follows
the same three steps, and this module is the shared implementation:
1. plan_json(path, data) / plan_text(path, text) render what WOULD be
written, in the file's own on-disk format, and diff it against the
current file. Nothing is written. The page shows plan["diff"].
2. commit(image_type, scope, plans, action) backs every changed file up
next to itself as <name>.bak-<YYYYmmdd-HHMMSS>-webui, writes the new
text atomically (temp file in the same dir + os.replace, original
owner and mode kept) and remembers the save as the undo point for
(image_type, scope). Pass expect={path: sha256} (from the preview) to
refuse the save when a file changed after the preview was shown.
3. undo(image_type, scope) puts the backups of the last commit back, but
only while every file still holds exactly what that commit wrote.
The on-disk format is never changed by a save: JSON keeps its indent, BOM,
line endings, trailing newline and ASCII escaping (detect_json_format).
A file that does not round-trip byte for byte (a hand-edited vendor file
with mixed tabs and spaces, say) is not re-rendered whole: for a top-level
JSON list the entries that did not change keep their exact original text
and only new or edited entries are rendered (_splice_list), so the diff
shows the real change. Only when that is impossible does a plan say
"reformat": True, so the page can warn that whitespace will move.
commit() and undo() hold one lock (UPDATES_DIR/webui-undo/.lock) from the
stale check to the last write, so two saves cannot interleave, and a
commit that fails half way puts the files it already wrote back.
Undo records live in UPDATES_DIR/webui-undo/<image>--<scope>.json; only
the last save per scope can be undone (the .bak files stay on the share
for anything older).
"""
import contextlib
import difflib
import fcntl
import hashlib
import json
import os
import re
import shutil
import tempfile
from datetime import datetime
from services import updates_common as uc
BACKUP_TAG = "webui"
_BOM = "\ufeff"
class Conflict(Exception):
"""A file changed on disk since the preview or since the last save."""
# --- format ------------------------------------------------------------------
def encode(text):
"""Bytes as written: UTF-8, a leading \\ufeff becomes a real BOM."""
if text.startswith(_BOM):
return b"\xef\xbb\xbf" + text[1:].encode("utf-8")
return text.encode("utf-8")
def sha256_text(text):
return hashlib.sha256(encode(text)).hexdigest()
def file_sha256(path):
"""sha256 of the file's bytes ("" hashed when missing)."""
try:
with open(path, "rb") as fh:
return hashlib.sha256(fh.read()).hexdigest()
except OSError:
return hashlib.sha256(b"").hexdigest()
def read_text(path):
"""Current file content as text ("" when missing). A BOM is kept as the
leading \\ufeff character so it survives a round trip."""
try:
with open(path, "rb") as fh:
raw = fh.read()
except OSError:
return ""
if raw.startswith(b"\xef\xbb\xbf"):
return _BOM + raw[3:].decode("utf-8", errors="replace")
return raw.decode("utf-8", errors="replace")
def detect_json_format(text):
"""How an existing JSON file is laid out. Defaults (for a new file) are
indent 4, no BOM, LF, trailing newline, non-ASCII kept as is."""
fmt = {"indent": 4, "bom": False, "newline": "\n", "trailing_newline": True,
"ensure_ascii": False}
if not text:
return fmt
fmt["bom"] = text.startswith(_BOM)
body = text[1:] if fmt["bom"] else text
fmt["newline"] = "\r\n" if "\r\n" in body else "\n"
fmt["trailing_newline"] = body.endswith("\n")
lines = body.splitlines()
if len(lines) < 2:
# One line: compact JSON, unless it is an empty "[]" / "{}", which
# says nothing about layout (keep the indent 4 default).
if not re.fullmatch(r"\s*(\[\s*\]|\{\s*\})\s*", body):
fmt["indent"] = None
else:
m = re.match(r"[ \t]+", lines[1])
if m:
ws = m.group(0)
fmt["indent"] = "\t" if ws.startswith("\t") else len(ws)
# \uXXXX escapes and no raw non-ASCII: the writer escaped everything.
fmt["ensure_ascii"] = bool(re.search(r"\\u[0-9a-fA-F]{4}", body)) and body.isascii()
return fmt
def render_json(data, fmt):
text = json.dumps(data, indent=fmt["indent"], ensure_ascii=fmt["ensure_ascii"])
if fmt["trailing_newline"]:
text += "\n"
if fmt["newline"] != "\n":
text = text.replace("\n", fmt["newline"])
if fmt["bom"]:
text = _BOM + text
return text
# --- plan ----------------------------------------------------------------------
def _diff(old, new, name):
return "".join(difflib.unified_diff(
old.splitlines(keepends=True), new.splitlines(keepends=True),
fromfile=f"a/{name}", tofile=f"b/{name}", n=3))
def plan_text(path, new_text, label=None):
"""What saving new_text to path would change. Keys: path, name, old,
new, changed, diff, sha_before, existed, reformat (always False here)."""
old = read_text(path)
name = label or os.path.basename(path)
return {
"path": path,
"name": name,
"old": old,
"new": new_text,
"changed": old != new_text,
"diff": _diff(old, new_text, name) if old != new_text else "",
"sha_before": file_sha256(path),
"existed": os.path.isfile(path),
"reformat": False,
}
_WS = re.compile(r"[ \t\r\n]*")
def _key(obj):
"""Order-sensitive identity of a JSON value (key order matters on disk)."""
return json.dumps(obj, ensure_ascii=False)
def _splice_object(s, span, old_obj, new_obj, fmt):
"""Text of new_obj made from the old object's text at span, when both
are objects with the same keys in the same order: every member keeps
its text and only changed values are replaced (rendered on one line
when scalar, else with fmt at that member's indentation). None when
the shapes differ."""
if not (isinstance(old_obj, dict) and isinstance(new_obj, dict)
and list(old_obj) == list(new_obj)):
return None
dec = json.JSONDecoder()
start, end = span
i = _WS.match(s, start + 1).end()
out, pos = [], start
try:
while s[i:i + 1] == '"':
key, k_end = dec.raw_decode(s, i)
j = _WS.match(s, k_end).end()
if s[j:j + 1] != ":":
return None
v_start = _WS.match(s, j + 1).end()
_val, v_end = dec.raw_decode(s, v_start)
if _key(old_obj.get(key)) != _key(new_obj.get(key)):
line_start = s.rfind("\n", 0, i) + 1
indent = re.match(r"[ \t]*", s[line_start:i]).group(0)
text = json.dumps(new_obj[key], indent=fmt["indent"],
ensure_ascii=fmt["ensure_ascii"])
out.append(s[pos:v_start])
out.append(text.replace("\n", fmt["newline"] + indent))
pos = v_end
j = _WS.match(s, v_end).end()
if s[j:j + 1] == ",":
i = _WS.match(s, j + 1).end()
else:
break
except ValueError:
return None
out.append(s[pos:end])
return "".join(out)
def _splice_list(old, data, fmt):
"""New text for a top-level JSON list that keeps the exact original
text of every entry that did not change (and the separators between
neighbours that stay neighbours); new or edited entries are rendered
with fmt at the first entry's indentation. None when old is not a
plain JSON list or the result would not parse back to data."""
if not isinstance(data, list) or not data:
return None
s = old
dec = json.JSONDecoder()
i = _WS.match(s, 1 if s.startswith(_BOM) else 0).end()
if s[i:i + 1] != "[":
return None
head_end = i + 1
i = _WS.match(s, head_end).end()
if s[i:i + 1] == "]":
return None
spans, items = [], []
try:
while True:
obj, end = dec.raw_decode(s, i)
spans.append((i, end))
items.append(obj)
j = _WS.match(s, end).end()
if s[j:j + 1] == ",":
i = _WS.match(s, j + 1).end()
continue
if s[j:j + 1] == "]":
close = j
break
return None
except ValueError:
return None
lead = s[head_end:spans[0][0]]
nl = fmt["newline"]
if "\n" in lead:
item_indent = lead.rsplit("\n", 1)[-1]
std_sep = "," + nl + item_indent
else:
item_indent, std_sep = "", ", "
sm = difflib.SequenceMatcher(None, [_key(o) for o in items], [_key(o) for o in data],
autojunk=False)
# src: old entry identical to new entry n; origin: the old entry new
# entry n stands in for (identical, or edited in a one-for-one replace).
src = [None] * len(data)
origin = [None] * len(data)
for tag, i1, i2, j1, j2 in sm.get_opcodes():
if tag == "equal" or (tag == "replace" and i2 - i1 == j2 - j1):
for k in range(j2 - j1):
origin[j1 + k] = i1 + k
if tag == "equal":
src[j1 + k] = i1 + k
def fresh(obj):
text = json.dumps(obj, indent=fmt["indent"], ensure_ascii=fmt["ensure_ascii"])
return text.replace("\n", nl + item_indent)
def edited(n, obj):
# An entry that is one of the old ones with values changed (same
# keys, same order) keeps its own layout: only the values move.
k = origin[n]
if src[n] is not None:
return s[spans[k][0]:spans[k][1]]
if k is not None:
text = _splice_object(s, spans[k], items[k], obj, fmt)
if text is not None:
return text
return fresh(obj)
out = [s[:head_end], lead]
for n, obj in enumerate(data):
if n:
a, b = origin[n - 1], origin[n]
out.append(s[spans[a][1]:spans[b][0]] if a is not None and b == a + 1 else std_sep)
out.append(edited(n, obj))
out.append(s[spans[-1][1]:close])
out.append(s[close:])
text = "".join(out)
try:
if _key(json.loads(text.lstrip(_BOM))) != _key(data):
return None
except ValueError:
return None
return text
def plan_json(path, data, label=None):
"""plan_text for a JSON document, rendered in the file's own format.
A file that does not round-trip keeps its unchanged entries verbatim
(_splice_list); "reformat" is set only when that is not possible."""
old = read_text(path)
fmt = detect_json_format(old)
if old:
try:
same = render_json(json.loads(old.lstrip(_BOM)), fmt) == old
except ValueError:
same = False
if not same:
spliced = _splice_list(old, data, fmt)
if spliced is not None:
return plan_text(path, spliced, label)
plan = plan_text(path, render_json(data, fmt), label)
plan["reformat"] = True
return plan
return plan_text(path, render_json(data, fmt), label)
def public(plan):
"""The part of a plan a page needs (no full file texts)."""
return {k: plan[k] for k in ("name", "changed", "diff", "sha_before", "existed", "reformat")}
# --- write ---------------------------------------------------------------------
def _stamp():
return datetime.now().strftime("%Y%m%d-%H%M%S")
def _match_owner(path, ref_stat):
if ref_stat is None:
return
try:
os.chown(path, ref_stat.st_uid, ref_stat.st_gid)
except (PermissionError, OSError):
pass # not root (local tests): the file stays ours
def _ref_stat(path):
"""Stat to copy owner/mode from: the file, else its directory."""
for p in (path, os.path.dirname(path)):
try:
return os.stat(p)
except OSError:
continue
return None
def atomic_write(path, text):
"""Write text (see encode) or bytes via a temp file in the same
directory, keeping the old file's owner and mode. A symlinked control
file is written through to its target (os.replace on the link itself
would turn it into a separate copy)."""
path = os.path.realpath(path)
folder = os.path.dirname(path)
os.makedirs(folder, exist_ok=True)
existed = os.path.isfile(path)
ref = _ref_stat(path)
data = text if isinstance(text, bytes) else encode(text)
fd, tmp = tempfile.mkstemp(dir=folder, prefix=".webui-")
try:
with os.fdopen(fd, "wb") as fh:
fh.write(data)
fh.flush()
os.fsync(fh.fileno())
if existed:
shutil.copymode(path, tmp)
else:
os.chmod(tmp, 0o664)
_match_owner(tmp, ref)
os.replace(tmp, path)
except BaseException:
if os.path.exists(tmp):
os.remove(tmp)
raise
def backup(path, tag=BACKUP_TAG, stamp=None):
"""Copy path to <path>.bak-<stamp>-<tag> (owner kept). None if missing."""
if not os.path.isfile(path):
return None
dst = f"{path}.bak-{stamp or _stamp()}-{tag}"
n = 1
while os.path.exists(dst): # two saves in one second
n += 1
dst = f"{path}.bak-{stamp or _stamp()}-{tag}{n}"
shutil.copy2(path, dst)
_match_owner(dst, _ref_stat(path))
return dst
# --- undo records ------------------------------------------------------------------
def _undo_path(image_type, scope):
safe = re.sub(r"[^A-Za-z0-9._-]", "_", f"{image_type}--{scope}")
return os.path.join(uc.state_dir("webui-undo"), safe + ".json")
@contextlib.contextmanager
def _locked():
"""One lock for every commit/undo (saves are rare; threads and worker
processes both honour it, flock is per open file)."""
lock = os.path.join(uc.state_dir("webui-undo"), ".lock")
with open(lock, "a") as fh:
fcntl.flock(fh, fcntl.LOCK_EX)
try:
yield
finally:
fcntl.flock(fh, fcntl.LOCK_UN)
def _restore(path, bak):
"""Put a file back as its backup holds it, or remove it (bak None)."""
if bak:
with open(bak, "rb") as fh:
atomic_write(path, fh.read())
elif os.path.lexists(path):
os.remove(path)
def last_save(image_type, scope):
"""The undo point for (image_type, scope): {"ts", "action", "files":
[{"path", "name", "backup", "existed", "sha_after"}]} or None."""
return uc.read_json(_undo_path(image_type, scope), None)
def commit(image_type, scope, plans, action="", expect=None):
"""Back up and write every changed plan; record the undo point.
Returns the list of changed plans (empty when nothing changed, and then
the previous undo point is kept). Raises Conflict when expect is given
and a file no longer has the previewed hash. When a write fails part
way, the files already written are put back before the error is raised."""
expect = expect or {}
with _locked():
changed = [p for p in plans if p["changed"]]
stale = [p["name"] for p in plans
if p["name"] in expect and expect[p["name"]] != file_sha256(p["path"])]
if stale:
raise Conflict("changed on disk since the preview: " + ", ".join(stale)
+ ". Reload the page and review again.")
if not changed:
return []
stamp = _stamp()
files = []
try:
for p in changed:
bak = backup(p["path"], stamp=stamp)
files.append({"path": p["path"], "name": p["name"], "backup": bak,
"existed": bak is not None, "sha_after": sha256_text(p["new"])})
atomic_write(p["path"], p["new"])
uc.write_json(_undo_path(image_type, scope),
{"ts": uc.now(), "action": action, "files": files})
except BaseException:
for f in reversed(files):
try:
_restore(f["path"], f["backup"])
except OSError:
pass # the .bak file is still next to it
raise
return changed
def undo(image_type, scope):
"""Restore the backups of the last commit for (image_type, scope).
The current files are themselves backed up (tag webui-undo) first, so
an undo can be undone by hand. Returns the record that was undone."""
with _locked():
rec = last_save(image_type, scope)
if not rec:
raise Conflict("nothing to undo")
moved = [f["name"] for f in rec["files"]
if file_sha256(f["path"]) != f["sha_after"]]
if moved:
raise Conflict("changed since that save, not undone: " + ", ".join(moved)
+ ". The backups are still next to the files.")
missing = [f["name"] for f in rec["files"]
if f["existed"] and not os.path.isfile(f["backup"] or "")]
if missing:
raise Conflict("backup missing for: " + ", ".join(missing))
stamp = _stamp()
for f in rec["files"]:
backup(f["path"], tag=BACKUP_TAG + "-undo", stamp=stamp)
_restore(f["path"], f["backup"] if f["existed"] else None)
os.remove(_undo_path(image_type, scope))
return rec

1422
webapp/services/dell_updates.py Normal file
View File

File diff suppressed because it is too large Load Diff

321
webapp/services/driver_lint.py Normal file
View File

@@ -0,0 +1,321 @@
"""Lint an image's HardwareDriver.json against PESetup's real driver matcher.
A port of scripts/lint-driver-catalogue.py for the webapp (the script is
left as it is: it still gates deploys from a workstation). Same matcher,
same checks, with three differences that come from running on the server:
* zips are looked up on the share directly, and a zip found only with
different letter case is a WARN, not an ERROR, when WinPE still finds
it: WinPE reads the share over SMB and Samba matches each folder name
case-insensitively for Windows clients (default "case sensitive =
auto"). It stays an ERROR when a folder with the exact spelling exists
and the zip sits in its other-case twin;
* a missing (null) modelswminame or family is CRITICAL, not just "no
tokens": the C# calls .Split / .ToLower on them, which throws for every
model the family filter lets through;
* "win10-first": a Win11 entry's model is answered by a Win10 entry
earlier in the list. Every GE image deploys Windows 11 and PESetup
has no OS filter, so those PCs get Windows 10 drivers.
Positions in messages are 1-based (#5 = fifth entry), as on the Drivers tab.
lint_entries(entries, image_type) -> findings, each a dict:
{"level", "check", "message", "index" (0-based or None),
"indexes" (every 0-based entry the finding is about)}
lint_image(image_type) reads the map from disk (used by the Overview tab).
"""
import json
import os
from pathlib import PurePosixPath
from services import fs, image_pages
CRITICAL, ERROR, WARN, INFO = "CRITICAL", "ERROR", "WARN", "INFO"
LEVELS = (CRITICAL, ERROR, WARN, INFO)
LEVEL_ORDER = {lvl: i for i, lvl in enumerate(LEVELS)}
# Short names for the checks, shown as badges on the Drivers tab.
CHECK_LABELS = {
"virtual-platform": "virtual platform",
"null-tokens": "no modelswminame",
"null-family": "no family",
"empty-token": "empty token",
"token-whitespace": "token whitespace",
"no-tokens": "no tokens",
"family-mismatch": "family mismatch",
"no-filename": "no file name",
"missing-zip": "zip missing",
"case-mismatch": "zip case differs",
"win10-first": "Win10 entry wins",
"shadowed": "shadowed",
"duplicate-token": "duplicate token",
"bad-entry": "not an object",
}
# --- PESetup's matcher (docs/PESETUP-INTERNALS.md, GetDriverByModel) ---------
def model_family(model):
"""Three Contains() tests, the last hit wins, "" when none hit."""
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_tokens(entry):
"""modelswminame split on comma. NOT trimmed: the C# does not trim.
A missing (null) modelswminame gives no tokens here; PESetup throws on
it instead, which the null-tokens check reports."""
if entry.get("modelswminame") is None:
return []
return str(entry.get("modelswminame")).split(",")
def entry_matches(entry, model):
family = model_family(model)
if family and family.lower() not in str(entry.get("family") or "").lower():
return False
# model.ToLower().Contains(token.ToLower()); "" is in every string.
return any(t.lower() in model.lower() for t in entry_tokens(entry))
def select_driver(entries, model):
"""First match wins, like GetDriverByModel: (index, entry) or (None, None)."""
for i, entry in enumerate(entries):
if entry_matches(entry, model):
return i, entry
return None, None
# --- helpers -------------------------------------------------------------------
def label(entry, index):
name = entry.get("modelsfriendlyname") or entry.get("modelswminame") or "?"
return "#%d %s" % (index + 1, name)
def _smb_lookup(root, rel):
"""Resolve rel the way Samba does for a Windows client: per folder, the
exact name when it exists, else the one name that differs only in case.
No backtracking: once a folder is picked, a file under its other-case
twin is not found (Dell_11 has held both Optiplex and OptiPlex)."""
cur = root
for part in PurePosixPath(rel).parts:
nxt = os.path.join(cur, part)
if not os.path.lexists(nxt):
try:
names = [n for n in os.listdir(cur) if n.lower() == part.lower()]
except OSError:
return None
if len(names) != 1:
return None
nxt = os.path.join(cur, names[0])
cur = nxt
return cur if os.path.isfile(cur) else None
def zip_state(image_type, entry):
"""Where an entry's zip is: {"rel", "status", "path"}. rel is relative
to the image root (Deploy/Out-of-box Drivers/...). status:
ok the exact path exists
case only with other letter case, and WinPE (SMB) still finds it
trap on disk under another-case folder that SMB does not pick
missing nowhere
none the entry names no file"""
rel = image_pages.hardware_coverage._zip_relpath(entry)
if not rel:
return {"rel": "", "status": "none", "path": None}
if ".." in PurePosixPath(rel).parts:
# Never look outside the image tree: /drivers/check takes any
# posted entry, and no real entry climbs out with "..".
return {"rel": rel, "status": "missing", "path": None}
root = fs.image_root(image_type)
exact = os.path.join(root, rel)
if os.path.isfile(exact):
return {"rel": rel, "status": "ok", "path": exact}
smb = _smb_lookup(root, rel)
if smb:
return {"rel": rel, "status": "case", "path": smb, "actual": os.path.relpath(smb, root)}
found = image_pages.zip_path(image_type, entry)
if found:
return {"rel": rel, "status": "trap", "path": found, "actual": os.path.relpath(found, root)}
return {"rel": rel, "status": "missing", "path": None}
def _add(out, level, check, message, index=None, indexes=None):
idx = list(indexes) if indexes is not None else ([] if index is None else [index])
out.append({"level": level, "check": check, "message": message,
"index": index, "indexes": idx})
# --- checks --------------------------------------------------------------------
def lint_entries(entries, image_type=None, zips=None):
"""Every finding for a list of entries. image_type enables the zip
checks; zips (a list of zip_state results, one per entry) saves
looking them up again."""
out = []
rows = [e if isinstance(e, dict) else {} for e in entries]
for i, e in enumerate(entries):
if not isinstance(e, dict):
_add(out, CRITICAL, "bad-entry",
"#%d is not a JSON object; PESetup cannot read the map." % (i + 1), i)
for i, entry in enumerate(rows):
tag = label(entry, i)
# Virtual platform hijack: one such entry anywhere flips PESetup into
# virtual-only mode and every physical machine fails.
manufacturer = str(entry.get("manufacturer") or "")
if "virtual platform" in manufacturer.lower():
_add(out, CRITICAL, "virtual-platform",
"%s manufacturer is %r. PESetup then takes the first entry as the driver and "
"REQUIRES the machine to look virtual, so every physical PC fails." % (tag, manufacturer), i)
# Null fields the C# dereferences without a check.
if entry.get("modelswminame") is None:
_add(out, CRITICAL, "null-tokens",
"%s has no modelswminame. PESetup calls Split() on it and throws for every "
"model the family filter lets through, so driver selection fails." % tag, i)
if entry.get("family") is None:
_add(out, CRITICAL, "null-family",
"%s has no family. PESetup calls ToLower() on it for every Latitude, OptiPlex "
"and Precision model and throws." % tag, i)
# Empty token: Contains("") is true for every model. An empty
# modelswminame is one empty token, so it matches everything too.
tokens = entry_tokens(entry)
if tokens == [""]:
_add(out, CRITICAL, "empty-token",
"%s modelswminame is empty. Split gives one empty token, which matches every "
"model, so this entry takes over the map." % tag, i)
elif any(t == "" for t in tokens):
_add(out, CRITICAL, "empty-token",
"%s modelswminame %r has an EMPTY token (trailing or doubled comma). An empty "
"token matches every model, so this entry takes over the map."
% (tag, entry.get("modelswminame")), i)
# Whitespace around a token is kept by Split(','), so it must be in the model.
for token in tokens:
if token and token != token.strip():
_add(out, ERROR, "token-whitespace",
"%s token %r carries whitespace. PESetup does not trim, so the model "
"string must contain the space too; this token will usually miss." % (tag, token), i)
if tokens and tokens != [""] and not str(entry.get("modelswminame")).strip():
_add(out, ERROR, "no-tokens",
"%s has an empty modelswminame, so no model can ever select it." % tag, i)
# The family filter runs on the MODEL first, then wants entry.family
# to contain it: a token implying another family can never match.
family_field = str(entry.get("family") or "")
for token in tokens:
token = token.strip()
implied = model_family(token) if token else ""
if implied and implied.lower() not in family_field.lower():
_add(out, ERROR, "family-mismatch",
"%s token %r implies family %s, but family is %r. The filter runs before "
"the token test, so this token can never match." % (tag, token, implied, family_field), i)
# The referenced zip.
if not str(entry.get("fileName") or entry.get("FileName") or ""):
_add(out, ERROR, "no-filename",
"%s has no fileName, so PESetup has nothing to unzip." % tag, i)
elif image_type:
z = zips[i] if zips is not None else zip_state(image_type, entry)
if z["status"] == "missing":
_add(out, ERROR, "missing-zip",
"%s references %s, which is not on the share. A driver miss is only a "
"warning in PESetup, so imaging finishes with no drivers." % (tag, z["rel"]), i)
elif z["status"] == "case":
_add(out, WARN, "case-mismatch",
"%s references %s but the share holds %s. WinPE finds it (Samba matches "
"case-insensitively), Linux tools and the lint script do not."
% (tag, z["rel"], z.get("actual", "")), i)
elif z["status"] == "trap":
_add(out, ERROR, "case-mismatch",
"%s references %s but the zip is at %s. A folder with the exact spelling "
"exists and does not hold it, so WinPE will not find it."
% (tag, z["rel"], z.get("actual", "")), i)
# Shadowing: replay each entry's own tokens through the matcher. When a
# Win10 entry answers a Win11 entry's model, that is worse than dead
# weight: Win11 PCs get Windows 10 drivers.
os_of = [image_pages.driver_os(e) for e in rows]
for i, entry in enumerate(rows):
for token in entry_tokens(entry):
token = token.strip()
if not token:
continue
win, _ = select_driver(rows, token)
if win is None or win == i:
continue
if os_of[i] == "Win11" and os_of[win] == "Win10":
_add(out, ERROR, "win10-first",
"%s token %r is answered by %s, a Windows 10 pack earlier in the list. "
"PESetup has no OS filter, so Windows 11 PCs get Windows 10 drivers. "
"Move the Win11 entry above it." % (label(entry, i), token, label(rows[win], win)),
i, indexes=[i, win])
else:
_add(out, WARN, "shadowed",
"%s token %r is answered by %s instead (first match wins)."
% (label(entry, i), token, label(rows[win], win)), i)
# One token claimed by several entries: the first one wins whatever the OS.
by_token = {}
for i, entry in enumerate(rows):
for token in entry_tokens(entry):
token = token.strip().lower()
if token:
by_token.setdefault(token, [])
if i not in by_token[token]:
by_token[token].append(i)
for token, idxs in sorted(by_token.items()):
if len(idxs) > 1:
detail = ", ".join("%s (%s, osId %s)" % (label(rows[i], i), os_of[i], rows[i].get("osId"))
for i in idxs)
_add(out, WARN, "duplicate-token",
"token %r is claimed by %d entries: %s. osId is not part of the match, so %s "
"wins whatever OS is applied." % (token, len(idxs), detail, label(rows[idxs[0]], idxs[0])),
None, indexes=idxs)
out.sort(key=lambda f: (LEVEL_ORDER[f["level"]], f["indexes"][0] if f["indexes"] else -1))
return out
def counts(findings):
"""{"total", "CRITICAL": n, "ERROR": n, "WARN": n, "INFO": n}."""
c = {lvl: 0 for lvl in LEVELS}
for f in findings:
c[f["level"]] += 1
c["total"] = sum(c[lvl] for lvl in LEVELS)
return c
def map_path(image_type):
return image_pages.control_file(image_type, "HardwareDriver.json")
def lint_image(image_type):
"""Findings for the map on disk. [] when the image has no map; one
CRITICAL finding when it does not parse or is not a list."""
path = map_path(image_type)
if not os.path.isfile(path):
return []
try:
with open(path, encoding="utf-8-sig") as fh:
entries = json.load(fh)
except (OSError, ValueError) as exc:
return [{"level": CRITICAL, "check": "parse", "index": None, "indexes": [],
"message": "HardwareDriver.json does not parse: %s" % exc}]
if not isinstance(entries, list):
return [{"level": CRITICAL, "check": "parse", "index": None, "indexes": [],
"message": "HardwareDriver.json is a %s, expected a list" % type(entries).__name__}]
return lint_entries(entries, image_type)

416
webapp/services/driver_map.py Normal file
View File

@@ -0,0 +1,416 @@
"""The Drivers tab's view of an image's HardwareDriver.json.
PESetup reads Deploy/Control/HardwareDriver.json and takes the FIRST entry
whose modelswminame token is in the PC's WMI model (no OS filter), so the
list order matters and is kept exactly as the user arranges it.
The page edits the whole map in the browser and posts it back as a list of
entries; every entry that came from the file carries "_orig" (its 0-based
position when the page was loaded), so the server can tell an edit from a
move, an addition or a removal. Keys starting with "_" are never written.
rows(image_type, entries) per-entry facts for the table (zip on disk,
OS, lint findings, Dell update hint)
orphans(image_type, entries) zips under the image's driver folder that
this map does not reference
staged(body) validate a posted map: (entries, origs)
changes(old, new, origs) per-entry summary + JSON diff for the review
delete_orphans(image_type, rels) zips no map, undo or rollback uses
save_upload(image_type, file) an uploaded zip, temp file + rename
Writes go through services/control_files (format kept, backup, undo).
"""
import difflib
import json
import os
import tempfile
import config
from services import control_files, driver_lint, fs, image_pages
FILE_NAME = "HardwareDriver.json"
DRIVERS_SUBDIR = "Out-of-box Drivers"
DEST_PREFIX = "*destinationdir*\\Deploy\\Out-of-box Drivers\\"
# Keys the C# reads as strings; a posted entry must not turn them into
# numbers or objects.
STRING_KEYS = ("manufacturer", "product", "family", "modelswminame", "modelsfriendlyname",
"fileName", "destinationDir", "FileName", "DestinationDir", "osId", "models")
def map_path(image_type):
return image_pages.control_file(image_type, FILE_NAME)
def load(image_type):
"""(entries, error). entries is [] and error a message when the file is
missing, does not parse or is not a list."""
path = map_path(image_type)
if not os.path.isfile(path):
return [], None
try:
with open(path, encoding="utf-8-sig") as fh:
data = json.load(fh)
except (OSError, ValueError) as exc:
return [], f"{FILE_NAME} does not parse: {exc}"
if not isinstance(data, list):
return [], f"{FILE_NAME} is a {type(data).__name__}, expected a list"
# The page edits entries as objects; a string or null entry would be
# turned into an object (or crash the page), so such a map is read-only.
bad = [i + 1 for i, e in enumerate(data) if not isinstance(e, dict)]
if bad:
return [], (f"{FILE_NAME} entry #{bad[0]} is not a JSON object "
f"({len(bad)} such entr{'y' if len(bad) == 1 else 'ies'})")
return data, None
def file_sha(image_type):
return control_files.file_sha256(map_path(image_type))
def _stat(path):
try:
st = os.stat(path)
return st.st_size, st.st_mtime
except (OSError, TypeError):
return None, None
def _dell_hints():
try:
from services import dell_updates
return dell_updates.update_hints()["drivers"]
except Exception: # no scan yet or the module failed: no hints
return {}
def rows(image_type, entries):
"""(rows, findings, counts) for a list of entries (from the file or
staged in the page). One row per entry, in order:
{"os", "zip": {"status", "rel", "actual", "size", "mtime"},
"findings": [finding, ...], "dell": newer Dell version or ""}"""
dicts = [e if isinstance(e, dict) else {} for e in entries]
zips = [driver_lint.zip_state(image_type, e) for e in dicts]
findings = driver_lint.lint_entries(entries, image_type, zips=zips)
per = [[] for _ in entries]
for f in findings:
for i in f["indexes"]:
if 0 <= i < len(per):
per[i].append({"level": f["level"], "check": f["check"], "message": f["message"],
"label": driver_lint.CHECK_LABELS.get(f["check"], f["check"])})
hints = _dell_hints()
out = []
for i, e in enumerate(dicts):
z = zips[i]
size, mtime = _stat(z["path"])
name = str(e.get("fileName") or e.get("FileName") or "")
out.append({
"os": image_pages.driver_os(e),
"zip": {"status": z["status"], "rel": z["rel"], "actual": z.get("actual", ""),
"size": size, "mtime": mtime},
"findings": per[i],
"dell": hints.get(name.lower(), ""),
})
return out, findings, driver_lint.counts(findings)
# --- orphans ------------------------------------------------------------------------
def _drivers_root(image_type):
"""Real path of the image's Out-of-box Drivers folder (a symlink into
_shared on the server), or None."""
path = os.path.join(fs.deploy_path(image_type), DRIVERS_SUBDIR)
real = os.path.realpath(path)
return real if os.path.isdir(real) else None
def _share_images():
"""Every image whose map can point into the shared driver tree: the
registered ones plus any other folder on the share with a map (an image
dropped from the registry can still be booted)."""
images = list(config.IMAGE_TYPES)
try:
names = sorted(os.listdir(config.SAMBA_SHARE))
except OSError:
names = []
for name in names:
if (not name.startswith("_") and name not in images
and os.path.isfile(map_path(name))):
images.append(name)
return images
def _restore_points():
"""[(image, label, map file)] for maps the app can still put back: the
backup behind each image's Undo last save, and the map backups of every
Dell update that can be rolled back. A zip those maps use must not be
deleted, or the undo / rollback restores an entry with no zip."""
out = []
for image in _share_images():
rec = control_files.last_save(image, "drivers") or {}
for f in rec.get("files", []):
if f.get("backup") and os.path.basename(f.get("path", "")) == FILE_NAME:
out.append((image, f"{image} (undo of last save)", f["backup"]))
try:
from services import dell_updates
history = dell_updates.history()
except Exception: # no Dell history yet or the module failed
history = []
for rec in history:
if rec.get("rolled_back"):
continue
for path, bak in (rec.get("backups") or {}).items():
if os.path.basename(path) != FILE_NAME or not bak:
continue
image = os.path.relpath(path, config.SAMBA_SHARE).split(os.sep)[0]
out.append((image, f"{image} (Dell update rollback)", bak))
return out
def _entries_of(path):
try:
with open(path, encoding="utf-8-sig") as fh:
data = json.load(fh)
except (OSError, ValueError):
return []
return [e for e in data if isinstance(e, dict)] if isinstance(data, list) else []
def _referenced():
"""{real zip path: [who uses it]}: every image's map, plus the maps an
undo or a Dell rollback would restore (see _restore_points)."""
refs = {}
sources = [(image, image, map_path(image)) for image in _share_images()]
for image, label, path in sources + _restore_points():
for e in _entries_of(path):
p = image_pages.zip_path(image, e)
if p:
refs.setdefault(os.path.realpath(p), []).append(label)
return refs
def orphans(image_type, refs=None):
"""Zips under this image's driver folder that its map does not use.
used_by lists the other images that do use it (deleting those would
break them, so the page only allows Add to map)."""
root = _drivers_root(image_type)
if not root:
return []
refs = _referenced() if refs is None else refs
out = []
# followlinks=False: Dell_11/OptiPlex can be a symlink to Optiplex, and
# following it would list every zip there twice.
for dirpath, dirnames, filenames in os.walk(root):
dirnames.sort()
for fn in sorted(filenames):
if not fn.lower().endswith(".zip"):
continue
full = os.path.join(dirpath, fn)
users = sorted(set(refs.get(os.path.realpath(full), [])))
if image_type in users:
continue
rel = os.path.relpath(full, root).replace(os.sep, "/")
folder = os.path.dirname(rel)
size, mtime = _stat(full)
out.append({
"name": fn,
"rel": rel,
"folder": folder,
"dest": DEST_PREFIX.rstrip("\\") + ("\\" + folder.replace("/", "\\") if folder else ""),
"size": size,
"mtime": mtime,
"used_by": users,
})
return out
def delete_orphans(image_type, rels):
"""Delete zips (paths relative to the driver folder) that NO image's
map references. Returns {"removed": [...], "refused": [(rel, why)]}."""
root = _drivers_root(image_type)
removed, refused = [], []
if not root:
return {"removed": removed, "refused": [(r, "no driver folder") for r in rels]}
refs = _referenced()
for rel in rels:
rel = str(rel or "")
full = os.path.realpath(os.path.join(root, rel))
if not full.startswith(root + os.sep) or not full.lower().endswith(".zip"):
refused.append((rel, "not a zip in the driver folder"))
elif os.path.islink(os.path.join(root, rel)):
# Unlinking the resolved path would delete the zip it points at.
refused.append((rel, "a symlink; remove it by hand"))
elif not os.path.isfile(full):
refused.append((rel, "not found"))
elif refs.get(full):
refused.append((rel, "used by " + ", ".join(sorted(set(refs[full])))))
else:
try:
os.unlink(full)
removed.append(rel)
except OSError as exc:
refused.append((rel, str(exc)))
return {"removed": removed, "refused": refused}
def save_upload(image_type, upload, overwrite=False):
"""Store an uploaded driver .zip at the top of the image's driver folder.
Written to a temp file next to it and renamed, so a PC copying the old
file never reads a half-written one; owner and group follow the folder
(the app runs as root, the share is pxe:pxe). Overwriting a zip any map,
undo or rollback uses is refused: its entry's hash and size would no
longer match. Nothing is added to the map (Add to map does that, through
the review step). Returns {"filename", "path", "replaced"}."""
from services import images # the existing upload name sanitizer
name = images._safe_filename(getattr(upload, "filename", ""))
if not name.lower().endswith(".zip"):
raise ValueError("Driver upload must be a .zip file")
root = _drivers_root(image_type)
if not root:
root = os.path.join(fs.deploy_path(image_type), DRIVERS_SUBDIR)
os.makedirs(root, exist_ok=True)
dst = os.path.join(root, name)
replaced = os.path.lexists(dst)
if replaced:
if not overwrite:
raise FileExistsError(f"{name} already exists in {DRIVERS_SUBDIR}/")
if os.path.islink(dst) or not os.path.isfile(dst):
raise ValueError(f"{name} in {DRIVERS_SUBDIR}/ is not a plain file; replace it by hand")
users = sorted(set(_referenced().get(os.path.realpath(dst), [])))
if users:
raise ValueError(f"{name} is in use ({', '.join(users)}); upload it under a new name "
"and point the entry at it instead")
fd, tmp = tempfile.mkstemp(dir=root, prefix=".webui-upload-", suffix=".part")
try:
with os.fdopen(fd, "wb") as fh:
upload.save(fh)
fh.flush()
os.fsync(fh.fileno())
os.chmod(tmp, 0o664)
try:
st = os.stat(root)
os.chown(tmp, st.st_uid, st.st_gid)
except OSError:
pass # not root (local tests)
os.replace(tmp, dst)
except BaseException:
if os.path.exists(tmp):
os.remove(tmp)
raise
return {"filename": name, "path": dst, "replaced": replaced}
# --- staged map -----------------------------------------------------------------------
def staged(body, old_len):
"""Validate a posted {"entries": [...]}. Returns (entries, origs, moved):
entries cleaned of "_" keys, origs[i] = position in the file the entry
came from (or None for a new one), moved[i] = the user moved it with
the arrows ("_moved"; only used to word the review). Raises ValueError."""
raw = body.get("entries") if isinstance(body, dict) else None
if not isinstance(raw, list) or not all(isinstance(e, dict) for e in raw):
raise ValueError("expected entries: a list of objects")
entries, origs, moved, seen = [], [], [], set()
for n, e in enumerate(raw, 1):
o = e.get("_orig")
if isinstance(o, bool) or not isinstance(o, int) or not 0 <= o < old_len or o in seen:
o = None
else:
seen.add(o)
clean = {k: v for k, v in e.items() if not str(k).startswith("_")}
for k in STRING_KEYS:
if k in clean and clean[k] is not None and not isinstance(clean[k], str):
raise ValueError(f"entry #{n}: {k} must be text")
if "aOsIds" in clean and not (isinstance(clean["aOsIds"], list)
and all(isinstance(x, str) for x in clean["aOsIds"])):
raise ValueError(f"entry #{n}: aOsIds must be a list of text")
entries.append(clean)
origs.append(o)
moved.append(bool(e.get("_moved")))
return entries, origs, moved
def _dump(entry):
return json.dumps(entry, indent=4, ensure_ascii=False)
def _kept_in_order(seq, weights):
"""Indexes (into seq) of the heaviest increasing run of values: the
entries that kept their relative order; the rest were moved. When the
user swaps two neighbours either one could be "the moved one", so the
entries the user did not touch weigh more and stay in the run."""
n = len(seq)
if not n:
return set()
best = list(weights)
prev = [-1] * n
for i in range(n):
for j in range(i):
if seq[j] < seq[i] and best[j] + weights[i] > best[i]:
best[i], prev[i] = best[j] + weights[i], j
i = max(range(n), key=lambda k: best[k])
keep = set()
while i != -1:
keep.add(i)
i = prev[i]
return keep
def changes(old, new, origs, user_moved=None):
"""What a save would do, entry by entry, for the review step:
[{"kind": "added"|"removed"|"edited"|"moved"|"edited+moved",
"label", "old_pos", "new_pos" (1-based or None), "file" (moves only),
"fields", "diff"}]
and a summary {"added", "removed", "edited", "moved"}."""
out = []
kept = [(i, o) for i, o in enumerate(origs) if o is not None]
user_moved = user_moved or [False] * len(new)
in_order = _kept_in_order([o for _, o in kept],
[1 if user_moved[i] else 1000 for i, _ in kept])
moved = {kept[k][0] for k in range(len(kept)) if k not in in_order}
summary = {"added": 0, "removed": 0, "edited": 0, "moved": 0}
for i, (entry, o) in enumerate(zip(new, origs)):
lbl = driver_lint.label(entry, i)
if o is None:
summary["added"] += 1
out.append({"kind": "added", "label": lbl, "old_pos": None, "new_pos": i + 1,
"fields": list(entry.keys()),
"diff": _udiff("", _dump(entry) + "\n", "(new)", f"#{i + 1}")})
continue
before = old[o] if isinstance(old[o], dict) else {}
edited = _dump(before) != _dump(entry)
was_moved = i in moved
if not edited and not was_moved:
continue
kind = "edited+moved" if edited and was_moved else ("edited" if edited else "moved")
if edited:
summary["edited"] += 1
if was_moved:
summary["moved"] += 1
fields = sorted({k for k in set(before) | set(entry) if before.get(k) != entry.get(k)}
| ({"(key order)"} if edited and list(before) != list(entry)
and set(before) == set(entry) else set()))
out.append({"kind": kind, "label": lbl, "old_pos": o + 1, "new_pos": i + 1,
"file": str(entry.get("fileName") or entry.get("FileName") or ""),
"fields": fields,
"diff": _udiff(_dump(before) + "\n", _dump(entry) + "\n",
f"#{o + 1}", f"#{i + 1}") if edited else ""})
gone = sorted(set(range(len(old))) - {o for o in origs if o is not None})
for o in gone:
before = old[o] if isinstance(old[o], dict) else {}
summary["removed"] += 1
out.append({"kind": "removed", "label": driver_lint.label(before, o), "old_pos": o + 1,
"new_pos": None, "fields": [],
"diff": _udiff(_dump(before) + "\n", "", f"#{o + 1}", "(removed)")})
out.sort(key=lambda c: (c["new_pos"] or c["old_pos"] or 0, c["kind"] != "removed"))
return out, summary
def _udiff(a, b, from_label, to_label):
# Whole entry as context: an entry is about 20 lines, and a reviewer
# wants to see which model a changed line belongs to.
return "".join(difflib.unified_diff(a.splitlines(keepends=True), b.splitlines(keepends=True),
fromfile=from_label, tofile=to_label, n=60))

6
webapp/services/hardware_coverage.py
View File

@@ -33,10 +33,12 @@ from pathlib import PurePosixPath
import config
# Images whose HardwareDriver.json is checked, with the Windows they deploy.
# All three deploy Windows 11 now (engineer moved off Win10), so every one
# counts toward a model's gaps.
IMAGES = (
("gea-standard", "Win11"),
("gea-shopfloor", "Win11 LTSC"),
("gea-engineer", "Win10"),
("gea-engineer", "Win11"),
)
KEEP = 10 # serials / versions remembered per model
@@ -239,7 +241,7 @@ def coverage(include_ok=True):
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"]
if d["status"] in ("no-entry", "zip-missing")]
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"])

348
webapp/services/image_pages.py Normal file
View File

@@ -0,0 +1,348 @@
"""Shared state for the per-image pages (GE Aerospace Images in the sidebar).
Each image has one page with tabs, all drawn under templates/_image_header.html:
tab id label endpoint URL
overview Overview images.overview /images/<t>
drivers Drivers images_drivers.page /images/<t>/drivers
unattend Unattend unattend_editor /images/<t>/unattend
os OS and Packages images.os_packages /images/<t>/os
IMAGE_TABS is the single place that maps a tab to its endpoint: the header
partial, the sidebar and every redirect use it (tab_url). A page that moves
to another endpoint changes its row here and nothing else.
header_info() is what the header partial shows; overview() feeds the
Overview tab. Both only read files, nothing here writes (saves go through
services/control_files.py).
"""
import glob
import os
from pathlib import PurePosixPath
from flask import url_for
import config
from services import control_files, fs, hardware_coverage
IMAGE_TABS = (
{"id": "overview", "label": "Overview", "icon": "bi-clipboard-data", "endpoint": "images.overview"},
{"id": "drivers", "label": "Drivers", "icon": "bi-hdd-network", "endpoint": "images_drivers.page"},
{"id": "unattend", "label": "Unattend", "icon": "bi-file-earmark-code", "endpoint": "unattend_editor"},
{"id": "os", "label": "OS and Packages", "icon": "bi-windows", "endpoint": "images.os_packages"},
)
TAB_IDS = tuple(t["id"] for t in IMAGE_TABS)
# PESetup loads the first of these that exists (docs/PESETUP-INTERNALS.md,
# "Paths it hardcodes"); the Unattend tab edits the fallback.
UNATTEND_PREFERRED = ("Deploy", "Tools", "GE", "XML", "FlatUnattendW10.xml")
# osId values that mean Windows 10 in the GE catalogue: 18 is the only id
# the Dell_10 packs carry; 19, 20 and 21 appear only on Win11 packs (20 =
# Win11 LTSC 24H2, 21 = Win11 Pro 24H2).
WIN10_OS_IDS = {"18"}
def tab_url(image_type, tab="overview", **kw):
"""URL of one tab of an image page."""
endpoint = next(t["endpoint"] for t in IMAGE_TABS if t["id"] == tab)
return url_for(endpoint, image_type=image_type, **kw)
def tab_for_endpoint(endpoint):
"""Tab id for a request endpoint, else None. Exact match first; then a
tab whose blueprint is its own (images_drivers.*) claims every endpoint
of that blueprint. The shared "images" blueprint only matches exactly."""
for t in IMAGE_TABS:
if endpoint == t["endpoint"]:
return t["id"]
bp = (endpoint or "").partition(".")[0]
for t in IMAGE_TABS:
own = t["endpoint"].partition(".")[0]
if "." in t["endpoint"] and own == bp and own != "images":
return t["id"]
return None
# --- file locations ----------------------------------------------------------
def control_file(image_type, name):
return os.path.join(fs.control_path(image_type), name)
def user_selections_path(image_type):
return os.path.join(fs.tools_path(image_type), "user_selections.json")
def preferred_unattend_path(image_type):
return os.path.join(fs.image_root(image_type), *UNATTEND_PREFERRED)
# --- small readers -------------------------------------------------------------
def _list(data):
return data if isinstance(data, list) else []
def _load(path):
"""fs.load_json that also survives a file that is not UTF-8 (a UTF-16
file saved by PowerShell 5.1 Out-File raises UnicodeDecodeError, which
fs.load_json lets through and which would fail every image tab)."""
try:
return fs.load_json(path)
except ValueError:
return []
def drivers(image_type):
"""HardwareDriver.json entries (the only driver file PESetup reads)."""
return _list(_load(control_file(image_type, "HardwareDriver.json")))
def user_selections(image_type):
"""(first object of user_selections.json, whole list) - the file is a
one-element JSON array written by Media Creator Lite."""
raw = _load(user_selections_path(image_type))
if isinstance(raw, list) and raw and isinstance(raw[0], dict):
return raw[0], raw
return {}, []
def driver_os(entry):
"""'Win11', 'Win10' or 'Other' for one driver entry. The pack file name
(win10_/win11_) decides first, then the Dell_10/Dell_11 folder, then
the osId list."""
name = str(entry.get("fileName") or entry.get("FileName") or "").lower()
dest = str(entry.get("destinationDir") or entry.get("DestinationDir") or "").lower()
if name.startswith("win10"):
return "Win10"
if name.startswith("win11"):
return "Win11"
if "\\dell_10\\" in dest or "/dell_10/" in dest:
return "Win10"
if "\\dell_11\\" in dest or "/dell_11/" in dest:
return "Win11"
ids = [i.strip() for i in str(entry.get("osId") or "").split(",") if i.strip()]
if not ids:
return "Other"
return "Win10" if all(i in WIN10_OS_IDS for i in ids) else "Win11"
def zip_path(image_type, entry):
"""Real path of an entry's zip (directories matched case-insensitively,
as on the live share), or None when it is not on disk."""
rel = hardware_coverage._zip_relpath(entry)
if not rel:
return None
return _find_ci(fs.image_root(image_type), PurePosixPath(rel).parts)
def _find_ci(cur, parts):
"""Depth-first case-insensitive lookup. Every case variant of a folder
is tried (Dell_11 can hold both Optiplex and OptiPlex as separate
folders), the exact spelling first."""
if not parts:
return cur if os.path.isfile(cur) else None
try:
names = os.listdir(cur)
except OSError:
return None
hits = sorted((n for n in names if n.lower() == parts[0].lower()), key=lambda n: n != parts[0])
for hit in hits:
found = _find_ci(os.path.join(cur, hit), parts[1:])
if found:
return found
return None
def _stat(path):
try:
st = os.stat(path)
return {"exists": True, "size": st.st_size, "mtime": st.st_mtime}
except OSError:
return {"exists": False, "size": None, "mtime": None}
def operating_systems(image_type):
"""OperatingSystem.json entries with the wim resolved on disk:
[{"entry", "osv", "wim_dir", "wim_path", "wim": _stat}]."""
out = []
for entry in _list(_load(control_file(image_type, "OperatingSystem.json"))):
osv = entry.get("operatingSystemVersion") or {}
wim = osv.get("wim") or {}
dest = wim.get("DestinationDir") or wim.get("destinationDir") or ""
wim_dir = fs.resolve_destination(dest, image_type)
wim_path = os.path.join(wim_dir, "install.wim") if wim_dir else ""
out.append({"entry": entry, "osv": osv, "dest": dest, "wim_path": wim_path,
"wim": _stat(wim_path) if wim_path else _stat("")})
return out
def packages(image_type):
"""packages.json entries with the file resolved on disk:
[{"entry", "path", "file": _stat}]."""
out = []
for entry in _list(_load(control_file(image_type, "packages.json"))):
name = entry.get("fileName") or entry.get("FileName") or ""
dest = entry.get("destinationDir") or entry.get("DestinationDir") or ""
folder = fs.resolve_destination(dest, image_type)
path = os.path.join(folder, name) if folder and name else ""
out.append({"entry": entry, "path": path, "file": _stat(path) if path else _stat("")})
return out
def _current_os(rows, selection):
"""The OS row imaging uses: the one user_selections.json selects, else
the first active one, else the first."""
want = str(selection if selection is not None else "")
return (next((r for r in rows if str(r["osv"].get("id")) == want), None)
or next((r for r in rows if r["osv"].get("isActive")), None)
or (rows[0] if rows else None))
def os_label(image_type):
"""'Windows 11 Professional 24H2 (osId 21)' for the selected OS, else
the first active entry, else ''."""
sel, _ = user_selections(image_type)
row = _current_os(operating_systems(image_type), sel.get("OperatingSystemSelection"))
if not row:
return ""
pick = row["osv"]
name = pick.get("productName") or pick.get("marketingName") or "Windows"
ver = pick.get("versionNumber") or ""
return f"{name} {ver} (osId {pick.get('id')})".replace(" ", " ")
# --- header + overview ---------------------------------------------------------------
def header_info(image_type):
"""What _image_header.html shows: name, OS, counts for the tab pills."""
entries = drivers(image_type)
return {
"image_type": image_type,
"friendly_name": config.FRIENDLY_NAMES.get(image_type, image_type),
"os_label": os_label(image_type),
"driver_count": len(entries),
"package_count": len(packages(image_type)),
"has_unattend": os.path.isfile(fs.unattend_path(image_type)),
}
def _lint_counts(image_type):
"""Findings of services/driver_lint (built by the Drivers tab) as
{"available", "total", "by_level"}; unavailable until that module
exists or when it fails."""
try:
from services import driver_lint # noqa: WPS433 (optional module)
findings = driver_lint.lint_image(image_type)
except Exception: # missing module or a bad map: show "not available"
return {"available": False, "total": 0, "by_level": {}, "map_errors": 0}
by_level = {}
map_errors = 0
for f in findings or []:
f = f if isinstance(f, dict) else vars(f)
level = str(f.get("level") or "INFO").upper()
by_level[level] = by_level.get(level, 0) + 1
# A missing zip is an ERROR finding too, but the Overview counts
# those on their own tile; map_errors is what is wrong in the map.
if level in ("CRITICAL", "ERROR") and f.get("check") != "missing-zip":
map_errors += 1
return {"available": True, "total": sum(by_level.values()), "by_level": by_level,
"map_errors": map_errors}
def _control_files(image_type):
"""The files the image pages edit, with size/mtime and backups."""
items = [
("HardwareDriver.json", control_file(image_type, "HardwareDriver.json"), "drivers",
"Driver map PESetup matches the PC model against"),
("OperatingSystem.json", control_file(image_type, "OperatingSystem.json"), "os",
"Windows build, install.wim location and index"),
("packages.json", control_file(image_type, "packages.json"), "os",
"Servicing packages applied offline (updates, .NET)"),
("user_selections.json", user_selections_path(image_type), "os",
"Media Creator Lite selections (OS and driver families)"),
("FlatUnattendW10.xml", fs.unattend_path(image_type), "unattend",
"Windows answer file (fallback location, the one this app edits)"),
]
out = []
for name, path, tab, what in items:
st = _stat(path)
backups = sorted(glob.glob(glob.escape(path) + ".bak-*"))
out.append(dict(st, name=name, path=path,
rel=os.path.relpath(path, fs.image_root(image_type)),
tab=tab, what=what, backups=len(backups)))
return out
def overview(image_type):
"""Everything the Overview tab shows, from data already on the server."""
entries = drivers(image_type)
by_os = {"Win11": 0, "Win10": 0, "Other": 0}
missing = []
for i, e in enumerate(entries, 1):
by_os[driver_os(e)] += 1
if not zip_path(image_type, e):
missing.append({"pos": i, "file": e.get("fileName") or e.get("FileName") or "",
"models": e.get("modelswminame") or e.get("models") or "",
"os": driver_os(e)})
sel, _ = user_selections(image_type)
oses = operating_systems(image_type)
pkgs = packages(image_type)
# Hardware seen by WinPE with no working driver entry in THIS image.
gaps = []
try:
for row in hardware_coverage.coverage(include_ok=False):
d = (row.get("drivers") or {}).get(image_type)
if d and d["status"] in ("no-entry", "zip-missing") and not row.get("dismissed"):
gaps.append({"model": row["model"], "status": d["status"],
"file": d.get("file", ""), "last_seen": row.get("last_seen"),
"skus": row.get("skus") or []})
except Exception:
pass
covered = image_type in dict(hardware_coverage.IMAGES)
# Newer Dell packs for files this map uses (last Updates > Dell scan).
dell = {"checked": None, "rows": []}
try:
from services import dell_updates
hints = dell_updates.update_hints()["drivers"]
dell["checked"] = (dell_updates.last_scan() or {}).get("checked")
seen = set()
for i, e in enumerate(entries, 1):
name = str(e.get("fileName") or e.get("FileName") or "")
if name.lower() in hints and name.lower() not in seen:
seen.add(name.lower())
dell["rows"].append({"pos": i, "file": name, "ver": hints[name.lower()],
"models": e.get("modelswminame") or ""})
except Exception:
pass
preferred = preferred_unattend_path(image_type)
hw_extra = control_file(image_type, "hw_drivers.json")
families = {str(e.get("family") or "") for e in entries}
models = sel.get("HardwareModelSelection") or []
return {
"drivers": {"total": len(entries), "by_os": by_os, "missing": missing},
"lint": _lint_counts(image_type),
"os": {"selection": sel.get("OperatingSystemSelection"), "rows": oses,
"current": _current_os(oses, sel.get("OperatingSystemSelection"))},
"packages": {
"total": len(pkgs),
"enabled": sum(1 for p in pkgs if p["entry"].get("enabled")),
"missing": [p for p in pkgs if p["entry"].get("enabled") and not p["file"]["exists"]],
},
"models": {"total": len(models),
"no_entry": [m for m in models if str(m.get("Id") or "") not in families]},
"files": _control_files(image_type),
"gaps": gaps,
"gaps_checked": covered,
"dell": dell,
"unattend_override": preferred if os.path.isfile(preferred) else None,
"hw_drivers_extra": hw_extra if os.path.isfile(hw_extra) else None,
"last_saves": {scope: control_files.last_save(image_type, scope)
for scope in ("drivers", "unattend", "os")},
}

724
webapp/services/ipxe_update.py Normal file
View File

@@ -0,0 +1,724 @@
"""Keep the Secure Boot iPXE chain on the TFTP root current.
What dnsmasq serves (see /etc/dnsmasq.conf):
* ipxe-shim.efi - Microsoft-signed shim (upstream x86_64-sb/shimx64.efi),
for the MACs tagged ipxe2; it always loads ipxe.efi from
the TFTP ROOT, whatever folder it came from;
* ipxe.efi - iPXE signed by the iPXE Secure Boot CA, loaded by the shim;
* ipxe-bcm.efi - the old 2021 Broadcom-signed build, everyone else.
Never touched here;
* ipxe.pxe - legacy BIOS iPXE (unsigned; BIOS has no Secure Boot).
Upstream publishes the signed set as one tarball on every release:
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
prints "Signature verification OK" even for a missing or unrelated cert.
Why it matters: the shim and the Broadcom build are signed by Microsoft's
UEFI CA 2011, whose signing period ended in 2026. Machines that only trust
the 2023 CA, or a revocation (DBX/SBAT) update, will refuse them; the fix is
a newer upstream release, installed here.
Staging runs as a background job (python3 -m services.ipxe_update stage);
install and rollback are a few MB of copying and run inline.
"""
from __future__ import annotations
import fcntl
import glob
import json
import os
import re
import shutil
import struct
import subprocess
import sys
import tarfile
import tempfile
from datetime import datetime
import config
from services import updates_common as uc
GITHUB_API = "https://api.github.com/repos/ipxe/ipxe/releases/latest"
RELEASES_PAGE = "https://github.com/ipxe/ipxe/releases"
LATEST_TARBALL_URL = "https://github.com/ipxe/ipxe/releases/latest/download/ipxeboot.tar.gz"
TARBALL = "ipxeboot.tar.gz"
JOB_UNIT = "pxe-webapp-ipxe-stage"
# TFTP name -> path inside the tarball, and whether it must carry a
# Secure Boot signature (and from whom).
MANAGED = {
"ipxe-shim.efi": ("ipxeboot/x86_64-sb/shimx64.efi", "microsoft"),
"ipxe.efi": ("ipxeboot/x86_64-sb/ipxe.efi", "ipxe"),
"ipxe.pxe": ("ipxeboot/x86_64/ipxe.pxe", None),
}
# Installed by default; ipxe.pxe (legacy BIOS) only when asked.
DEFAULT_INSTALL = ("ipxe-shim.efi", "ipxe.efi")
# Shown on the page for reference, never written.
READ_ONLY = ("ipxe-bcm.efi",)
NEED_LABEL = {"microsoft": "a Microsoft UEFI CA", "ipxe": "the iPXE Secure Boot CA"}
# Issuer CNs, as sbverify prints them.
MS_CA_2011 = "Microsoft Corporation UEFI CA 2011"
MS_CA_2023 = "Microsoft UEFI CA 2023"
IPXE_CA = "iPXE Secure Boot CA"
def _dir():
return uc.state_dir("ipxe")
def _job_file():
return os.path.join(_dir(), "job.json")
# --- reading binaries -------------------------------------------------------
def signers(path):
"""[(subject CN, issuer CN), ...] from the Authenticode chain, [] if none.
Raises FileNotFoundError when sbverify is not installed."""
r = subprocess.run(["sbverify", "--list", path], capture_output=True, text=True, timeout=60)
pairs, subject = [], None
for line in r.stdout.splitlines():
line = line.strip()
m = re.match(r"- subject: .*?CN=([^/]+)", line)
if m:
subject = m.group(1).strip()
continue
m = re.match(r"issuer: .*?CN=([^/]+)", line)
if m and subject is not None:
pairs.append((subject, m.group(1).strip()))
subject = None
return pairs
def signature_kind(pairs):
"""microsoft-2023 | microsoft-2011 | microsoft | ipxe | other | unsigned.
A file signed under both Microsoft CAs counts as 2023."""
issuers = {i for _, i in pairs}
names = " ".join(s + " " + i for s, i in pairs)
if MS_CA_2023 in issuers:
return "microsoft-2023"
if MS_CA_2011 in issuers:
return "microsoft-2011"
if "Microsoft" in names:
return "microsoft"
if IPXE_CA in issuers:
return "ipxe"
return "unsigned" if not pairs else "other"
def _sbat(data):
"""The SBAT CSV a shim or iPXE build carries, as {component: [fields]}."""
i = data.find(b"sbat,1,SBAT Version,")
if i < 0:
return {}
text = data[i:i + 4096].split(b"\0", 1)[0].decode("ascii", errors="replace")
out = {}
for line in text.splitlines():
fields = line.split(",")
if len(fields) >= 2 and fields[0] != "sbat":
out[fields[0]] = fields
return out
def version_of(path):
try:
with open(path, "rb") as f:
data = f.read()
except OSError:
return ""
# A shim says so in its build ident; the shim's SBAT also names iPXE
# (shim.ipxe), so test this first.
m = re.search(rb"\$Version: ([^ $]{1,30})", data)
if m:
return "shim " + m.group(1).decode(errors="replace")
sbat = _sbat(data)
if "ipxe" in sbat and len(sbat["ipxe"]) >= 5:
return "iPXE " + sbat["ipxe"][4]
m = re.search(rb"([0-9]+\.[0-9]+\.[0-9]+\+? ?\(g[0-9a-f]+\))", data)
if m:
return "iPXE " + m.group(1).decode(errors="replace")
return ""
def sbat_generations(path):
"""'shim 4, shim.ipxe 1': the generation numbers a revocation compares."""
try:
with open(path, "rb") as f:
sbat = _sbat(f.read())
except OSError:
return ""
return ", ".join(f"{k} {v[1]}" for k, v in sbat.items())
def vendor_cert_cn(path):
"""Subject CN of the certificate a shim trusts for its second stage (the
PE .vendor_cert section), or None. iPXE's shim embeds the iPXE Secure
Boot CA, so ipxe.efi must chain to that name."""
try:
with open(path, "rb") as f:
d = f.read()
pe = struct.unpack_from("<I", d, 0x3C)[0]
if d[pe:pe + 4] != b"PE\0\0":
return None
nsec = struct.unpack_from("<H", d, pe + 6)[0]
symptr, nsym = struct.unpack_from("<II", d, pe + 12)
optsz = struct.unpack_from("<H", d, pe + 20)[0]
strtab = symptr + nsym * 18
off = pe + 24 + optsz
for i in range(nsec):
raw, _vsize, _vaddr, _rsize, rptr = struct.unpack_from("<8sIIII", d, off + 40 * i)
name = raw.rstrip(b"\0")
if name.startswith(b"/") and symptr: # long name, in the COFF string table
o = strtab + int(name[1:])
name = d[o:d.index(b"\0", o)]
if name != b".vendor_cert":
continue
size, _dsize, cert_off, _doff = struct.unpack_from("<IIII", d, rptr)
cert = d[rptr + cert_off:rptr + cert_off + size]
# DER: every CN is OID 2.5.4.3 then a short string. Issuer comes
# before subject, so the last one is the subject.
cns = [cert[m.end() + 2:m.end() + 2 + cert[m.end() + 1]].decode(errors="replace")
for m in re.finditer(rb"\x06\x03\x55\x04\x03(?=[\x0c\x13\x16])", cert)]
return cns[-1] if cns else None
except (OSError, struct.error, ValueError, IndexError):
return None
return None
def describe(path):
if not os.path.isfile(path):
return None
st = os.stat(path)
efi = path.endswith(".efi")
pairs, error = [], ""
if efi:
try:
pairs = signers(path)
except (OSError, subprocess.SubprocessError) as ex:
error = f"sbverify failed: {ex}"
return {
"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",
"trusts": vendor_cert_cn(path) if efi else None,
"sig_error": error,
}
def _installed_log():
"""{name: {tag, sha256, at}}: which release each installed file came from."""
log = uc.read_json(os.path.join(_dir(), "installed.json"), {})
return log if isinstance(log, dict) else {}
def installed():
log = _installed_log()
rows = []
for name in list(MANAGED) + list(READ_ONLY):
info = describe(os.path.join(config.TFTP_ROOT, name))
origin = log.get(name) or {}
if info and origin.get("sha256") == info["sha256"]:
info["release"] = origin.get("tag")
rows.append({"name": name, "managed": name in MANAGED, "info": info})
# The shim only starts an ipxe.efi signed under the CA it embeds.
by = {r["name"]: r["info"] for r in rows}
shim, ipxe = by.get("ipxe-shim.efi"), by.get("ipxe.efi")
if shim and ipxe and shim.get("trusts"):
ipxe["chain_ok"] = shim["trusts"] in {i for _, i in ipxe["signers"]}
ipxe["trusts_needed"] = shim["trusts"]
return rows
# --- dnsmasq ----------------------------------------------------------------
MAC_RE = re.compile(r"^(?:[0-9a-fA-F*]{1,2}[:-]){5}[0-9a-fA-F*]{1,2}$")
# Debian/Ubuntu start dnsmasq with -7 /etc/dnsmasq.d,.dpkg-dist,.dpkg-old,
# .dpkg-new (CONFIG_DIR in /etc/default/dnsmasq), so that folder is read even
# when dnsmasq.conf never names it.
DEBIAN_CONF_DIR_EXCLUDE = (".dpkg-dist", ".dpkg-old", ".dpkg-new")
def _conf_dir_lines(folder, wanted, depth, seen):
"""dnsmasq conf-dir rules: '*.conf' entries are the only names read
(if any are given), other entries are suffixes to skip, and dot files
and '~' backups are always skipped."""
include = [w[1:] for w in wanted if w.startswith("*")]
exclude = [w for w in wanted if w and not w.startswith("*")]
out = []
for p in sorted(glob.glob(os.path.join(folder, "*"))):
base = os.path.basename(p)
if base.startswith(".") or base.endswith("~") or not os.path.isfile(p):
continue
if include and not any(base.endswith(w) for w in include):
continue
if any(base.endswith(w) for w in exclude):
continue
out += _conf_lines(p, depth + 1, seen)
return out
def _conf_lines(path, depth=0, seen=None):
"""Lines of dnsmasq.conf plus any conf-file/conf-dir it pulls in. Each
file is read once, so an include named twice does not list MACs twice."""
seen = set() if seen is None else seen
real = os.path.realpath(path)
if real in seen:
return []
seen.add(real)
try:
with open(path, encoding="utf-8", errors="replace") as f:
lines = f.read().splitlines()
except OSError:
return []
out = []
for line in lines:
s = line.split("#", 1)[0].strip()
if not s:
continue
out.append((path, s))
if depth > 2:
continue
if s.startswith("conf-file="):
out += _conf_lines(s.split("=", 1)[1].strip(), depth + 1, seen)
elif s.startswith("conf-dir="):
parts = s.split("=", 1)[1].split(",")
out += _conf_dir_lines(parts[0].strip(), [p.strip() for p in parts[1:]], depth, seen)
if depth == 0:
implicit = os.path.join(os.path.dirname(path), "dnsmasq.d")
if os.path.isdir(implicit):
out += _conf_dir_lines(implicit, list(DEBIAN_CONF_DIR_EXCLUDE), depth, seen)
return out
def dnsmasq_routing():
"""Which boot file each class of client gets, read from dnsmasq.conf.
Read-only: this module never edits dnsmasq."""
tag_macs, boots, raw = {}, [], []
for src, s in _conf_lines(config.DNSMASQ_CONF):
key, _, val = s.partition("=")
key = key.strip()
if key == "dhcp-host":
fields = [f.strip() for f in val.split(",")]
macs = [f.lower() for f in fields if MAC_RE.match(f)]
for f in fields:
if f.startswith("set:"):
tag_macs.setdefault(f[4:], []).extend(macs)
elif key == "dhcp-boot":
fields = [f.strip() for f in val.split(",")]
need = [f[4:] for f in fields if f.startswith("tag:") and not f.startswith("tag:!")]
exclude = [f[5:] for f in fields if f.startswith("tag:!")]
rest = [f for f in fields if not f.startswith("tag:")]
boots.append({"file": rest[0] if rest else "", "need": need, "exclude": exclude,
"line": s, "source": src})
if key in ("dhcp-boot", "dhcp-host", "dhcp-match", "dhcp-userclass", "dhcp-vendorclass",
"tag-if", "enable-tftp", "tftp-root"):
if key != "dhcp-host" or "set:" in val:
raw.append(s)
for b in boots:
b["macs"] = sorted({m for t in b["need"] for m in tag_macs.get(t, [])})
# A path like ipxe2/ipxe-shim.efi is another copy, not the file in
# the TFTP root that this page installs.
rel = b["file"].lstrip("/")
b["managed"] = rel in MANAGED
b["read_only"] = rel in READ_ONLY
b["other_copy"] = rel not in MANAGED and os.path.basename(rel) in MANAGED
return {"tags": tag_macs, "boots": boots, "lines": raw,
"shim_macs": sorted({m for b in boots if os.path.basename(b["file"]) == "ipxe-shim.efi"
for m in b["macs"]})}
# --- upstream ---------------------------------------------------------------
def _version_tuple(text):
m = re.search(r"(\d+)\.(\d+)(?:\.(\d+))?", text or "")
return tuple(int(x or 0) for x in m.groups()) if m else None
def compare(tag, installed_version):
"""'update' | 'current' | 'newer' (installed is ahead) | 'unknown'."""
a, b = _version_tuple(tag), _version_tuple(installed_version)
if not a or not b:
return "unknown"
return "update" if a > b else "current" if a == b else "newer"
def check_latest(timeout=20):
"""Latest GitHub release: tag, date, tarball URL and sha256. Cached in
latest.json so the page and the stage job agree on what to fetch."""
try:
rel = json.loads(uc.fetch_text(GITHUB_API, timeout=timeout))
except ValueError:
raise RuntimeError("GitHub returned something that is not JSON (proxy login page?)")
asset = next((a for a in rel.get("assets", []) if a.get("name") == TARBALL), None)
if not rel.get("tag_name"):
raise RuntimeError("GitHub answer has no release tag")
if not asset:
raise RuntimeError(f"release {rel.get('tag_name')} has no {TARBALL}")
digest = str(asset.get("digest") or "")
page = str(rel.get("html_url") or "") # shown as a link on the page
info = {
"tag": rel.get("tag_name"), "name": rel.get("name") or "",
"published": rel.get("published_at"),
"page": page if page.startswith("https://") else RELEASES_PAGE,
"url": asset.get("browser_download_url"),
"size": asset.get("size"),
"sha256": digest.split(":", 1)[1].lower() if digest.startswith("sha256:") else "",
"checked": uc.now(),
}
uc.write_json(os.path.join(_dir(), "latest.json"), info)
return info
def last_check():
return uc.read_json(os.path.join(_dir(), "latest.json"))
def _safe_tag(tag):
tag = re.sub(r"[^\w.-]", "_", str(tag or ""))
if not tag or tag in (".", ".."):
raise ValueError("bad release tag")
return tag
def _resolve(members, name):
"""Follow symlinks/hardlinks inside the archive to a regular member."""
for _ in range(8):
m = members.get(name)
if m is None or m.isfile():
return m
if m.issym():
name = os.path.normpath(os.path.join(os.path.dirname(name), m.linkname))
elif m.islnk():
name = os.path.normpath(m.linkname)
else:
return None
return None
def _progress(**fields):
data = uc.read_json(_job_file(), {}) or {}
data.update(fields, updated=uc.now())
uc.write_json(_job_file(), data)
def stage(info=None):
"""Download the tarball, verify it, unpack the managed files to
staged/<tag>/ and check each one's signature. Returns the staged record."""
info = info or last_check() or check_latest()
if not info.get("sha256"):
raise RuntimeError("GitHub did not publish a sha256 for this release; stage it by hand")
folder = os.path.join(_dir(), "staged", _safe_tag(info["tag"]))
os.makedirs(folder, exist_ok=True)
tar_path = os.path.join(folder, TARBALL)
_progress(step="download", tag=info["tag"], size=info.get("size"), tarball=tar_path)
if os.path.isfile(tar_path) and uc.file_hash(tar_path).lower() != info["sha256"].lower():
# A partial is resumed. A wrong file at full size or more cannot be
# (curl fails with 416/36 on every retry), so start it over.
size = info.get("size") or 0
if not size or os.path.getsize(tar_path) >= size:
os.remove(tar_path)
if not (os.path.isfile(tar_path) and uc.file_hash(tar_path).lower() == info["sha256"].lower()):
# Room for a slow cap: twice the time the cap allows, at least 10 min.
cap = float(uc.settings().get("max_mbps") or 0)
size = float(info.get("size") or 0)
timeout = max(600, int(2 * size / (cap * 1024 * 1024))) if cap and size else 600
uc.download(info["url"], tar_path, expect=info["sha256"], timeout=timeout)
_progress(step="verify")
files = {}
with tarfile.open(tar_path) as tar:
members = {os.path.normpath(m.name[2:] if m.name.startswith("./") else m.name): m
for m in tar.getmembers()}
for name, (inner, need) in MANAGED.items():
m = _resolve(members, inner)
if m is None:
files[name] = {"ok": False, "error": f"{inner} not in the tarball"}
continue
dest = os.path.join(folder, name)
tmp = dest + ".part"
with tar.extractfile(m) as src, open(tmp, "wb") as out:
shutil.copyfileobj(src, out)
os.replace(tmp, dest)
d = describe(dest)
if d["sig_error"]:
ok, error = False, d["sig_error"]
elif need and not d["kind"].startswith(need):
ok, error = False, f"expected a signature under {NEED_LABEL[need]}, found {d['kind']}"
elif not d["size"]:
ok, error = False, "empty file"
else:
ok, error = True, ""
files[name] = dict(d, ok=ok, error=error, source=inner)
# The shim from this release must trust the CA its ipxe.efi is signed by.
shim, ipxe = files.get("ipxe-shim.efi") or {}, files.get("ipxe.efi") or {}
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")
record = {"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
def staged(tag=None):
root = os.path.join(_dir(), "staged")
if not os.path.isdir(root):
return []
out = []
for t in sorted(os.listdir(root), reverse=True):
rec = uc.read_json(os.path.join(root, t, "staged.json"))
if not (isinstance(rec, dict) and rec.get("tag") and isinstance(rec.get("files"), dict)
and rec["files"]):
continue # a hand-edited or half-written record has nothing to install
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)
out.append(rec)
out.sort(key=lambda r: r.get("staged") or "", reverse=True)
return out
# --- background stage job ---------------------------------------------------
def job_state():
"""Last stage job's progress, with 'running' from systemd and, while
downloading, a percent from the partial file size."""
data = uc.read_json(_job_file(), {}) or {}
try:
running = uc.job_running(JOB_UNIT)
except OSError:
running = False
data["running"] = running
if data.get("state") in ("queued", "running") and not running:
# Give systemd-run a moment to bring the unit up before calling it dead.
try:
age = datetime.now().timestamp() - os.path.getmtime(_job_file())
except OSError:
age = 0
if age > 20:
data["state"] = "interrupted"
if data.get("step") == "download" and data.get("size") and data.get("tarball"):
try:
done = os.path.getsize(data["tarball"])
data["percent"] = min(100, int(100 * done / data["size"]))
except OSError:
data["percent"] = 0
return data
def start_stage():
"""Queue the stage job (systemd-run). The page polls job_state()."""
if uc.job_running(JOB_UNIT):
raise RuntimeError("a download is already running")
info = last_check()
if not info:
raise RuntimeError("check for updates first")
uc.write_json(_job_file(), {"state": "queued", "tag": info["tag"], "started": uc.now(),
"updated": uc.now()})
try:
uc.start_job(JOB_UNIT, ["services.ipxe_update", "stage"])
except Exception as ex:
uc.write_json(_job_file(), {"state": "error", "tag": info["tag"], "error": str(ex),
"updated": uc.now()})
raise
def run_stage_job():
"""Entry point of the background job. Never raises; the outcome goes to job.json."""
_progress(state="running", step="start", error="", started=uc.now())
try:
rec = stage(last_check())
bad = [n for n, f in rec["files"].items() if not f.get("ok")]
_progress(state="done", step="done", tag=rec["tag"], finished=uc.now(),
failed=bad)
return 0
except Exception as ex: # job boundary: record any failure for the page
_progress(state="error", error=str(ex)[:500], finished=uc.now())
return 1
# --- install / rollback -----------------------------------------------------
def _new_folder(root):
"""A fresh timestamped folder; never reuses one (a rollback in the same
second as its backup must not overwrite the files it restores)."""
base = os.path.join(root, datetime.now().strftime("%Y%m%d-%H%M%S"))
folder, n = base, 1
while True:
try:
os.makedirs(folder)
return folder
except FileExistsError:
folder, n = f"{base}-{n}", n + 1
def _backup(names, reason):
folder = _new_folder(os.path.join(_dir(), "backups"))
saved, absent = [], []
for name in names:
src = os.path.join(config.TFTP_ROOT, name)
if os.path.isfile(src):
shutil.copy2(src, os.path.join(folder, name))
saved.append(name)
else:
absent.append(name)
uc.write_json(os.path.join(folder, "backup.json"),
{"reason": reason, "files": saved, "absent": absent, "at": uc.now()})
return folder
class _Lock:
"""One install or rollback at a time: two at once would back up each
other's half-placed files."""
def __enter__(self):
self.f = open(os.path.join(_dir(), ".install.lock"), "w")
try:
fcntl.flock(self.f, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
self.f.close()
raise RuntimeError("another install or rollback is running; try again") from None
return self
def __exit__(self, *exc):
self.f.close()
def _prepare(src, name, sha256=None):
"""Copy src next to its TFTP name under a temporary name, fsynced, mode
0644 and with the current file's owner. With sha256, the copy is checked.
Returns the temporary path; nothing is live yet."""
dest = os.path.join(config.TFTP_ROOT, name)
fd, tmp = tempfile.mkstemp(dir=config.TFTP_ROOT, prefix=f".{name}.", suffix=".new")
try:
with open(src, "rb") as s, os.fdopen(fd, "wb") as out:
shutil.copyfileobj(s, out)
out.flush()
os.fsync(out.fileno())
if sha256 and uc.file_hash(tmp).lower() != sha256.lower():
raise RuntimeError(f"{name}: copy does not match its sha256, nothing replaced")
os.chmod(tmp, 0o644)
try:
st = os.stat(dest)
os.chown(tmp, st.st_uid, st.st_gid)
except OSError:
pass # no current file, or not root (local tests)
return tmp
except BaseException:
os.remove(tmp)
raise
def _place_all(items):
"""[(src, name, sha256)] into the TFTP root. Every copy is made and
checked first, then each is renamed into place (atomic per file, so a
booting client never reads half a file). A failed copy leaves all the
live files as they were."""
ready = []
try:
for src, name, sha256 in items:
ready.append((_prepare(src, name, sha256), name))
except BaseException:
for tmp, _ in ready:
os.remove(tmp)
raise
for tmp, name in ready:
os.replace(tmp, os.path.join(config.TFTP_ROOT, name))
def install(tag, names=DEFAULT_INSTALL):
"""Copy staged files for `tag` into the TFTP root after backing up the
current ones. Refuses a set with a failed signature check, and refuses to
install ipxe-shim.efi without its matching ipxe.efi (the shim only trusts
the iPXE CA the same release signs with). Returns the backup folder."""
rec = next(iter(staged(tag)), None)
if not rec:
raise ValueError(f"{tag} is not staged")
names = [n for n in MANAGED if n in names]
if not names:
raise ValueError("nothing selected")
if "ipxe-shim.efi" in names and "ipxe.efi" not in names:
raise ValueError("install ipxe-shim.efi together with ipxe.efi")
if uc.job_running(JOB_UNIT):
raise RuntimeError("a download is running; wait for it to finish")
with _Lock():
return _install(tag, rec, names)
def _install(tag, rec, names):
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"]:
raise ValueError(f"{n}: staged copy changed since it was verified; download again")
if "ipxe.efi" in names and "ipxe-shim.efi" not in names:
# ipxe.efi alone must still be one the installed shim will start.
trusts = vendor_cert_cn(os.path.join(config.TFTP_ROOT, "ipxe-shim.efi"))
issuers = {i for _, i in rec["files"]["ipxe.efi"].get("signers", [])}
if trusts and trusts not in issuers:
raise ValueError(f"the installed shim trusts {trusts}; this ipxe.efi is not signed "
"under it, install the shim from the same release too")
backup = _backup(names, f"before installing {tag}")
_place_all([(os.path.join(rec["folder"], n), n, rec["files"][n]["sha256"]) for n in names])
log = _installed_log()
for n in names:
log[n] = {"tag": tag, "sha256": rec["files"][n]["sha256"], "at": uc.now()}
uc.write_json(os.path.join(_dir(), "installed.json"), log)
return backup
def backups():
root = os.path.join(_dir(), "backups")
if not os.path.isdir(root):
return []
out = []
for b in sorted(os.listdir(root), reverse=True):
folder = os.path.join(root, b)
if not os.path.isdir(folder):
continue
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))]
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}})
return out
def rollback(backup_id):
"""Put the files of one backup back, after backing up what is there now."""
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):
raise ValueError("no such backup")
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")
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)))
for f in files])
log = _installed_log()
for f in files:
log.pop(f, None)
uc.write_json(os.path.join(_dir(), "installed.json"), log)
return files
if __name__ == "__main__":
if sys.argv[1:2] == ["stage"]:
sys.exit(run_stage_job())
print("usage: python3 -m services.ipxe_update stage", file=sys.stderr)
sys.exit(2)

371
webapp/services/mcl_ppkg.py Normal file
View File

@@ -0,0 +1,371 @@
"""MCL provisioning packages for gea-standard and gea-engineer.
The vendor's bulk-enroll scripts in _shared/Applications/BPRT (every image's
Deploy/Applications/BPRT is a symlink to it) pick the package by filename:
* the name must match ^((?:\\w|-)+?_){5,}v\\d+\\.\\d+.ppkg$ (PowerShell
-Match, so case-insensitive), e.g. GCCH_Prod_MCL-PERS_Exp_XXXXXXXX_v2.19.ppkg;
* Insert-BPRTToPPKG.ps1 then keeps the names that are -like
"<Tenant>_<Environment>_<Purpose>*" (a case-insensitive startswith, with
no trailing underscore) and FAILS enrollment when more than one is left.
So an upload replaces the package with the same first three fields, and the
old one moves to BPRT-archive/<timestamp>/ - outside BPRT, because PESetup
copies the whole folder to every PC. Restore swaps an archived package back
the same way. Either is refused when the result would leave two packages that
one prefix claims (e.g. MCL-SH and MCL-SHX).
A package is a WIM (Insert-BPRTToPPKG.ps1 renames it to .wim and mounts it),
so an upload must start with the WIM magic. The upload is written under the
archive folder first and renamed into BPRT only when complete, so a bay
copying BPRT mid-upload never sees a partial file.
"""
from __future__ import annotations
import errno
import os
import re
import shutil
import threading
import time
from datetime import datetime
import config
# The vendor's own pattern (Insert-BPRTToPPKG.ps1 / Start-BulkEnrollOrchestrator.ps1).
VENDOR_PATTERN = re.compile(r"^((?:\w|-)+?_){5,}v\d+\.\d+.ppkg$", re.IGNORECASE)
_VERSION = re.compile(r"_(v\d+\.\d+)\.ppkg$", re.IGNORECASE)
# Insert-BPRTToPPKG.ps1 swaps this placeholder for the token's expiry date on
# the PC's copy; a name without it is a package already used on a PC.
_PLACEHOLDER = "_exp_xxxxxxxx_"
_BATCH = re.compile(r"^\d{8}-\d{6}(-\d+)?$")
WIM_MAGIC = b"MSWIM\0\0\0"
# Serialises the archive-and-place step (uploads and restores).
_lock = threading.Lock()
def bprt_dir():
return os.path.join(config.SHARED_DIR, "Applications", "BPRT")
def archive_dir():
return os.path.join(config.SHARED_DIR, "Applications", "BPRT-archive")
def clean_name(name):
"""Bare file name; some browsers send a full Windows path."""
return os.path.basename(str(name or "").replace("\\", "/")).strip()
def is_mcl_name(name):
"""True for a name the bulk-enroll scripts would treat as an MCL package
(the shopfloor SFLD names, e.g. GCCH_Prod_SFLD_v4.18.ppkg, are not)."""
return bool(VENDOR_PATTERN.match(clean_name(name)))
def parse(name, strict=True):
"""{"prefix", "version"} for a usable package name, else ValueError.
The vendor regex ends in `.ppkg` with an unescaped dot, so it also takes
e.g. ..._v2.19-ppkg; strict (uploads, restores) wants the real extension,
while the listing (strict=False) shows what the enroll scripts would see."""
if not VENDOR_PATTERN.match(name) or (strict and not name.lower().endswith(".ppkg")):
raise ValueError(
f"{name} does not match the vendor naming the enroll scripts need, "
"e.g. GCCH_Prod_MCL-PERS_Exp_XXXXXXXX_v2.19.ppkg")
m = _VERSION.search(name)
return {"prefix": "_".join(name.split("_")[:3]),
"version": m.group(1) if m else ""}
def _claims(prefix, name):
"""What the vendor's `-like "$packagePrefix*"` does."""
return name.lower().startswith(prefix.lower())
def _clashes(names):
"""{name: [other names its prefix also claims]} for a set of vendor-valid
names; empty when every prefix finds exactly one package."""
out = {}
for n in names:
others = [m for m in names if m != n and _claims(parse(n, strict=False)["prefix"], m)]
if others:
out[n] = others
return out
def _check_unique(names, name):
"""Refuse when `name` would be part of a clash: its prefix claims another
package, or another's prefix claims it. A clash among other packages
(made by hand) is flagged on the page but does not block this one."""
clashes = {n: o for n, o in _clashes(names).items() if n == name or name in o}
if clashes:
n, others = next(iter(clashes.items()))
raise ValueError(
f"{parse(n, strict=False)['prefix']} would match {n} and {', '.join(others)}; "
"the enroll scripts fail when one prefix finds two packages")
def _pkg(folder, name):
st = os.stat(os.path.join(folder, name))
info = {"filename": name, "size": st.st_size, "modified": st.st_mtime,
"placeholder": _PLACEHOLDER in name.lower()}
try:
info.update(parse(name, strict=False))
except ValueError:
info.update(prefix="", version="")
return info
def _listed(folder, name):
return ((name.lower().endswith(".ppkg") or VENDOR_PATTERN.match(name))
and os.path.isfile(os.path.join(folder, name)))
def packages():
"""Current packages in BPRT, sorted by prefix."""
folder = bprt_dir()
out = []
if os.path.isdir(folder):
for name in os.listdir(folder):
if _listed(folder, name):
out.append(_pkg(folder, name))
out.sort(key=lambda p: (p["prefix"].lower(), p["filename"].lower()))
# A prefix that finds two packages breaks enrollment for that tenant/purpose.
clashes = _clashes([p["filename"] for p in out if p["prefix"]])
for p in out:
p["conflict_with"] = clashes.get(p["filename"], [])
p["conflict"] = bool(p["conflict_with"])
return out
def archived():
"""Archived packages, newest batch first: [{"batch", "filename", ...}]."""
root = archive_dir()
out = []
if os.path.isdir(root):
for batch in sorted(os.listdir(root), reverse=True):
folder = os.path.join(root, batch)
if not _BATCH.match(batch) or not os.path.isdir(folder):
continue # .incoming (uploads in progress) and strays
for name in sorted(os.listdir(folder)):
if _listed(folder, name):
out.append(dict(_pkg(folder, name), batch=batch))
return out
# --- moving files -----------------------------------------------------------
def _move(src, dst):
"""Rename src onto dst. BPRT and its archive sit on one filesystem, so this
is a rename; the copy fallback only covers a share split across mounts."""
try:
os.replace(src, dst)
except OSError as exc:
if exc.errno != errno.EXDEV:
raise
tmp = os.path.join(os.path.dirname(dst), f".{os.path.basename(dst)}.part")
try:
shutil.copy2(src, tmp)
os.replace(tmp, dst)
finally:
if os.path.exists(tmp):
os.remove(tmp)
os.remove(src)
def _new_batch():
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
path = os.path.join(archive_dir(), stamp)
n = 1
while os.path.exists(path): # two changes in one second must not share a batch
n += 1
path = os.path.join(archive_dir(), f"{stamp}-{n}")
os.makedirs(path)
_match_perms(path, directory=True)
return path
def _archive(names):
"""Move these BPRT packages into a new archive batch. Returns the batch
path (None when there was nothing to move)."""
if not names:
return None
batch = _new_batch()
done = []
try:
for name in names:
_move(os.path.join(bprt_dir(), name), os.path.join(batch, name))
done.append(name)
except OSError:
_unarchive(batch, done)
raise
return batch
def _unarchive(batch, names):
"""Undo _archive after a failed placement: best effort, every file back."""
for name in names:
try:
_move(os.path.join(batch, name), os.path.join(bprt_dir(), name))
except OSError:
pass
try:
os.rmdir(batch)
except OSError:
pass
def _same_prefix(prefix):
"""Valid BPRT package names whose own prefix equals this one."""
return [p["filename"] for p in packages()
if p["prefix"] and p["prefix"].lower() == prefix.lower()]
def _plan(name, info):
"""(names to archive) for placing `name`, or ValueError if the result
would leave a prefix finding two packages."""
current = [p["filename"] for p in packages() if p["prefix"]]
replaced = _same_prefix(info["prefix"])
_check_unique([n for n in current if n not in replaced] + [name], name)
return replaced
def _place(src, name, info):
"""Archive what `name` replaces, then rename src into BPRT. Returns the
archived names."""
with _lock:
replaced = _plan(name, info)
batch = _archive(replaced)
dest = os.path.join(bprt_dir(), name)
try:
_move(src, dest)
except OSError:
if batch:
_unarchive(batch, replaced)
raise
return replaced
# --- upload and restore -----------------------------------------------------
def check_upload(name):
"""Validate a name before the body is saved: (clean name, info)."""
name = clean_name(name)
info = parse(name)
_plan(name, info)
return name, info
def install_upload(fileobj, name):
"""Save an uploaded package into BPRT, archiving the one it replaces.
Returns (info, archived_names); info["filename"] is the name used and
info["warnings"] lists things worth telling the user."""
name, info = check_upload(name)
os.makedirs(bprt_dir(), exist_ok=True)
incoming = os.path.join(archive_dir(), ".incoming")
os.makedirs(incoming, exist_ok=True)
_sweep(incoming)
_match_perms(archive_dir(), directory=True)
_match_perms(incoming, directory=True)
tmp = os.path.join(incoming, f"{os.getpid()}-{threading.get_ident()}-{name}.part")
try:
fileobj.save(tmp)
if os.path.getsize(tmp) == 0:
raise ValueError("uploaded file is empty")
with open(tmp, "rb") as f:
if f.read(len(WIM_MAGIC)) != WIM_MAGIC:
raise ValueError(f"{name} is not a provisioning package (no WIM header)")
moved = _place(tmp, name, info)
_match_perms(os.path.join(bprt_dir(), name))
finally:
if os.path.exists(tmp):
os.remove(tmp)
info["filename"] = name
info["warnings"] = []
if _PLACEHOLDER not in name.lower():
info["warnings"].append(
f"{name} has no Exp_XXXXXXXX placeholder; a package already used on a PC "
"carries its BPRT and the enroll script cannot inject a new one")
return info, moved
def _sweep(incoming, age=86400):
"""Drop temp files a crashed upload left behind (300 MB each, invisible
on the page). A live upload's file is seconds old."""
cutoff = time.time() - age
for n in os.listdir(incoming):
path = os.path.join(incoming, n)
try:
if n.endswith(".part") and os.path.isfile(path) and os.path.getmtime(path) < cutoff:
os.remove(path)
except OSError:
pass
def archive(name):
"""Move one package out of BPRT into a new archive batch (clears a
duplicate made by hand). Returns the bare name."""
name = clean_name(name)
with _lock:
if name not in [p["filename"] for p in packages()]:
raise ValueError(f"{name or '(blank)'} is not a package in BPRT")
_archive([name])
return name
def restore(batch, name):
"""Put an archived package back, archiving whatever holds its prefix now."""
batch = str(batch or "")
if not _BATCH.match(batch):
raise ValueError(f"{batch or '(blank)'} is not an archive batch")
name = clean_name(name)
folder = os.path.join(archive_dir(), batch)
src = os.path.join(folder, name)
if not name or not os.path.isfile(src):
raise ValueError(f"{batch}/{name} is not in the archive")
info = parse(name)
moved = _place(src, name, info)
try:
os.rmdir(folder)
except OSError:
pass # batch still holds other packages
info["filename"] = name
return info, moved
def _match_perms(path, directory=False):
"""Uploads run as root; keep the share's pxe ownership so the Samba side
can still manage the file."""
try:
st = os.stat(bprt_dir())
os.chown(path, st.st_uid, st.st_gid)
os.chmod(path, 0o755 if directory else 0o744)
except OSError:
pass
def image_links():
"""[{"image", "state", "target"}] for every image with a
Deploy/Applications folder: state "ok" when its BPRT is the link to the
shared folder, "copy" when it is a real folder (an image import rsync
replaces the link with one, and then that image's PCs never see a
package changed here), "elsewhere" when it links somewhere else or
dangles, "none" when the image has no BPRT."""
shared = os.path.realpath(bprt_dir())
out = []
for image in config.IMAGE_TYPES:
apps = os.path.join(config.SAMBA_SHARE, image, "Deploy", "Applications")
if not os.path.isdir(apps):
continue
path = os.path.join(apps, "BPRT")
target = ""
if os.path.islink(path):
target = os.readlink(path)
state = "ok" if os.path.realpath(path) == shared and os.path.isdir(path) else "elsewhere"
elif os.path.isdir(path):
state = "copy"
else:
state = "none"
out.append({"image": image, "state": state, "target": target, "path": path})
return out

1042
webapp/services/unattend.py
View File

File diff suppressed because it is too large Load Diff

184
webapp/services/updates_common.py Normal file
View File

@@ -0,0 +1,184 @@
"""Shared plumbing for the Updates pages (Dell drivers/BIOS and iPXE).
* settings: the download speed cap, one small JSON file;
* download(): curl with that cap, resume, and a hash check;
* background jobs: long downloads run as a transient systemd unit
(systemd-run), so a webapp restart or a closed browser does not kill a
40-minute driver pull. Each job writes its progress to a JSON state file
the page polls.
The server reaches the internet through its corporate NIC (enp0s31f6); GE's
Zscaler re-signs GitHub, so the Zscaler root CA must be in the system trust
store (the playbook installs it). downloads.dell.com is not intercepted.
"""
from __future__ import annotations
import hashlib
import json
import math
import os
import subprocess
import sys
import tempfile
from datetime import datetime, timezone
import config
DEFAULT_SETTINGS = {
# MB/s; 0 = no cap. 10 MB/s is about 80 Mbps, a fifth of the site link.
"max_mbps": 10,
}
def state_dir(*parts):
path = os.path.join(config.UPDATES_DIR, *parts)
os.makedirs(path, exist_ok=True)
return path
def now():
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def read_json(path, default=None):
try:
with open(path, encoding="utf-8") as f:
return json.load(f)
except (OSError, ValueError):
return default
def write_json(path, data):
folder = os.path.dirname(path)
os.makedirs(folder, exist_ok=True)
fd, tmp = tempfile.mkstemp(dir=folder, prefix=".tmp-")
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(data, f, indent=1)
os.chmod(tmp, 0o644) # mkstemp makes 0600
os.replace(tmp, path)
except BaseException:
if os.path.exists(tmp):
os.remove(tmp)
raise
# --- settings ---------------------------------------------------------------
def settings():
data = dict(DEFAULT_SETTINGS)
saved = read_json(os.path.join(state_dir(), "settings.json"), {})
if isinstance(saved, dict):
data.update(saved)
try:
cap = float(data.get("max_mbps"))
except (TypeError, ValueError):
cap = -1
if not math.isfinite(cap) or cap < 0:
data["max_mbps"] = DEFAULT_SETTINGS["max_mbps"] # hand-edited junk
return data
def save_settings(max_mbps):
try:
value = float(str(max_mbps).strip())
except ValueError:
raise ValueError("speed cap must be a number of MB/s") from None
# float() takes "nan", which passes both comparisons below.
if not math.isfinite(value) or value < 0 or value > 1000:
raise ValueError("speed cap must be between 0 and 1000 MB/s")
data = settings()
# A tiny cap must not round to 0, which means unlimited.
data["max_mbps"] = max(round(value, 1), 0.1) if value > 0 else 0
write_json(os.path.join(state_dir(), "settings.json"), data)
return data
# --- downloads --------------------------------------------------------------
def file_hash(path, algorithm="sha256"):
h = hashlib.new(algorithm)
with open(path, "rb") as f:
for block in iter(lambda: f.read(1 << 22), b""):
h.update(block)
return h.hexdigest()
def download(url, dest, expect=None, algorithm="sha256", timeout=None):
"""Fetch url to dest at the configured speed cap, resuming a partial file.
expect: hex digest to verify (case-insensitive). Raises on failure.
Without expect nothing catches a stale file resumed into, so callers
that pass no hash should remove dest first."""
os.makedirs(os.path.dirname(os.path.abspath(dest)), exist_ok=True)
cmd = ["curl", "-sS", "-L", "--fail", "--retry", "5", "--retry-delay", "5",
"-C", "-", "-o", dest]
cap = settings().get("max_mbps") or 0
if cap:
cmd += ["--limit-rate", f"{max(int(float(cap) * 1024 * 1024), 1024)}"]
if timeout:
cmd += ["-m", str(int(timeout))]
for attempt in range(2):
r = subprocess.run(cmd + [url], capture_output=True, text=True)
if r.returncode == 33:
# The server will not resume: the partial on disk is not the
# file, so throw it away and fetch from the start.
if os.path.exists(dest):
os.remove(dest)
continue
if r.returncode:
raise RuntimeError(f"download failed ({r.returncode}): {r.stderr.strip()[:300]}")
if not expect or file_hash(dest, algorithm).lower() == expect.strip().lower():
return dest
# A corrupt partial resumes into a corrupt whole: start over once.
os.remove(dest)
if expect and os.path.exists(dest):
os.remove(dest)
raise RuntimeError(f"could not download a verified copy of {os.path.basename(dest)}"
+ (f" ({algorithm} mismatch)" if expect else ""))
def fetch_text(url, timeout=30):
r = subprocess.run(["curl", "-sS", "-L", "--fail", "-m", str(timeout), url],
capture_output=True)
if r.returncode:
raise RuntimeError(f"cannot fetch {url}: {r.stderr.decode(errors='replace').strip()[:300]}")
return r.stdout
# --- background jobs --------------------------------------------------------
def job_running(unit):
r = subprocess.run(["systemctl", "is-active", "--quiet", unit])
return r.returncode == 0
def _job_env():
"""--setenv flags for the job: a transient unit starts with an empty
environment, but pxe-webapp.service sets the share paths (SAMBA_SHARE,
WEB_ROOT, AUDIT_LOG, ...) and TZ, and the job must see the same config."""
keep = {"TZ", "LANG", "PYTHONPATH"}
# Never secrets: systemd-run's argv is readable by every local user.
secret = ("KEY", "SECRET", "PASS", "TOKEN")
return [f"--setenv={k}={v}" for k, v in sorted(os.environ.items())
if (k in keep or (k.isupper() and hasattr(config, k)))
and not any(w in k for w in secret) and "\n" not in v]
def start_job(unit, args):
"""Run `python3 -m <args>` from the webapp directory as a transient
systemd unit, with the webapp's own interpreter and config environment.
Refuses when the unit is already running."""
if job_running(unit):
raise RuntimeError("a job of this kind is already running")
subprocess.run(["systemctl", "reset-failed", unit], capture_output=True)
here = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
r = subprocess.run(
["systemd-run", f"--unit={unit}", "--collect", f"--working-directory={here}"]
+ _job_env() + [sys.executable or "/usr/bin/python3", "-m"] + list(args),
capture_output=True, text=True)
if r.returncode:
raise RuntimeError(f"systemd-run failed: {r.stderr.strip()[:300]}")
def stop_job(unit):
subprocess.run(["systemctl", "stop", unit], capture_output=True)