2bcc9d8016
Blocker - /reports/export returned 500: REPORT_EXPORT_RANGES and _months_ago were deleted with the old reports route. Restored. Major - A session without stage_index/stage_total took down /, /imaging and /imaging/tiles (int filter on Undefined). The templates default both; saving a LAPS password for a session that no longer exists is refused (it created a bare session). - /reports/todo parsed every report four times (23 s at 700 reports). blancco_report.parse() now caches by path, mtime and size: 0.09 s. - Stored XSS through the session serial in the Clear confirm (the browser decodes ' before running an inline handler). The message moves to data-confirm, and update_session() only accepts serials of letters, digits, '.', '_' and '-' (400 from /imaging/status otherwise). - Reports without a document id and with odd characters in the file name got a key the mark routes rejected, so they could never be cleared. report_key() now always produces a valid key. Minor - failure_note() handles a list-valued erasure target (500 on the report and to-do pages). - _recently_marked() tolerates naive, 'Z' and non-dict state entries. - Key regexes use fullmatch (a trailing newline was accepted). - Unreadable reports are no longer counted as to tick / to destroy; the reports page already flags them as Unreadable. - Rollback review recounts restorable files after reading the backups. - Enrollment: the startnet fallback package (no or empty ppkg.conf) and a quoted SOURCE_PPKG are protected too. - Backup upload overwrite check is case-insensitive (SMB share) and replaces the existing file rather than adding a case twin. - Overview polling keeps the last good rows on an error and pauses while the tab is hidden. A shown or typed LAPS password survives the floor's refresh. - hardware_dismiss only redirects to same-site paths; tiles ?limit is clamped; the Updates overview survives an unreadable Dell scan or iPXE state; describe() uses aversion(); a client cannot label its session 'inferred'.
1525 lines
60 KiB
Python
1525 lines
60 KiB
Python
"""Keep the Dell Win11 driver packs and the WinPE BIOS files current.
|
|
|
|
What it manages:
|
|
* driver packs: every Win11 entry in the three images' HardwareDriver.json
|
|
(gea-standard, gea-engineer, gea-shopfloor; standard and engineer are
|
|
byte-identical and share the zips under _shared/Out-of-box Drivers);
|
|
* BIOS: the WinPE manifest models.txt in BIOS_DIR
|
|
(`ModelSubstring|BIOSFile|Version`, first substring match wins in
|
|
check-bios.cmd, so lines are never reordered).
|
|
|
|
Where the answers come from (both cabs are plain XML inside, read with 7z):
|
|
* https://downloads.dell.com/catalog/DriverPackCatalog.cab: one
|
|
DriverPackage per pack, with OS codes, supported models and hashes;
|
|
* https://downloads.dell.com/catalog/CatalogIndexPC.cab: one GroupManifest
|
|
per system ID pointing at a small per-model cab whose BIOS
|
|
SoftwareComponent is the latest BIOS. Per-model cabs are cached and
|
|
checked against the sha256 the index gives. (CatalogPC.cab is stale and
|
|
is not used.)
|
|
|
|
The flow, each step a function here and a button on /updates/dell:
|
|
1. check() background job: fetch catalogs, write scan.json;
|
|
2. download() background job: fetch the selected packs/BIOS at the speed
|
|
cap, verify Dell's sha256 (md5 fallback), turn each driver
|
|
exe into the deflate zip PESetup expects, stage it;
|
|
3. preview()/apply(): place staged files next to the files they replace
|
|
(rename; staging and share are one filesystem), rewrite all
|
|
three maps and models.txt with backups, keep history;
|
|
4. rollback(): put the maps / models.txt back from those backups.
|
|
Old zips and exes are never deleted.
|
|
|
|
Run as a background job: python3 -m services.dell_updates check|download|apply
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import fcntl
|
|
import glob
|
|
import difflib
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import xml.etree.ElementTree as ET
|
|
from datetime import datetime
|
|
from pathlib import PurePosixPath
|
|
|
|
import config
|
|
from services import updates_common as uc
|
|
|
|
NS = "{openmanage/cm/dm}"
|
|
DELL = "https://downloads.dell.com/"
|
|
CATALOGS = {
|
|
"drivers": "catalog/DriverPackCatalog.cab",
|
|
"index": "catalog/CatalogIndexPC.cab",
|
|
}
|
|
# Images whose driver maps are scanned and rewritten (all deploy Win11).
|
|
IMAGES = ("gea-standard", "gea-engineer", "gea-shopfloor")
|
|
# PESetup osId values that mean Windows 11 (18 is Windows 10).
|
|
WIN11_OS_IDS = {"19", "20", "21"}
|
|
# Background job unit names (systemd-run --unit).
|
|
UNITS = {"check": "pxe-dell-check", "download": "pxe-dell-download"}
|
|
# Index brands that are never on this site's floor; skipping them keeps
|
|
# substrings like "7090" from matching consumer models.
|
|
_SKIP_BRANDS = re.compile(r"inspiron|xps|vostro|alienware|g series|chromebook|wyse|edge|poweredge", re.I)
|
|
# Seen models that have no Dell packs by design.
|
|
_SKIP_MODELS = ("virtual", "vmware", "qemu", "hyper-v", "kvm")
|
|
|
|
|
|
# --- paths and small helpers --------------------------------------------------
|
|
|
|
def _dir(*parts):
|
|
return uc.state_dir("dell", *parts)
|
|
|
|
|
|
def _path(name):
|
|
return os.path.join(_dir(), name)
|
|
|
|
|
|
def _stamp():
|
|
return datetime.now().strftime("%Y%m%d-%H%M%S")
|
|
|
|
|
|
def _sha256(path):
|
|
return uc.file_hash(path, "sha256").upper()
|
|
|
|
|
|
def _alnum(s):
|
|
return re.sub(r"[^a-z0-9]", "", str(s or "").lower())
|
|
|
|
|
|
def _nice():
|
|
"""Prefix for CPU/disk heavy commands so imaging traffic keeps priority."""
|
|
cmd = []
|
|
if shutil.which("ionice"):
|
|
cmd += ["ionice", "-c", "3"]
|
|
if shutil.which("nice"):
|
|
cmd += ["nice", "-n", "15"]
|
|
return cmd
|
|
|
|
|
|
def _cab_root(cab):
|
|
"""Parse the XML inside a Dell cab (UTF-8 or UTF-16, 7z extracts both)."""
|
|
r = subprocess.run(["7z", "e", "-so", cab], capture_output=True)
|
|
if r.returncode or not r.stdout:
|
|
raise RuntimeError(f"cannot extract {os.path.basename(cab)}: "
|
|
f"{r.stderr.decode(errors='replace').strip()[:200]}")
|
|
return ET.fromstring(r.stdout)
|
|
|
|
|
|
def _hashes(elem):
|
|
"""{"SHA256": hex, "MD5": hex, ...} from an element's own Cryptography."""
|
|
out = {}
|
|
crypt = elem.find(NS + "Cryptography") if elem is not None else None
|
|
if crypt is not None:
|
|
for h in crypt.findall(NS + "Hash"):
|
|
if h.text:
|
|
out[(h.get("algorithm") or "").upper()] = h.text.strip()
|
|
return out
|
|
|
|
|
|
def _int(v):
|
|
try:
|
|
return int(v)
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
|
|
# check-bios.cmd expands the BIOSFile and Version fields unquoted in cmd
|
|
# (`set "BIOSFILE=%%B"`, `find /I "%TARGETVER%"`), so a name with & ^ % ! ( )
|
|
# or a space would break every WinPE boot. Every real Dell BIOS name fits.
|
|
_CMD_SAFE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]*")
|
|
|
|
|
|
def bios_name_ok(name, version):
|
|
return bool(_CMD_SAFE.fullmatch(str(name or "")) and _CMD_SAFE.fullmatch(str(version or "")))
|
|
|
|
|
|
def file_name_ok(name):
|
|
"""A plain file name from Dell's catalog: no path, no leading dot. Some
|
|
Win11 CAB packs have spaces ('7390 2in1-win11-A00-4PY3P.CAB'), which is
|
|
fine for a staging file run through argv lists."""
|
|
name = str(name or "")
|
|
return (bool(name) and name.isascii() and name.isprintable() and not name.startswith(".")
|
|
and "/" not in name and "\\" not in name)
|
|
|
|
|
|
# --- name and version rules -----------------------------------------------------
|
|
|
|
def norm(s):
|
|
"""Model name for matching: lower case, no 'Dell', AIO spelled one way."""
|
|
s = str(s or "").lower().replace("all-in-one", "aio").replace("all in one", "aio")
|
|
s = s.replace("optiplex", "optiplex ")
|
|
s = re.sub(r"\bdell\b", "", s)
|
|
s = re.sub(r"[^a-z0-9]+", " ", s)
|
|
return " ".join(s.split())
|
|
|
|
|
|
def aversion(name):
|
|
"""Dell A-number from a pack or zip name: ..._a08.zip -> 8, None if absent."""
|
|
m = re.search(r"(?i)(?:^|[_-])a(\d+)(?=[_.\-(]|$)", str(name or ""))
|
|
return int(m.group(1)) if m else None
|
|
|
|
|
|
def pack_code(file_name, release_id=""):
|
|
"""Model code of a Dell pack: PB13250-HT8VV_Win11_1.0_A08.exe -> pb13250,
|
|
OptiPlex-3000-JN664_Win11_... -> optiplex3000, and the older CAB packs
|
|
'7390 2in1-win11-A00-4PY3P.CAB' -> 73902in1."""
|
|
base = os.path.basename(file_name)
|
|
m = re.match(r"(?i)(.+?)[_-]win1[01]", base)
|
|
stem = m.group(1) if m else base.split("_")[0]
|
|
if release_id and stem.lower().endswith("-" + release_id.lower()):
|
|
stem = stem[: -len(release_id) - 1]
|
|
elif not release_id:
|
|
stem = re.sub(r"-[A-Za-z0-9]{5}$", "", stem)
|
|
return _alnum(stem)
|
|
|
|
|
|
def zip_name(file_name, release_id, dell_version):
|
|
"""win11_<code>_<ver>.zip, the name the driver zip is staged under."""
|
|
return f"win11_{pack_code(file_name, release_id)}_{_alnum(dell_version)}.zip"
|
|
|
|
|
|
def vtuple(v):
|
|
"""Comparable BIOS version: '1.45.0' -> (1, 45, 0), '02.12.00' -> (2, 12, 0),
|
|
'A22' -> (22,)."""
|
|
v = str(v or "").strip()
|
|
m = re.fullmatch(r"(?i)a(\d+)", v)
|
|
if m:
|
|
return (int(m.group(1)),)
|
|
return tuple(int(x) for x in re.findall(r"\d+", v))
|
|
|
|
|
|
def norm_version(v):
|
|
"""Version as models.txt writes it: dotted numbers without leading zeros."""
|
|
v = str(v or "").strip()
|
|
if re.fullmatch(r"[\d._]+", v):
|
|
return ".".join(str(int(x)) for x in re.split(r"[._]", v) if x)
|
|
return v
|
|
|
|
|
|
def bios_stem(file_name):
|
|
"""BIOS file family: the name with its version taken out, lower case.
|
|
Latitude_5430_7330_Rugged_1.45.0.exe -> latitude_5430_7330_rugged_#.exe"""
|
|
base = os.path.basename(str(file_name or "")).lower()
|
|
hits = list(re.finditer(r"\d+(?:[._]\d+){1,3}", base))
|
|
if hits:
|
|
m = hits[-1]
|
|
return base[: m.start()] + "#" + base[m.end():]
|
|
return re.sub(r"a\d+(?=\.exe$)", "#", base)
|
|
|
|
|
|
# --- reading the share ----------------------------------------------------------
|
|
|
|
def map_path(image):
|
|
return os.path.join(config.SAMBA_SHARE, image, "Deploy", "Control", "HardwareDriver.json")
|
|
|
|
|
|
def read_maps(errors=None):
|
|
"""{image: {"path", "raw", "entries"}} for every image map present.
|
|
A map that exists but cannot be read or parsed is left out and, when
|
|
errors is a list, reported there: apply must not quietly update two of
|
|
three images."""
|
|
out = {}
|
|
for image in IMAGES:
|
|
path = map_path(image)
|
|
if not os.path.lexists(path):
|
|
continue # image not on this server
|
|
try:
|
|
with open(path, "rb") as f:
|
|
raw = f.read()
|
|
entries = json.loads(raw.decode("utf-8-sig"))
|
|
except (OSError, ValueError) as ex:
|
|
if errors is not None:
|
|
errors.append(f"{image} HardwareDriver.json cannot be read ({str(ex)[:120]}); "
|
|
"that image is not updated")
|
|
continue
|
|
if isinstance(entries, list):
|
|
out[image] = {"path": path, "raw": raw, "entries": entries}
|
|
elif errors is not None:
|
|
errors.append(f"{image} HardwareDriver.json is not a list; that image is not updated")
|
|
return out
|
|
|
|
|
|
def is_win11(entry):
|
|
ids = set(str(entry.get("osId") or "").replace(" ", "").split(","))
|
|
return bool(ids & WIN11_OS_IDS)
|
|
|
|
|
|
def tokens(entry):
|
|
return [t.strip() for t in str(entry.get("modelswminame") or "").split(",") if t.strip()]
|
|
|
|
|
|
def entry_file(entry):
|
|
return str(entry.get("fileName") or entry.get("FileName") or "")
|
|
|
|
|
|
def entry_dir(entry):
|
|
"""destinationDir as a path relative to SHARED_DIR ("Out-of-box Drivers/...")."""
|
|
d = str(entry.get("destinationDir") or entry.get("DestinationDir") or "")
|
|
d = re.sub(r"(?i)^\*destinationdir\*\\?", "", d).replace("\\", "/").strip("/")
|
|
if d.lower().startswith("deploy/"):
|
|
d = d[len("deploy/"):]
|
|
return d
|
|
|
|
|
|
def models_txt_path():
|
|
return os.path.join(config.BIOS_DIR, "models.txt")
|
|
|
|
|
|
def read_models_txt():
|
|
"""[(index, raw line without CR/LF, substring, file, version)] for each
|
|
manifest line, plus the raw text."""
|
|
try:
|
|
with open(models_txt_path(), "rb") as f:
|
|
text = f.read().decode("ascii", errors="replace")
|
|
except OSError:
|
|
return [], ""
|
|
rows = []
|
|
for i, line in enumerate(text.split("\n")):
|
|
line = line.rstrip("\r")
|
|
if not line.strip() or line.lstrip().startswith("#") or line.count("|") < 2:
|
|
continue
|
|
sub, bfile, ver = (x.strip() for x in line.split("|")[:3])
|
|
if sub:
|
|
rows.append((i, line, sub, bfile, ver))
|
|
return rows, text
|
|
|
|
|
|
def resolve_dir_ci(root, relpath, create=False):
|
|
"""root/relpath matching each existing component case-insensitively (the
|
|
share has Dell_11/Optiplex and an OptiPlex symlink to it). With create,
|
|
missing components are made (owned like their parent). None when absent."""
|
|
cur = root
|
|
for part in PurePosixPath(relpath).parts:
|
|
try:
|
|
names = os.listdir(cur)
|
|
except OSError:
|
|
names = []
|
|
hit = next((n for n in names if n == part), None) or \
|
|
next((n for n in names if n.lower() == part.lower()), None)
|
|
if hit is None:
|
|
if not create:
|
|
return None
|
|
new = os.path.join(cur, part)
|
|
os.makedirs(new, exist_ok=True)
|
|
_chown_like(new, cur)
|
|
cur = new
|
|
else:
|
|
cur = os.path.join(cur, hit)
|
|
return cur if os.path.isdir(cur) else None
|
|
|
|
|
|
def _chown_like(path, ref):
|
|
try:
|
|
st = os.stat(ref)
|
|
os.chown(path, st.st_uid, st.st_gid)
|
|
except OSError:
|
|
pass # not root (local tests): leave as is
|
|
|
|
|
|
# --- Dell catalogs --------------------------------------------------------------
|
|
|
|
def _catalog_file(kind):
|
|
return os.path.join(_dir("catalog"), os.path.basename(CATALOGS[kind]))
|
|
|
|
|
|
def fetch_catalogs():
|
|
"""Download both catalog cabs into the cache (they carry no hash of their
|
|
own; the per-model cabs they list do)."""
|
|
for kind, rel in CATALOGS.items():
|
|
dest = _catalog_file(kind)
|
|
tmp = dest + ".part"
|
|
if os.path.exists(tmp):
|
|
os.remove(tmp)
|
|
uc.download(DELL + rel, tmp, timeout=300)
|
|
_cab_root(tmp) # refuse a cab that does not parse before replacing
|
|
os.replace(tmp, dest)
|
|
|
|
|
|
def driver_catalog(root):
|
|
"""Win11 driver packs: [{models, sids, ver, path, name, date, size, sha256,
|
|
md5, release_id}], plus the catalog's own date."""
|
|
packs = []
|
|
for p in root.findall(NS + "DriverPackage"):
|
|
oss = {o.get("osCode") for o in p.iter(NS + "OperatingSystem")}
|
|
if "Windows11" not in oss:
|
|
continue
|
|
h = _hashes(p)
|
|
path = p.get("path") or ""
|
|
packs.append({
|
|
"models": [m.get("name") or "" for m in p.iter(NS + "Model")],
|
|
"sids": sorted({(m.get("systemID") or "").upper() for m in p.iter(NS + "Model")} - {""}),
|
|
"ver": p.get("dellVersion") or "",
|
|
"path": path,
|
|
"name": path.rsplit("/", 1)[-1],
|
|
"date": (p.get("dateTime") or "")[:10],
|
|
"size": _int(p.get("size")),
|
|
"sha256": h.get("SHA256", ""),
|
|
"md5": h.get("MD5", "") or (p.get("hashMD5") or ""),
|
|
"release_id": p.get("releaseID") or "",
|
|
})
|
|
return packs
|
|
|
|
|
|
def catalog_index(root):
|
|
"""Per-model BIOS catalogs: [{brand, name, full, sid, path, sha256, size}]."""
|
|
idx = []
|
|
for g in root.findall(NS + "GroupManifest"):
|
|
b = g.find(f".//{NS}Brand/{NS}Display")
|
|
brand = (b.text or "").strip() if b is not None else ""
|
|
if _SKIP_BRANDS.search(brand):
|
|
continue
|
|
mi = g.find(NS + "ManifestInformation")
|
|
if mi is None or not mi.get("path"):
|
|
continue
|
|
h = _hashes(mi)
|
|
for m in g.iter(NS + "Model"):
|
|
d = m.find(NS + "Display")
|
|
name = (d.text or "").strip() if d is not None else ""
|
|
idx.append({"brand": brand, "name": name, "full": norm(brand + " " + name),
|
|
"sid": (m.get("systemID") or "").upper(), "path": mi.get("path"),
|
|
"sha256": h.get("SHA256", ""), "size": _int(mi.get("size"))})
|
|
return idx
|
|
|
|
|
|
class ModelCabs:
|
|
"""Latest BIOS per per-model cab, downloaded once and hash checked."""
|
|
|
|
def __init__(self, fetch=True):
|
|
self.fetch = fetch
|
|
self.cache = {}
|
|
self.errors = {}
|
|
|
|
def latest_bios(self, ix):
|
|
path = ix["path"]
|
|
if path in self.cache:
|
|
return self.cache[path]
|
|
best = None
|
|
try:
|
|
local = os.path.join(_dir("catalog", "models"), path.replace("/", "_"))
|
|
fresh = os.path.exists(local) and (
|
|
not ix["sha256"] or uc.file_hash(local).lower() == ix["sha256"].lower())
|
|
if not fresh:
|
|
if not self.fetch:
|
|
raise RuntimeError("per-model catalog not cached")
|
|
if os.path.exists(local):
|
|
os.remove(local)
|
|
uc.download(DELL + path, local, expect=ix["sha256"] or None, timeout=120)
|
|
root = _cab_root(local)
|
|
for c in root.iter(NS + "SoftwareComponent"):
|
|
ct = c.find(NS + "ComponentType")
|
|
if ct is None or ct.get("value") != "BIOS":
|
|
continue
|
|
cpath = c.get("path") or ""
|
|
version = c.get("vendorVersion") or c.get("dellVersion") or ""
|
|
h = _hashes(c)
|
|
cand = {
|
|
"file": cpath.rsplit("/", 1)[-1], "path": cpath, "url": DELL + cpath,
|
|
"version": norm_version(version), "date": c.get("releaseDate") or "",
|
|
"size": _int(c.get("size")), "sha256": h.get("SHA256", "").upper(),
|
|
"md5": h.get("MD5", "").upper(),
|
|
}
|
|
if best is None or vtuple(cand["version"]) > vtuple(best["version"]):
|
|
best = cand
|
|
except Exception as ex: # one bad cab must not sink the whole scan
|
|
self.errors[path] = str(ex)[:200]
|
|
self.cache[path] = best
|
|
return best
|
|
|
|
|
|
# --- matching -------------------------------------------------------------------
|
|
|
|
def _rank(t_words, c_words):
|
|
"""How well a catalog model name fits a modelswminame token, or None.
|
|
The model numbers (words with a digit) of the token must all be in the
|
|
catalog name; then an exact word set beats one name containing the
|
|
other ('OptiPlex Tower Plus 7020' vs 'OptiPlex 7020 Tower'), which beats
|
|
a bare number match. Fewer extra words is better."""
|
|
t, c = set(t_words), set(c_words)
|
|
digits = {w for w in t if any(ch.isdigit() for ch in w)}
|
|
if not digits or not digits <= c:
|
|
return None
|
|
if t == c:
|
|
return (3, 0)
|
|
if t <= c or c <= t:
|
|
return (2, -len(t ^ c))
|
|
if (t - digits) & c:
|
|
return (1, -len(t ^ c))
|
|
return None
|
|
|
|
|
|
def _pack_hits(token, packs):
|
|
"""Win11 packs whose model names best fit one modelswminame token."""
|
|
nn = norm(token).split()
|
|
if not nn:
|
|
return []
|
|
scored = []
|
|
for c in packs:
|
|
ranks = [r for r in (_rank(nn, norm(x).split()) for x in c["models"]) if r]
|
|
if ranks:
|
|
scored.append((max(ranks), c))
|
|
if not scored:
|
|
return []
|
|
top = max(r for r, _ in scored)
|
|
return [c for r, c in scored if r == top]
|
|
|
|
|
|
def _best(packs):
|
|
return max(packs, key=lambda c: (aversion("_" + c["ver"]) or 0, c["date"])) if packs else None
|
|
|
|
|
|
def _pack_view(c):
|
|
return {
|
|
"key": pack_code(c["name"], c["release_id"]) + "-" + _alnum(c["ver"]),
|
|
"path": c["path"], "url": DELL + c["path"], "name": c["name"], "ver": c["ver"],
|
|
"date": c["date"], "size": c["size"], "sha256": c["sha256"].upper(),
|
|
"md5": c["md5"].upper(), "release_id": c["release_id"],
|
|
"zip": zip_name(c["name"], c["release_id"], c["ver"]),
|
|
"code": pack_code(c["name"], c["release_id"]),
|
|
"models": sorted(set(c["models"])),
|
|
}
|
|
|
|
|
|
def scan_drivers(maps, packs):
|
|
"""One row per distinct current file in the Win11 entries of all maps."""
|
|
groups = {}
|
|
for image, m in maps.items():
|
|
for e in m["entries"]:
|
|
if not isinstance(e, dict) or not is_win11(e):
|
|
continue
|
|
fn = entry_file(e)
|
|
if not fn:
|
|
continue
|
|
g = groups.setdefault(fn, {"tokens": [], "images": [], "dirs": []})
|
|
for t in tokens(e):
|
|
if t not in g["tokens"]:
|
|
g["tokens"].append(t)
|
|
if image not in g["images"]:
|
|
g["images"].append(image)
|
|
d = entry_dir(e)
|
|
if d not in g["dirs"]:
|
|
g["dirs"].append(d)
|
|
rows = []
|
|
for fn, g in groups.items():
|
|
per_token, codes = {}, set()
|
|
for t in g["tokens"]:
|
|
hits = _pack_hits(t, packs)
|
|
codes |= {pack_code(c["name"], c["release_id"]) for c in hits}
|
|
b = _best(hits)
|
|
if b:
|
|
per_token[t] = b
|
|
best = _best(list(per_token.values()))
|
|
ours = aversion(fn)
|
|
warnings, notes = [], []
|
|
if not best:
|
|
status = "no-match"
|
|
elif ours is None:
|
|
status = "unknown-version"
|
|
elif (aversion("_" + best["ver"]) or 0) > ours:
|
|
status = "update"
|
|
else:
|
|
status = "current"
|
|
pack = _pack_view(best) if best else None
|
|
if pack:
|
|
if len(codes) > 1:
|
|
warnings.append(
|
|
f"The model names in this entry match {len(codes)} different Dell packs "
|
|
f"({', '.join(sorted(codes))}); one zip cannot cover them all, so check "
|
|
"which one fits or split the entry by hand.")
|
|
if fn.lower().startswith("win10") or "_win10_" in fn.lower():
|
|
warnings.append("The current file is a Win10 pack, so its A-number is not comparable.")
|
|
elif pack["code"] and pack["code"] not in _alnum(fn):
|
|
warnings.append(
|
|
f"The current file looks like a GE/MCL family pack (its name lacks "
|
|
f"'{pack['code']}'), so its A-number may not be comparable. Decide by hand.")
|
|
if "nvidia" in fn.lower():
|
|
notes.append("The current pack carries NVIDIA drivers; the update is skipped "
|
|
"if the Dell pack has no NVIDIA INFs.")
|
|
rows.append({
|
|
"id": "d-" + hashlib.sha1(fn.encode("utf-8")).hexdigest()[:10],
|
|
"current": fn, "cur_ver": f"A{ours:02d}" if ours is not None else "",
|
|
"tokens": g["tokens"], "images": g["images"], "dirs": g["dirs"],
|
|
"status": status, "pack": pack, "warnings": warnings, "notes": notes,
|
|
"default": status == "update" and not warnings,
|
|
})
|
|
order = {"update": 0, "unknown-version": 1, "no-match": 2, "current": 3}
|
|
rows.sort(key=lambda r: (order.get(r["status"], 9), ", ".join(r["tokens"]).lower()))
|
|
return rows
|
|
|
|
|
|
def _variants(ix):
|
|
"""The model names one index entry stands for, lower case, as check-bios
|
|
would see them: 'Dell Pro 7 13 P713260/Dell Pro 7 13 2-in-1 P703260' is
|
|
two machines sharing one per-model catalog."""
|
|
out = []
|
|
for v in ix["name"].split("/"):
|
|
v = " ".join(v.split()).lower()
|
|
if v:
|
|
out.append(v)
|
|
out.append((ix["brand"].lower() + " " + v).strip())
|
|
return out
|
|
|
|
|
|
def _line_owner(lines, texts):
|
|
"""Index into lines of the first manifest line check-bios.cmd would pick
|
|
for a model called any of texts (literal substring, case-insensitive)."""
|
|
for n, (_, _, sub, _, _) in enumerate(lines):
|
|
s = sub.lower()
|
|
if any(s in t for t in texts):
|
|
return n
|
|
return None
|
|
|
|
|
|
def _merge_by_family(cands):
|
|
"""One candidate per file family (stem), the highest version, models merged."""
|
|
fam = {}
|
|
for c in cands:
|
|
k = bios_stem(c["file"])
|
|
cur = fam.get(k)
|
|
if cur is None or vtuple(c["version"]) > vtuple(cur["version"]):
|
|
merged = dict(c, models=sorted(set(c["models"]) | set(cur["models"] if cur else [])))
|
|
fam[k] = merged
|
|
else:
|
|
cur["models"] = sorted(set(cur["models"]) | set(c["models"]))
|
|
return list(fam.values())
|
|
|
|
|
|
def scan_bios(lines, index, cabs):
|
|
"""One row per models.txt line. Each Dell model is credited to the first
|
|
line that would catch it, as check-bios.cmd does, so '7090' does not
|
|
report the models '7090 Ultra' (an earlier line) already owns. A line no
|
|
Dell name contains literally (Dell lists 'OptiPlex Micro 7010' as just
|
|
'7010') falls back to a model-number match that prefers our own file
|
|
family."""
|
|
owned = {}
|
|
for ix in index:
|
|
for v in ix["name"].split("/") or [ix["name"]]:
|
|
n = _line_owner(lines, [t for t in _variants(dict(ix, name=v))])
|
|
if n is not None and ix not in owned.setdefault(n, []):
|
|
owned[n].append(ix)
|
|
rows = []
|
|
for n, (i, raw, sub, bfile, ver) in enumerate(lines):
|
|
loose = n not in owned
|
|
models = owned.get(n, [])
|
|
if loose:
|
|
digits = {w for w in norm(sub).split() if any(ch.isdigit() for ch in w)}
|
|
models = [ix for ix in index if digits and digits <= set(norm(ix["brand"] + " " + ix["name"]).split())]
|
|
cands = []
|
|
for ix in models:
|
|
b = cabs.latest_bios(ix)
|
|
if b:
|
|
cands.append(dict(b, models=[(ix["brand"] + " " + ix["name"]).strip()]))
|
|
ours_stem = bios_stem(bfile)
|
|
fam = _merge_by_family(cands)
|
|
warnings, notes = [], []
|
|
if loose and fam:
|
|
mine = [c for c in fam if bios_stem(c["file"]) == ours_stem]
|
|
notes.append("No Dell model name contains this line; matched by model number only.")
|
|
if mine:
|
|
fam = mine
|
|
fam.sort(key=lambda c: (bios_stem(c["file"]) != ours_stem, tuple(-x for x in vtuple(c["version"]))))
|
|
if not fam:
|
|
errs = [cabs.errors[ix["path"]] for ix in models if ix["path"] in cabs.errors]
|
|
status = "error" if errs else "no-match"
|
|
if errs:
|
|
warnings.append("Per-model catalog failed: " + errs[0])
|
|
elif len(fam) > 1:
|
|
status = "ambiguous"
|
|
warnings.append(f"{len(fam)} different BIOS file families match this line; "
|
|
"check-bios.cmd would flash the same file on all of them. Pick one.")
|
|
else:
|
|
c = fam[0]
|
|
if bios_stem(c["file"]) != ours_stem:
|
|
status = "different-file-family"
|
|
warnings.append("Dell's latest BIOS for these models is a different file family.")
|
|
elif vtuple(c["version"]) > vtuple(ver):
|
|
status = "update"
|
|
else:
|
|
status = "current"
|
|
rows.append({
|
|
"id": f"b-{i}", "line_no": i, "line": raw, "sub": sub, "current": bfile,
|
|
"cur_ver": ver, "status": status, "candidates": fam, "pick": 0,
|
|
"warnings": warnings, "notes": notes, "default": status == "update",
|
|
})
|
|
order = {"update": 0, "different-file-family": 1, "ambiguous": 2, "error": 3,
|
|
"no-match": 4, "current": 5}
|
|
rows.sort(key=lambda r: (order.get(r["status"], 9), r["line_no"]))
|
|
return rows
|
|
|
|
|
|
def _entry_matches(entry, model):
|
|
"""PESetup's matcher (see services/hardware_coverage.py)."""
|
|
try:
|
|
from services import hardware_coverage
|
|
return hardware_coverage._entry_matches(entry, model)
|
|
except (ImportError, AttributeError):
|
|
return any(t.lower() in model.lower()
|
|
for t in str(entry.get("modelswminame") or "").split(","))
|
|
|
|
|
|
def scan_missing(maps, lines, packs, index, cabs):
|
|
"""Models WinPE has seen with no Win11 driver entry or no BIOS line, and
|
|
the Dell pack/BIOS that would cover them. Informational only."""
|
|
state = uc.read_json(config.HARDWARE_STATE, {}) or {}
|
|
out = []
|
|
for model, rec in sorted((state.get("models") or {}).items()):
|
|
if any(s in model.lower() for s in _SKIP_MODELS):
|
|
continue
|
|
no_drv = [img for img, m in maps.items()
|
|
if not any(isinstance(e, dict) and is_win11(e) and _entry_matches(e, model)
|
|
for e in m["entries"])]
|
|
no_bios = not any(sub.lower() in model.lower() for _, _, sub, _, _ in lines)
|
|
if not no_drv and not no_bios:
|
|
continue
|
|
skus = [s.upper() for s in rec.get("skus") or [] if s]
|
|
pack = bios = None
|
|
if no_drv:
|
|
hits = [c for c in packs if set(c["sids"]) & set(skus)] or _pack_hits(model, packs)
|
|
b = _best(hits)
|
|
pack = _pack_view(b) if b else None
|
|
if no_bios:
|
|
want = norm(model)
|
|
ixs = [ix for ix in index if ix["sid"] in skus] or \
|
|
[ix for ix in index if want in {norm(v) for v in _variants(ix)}]
|
|
for ix in ixs:
|
|
bios = cabs.latest_bios(ix)
|
|
if bios:
|
|
break
|
|
out.append({"model": model, "skus": skus, "no_drivers": no_drv,
|
|
"no_bios": no_bios, "pack": pack, "bios": bios,
|
|
"last_seen": rec.get("last_seen", "")})
|
|
return out
|
|
|
|
|
|
# --- 1. check -------------------------------------------------------------------
|
|
|
|
def check_state():
|
|
return uc.read_json(_path("check.json"), {}) or {}
|
|
|
|
|
|
def _check_progress(**kw):
|
|
state = check_state()
|
|
state.update(kw)
|
|
uc.write_json(_path("check.json"), state)
|
|
|
|
|
|
def last_scan():
|
|
return uc.read_json(_path("scan.json"), None)
|
|
|
|
|
|
def check(fetch=True):
|
|
"""Fetch the catalogs (unless fetch is False: cached cabs only), compare
|
|
everything, write scan.json and return it."""
|
|
uc.write_json(_path("check.json"), {"state": "running", "started": uc.now(),
|
|
"step": "Fetching Dell catalogs"})
|
|
try:
|
|
if fetch:
|
|
fetch_catalogs()
|
|
_check_progress(step="Reading catalogs")
|
|
droot = _cab_root(_catalog_file("drivers"))
|
|
iroot = _cab_root(_catalog_file("index"))
|
|
packs = driver_catalog(droot)
|
|
index = catalog_index(iroot)
|
|
map_errors = []
|
|
maps = read_maps(map_errors)
|
|
lines, _ = read_models_txt()
|
|
_check_progress(step="Comparing driver packs")
|
|
drivers = scan_drivers(maps, packs)
|
|
cabs = ModelCabs(fetch=fetch)
|
|
_check_progress(step="Reading per-model BIOS catalogs")
|
|
bios = scan_bios(lines, index, cabs)
|
|
missing = scan_missing(maps, lines, packs, index, cabs)
|
|
result = {
|
|
"checked": uc.now(),
|
|
"catalogs": {"drivers": droot.get("dateTime") or droot.get("version") or "",
|
|
"index": iroot.get("creationDateTime") or iroot.get("version") or ""},
|
|
"images": sorted(maps),
|
|
"drivers": drivers, "bios": bios, "missing": missing,
|
|
"cab_errors": cabs.errors, "map_errors": map_errors,
|
|
}
|
|
uc.write_json(_path("scan.json"), result)
|
|
_check_progress(state="done", finished=uc.now(), step="")
|
|
return result
|
|
except Exception as ex:
|
|
_check_progress(state="error", finished=uc.now(), error=str(ex)[:400])
|
|
raise
|
|
|
|
|
|
# --- 2. download ----------------------------------------------------------------
|
|
|
|
def staged_state():
|
|
return uc.read_json(_path("staged.json"), {}) or {}
|
|
|
|
|
|
def _save_staged(state):
|
|
uc.write_json(_path("staged.json"), state)
|
|
|
|
|
|
def _set_item(key, **kw):
|
|
state = staged_state()
|
|
state.setdefault("items", {}).setdefault(key, {}).update(kw)
|
|
_save_staged(state)
|
|
|
|
|
|
def select(driver_ids, bios_picks):
|
|
"""Turn checked rows of the last scan into a download list.
|
|
driver_ids: scan row ids; bios_picks: {row id: candidate index}.
|
|
Rows sharing one Dell file become one item. Returns the item count."""
|
|
scan = last_scan()
|
|
if not scan:
|
|
raise ValueError("no scan yet: run a check first")
|
|
items = {}
|
|
for row in scan["drivers"]:
|
|
if row["id"] not in driver_ids or not row.get("pack"):
|
|
continue
|
|
p = row["pack"]
|
|
if not file_name_ok(p["name"]) or not _CMD_SAFE.fullmatch(p["zip"]):
|
|
raise ValueError(f"refusing Dell pack with an unsafe file name: {p['name']!r}")
|
|
it = items.setdefault(p["key"], {
|
|
"key": p["key"], "kind": "driver", "url": p["url"], "path": p["path"],
|
|
"name": p["name"], "size": p["size"], "sha256": p["sha256"], "md5": p["md5"],
|
|
"ver": p["ver"], "date": p["date"], "zip": p["zip"], "replaces": []})
|
|
if row["current"] not in it["replaces"]:
|
|
it["replaces"].append(row["current"])
|
|
for row in scan["bios"]:
|
|
if row["id"] not in bios_picks or not row["candidates"]:
|
|
continue
|
|
n = bios_picks[row["id"]]
|
|
if not 0 <= n < len(row["candidates"]):
|
|
n = 0
|
|
c = row["candidates"][n]
|
|
if not bios_name_ok(c["file"], c["version"]):
|
|
raise ValueError(f"refusing BIOS {c['file']!r} version {c['version']!r}: "
|
|
"not safe in models.txt / check-bios.cmd")
|
|
key = "bios-" + _alnum(c["file"])
|
|
it = items.setdefault(key, {
|
|
"key": key, "kind": "bios", "url": c["url"], "path": c["path"],
|
|
"name": c["file"], "size": c["size"], "sha256": c["sha256"], "md5": c["md5"],
|
|
"ver": c["version"], "date": c["date"], "lines": []})
|
|
it["lines"].append({"line_no": row["line_no"], "line": row["line"],
|
|
"sub": row["sub"], "current": row["current"]})
|
|
if not items:
|
|
raise ValueError("nothing selected")
|
|
uc.write_json(_path("selection.json"), {"created": uc.now(), "items": list(items.values())})
|
|
return len(items)
|
|
|
|
|
|
def _free_bytes(path):
|
|
try:
|
|
return shutil.disk_usage(path).free
|
|
except OSError:
|
|
return 0
|
|
|
|
|
|
def _is_nvidia_inf(path, rel):
|
|
"""An NVIDIA INF: under a folder named NVIDIA..., an nv* INF whose text
|
|
names NVIDIA, or any INF whose Provider line does (INFs are ANSI or
|
|
UTF-16). A bare 'nv' prefix is not enough: nvme*.inf storage drivers
|
|
would make a pack without NVIDIA look like it has some, and the NVIDIA
|
|
rule would then replace a GE pack that carries them."""
|
|
if any(part.lower().startswith("nvidia") for part in rel.split(os.sep)[:-1]):
|
|
return True
|
|
try:
|
|
with open(path, "rb") as f:
|
|
data = f.read(1 << 20)
|
|
except OSError:
|
|
return False
|
|
utf16 = data[:2] in (b"\xff\xfe", b"\xfe\xff") or data[1:200:2].count(0) > 50
|
|
text = data.decode("utf-16" if utf16 else "latin-1", errors="replace").lower()
|
|
if os.path.basename(rel).lower().startswith("nv") and "nvidia" in text:
|
|
return True
|
|
return bool(re.search(r"^\s*provider\s*=[^\n]*nvidia", text, re.M))
|
|
|
|
|
|
def _count_infs(folder):
|
|
infs = [p for p in glob.glob(os.path.join(folder, "**", "*"), recursive=True)
|
|
if p.lower().endswith(".inf")]
|
|
nv = [p for p in infs if _is_nvidia_inf(p, os.path.relpath(p, folder))]
|
|
return len(infs), len(nv)
|
|
|
|
|
|
def _stage_one(it):
|
|
key = it["key"]
|
|
if not file_name_ok(it["name"]) or not _CMD_SAFE.fullmatch(key):
|
|
raise RuntimeError(f"unsafe file name {it['name']!r}; refusing it")
|
|
folder = _dir("staged", key)
|
|
exe = os.path.join(folder, it["name"])
|
|
algo, expect = ("sha256", it["sha256"]) if it.get("sha256") else ("md5", it.get("md5"))
|
|
if not expect:
|
|
raise RuntimeError("Dell lists no hash for this file; refusing it")
|
|
need = it["size"] * (5 if it["kind"] == "driver" else 1) + (1 << 30)
|
|
have = os.path.getsize(exe) if os.path.exists(exe) else 0
|
|
if _free_bytes(folder) + have < need:
|
|
raise RuntimeError(f"not enough disk space in {folder} (need about {need // (1 << 30)} GB)")
|
|
if not (have == it["size"] and uc.file_hash(exe, algo).lower() == expect.lower()):
|
|
_set_item(key, status="downloading", partial=exe, error="")
|
|
uc.download(it["url"], exe, expect=expect, algorithm=algo)
|
|
_set_item(key, status="verified", partial="")
|
|
if it["kind"] == "bios":
|
|
_set_item(key, status="ready", file=exe, sha256_file=_sha256(exe),
|
|
size_file=os.path.getsize(exe), finished=uc.now())
|
|
return
|
|
# Driver pack: 7z-extract the exe and deflate-zip its top folder(s).
|
|
work = os.path.join(folder, "x")
|
|
shutil.rmtree(work, ignore_errors=True)
|
|
os.makedirs(work)
|
|
try:
|
|
_set_item(key, status="extracting")
|
|
r = subprocess.run(_nice() + ["7z", "x", "-y", "-o" + work, exe],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True)
|
|
if r.returncode:
|
|
raise RuntimeError(f"7z failed ({r.returncode}): {r.stderr.strip()[:200]}")
|
|
tops = sorted(os.listdir(work))
|
|
infs, nvidia = _count_infs(work)
|
|
if not tops or not infs:
|
|
raise RuntimeError("the pack extracted to no INF files")
|
|
_set_item(key, status="zipping", tops=tops, infs=infs, nvidia_infs=nvidia)
|
|
final = os.path.join(folder, it["zip"])
|
|
tmp = os.path.join(folder, ".tmp-" + it["zip"])
|
|
if os.path.exists(tmp):
|
|
os.remove(tmp)
|
|
r = subprocess.run(_nice() + ["zip", "-r", "-q", "-X", tmp] + tops, cwd=work,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True)
|
|
if r.returncode:
|
|
raise RuntimeError(f"zip failed ({r.returncode}): {r.stderr.strip()[:200]}")
|
|
os.replace(tmp, final)
|
|
finally:
|
|
shutil.rmtree(work, ignore_errors=True)
|
|
_set_item(key, status="ready", file=final, sha256_file=_sha256(final),
|
|
size_file=os.path.getsize(final), finished=uc.now())
|
|
os.remove(exe) # the zip is what gets deployed; the exe can be fetched again
|
|
|
|
|
|
def download():
|
|
"""Background job: stage every item of selection.json. Resumable: ready
|
|
items are skipped, a partial exe is resumed."""
|
|
sel = uc.read_json(_path("selection.json"), {}) or {}
|
|
state = staged_state()
|
|
state.update(state="running", started=uc.now(), finished="", error="")
|
|
items = state.setdefault("items", {})
|
|
for it in sel.get("items", []):
|
|
cur = items.get(it["key"], {})
|
|
if cur.get("status") == "ready" and os.path.exists(cur.get("file", "")):
|
|
# Already staged: just widen what it replaces.
|
|
for k in ("replaces", "lines"):
|
|
if k in it:
|
|
merged = cur.get(k, [])
|
|
for v in it[k]:
|
|
if v not in merged:
|
|
merged.append(v)
|
|
cur[k] = merged
|
|
continue
|
|
items[it["key"]] = dict(it, status="queued", error="", queued=uc.now())
|
|
_save_staged(state)
|
|
failed = 0
|
|
for it in sel.get("items", []):
|
|
if staged_state()["items"][it["key"]].get("status") == "ready":
|
|
continue
|
|
try:
|
|
_stage_one(it)
|
|
except Exception as ex:
|
|
failed += 1
|
|
_set_item(it["key"], status="error", error=str(ex)[:300], partial="")
|
|
state = staged_state()
|
|
state.update(state="done", finished=uc.now(),
|
|
error=f"{failed} item(s) failed" if failed else "")
|
|
_save_staged(state)
|
|
_audit("DELL_DOWNLOAD", f"{len(sel.get('items', []))} item(s), {failed} failed")
|
|
|
|
|
|
def progress():
|
|
"""staged.json with live percent for items still downloading, and whether
|
|
the jobs are running."""
|
|
state = staged_state()
|
|
running = uc.job_running(UNITS["download"])
|
|
if state.get("state") == "running" and not running:
|
|
state["state"] = "interrupted"
|
|
for it in (state.get("items") or {}).values():
|
|
pct = None
|
|
if it.get("status") == "downloading" and it.get("partial") and it.get("size"):
|
|
try:
|
|
pct = min(100.0, round(100.0 * os.path.getsize(it["partial"]) / it["size"], 1))
|
|
except OSError:
|
|
pct = 0.0
|
|
elif it.get("status") in ("verified", "extracting", "zipping", "ready", "applied"):
|
|
pct = 100.0
|
|
it["percent"] = pct
|
|
if not running and it.get("status") in ("queued", "downloading", "verified",
|
|
"extracting", "zipping"):
|
|
it["status"] = "interrupted"
|
|
chk = check_state()
|
|
if chk.get("state") == "running" and not uc.job_running(UNITS["check"]):
|
|
chk["state"] = "interrupted"
|
|
return {"download": state, "check": chk, "download_running": running,
|
|
"check_running": uc.job_running(UNITS["check"])}
|
|
|
|
|
|
def cancel_download():
|
|
uc.stop_job(UNITS["download"])
|
|
state = staged_state()
|
|
for it in (state.get("items") or {}).values():
|
|
if it.get("status") in ("queued", "downloading", "verified", "extracting", "zipping"):
|
|
it["status"] = "cancelled"
|
|
if state.get("state") == "running":
|
|
state["state"] = "cancelled"
|
|
state["finished"] = uc.now()
|
|
_save_staged(state)
|
|
|
|
|
|
def discard(keys):
|
|
"""Drop staged items and their files (staging only; never the share)."""
|
|
state = staged_state()
|
|
items = state.get("items") or {}
|
|
gone = []
|
|
for key in keys:
|
|
if key not in items or "/" in key or key.startswith("."):
|
|
continue
|
|
shutil.rmtree(os.path.join(_dir("staged"), key), ignore_errors=True)
|
|
items.pop(key)
|
|
gone.append(key)
|
|
_save_staged(state)
|
|
return gone
|
|
|
|
|
|
# --- 3. apply -------------------------------------------------------------------
|
|
|
|
def _ready(keys):
|
|
items = (staged_state().get("items") or {})
|
|
return [items[k] for k in keys if k in items and items[k].get("status") == "ready"
|
|
and os.path.exists(items[k].get("file", ""))]
|
|
|
|
|
|
def _new_line(raw, bfile, ver):
|
|
parts = raw.split("|")
|
|
parts[1], parts[2] = bfile, ver
|
|
return "|".join(parts)
|
|
|
|
|
|
def preview(keys):
|
|
"""What apply(keys) would change, without changing anything."""
|
|
map_errors = []
|
|
maps = read_maps(map_errors)
|
|
lines_by_no = {}
|
|
_, text = read_models_txt()
|
|
for n, line in enumerate(text.split("\n")):
|
|
lines_by_no[n] = line.rstrip("\r")
|
|
out = {"drivers": {img: [] for img in maps}, "bios": [], "skipped": [], "missing": [],
|
|
"errors": map_errors}
|
|
items = _ready(keys)
|
|
out["missing"] = [k for k in keys if k not in {i["key"] for i in items}]
|
|
for it in items:
|
|
if it["kind"] == "driver":
|
|
for cur in it.get("replaces", []):
|
|
if "nvidia" in cur.lower() and not it.get("nvidia_infs"):
|
|
out["skipped"].append(f"{cur}: has NVIDIA drivers, Dell pack {it['zip']} has none")
|
|
continue
|
|
hit = False
|
|
for img, m in maps.items():
|
|
for e in m["entries"]:
|
|
if isinstance(e, dict) and entry_file(e) == cur:
|
|
hit = True
|
|
out["drivers"][img].append({
|
|
"models": e.get("modelswminame", ""), "old": cur,
|
|
"new": it["zip"], "dir": entry_dir(e)})
|
|
if not hit:
|
|
out["skipped"].append(f"{cur}: no map entry uses it any more")
|
|
else:
|
|
if not bios_name_ok(it["name"], it["ver"]):
|
|
out["errors"].append(f"{it['name']}: file name or version not safe for check-bios.cmd")
|
|
continue
|
|
for ln in it.get("lines", []):
|
|
n = _find_line(lines_by_no, ln)
|
|
if n is None:
|
|
out["skipped"].append(f"models.txt line '{ln['line']}' changed since the check")
|
|
continue
|
|
out["bios"].append({"old": lines_by_no[n],
|
|
"new": _new_line(lines_by_no[n], it["name"], it["ver"])})
|
|
return out
|
|
|
|
|
|
def _find_line(lines_by_no, ln):
|
|
"""The line's current index: where the scan saw it, or its unique copy."""
|
|
if lines_by_no.get(ln["line_no"]) == ln["line"]:
|
|
return ln["line_no"]
|
|
hits = [n for n, t in lines_by_no.items() if t == ln["line"]]
|
|
return hits[0] if len(hits) == 1 else None
|
|
|
|
|
|
def _same_file(path, size, sha256):
|
|
return os.path.getsize(path) == size and _sha256(path) == sha256.upper()
|
|
|
|
|
|
def _place(src, dest_dir, name, size, sha256, move):
|
|
"""Put src into dest_dir/name. Rename when move and on one filesystem,
|
|
else copy and verify. An existing file is accepted only if identical.
|
|
Returns the placed path."""
|
|
dst = os.path.join(dest_dir, name)
|
|
if os.path.exists(dst):
|
|
if _same_file(dst, size, sha256):
|
|
return dst
|
|
raise RuntimeError(f"{dst} already exists with different content; not overwritten")
|
|
if move:
|
|
try:
|
|
os.rename(src, dst)
|
|
_chown_like(dst, dest_dir)
|
|
return dst
|
|
except OSError as ex:
|
|
if ex.errno != 18: # EXDEV: other filesystem, copy instead
|
|
raise
|
|
fd, tmp = tempfile.mkstemp(dir=dest_dir, prefix=".tmp-")
|
|
os.close(fd)
|
|
try:
|
|
shutil.copyfile(src, tmp)
|
|
if not _same_file(tmp, size, sha256):
|
|
raise RuntimeError(f"copy of {name} to {dest_dir} failed verification")
|
|
os.chmod(tmp, 0o644)
|
|
_chown_like(tmp, dest_dir)
|
|
os.replace(tmp, dst)
|
|
finally:
|
|
if os.path.exists(tmp):
|
|
os.remove(tmp)
|
|
return dst
|
|
|
|
|
|
def _write_like(path, data, ts, tag="pre-webupdate"):
|
|
"""Back up path to path.bak-<ts>-<tag>, then atomically replace it with
|
|
data (bytes), keeping owner and mode. Returns the backup path."""
|
|
real = os.path.realpath(path)
|
|
backup = f"{path}.bak-{ts}-{tag}"
|
|
n = 1
|
|
while os.path.lexists(backup): # never overwrite an earlier backup
|
|
n += 1
|
|
backup = f"{path}.bak-{ts}-{tag}-{n}"
|
|
shutil.copy2(real, backup)
|
|
st = os.stat(real)
|
|
folder = os.path.dirname(real)
|
|
fd, tmp = tempfile.mkstemp(dir=folder, prefix=".tmp-")
|
|
try:
|
|
with os.fdopen(fd, "wb") as f:
|
|
f.write(data)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
os.chmod(tmp, st.st_mode & 0o7777)
|
|
try:
|
|
os.chown(tmp, st.st_uid, st.st_gid)
|
|
os.chown(backup, st.st_uid, st.st_gid)
|
|
except OSError:
|
|
pass
|
|
os.replace(tmp, real)
|
|
finally:
|
|
if os.path.exists(tmp):
|
|
os.remove(tmp)
|
|
return backup
|
|
|
|
|
|
def dump_map(entries, raw):
|
|
"""Serialise a map the way the originals are written: indent 4, UTF-8,
|
|
a trailing newline only if the original had one, BOM only if it had one."""
|
|
text = json.dumps(entries, indent=4, ensure_ascii=False)
|
|
if raw.endswith(b"\n"):
|
|
text += "\n"
|
|
data = text.encode("utf-8")
|
|
return (b"\xef\xbb\xbf" + data) if raw.startswith(b"\xef\xbb\xbf") else data
|
|
|
|
|
|
def history():
|
|
return uc.read_json(_path("applied.json"), []) or []
|
|
|
|
|
|
class _Busy:
|
|
"""Non-blocking exclusive lock so two applies/rollbacks (a double click,
|
|
two tabs) never interleave their reads and rewrites of the same maps."""
|
|
|
|
def __enter__(self):
|
|
self.fd = os.open(_path(".apply.lock"), os.O_CREAT | os.O_RDWR, 0o644)
|
|
try:
|
|
fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except OSError:
|
|
os.close(self.fd)
|
|
raise RuntimeError("another apply or rollback is already running; try again "
|
|
"when it has finished") from None
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
fcntl.flock(self.fd, fcntl.LOCK_UN)
|
|
os.close(self.fd)
|
|
return False
|
|
|
|
|
|
def apply(keys):
|
|
"""Install the staged items named by keys. Returns a report dict."""
|
|
with _Busy():
|
|
return _apply(keys)
|
|
|
|
|
|
def _apply(keys):
|
|
ts = _stamp()
|
|
report = {"id": ts, "at": uc.now(), "drivers": [], "bios": [], "skipped": [],
|
|
"errors": [], "backups": {}, "after": {}}
|
|
maps = read_maps(report["errors"])
|
|
changed_maps = set()
|
|
items = _ready(list(dict.fromkeys(keys)))
|
|
staged_items = {} # only the items this apply changed are written back
|
|
|
|
# Driver zips first: a map only changes once its zip is in place.
|
|
for it in [i for i in items if i["kind"] == "driver"]:
|
|
todo = []
|
|
for cur in it.get("replaces", []):
|
|
if "nvidia" in cur.lower() and not it.get("nvidia_infs"):
|
|
report["skipped"].append(f"{cur}: has NVIDIA drivers, Dell pack {it['zip']} has none")
|
|
continue
|
|
hits = [(img, e) for img, m in maps.items() for e in m["entries"]
|
|
if isinstance(e, dict) and entry_file(e) == cur]
|
|
if not hits:
|
|
report["skipped"].append(f"{cur}: no map entry uses it any more")
|
|
continue
|
|
todo.append((cur, hits))
|
|
if not todo:
|
|
continue
|
|
placed = []
|
|
try:
|
|
if not _CMD_SAFE.fullmatch(it["zip"]):
|
|
raise RuntimeError("unsafe zip name")
|
|
dirs = []
|
|
for _, hits in todo:
|
|
for _, e in hits:
|
|
rel = entry_dir(e)
|
|
if ".." in PurePosixPath(rel).parts:
|
|
raise RuntimeError(f"destinationDir {rel} leaves the share")
|
|
d = resolve_dir_ci(config.SHARED_DIR, rel, create=True)
|
|
if d is None:
|
|
raise RuntimeError(f"cannot resolve {rel}")
|
|
if os.path.realpath(d) not in {os.path.realpath(x) for x in dirs}:
|
|
dirs.append(d)
|
|
src, move = it["file"], True
|
|
for d in dirs:
|
|
p = _place(src, d, it["zip"], it["size_file"], it["sha256_file"], move)
|
|
placed.append(p)
|
|
src, move = p, False # later directories get verified copies
|
|
except Exception as ex:
|
|
report["errors"].append(f"{it['zip']}: {ex}")
|
|
if placed and not os.path.exists(it["file"]):
|
|
# The staged zip was renamed onto the share before a later
|
|
# folder failed: point the item there so a retry can finish.
|
|
staged_items.setdefault(it["key"], {})["file"] = placed[0]
|
|
continue
|
|
for cur, hits in todo:
|
|
for img, e in hits:
|
|
for k in ("fileName", "FileName"):
|
|
if k in e:
|
|
e[k] = it["zip"]
|
|
e["url"] = it["url"]
|
|
e["hash"] = it["sha256_file"]
|
|
e["size"] = it["size_file"]
|
|
changed_maps.add(img)
|
|
report["drivers"].append({"old": cur, "new": it["zip"], "placed": placed,
|
|
"images": sorted({img for img, _ in hits})})
|
|
staged_items.setdefault(it["key"], {}).update(status="applied", applied=ts, placed=placed)
|
|
|
|
# One write per real file: if an image's map is a symlink to another's,
|
|
# writing it twice would back up the already-updated copy the second time.
|
|
written = set()
|
|
for img in sorted(changed_maps, key=lambda i: (os.path.islink(maps[i]["path"]), i)):
|
|
m = maps[img]
|
|
real = os.path.realpath(m["path"])
|
|
if real in written:
|
|
continue
|
|
written.add(real)
|
|
data = dump_map(m["entries"], m["raw"])
|
|
try:
|
|
report["backups"][m["path"]] = _write_like(m["path"], data, ts)
|
|
report["after"][m["path"]] = hashlib.sha256(data).hexdigest()
|
|
except OSError as ex:
|
|
report["errors"].append(f"{img} HardwareDriver.json not written ({ex}); its zip is on "
|
|
"the share but this image still uses the old file")
|
|
|
|
# BIOS: only the file|version fields of its lines; the exe goes into
|
|
# BIOS_DIR only when at least one line will point at it.
|
|
bios_items = [i for i in items if i["kind"] == "bios"]
|
|
if bios_items:
|
|
mpath = models_txt_path()
|
|
try:
|
|
with open(mpath, "rb") as f:
|
|
raw = f.read()
|
|
text = raw.decode("ascii")
|
|
except (OSError, UnicodeDecodeError) as ex:
|
|
report["errors"].append(f"models.txt: {ex}")
|
|
text = None
|
|
if text is not None:
|
|
split = text.split("\n")
|
|
lines_by_no = {n: t.rstrip("\r") for n, t in enumerate(split)}
|
|
edits, done = {}, []
|
|
for it in bios_items:
|
|
if not bios_name_ok(it["name"], it["ver"]):
|
|
report["errors"].append(f"{it['name']}: file name or version not safe for "
|
|
"check-bios.cmd; not applied")
|
|
continue
|
|
mine = {}
|
|
for ln in it.get("lines", []):
|
|
n = _find_line(lines_by_no, ln)
|
|
if n is None:
|
|
report["skipped"].append(f"models.txt line '{ln['line']}' changed since the check")
|
|
continue
|
|
mine[n] = _new_line(lines_by_no[n], it["name"], it["ver"])
|
|
if not mine:
|
|
continue
|
|
try:
|
|
placed = _place(it["file"], config.BIOS_DIR, it["name"], it["size_file"],
|
|
it["sha256_file"], True)
|
|
except Exception as ex:
|
|
report["errors"].append(f"{it['name']}: {ex}")
|
|
continue
|
|
edits.update(mine)
|
|
done.append((it, placed, mine))
|
|
if edits:
|
|
for n, new in edits.items():
|
|
cr = "\r" if split[n].endswith("\r") else ""
|
|
split[n] = new + cr
|
|
data = "\n".join(split).encode("ascii")
|
|
try:
|
|
report["backups"][mpath] = _write_like(mpath, data, ts)
|
|
report["after"][mpath] = hashlib.sha256(data).hexdigest()
|
|
except OSError as ex:
|
|
report["errors"].append(f"models.txt not written ({ex}); the BIOS files are in "
|
|
"place but no line points at them yet")
|
|
for it, placed, _ in done: # keep them applicable for a retry
|
|
staged_items.setdefault(it["key"], {})["file"] = placed
|
|
done = []
|
|
for it, placed, mine in done:
|
|
for n, new in mine.items():
|
|
report["bios"].append({"old": lines_by_no[n], "new": new})
|
|
staged_items.setdefault(it["key"], {}).update(status="applied", applied=ts,
|
|
placed=[placed])
|
|
|
|
# Merge field by field into a fresh read: a download job may have
|
|
# written other items while this ran.
|
|
state = staged_state()
|
|
for k, v in staged_items.items():
|
|
if k in state.get("items", {}):
|
|
state["items"][k].update(v)
|
|
_save_staged(state)
|
|
if report["drivers"] or report["bios"] or report["errors"]:
|
|
hist = history()
|
|
hist.insert(0, report)
|
|
uc.write_json(_path("applied.json"), hist[:50])
|
|
_audit("DELL_APPLY", f"{len(report['drivers'])} driver replacement(s), "
|
|
f"{len(report['bios'])} BIOS line(s), {len(report['skipped'])} skipped, "
|
|
f"{len(report['errors'])} error(s)")
|
|
_invalidate()
|
|
return report
|
|
|
|
|
|
def rollback(apply_id):
|
|
"""Restore the maps and models.txt saved before apply apply_id. A file
|
|
changed again since is left alone and reported. Placed zips/exes stay."""
|
|
with _Busy():
|
|
return _rollback(apply_id)
|
|
|
|
|
|
def _rollback(apply_id):
|
|
rec = next((h for h in history() if h.get("id") == apply_id), None)
|
|
if rec is None:
|
|
raise ValueError("unknown apply")
|
|
ts = _stamp()
|
|
restored, refused = [], []
|
|
for path, backup in rec.get("backups", {}).items():
|
|
try:
|
|
with open(path, "rb") as f:
|
|
now_hash = hashlib.sha256(f.read()).hexdigest()
|
|
if now_hash != rec.get("after", {}).get(path):
|
|
refused.append(f"{path} changed since that apply")
|
|
continue
|
|
with open(backup, "rb") as f:
|
|
data = f.read()
|
|
_write_like(path, data, ts, tag="pre-rollback")
|
|
restored.append(path)
|
|
except OSError as ex:
|
|
refused.append(f"{path}: {ex}")
|
|
hist = history()
|
|
for h in hist:
|
|
if h.get("id") == apply_id:
|
|
h["rolled_back"] = uc.now() if restored else h.get("rolled_back")
|
|
uc.write_json(_path("applied.json"), hist)
|
|
_audit("DELL_ROLLBACK", f"{apply_id}: restored {len(restored)}, refused {len(refused)}")
|
|
_invalidate()
|
|
return restored, refused
|
|
|
|
|
|
# --- history for the Updates pages -------------------------------------------------
|
|
|
|
def _file_label(path):
|
|
"""Short name of a file an apply wrote: 'gea-standard/.../HardwareDriver.json'
|
|
or 'BIOS/models.txt'."""
|
|
parts = path.replace("\\", "/").split("/")
|
|
if parts[-1] == "HardwareDriver.json":
|
|
image = next((p for p in parts if p in IMAGES), parts[-4] if len(parts) > 3 else "")
|
|
return f"{image}/.../HardwareDriver.json"
|
|
return "/".join(parts[-2:])
|
|
|
|
|
|
def file_status(rec):
|
|
"""Per file an apply wrote: does it still hold exactly what the apply
|
|
wrote (content hash, the test rollback uses)?
|
|
[{"path", "label", "backup", "state": unchanged|changed|missing, "mtime"}]"""
|
|
out = []
|
|
for path, backup in (rec.get("backups") or {}).items():
|
|
row = {"path": path, "label": _file_label(path), "backup": backup, "state": "missing", "mtime": None}
|
|
try:
|
|
with open(path, "rb") as f:
|
|
now_hash = hashlib.sha256(f.read()).hexdigest()
|
|
row["mtime"] = os.stat(path).st_mtime
|
|
row["state"] = "unchanged" if now_hash == (rec.get("after") or {}).get(path) else "changed"
|
|
except OSError:
|
|
pass
|
|
out.append(row)
|
|
return out
|
|
|
|
|
|
def _model_for_zip(zip_name, maps):
|
|
"""First model token of the map entry that uses zip_name, or ''."""
|
|
for m in maps.values():
|
|
for e in m["entries"]:
|
|
if isinstance(e, dict) and str(e.get("fileName") or e.get("FileName") or "").lower() == str(zip_name).lower():
|
|
return str(e.get("modelswminame") or "").split(",")[0].strip()
|
|
return ""
|
|
|
|
|
|
def describe(rec, maps=None):
|
|
"""One line per change of an apply, for history lists: a driver pack
|
|
'Latitude 5350 pack A11 to A12' (model from the maps, version from the
|
|
zip names) or a BIOS manifest line."""
|
|
maps = read_maps() if maps is None else maps
|
|
out = []
|
|
for d in rec.get("drivers") or []:
|
|
model = _model_for_zip(d.get("new", ""), maps) or _model_for_zip(d.get("old", ""), maps)
|
|
a, b = aversion(d.get("old", "")), aversion(d.get("new", ""))
|
|
if model and a is not None and b is not None:
|
|
out.append(f"{model} pack A{a:02d} to A{b:02d}")
|
|
else:
|
|
out.append(f"{d.get('old', '')} to {d.get('new', '')}")
|
|
for b in rec.get("bios") or []:
|
|
out.append(f"BIOS line {str(b.get('new', '')).strip()}")
|
|
return out
|
|
|
|
|
|
def history_view(limit=15):
|
|
"""history() with, per apply: "changes" (describe), "title" (first change
|
|
and "and n more"), "files" (file_status), "unchanged" and "total"."""
|
|
maps = read_maps()
|
|
out = []
|
|
for rec in history()[:limit]:
|
|
r = dict(rec)
|
|
r["changes"] = describe(rec, maps)
|
|
n = len(r["changes"])
|
|
r["title"] = (r["changes"][0] + (f" and {n - 1} more" if n > 1 else "")) if n else "No change"
|
|
r["files"] = file_status(rec)
|
|
r["unchanged"] = sum(1 for f in r["files"] if f["state"] == "unchanged")
|
|
r["total"] = len(r["files"])
|
|
out.append(r)
|
|
return out
|
|
|
|
|
|
def rollback_preview(apply_id, max_lines=400):
|
|
"""What a rollback of apply_id would do, without doing it: the apply
|
|
record, file_status, and for each unchanged file a unified diff from
|
|
the current content to the backup it would restore."""
|
|
rec = next((h for h in history_view(limit=1000) if h.get("id") == apply_id), None)
|
|
if rec is None:
|
|
raise ValueError("unknown apply")
|
|
for f in rec["files"]:
|
|
f["diff"] = ""
|
|
if f["state"] != "unchanged":
|
|
continue
|
|
try:
|
|
with open(f["path"], encoding="utf-8-sig", errors="replace") as cur, \
|
|
open(f["backup"], encoding="utf-8-sig", errors="replace") as old:
|
|
lines = list(difflib.unified_diff(cur.read().splitlines(), old.read().splitlines(),
|
|
"now", "after rollback", n=2, lineterm=""))
|
|
except OSError as ex:
|
|
f["state"], f["error"] = "missing", str(ex)
|
|
continue
|
|
if len(lines) > max_lines:
|
|
lines = lines[:max_lines] + [f"... {len(lines) - max_lines} more lines"]
|
|
f["diff"] = "\n".join(lines)
|
|
# A file whose backup could not be read above is no longer restorable.
|
|
rec["unchanged"] = sum(1 for f in rec["files"] if f["state"] == "unchanged")
|
|
return rec
|
|
|
|
|
|
def _audit(action, detail):
|
|
try:
|
|
from services.audit import audit
|
|
audit(action, detail)
|
|
except Exception:
|
|
pass # audit log not writable (local tests): the history file still has it
|
|
|
|
|
|
def _invalidate():
|
|
try:
|
|
from services import hardware_coverage
|
|
hardware_coverage.invalidate()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# --- hints for /hardware ----------------------------------------------------------
|
|
|
|
def update_hints():
|
|
"""{"drivers": {current zip (lower): Dell version}, "bios": {current BIOS
|
|
file (lower): Dell version}} for rows of the last scan with an update.
|
|
Keyed by file so /hardware can match what each model actually resolves
|
|
to (first matching map entry / models.txt line) without re-matching."""
|
|
scan = last_scan() or {}
|
|
out = {"drivers": {}, "bios": {}}
|
|
for r in scan.get("drivers", []):
|
|
if r.get("status") == "update" and r.get("current"):
|
|
out["drivers"][r["current"].lower()] = (r.get("pack") or {}).get("ver", "")
|
|
for r in scan.get("bios", []):
|
|
if r.get("status") == "update" and r.get("current"):
|
|
cands = r.get("candidates") or []
|
|
pick = r.get("pick") or 0
|
|
ver = cands[pick].get("version", "") if 0 <= pick < len(cands) else ""
|
|
out["bios"].setdefault(r["current"].lower(), ver)
|
|
return out
|
|
|
|
|
|
# --- job entry point --------------------------------------------------------------
|
|
|
|
def start(kind):
|
|
"""Start the check or download job (systemd-run). The two never overlap."""
|
|
other = "download" if kind == "check" else "check"
|
|
if uc.job_running(UNITS[other]):
|
|
raise RuntimeError(f"the {other} job is still running")
|
|
if kind == "check":
|
|
if uc.job_running(UNITS["check"]):
|
|
raise RuntimeError("a check is already running")
|
|
uc.write_json(_path("check.json"), {"state": "running", "started": uc.now(),
|
|
"step": "Starting"})
|
|
try:
|
|
uc.start_job(UNITS[kind], ["services.dell_updates", kind])
|
|
except Exception as ex:
|
|
if kind == "check": # else the page would show a check that never ran
|
|
_check_progress(state="error", finished=uc.now(), error=str(ex)[:400])
|
|
raise
|
|
|
|
|
|
def main(argv):
|
|
if not argv or argv[0] not in ("check", "download", "apply"):
|
|
print("usage: python3 -m services.dell_updates check|download|apply KEY...", file=sys.stderr)
|
|
return 2
|
|
if argv[0] == "check":
|
|
check(fetch="--offline" not in argv)
|
|
elif argv[0] == "download":
|
|
download()
|
|
else:
|
|
print(json.dumps(apply(argv[1:]), indent=1))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv[1:]))
|