397063f987
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.
372 lines
13 KiB
Python
372 lines
13 KiB
Python
"""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
|