606659bf07
- imaging: coerce client payload and stored session fields to str/int, drop lists/dicts; tiles template guards pctype/current_stage; POST rejects non-object bodies and non-string serials. - nic_status: normalise every check-file entry; load nic.conf as-is with a [nic] fallback; split interface= lists; keep the check's binding while fresh, resolve wildcard sockets from the dnsmasq pin (never True for an excluded NIC); mark vanished NICs down; include role bridges; stale or future-dated files downgrade to warnings and drop evidence. - Overview renders the network pill in its own try and shows 'Network: unknown' when status fails. - ipxe_update: read every Authenticode signature (all WIN_CERTIFICATE entries plus nested SPC_NESTED_SIGNATURE) with openssl; shim swap and install need both Microsoft UEFI CA 2011 and 2023; install re-reads each file (snponly ipxe.efi only); staged.json schema version, older records shown stale and refused; uhci counts as native USB. - backups describe their files; risky rollbacks need a confirm flag. - mac_folders: guarded listdir, cached describe, configured TFTP root, warning for a shim without ipxe.efi. - evidence pack: years 2000..2100, 400 days inclusive. - a11y role=group on quick ranges; progress bar stays full during 'shim'. - tests/test_review_fixes.py: 29 unittest cases incl. both real shims.
1058 lines
43 KiB
Python
1058 lines
43 KiB
Python
"""Keep the Secure Boot iPXE chain on the TFTP root current.
|
|
|
|
What dnsmasq serves (see /etc/dnsmasq.conf), since 2026-10-14:
|
|
* ipxe-shim.efi - iPXE's Microsoft-signed shim, every UEFI client. Over
|
|
TFTP it always loads ipxe.efi from the TFTP root (it does
|
|
not derive the name from its own, whatever the upstream
|
|
docs say);
|
|
* ipxe.efi - iPXE signed by the iPXE Secure Boot CA, loaded by the
|
|
shim. It is the release's SNPONLY build (x86_64-sb/
|
|
snponly.efi): firmware network and USB drivers. The
|
|
release's full ipxe.efi drives USB itself, which kills a
|
|
USB keyboard at the iPXE menu, so it is never installed;
|
|
* ipxe-bcm.efi - the old 2021 Broadcom-signed 1.21.1 build, kept as the
|
|
rollback (point dnsmasq's UEFI line back at it). Never
|
|
touched here;
|
|
* ipxe.pxe - legacy BIOS iPXE (unsigned; BIOS has no Secure Boot).
|
|
|
|
Per-MAC folders: with tftp-unique-root=mac, /srv/tftp/<aa-bb-cc-dd-ee-ff>/
|
|
wins over the root for that client, with NO fallback for a file missing
|
|
from it. They are shown here read-only and never written.
|
|
|
|
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, plus every
|
|
signature in the PE security directory, nested ones included, read with
|
|
openssl for the shim's Microsoft CAs); the cryptographic trust comes from
|
|
the GitHub digest. Do not swap in `sbverify --cert`: it
|
|
prints "Signature verification OK" even for a missing or unrelated cert.
|
|
|
|
Why it matters: the Broadcom build, and the shim in the iPXE tarball, are
|
|
signed only 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. iPXE publishes a shim dual-signed with the 2011 and 2023
|
|
CAs as a separate release (github.com/ipxe/shim); staging fetches it, checks
|
|
its digest and signatures, and uses it instead of the tarball's shim when it
|
|
trusts the same iPXE CA.
|
|
|
|
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"
|
|
SHIM_API = "https://api.github.com/repos/ipxe/shim/releases/latest"
|
|
SHIM_ASSET = "ipxe-shimx64.efi"
|
|
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"),
|
|
# The snponly build, installed under the name the shim loads.
|
|
"ipxe.efi": ("ipxeboot/x86_64-sb/snponly.efi", "ipxe"),
|
|
"ipxe.pxe": ("ipxeboot/x86_64/ipxe.pxe", None),
|
|
}
|
|
SHIMS = ("ipxe-shim.efi",)
|
|
# 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"
|
|
MS_CAS = (MS_CA_2011, MS_CA_2023)
|
|
|
|
# Bumped when the staged.json layout or the checks behind it change. A
|
|
# record written by an older version was verified by weaker checks and is
|
|
# shown as stale (download it again) and never installed.
|
|
STAGED_SCHEMA = 2
|
|
|
|
|
|
def _dir():
|
|
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
|
|
|
|
|
|
# Authenticode nested signature (SPC_NESTED_SIGNATURE, 1.3.6.1.4.1.311.2.4.1),
|
|
# DER-encoded OID. A second signature can sit in a second WIN_CERTIFICATE
|
|
# entry or nested inside the first one's unsigned attributes; sbverify may
|
|
# list only the outer one, so both places are read here.
|
|
NESTED_SIG_OID = bytes.fromhex("060a2b060104018237020401")
|
|
WIN_CERT_TYPE_PKCS_SIGNED_DATA = 0x0002
|
|
|
|
|
|
def _der_tlv(buf, i):
|
|
"""(content start, content end) of the DER element at buf[i]; raises
|
|
ValueError when it runs past the buffer."""
|
|
if i + 2 > len(buf):
|
|
raise ValueError("short DER")
|
|
n = buf[i + 1]
|
|
if n < 0x80:
|
|
start, length = i + 2, n
|
|
else:
|
|
k = n & 0x7F
|
|
if not 1 <= k <= 4 or i + 2 + k > len(buf):
|
|
raise ValueError("bad DER length")
|
|
start, length = i + 2 + k, int.from_bytes(buf[i + 2:i + 2 + k], "big")
|
|
if start + length > len(buf):
|
|
raise ValueError("DER runs past the buffer")
|
|
return start, start + length
|
|
|
|
|
|
def _pe_signatures(data):
|
|
"""The PKCS#7 blobs in a PE file's security directory, [] if none."""
|
|
try:
|
|
pe = struct.unpack_from("<I", data, 0x3C)[0]
|
|
if data[pe:pe + 4] != b"PE\0\0":
|
|
return []
|
|
magic = struct.unpack_from("<H", data, pe + 24)[0]
|
|
dirs = pe + 24 + (96 if magic == 0x10B else 112)
|
|
off, size = struct.unpack_from("<II", data, dirs + 4 * 8)
|
|
except struct.error:
|
|
return []
|
|
out, i, end = [], off, min(off + size, len(data))
|
|
while off and i + 8 <= end:
|
|
length, _rev, ctype = struct.unpack_from("<IHH", data, i)
|
|
if length < 8:
|
|
break
|
|
if ctype == WIN_CERT_TYPE_PKCS_SIGNED_DATA:
|
|
blob = data[i + 8:min(i + length, end)]
|
|
try:
|
|
out.append(blob[:_der_tlv(blob, 0)[1]]) # drop the 8-byte padding
|
|
except ValueError:
|
|
pass
|
|
i += (length + 7) & ~7
|
|
return out
|
|
|
|
|
|
def _nested_signatures(blob, depth=0):
|
|
"""ContentInfo blobs nested in `blob` under SPC_NESTED_SIGNATURE, deep
|
|
nests included."""
|
|
out = []
|
|
pos = blob.find(NESTED_SIG_OID)
|
|
while pos >= 0 and depth < 4:
|
|
try:
|
|
i = pos + len(NESTED_SIG_OID)
|
|
if blob[i] == 0x31: # SET OF ContentInfo
|
|
start, end = _der_tlv(blob, i)
|
|
j = start
|
|
while j < end:
|
|
_cs, ce = _der_tlv(blob, j)
|
|
inner = blob[j:ce]
|
|
if blob[j] == 0x30:
|
|
out.append(inner)
|
|
out += _nested_signatures(inner, depth + 1)
|
|
j = ce
|
|
except (ValueError, IndexError):
|
|
pass
|
|
pos = blob.find(NESTED_SIG_OID, pos + 1)
|
|
return out
|
|
|
|
|
|
def _cert_issuers(blob):
|
|
"""Issuer CNs of every certificate in one PKCS#7 blob (openssl)."""
|
|
r = subprocess.run(["openssl", "pkcs7", "-inform", "DER", "-print_certs", "-noout"],
|
|
input=blob, capture_output=True, timeout=30)
|
|
text = r.stdout.decode("utf-8", errors="replace")
|
|
return {m.strip() for m in re.findall(r"^issuer=.*?CN\s*=\s*([^,/\n]+)", text, re.M)}
|
|
|
|
|
|
def microsoft_cas(path):
|
|
"""Which of the Microsoft UEFI CAs (2011, 2023) sign this file, looking
|
|
at every signature: each WIN_CERTIFICATE entry and any nested inside one.
|
|
Raises FileNotFoundError when openssl is not installed."""
|
|
try:
|
|
with open(path, "rb") as f:
|
|
data = f.read()
|
|
except OSError:
|
|
return []
|
|
issuers = set()
|
|
for blob in _pe_signatures(data):
|
|
for b in [blob] + _nested_signatures(blob):
|
|
issuers |= _cert_issuers(b)
|
|
return [ca for ca in MS_CAS if ca in issuers]
|
|
|
|
|
|
def dual_signed(info):
|
|
"""True when a described file carries both Microsoft UEFI CAs."""
|
|
return bool(info) and all(ca in (info.get("ms_cas") or []) for ca in MS_CAS)
|
|
|
|
|
|
def signature_kind(pairs):
|
|
"""microsoft-2023 | microsoft-2011 | microsoft | ipxe | other | unsigned.
|
|
A file signed under both Microsoft CAs counts as 2023."""
|
|
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 drivers_of(path):
|
|
"""'native' when an iPXE build carries its own USB host drivers (it takes
|
|
the USB controller from the firmware, so a USB keyboard stops working),
|
|
'firmware' when it has none (snponly), '' for anything else."""
|
|
try:
|
|
with open(path, "rb") as f:
|
|
data = f.read()
|
|
except OSError:
|
|
return ""
|
|
if b"iPXE" not in data or b"$Version: " in data:
|
|
return "" # not iPXE, or a shim
|
|
return "native" if re.search(rb"\0(?:xhci|ehci|uhci)\0", data) else "firmware"
|
|
|
|
|
|
_DESCRIBE_CACHE = {}
|
|
|
|
|
|
def describe(path):
|
|
"""Size, hash, version, signers and drivers of one boot file, or None.
|
|
Cached per (path, size, mtime): reading and sbverify-ing every file on
|
|
each page load is slow. Returns a copy the caller may change."""
|
|
try:
|
|
st = os.stat(path)
|
|
except OSError:
|
|
return None
|
|
if not os.path.isfile(path):
|
|
return None
|
|
key = (os.path.abspath(path), st.st_size, st.st_mtime_ns)
|
|
hit = _DESCRIBE_CACHE.get(key)
|
|
if hit is None:
|
|
hit = _describe(path, st)
|
|
if len(_DESCRIBE_CACHE) > 256:
|
|
_DESCRIBE_CACHE.clear()
|
|
_DESCRIBE_CACHE[key] = hit
|
|
return dict(hit)
|
|
|
|
|
|
def _describe(path, st):
|
|
efi = path.endswith(".efi")
|
|
pairs, error, ms_cas = [], "", []
|
|
if efi:
|
|
try:
|
|
pairs = signers(path)
|
|
except (OSError, subprocess.SubprocessError) as ex:
|
|
error = f"sbverify failed: {ex}"
|
|
kind = (signature_kind(pairs) if not error else "error") if efi else "bios"
|
|
if kind.startswith("microsoft"):
|
|
try:
|
|
ms_cas = microsoft_cas(path)
|
|
except (OSError, subprocess.SubprocessError) as ex:
|
|
error = f"openssl failed: {ex}"
|
|
kind = "error"
|
|
info = {
|
|
"size": st.st_size, "modified": st.st_mtime,
|
|
"sha256": uc.file_hash(path), "version": version_of(path),
|
|
"sbat": sbat_generations(path) if efi else "",
|
|
"signers": pairs, "kind": kind,
|
|
"trusts": vendor_cert_cn(path) if efi else None,
|
|
"drivers": drivers_of(path) if efi else "",
|
|
"sig_error": error, "ms_cas": ms_cas,
|
|
}
|
|
info["dual"] = dual_signed(info)
|
|
return info
|
|
|
|
|
|
def risks(name, info):
|
|
"""Why a boot file should not go live, [] when none: an ipxe.efi that is
|
|
not the snponly (firmware drivers) build, or a shim that does not carry
|
|
both Microsoft UEFI CAs."""
|
|
if not info:
|
|
return []
|
|
if name == "ipxe.efi" and info.get("drivers") != "firmware":
|
|
if info.get("drivers") == "native":
|
|
return ["ipxe.efi drives USB itself (not snponly): USB keyboards stop working at the iPXE menu"]
|
|
return ["ipxe.efi is not recognised as an iPXE snponly build"]
|
|
if name in SHIMS and not info.get("dual"):
|
|
have = " and ".join(info.get("ms_cas") or []) or info.get("kind") or "no Microsoft CA"
|
|
return [f"{name} is not dual-signed (2011 + 2023): it carries {have}"]
|
|
return []
|
|
|
|
|
|
def _installed_log():
|
|
"""{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 signed under the CA it embeds.
|
|
by = {r["name"]: r["info"] for r in rows}
|
|
_check_chain(by.get("ipxe-shim.efi"), [by.get("ipxe.efi")])
|
|
return rows
|
|
|
|
|
|
def _check_chain(shim, ipxes):
|
|
for ipxe in ipxes:
|
|
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"]
|
|
|
|
|
|
MAC_DIR_RE = re.compile(r"^[0-9a-f]{2}(-[0-9a-f]{2}){5}$")
|
|
|
|
|
|
def mac_folders():
|
|
"""Per-client TFTP folders (tftp-unique-root=mac): [{mac, files}], each
|
|
file described. Read-only; install never writes into them."""
|
|
out = []
|
|
try:
|
|
names = sorted(os.listdir(config.TFTP_ROOT))
|
|
except OSError:
|
|
return out
|
|
for d in names:
|
|
folder = os.path.join(config.TFTP_ROOT, d)
|
|
if not (MAC_DIR_RE.match(d) and os.path.isdir(folder)):
|
|
continue
|
|
files, warnings = [], []
|
|
try:
|
|
entries = sorted(os.listdir(folder))
|
|
except OSError as ex:
|
|
entries = []
|
|
warnings.append(f"cannot read the folder: {ex.strerror or ex}")
|
|
for f in entries:
|
|
info = describe(os.path.join(folder, f))
|
|
if info:
|
|
files.append({"name": f, "info": info})
|
|
by = {f["name"]: f["info"] for f in files}
|
|
shim = next((by[n] for n in SHIMS if n in by), None)
|
|
if shim is None:
|
|
shim = next((f["info"] for f in files if (f["info"].get("version") or "").startswith("shim ")), None)
|
|
_check_chain(shim, [by.get("ipxe.efi")])
|
|
if shim is not None and "ipxe.efi" not in by:
|
|
warnings.append("This folder has a shim but no ipxe.efi. The shim loads ipxe.efi from this folder "
|
|
"(no fallback to the root), so this client stops after the shim.")
|
|
out.append({"mac": d.replace("-", ":"), "dir": d, "files": files, "warnings": warnings})
|
|
return out
|
|
|
|
|
|
# --- 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", "tftp-unique-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,
|
|
"unique_root": any(s.startswith("tftp-unique-root") for s in raw),
|
|
"shim_macs": sorted({m for b in boots if os.path.basename(b["file"]) in SHIMS
|
|
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)
|
|
_progress(step="shim")
|
|
_use_dual_signed_shim(folder, files)
|
|
# The shim must trust the CA this release's iPXE builds are signed by.
|
|
shim = files.get("ipxe-shim.efi") or {}
|
|
ipxe = 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")
|
|
if ipxe.get("ok") and ipxe.get("drivers") != "firmware":
|
|
# The full build would kill USB keyboards on every client; anything
|
|
# not recognised as snponly is refused too.
|
|
ipxe.update(ok=False, error=("this build drives USB itself (not snponly); refusing it as ipxe.efi"
|
|
if ipxe.get("drivers") == "native" else
|
|
"not recognised as an iPXE snponly build; refusing it as ipxe.efi"))
|
|
for n in SHIMS:
|
|
f = files.get(n) or {}
|
|
if f.get("ok") and not f.get("dual"):
|
|
# Kept on disk and shown, but never installed: a 2011-only shim
|
|
# stops booting once machines drop or revoke the 2011 CA.
|
|
have = " and ".join(f.get("ms_cas") or []) or f.get("kind")
|
|
f.update(ok=False, error=f"not dual-signed (needs {MS_CA_2011} and {MS_CA_2023}; has {have}). "
|
|
+ (f.get("note") or ""))
|
|
record = {"schema": STAGED_SCHEMA, "tag": info["tag"], "published": info.get("published"), "folder": folder,
|
|
"tarball_sha256": info["sha256"], "files": files, "staged": uc.now()}
|
|
uc.write_json(os.path.join(folder, "staged.json"), record)
|
|
return record
|
|
|
|
|
|
def _use_dual_signed_shim(folder, files):
|
|
"""Swap the tarball's shim for the ipxe/shim release build when that one
|
|
carries both Microsoft UEFI CAs (2011 and 2023, read from every
|
|
signature, nested ones included) and trusts the same iPXE CA. Any
|
|
problem leaves the tarball shim in place, with the reason noted."""
|
|
tar_shim = files.get("ipxe-shim.efi") or {}
|
|
if not tar_shim.get("ok") or tar_shim.get("dual"):
|
|
return
|
|
try:
|
|
rel = json.loads(uc.fetch_text(SHIM_API, timeout=20))
|
|
asset = next((a for a in rel.get("assets", []) if a.get("name") == SHIM_ASSET), None)
|
|
digest = str((asset or {}).get("digest") or "")
|
|
if not asset or not digest.startswith("sha256:"):
|
|
raise RuntimeError(f"{rel.get('tag_name')} lists no sha256 for {SHIM_ASSET}")
|
|
sha = digest.split(":", 1)[1].lower()
|
|
dest = os.path.join(folder, SHIM_ASSET)
|
|
if not (os.path.isfile(dest) and uc.file_hash(dest).lower() == sha):
|
|
if os.path.exists(dest):
|
|
os.remove(dest)
|
|
uc.download(asset["browser_download_url"], dest, expect=sha, timeout=300)
|
|
d = describe(dest)
|
|
if d["sig_error"]:
|
|
raise RuntimeError(f"{SHIM_ASSET}: {d['sig_error']}")
|
|
if not d["dual"]:
|
|
have = " and ".join(d["ms_cas"]) or d["kind"]
|
|
raise RuntimeError(f"{SHIM_ASSET} is not signed under both {MS_CA_2011} and {MS_CA_2023} "
|
|
f"(found {have})")
|
|
if d.get("trusts") != tar_shim.get("trusts"):
|
|
raise RuntimeError(f"{SHIM_ASSET} trusts {d.get('trusts')}, the tarball's iPXE needs "
|
|
f"{tar_shim.get('trusts')}")
|
|
except Exception as ex: # optional upgrade: keep the tarball shim
|
|
note = f"dual-signed shim not used: {str(ex)[:200]}"
|
|
for n in SHIMS:
|
|
if n in files:
|
|
files[n]["note"] = note
|
|
return
|
|
for n in SHIMS:
|
|
if n in files:
|
|
shutil.copy2(dest, os.path.join(folder, n))
|
|
files[n] = dict(d, ok=True, error="", source=f"github.com/ipxe/shim {rel.get('tag_name')} "
|
|
f"{SHIM_ASSET} (dual-signed: {' + '.join(d['ms_cas'])})")
|
|
|
|
|
|
def _record_stale(rec):
|
|
try:
|
|
schema = int(rec.get("schema") or 0)
|
|
except (TypeError, ValueError):
|
|
schema = 0
|
|
if schema < STAGED_SCHEMA:
|
|
return True
|
|
return any(isinstance(f, dict) and f.get("signers") is not None and "drivers" not in f
|
|
for f in rec["files"].values())
|
|
|
|
|
|
def staged(tag=None):
|
|
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)
|
|
if _record_stale(rec):
|
|
# Verified by older, weaker checks: show it, never install it.
|
|
rec["stale"] = True
|
|
for f in rec["files"].values():
|
|
if isinstance(f, dict):
|
|
f.update(ok=False, error="verified by an older version of this page; download it again")
|
|
out.append(rec)
|
|
out.sort(key=lambda r: r.get("staged") or "", reverse=True)
|
|
return out
|
|
|
|
|
|
# --- 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")
|
|
if rec.get("stale"):
|
|
raise ValueError(f"{tag} was verified by an older version of this page; download it again")
|
|
names = [n for n in MANAGED if n in names]
|
|
if not names:
|
|
raise ValueError("nothing selected")
|
|
if any(n in names for n in SHIMS) and "ipxe.efi" not in names:
|
|
raise ValueError("install a shim 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):
|
|
if _record_stale(rec):
|
|
raise ValueError(f"{tag} was verified by an older version of this page; download it again")
|
|
for n in names:
|
|
f = rec["files"].get(n) or {}
|
|
if not f.get("ok"):
|
|
raise ValueError(f"{n}: {f.get('error') or 'not staged'}")
|
|
path = os.path.join(rec["folder"], n)
|
|
if not os.path.isfile(path) or uc.file_hash(path) != f.get("sha256"):
|
|
raise ValueError(f"{n}: staged copy changed since it was verified; download again")
|
|
# Re-read the file itself: the record says what was true at staging.
|
|
need = MANAGED[n][1]
|
|
if need:
|
|
now = _describe(path, os.stat(path)) # uncached: this gates going live
|
|
if now.get("sig_error") or not str(now.get("kind") or "").startswith(need):
|
|
raise ValueError(f"{n}: expected a signature under {NEED_LABEL[need]}, found "
|
|
f"{now.get('sig_error') or now.get('kind')}")
|
|
bad = risks(n, now)
|
|
if bad:
|
|
raise ValueError(f"{n}: {bad[0]}; refusing to install it")
|
|
if not any(n in names for n in SHIMS):
|
|
# iPXE alone must still be one the installed shim will start.
|
|
trusts = vendor_cert_cn(os.path.join(config.TFTP_ROOT, "ipxe-shim.efi"))
|
|
if "ipxe.efi" in names:
|
|
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))]
|
|
infos = {f: describe(os.path.join(folder, f)) for f in files}
|
|
out.append({"id": b, "reason": meta.get("reason", ""), "at": meta.get("at", ""),
|
|
"files": files, "absent": meta.get("absent", []),
|
|
"info": infos,
|
|
"versions": {f: (infos[f] or {}).get("version", "") for f in files},
|
|
"warnings": [w for f in files for w in risks(f, infos[f])]})
|
|
return out
|
|
|
|
|
|
def rollback(backup_id, confirm=False):
|
|
"""Put the files of one backup back, after backing up what is there now.
|
|
A backup holding a native-USB ipxe.efi or a shim that is not dual-signed
|
|
is only restored with confirm=True."""
|
|
bid = os.path.basename(str(backup_id or ""))
|
|
folder = os.path.join(_dir(), "backups", bid)
|
|
if not bid or bid.startswith(".") or not os.path.isdir(folder):
|
|
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")
|
|
warnings = [w for f in files
|
|
for w in risks(f, _describe(os.path.join(folder, f), os.stat(os.path.join(folder, f))))]
|
|
if warnings and not confirm:
|
|
raise ValueError("; ".join(warnings) + ". Tick the confirmation to restore it anyway")
|
|
with _Lock():
|
|
_backup(files, f"before rolling back to {bid}")
|
|
_place_all([(os.path.join(folder, f), f, uc.file_hash(os.path.join(folder, f)))
|
|
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)
|