81c2d85301
Staging fetches ipxe-shimx64.efi from the ipxe/shim release (sha256 from GitHub), and uses it instead of the tarball's 2011-only shim when it is signed under the Microsoft UEFI CA 2023 and trusts the same iPXE CA; otherwise the tarball shim stays, with the reason shown. snponly.efi and snponly-shim.efi are managed alongside ipxe.efi and ipxe-shim.efi, each iPXE build says whether it drives USB itself (kills USB keyboards), and tftp-unique-root per-MAC folders are listed read-only.
837 lines
34 KiB
Python
837 lines
34 KiB
Python
"""Keep the Secure Boot iPXE chain on the TFTP root current.
|
|
|
|
What dnsmasq serves (see /etc/dnsmasq.conf):
|
|
* ipxe-shim.efi - Microsoft-signed shim, for the MACs tagged ipxe2. 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);
|
|
* snponly-shim.efi - the same shim under the name dnsmasq hands the
|
|
ipxesnp-tagged MACs;
|
|
* ipxe.efi - iPXE signed by the iPXE Secure Boot CA, loaded by the
|
|
shim. The full build drives USB itself, which kills a
|
|
USB keyboard at the iPXE menu;
|
|
* snponly.efi - the same release built to use the firmware's network
|
|
and USB drivers (keyboard keeps working). Served to a
|
|
client as ipxe.efi from its per-MAC folder;
|
|
* ipxe-bcm.efi - the old 2021 Broadcom-signed build, everyone else.
|
|
Never touched here;
|
|
* ipxe.pxe - legacy BIOS iPXE (unsigned; BIOS has no Secure Boot).
|
|
|
|
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); 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"),
|
|
"snponly-shim.efi": ("ipxeboot/x86_64-sb/shimx64.efi", "microsoft"),
|
|
"ipxe.efi": ("ipxeboot/x86_64-sb/ipxe.efi", "ipxe"),
|
|
"snponly.efi": ("ipxeboot/x86_64-sb/snponly.efi", "ipxe"),
|
|
"ipxe.pxe": ("ipxeboot/x86_64/ipxe.pxe", None),
|
|
}
|
|
SHIMS = ("ipxe-shim.efi", "snponly-shim.efi")
|
|
# Installed by default; ipxe.pxe (legacy BIOS) only when asked.
|
|
DEFAULT_INSTALL = ("ipxe-shim.efi", "snponly-shim.efi", "ipxe.efi", "snponly.efi")
|
|
# Shown on the page for reference, never written.
|
|
READ_ONLY = ("ipxe-bcm.efi",)
|
|
|
|
NEED_LABEL = {"microsoft": "a Microsoft UEFI CA", "ipxe": "the iPXE Secure Boot CA"}
|
|
|
|
# Issuer CNs, as sbverify prints them.
|
|
MS_CA_2011 = "Microsoft Corporation UEFI CA 2011"
|
|
MS_CA_2023 = "Microsoft UEFI CA 2023"
|
|
IPXE_CA = "iPXE Secure Boot CA"
|
|
|
|
|
|
def _dir():
|
|
return uc.state_dir("ipxe")
|
|
|
|
|
|
def _job_file():
|
|
return os.path.join(_dir(), "job.json")
|
|
|
|
|
|
# --- reading binaries -------------------------------------------------------
|
|
|
|
def signers(path):
|
|
"""[(subject CN, issuer CN), ...] from the Authenticode chain, [] if none.
|
|
Raises FileNotFoundError when sbverify is not installed."""
|
|
r = subprocess.run(["sbverify", "--list", path], capture_output=True, text=True, timeout=60)
|
|
pairs, subject = [], None
|
|
for line in r.stdout.splitlines():
|
|
line = line.strip()
|
|
m = re.match(r"- subject: .*?CN=([^/]+)", line)
|
|
if m:
|
|
subject = m.group(1).strip()
|
|
continue
|
|
m = re.match(r"issuer: .*?CN=([^/]+)", line)
|
|
if m and subject is not None:
|
|
pairs.append((subject, m.group(1).strip()))
|
|
subject = None
|
|
return pairs
|
|
|
|
|
|
def signature_kind(pairs):
|
|
"""microsoft-2023 | microsoft-2011 | microsoft | ipxe | other | unsigned.
|
|
A file signed under both Microsoft CAs counts as 2023."""
|
|
issuers = {i for _, i in pairs}
|
|
names = " ".join(s + " " + i for s, i in pairs)
|
|
if MS_CA_2023 in issuers:
|
|
return "microsoft-2023"
|
|
if MS_CA_2011 in issuers:
|
|
return "microsoft-2011"
|
|
if "Microsoft" in names:
|
|
return "microsoft"
|
|
if IPXE_CA in issuers:
|
|
return "ipxe"
|
|
return "unsigned" if not pairs else "other"
|
|
|
|
|
|
def _sbat(data):
|
|
"""The SBAT CSV a shim or iPXE build carries, as {component: [fields]}."""
|
|
i = data.find(b"sbat,1,SBAT Version,")
|
|
if i < 0:
|
|
return {}
|
|
text = data[i:i + 4096].split(b"\0", 1)[0].decode("ascii", errors="replace")
|
|
out = {}
|
|
for line in text.splitlines():
|
|
fields = line.split(",")
|
|
if len(fields) >= 2 and fields[0] != "sbat":
|
|
out[fields[0]] = fields
|
|
return out
|
|
|
|
|
|
def version_of(path):
|
|
try:
|
|
with open(path, "rb") as f:
|
|
data = f.read()
|
|
except OSError:
|
|
return ""
|
|
# A shim says so in its build ident; the shim's SBAT also names iPXE
|
|
# (shim.ipxe), so test this first.
|
|
m = re.search(rb"\$Version: ([^ $]{1,30})", data)
|
|
if m:
|
|
return "shim " + m.group(1).decode(errors="replace")
|
|
sbat = _sbat(data)
|
|
if "ipxe" in sbat and len(sbat["ipxe"]) >= 5:
|
|
return "iPXE " + sbat["ipxe"][4]
|
|
m = re.search(rb"([0-9]+\.[0-9]+\.[0-9]+\+? ?\(g[0-9a-f]+\))", data)
|
|
if m:
|
|
return "iPXE " + m.group(1).decode(errors="replace")
|
|
return ""
|
|
|
|
|
|
def sbat_generations(path):
|
|
"""'shim 4, shim.ipxe 1': the generation numbers a revocation compares."""
|
|
try:
|
|
with open(path, "rb") as f:
|
|
sbat = _sbat(f.read())
|
|
except OSError:
|
|
return ""
|
|
return ", ".join(f"{k} {v[1]}" for k, v in sbat.items())
|
|
|
|
|
|
def vendor_cert_cn(path):
|
|
"""Subject CN of the certificate a shim trusts for its second stage (the
|
|
PE .vendor_cert section), or None. iPXE's shim embeds the iPXE Secure
|
|
Boot CA, so ipxe.efi must chain to that name."""
|
|
try:
|
|
with open(path, "rb") as f:
|
|
d = f.read()
|
|
pe = struct.unpack_from("<I", d, 0x3C)[0]
|
|
if d[pe:pe + 4] != b"PE\0\0":
|
|
return None
|
|
nsec = struct.unpack_from("<H", d, pe + 6)[0]
|
|
symptr, nsym = struct.unpack_from("<II", d, pe + 12)
|
|
optsz = struct.unpack_from("<H", d, pe + 20)[0]
|
|
strtab = symptr + nsym * 18
|
|
off = pe + 24 + optsz
|
|
for i in range(nsec):
|
|
raw, _vsize, _vaddr, _rsize, rptr = struct.unpack_from("<8sIIII", d, off + 40 * i)
|
|
name = raw.rstrip(b"\0")
|
|
if name.startswith(b"/") and symptr: # long name, in the COFF string table
|
|
o = strtab + int(name[1:])
|
|
name = d[o:d.index(b"\0", o)]
|
|
if name != b".vendor_cert":
|
|
continue
|
|
size, _dsize, cert_off, _doff = struct.unpack_from("<IIII", d, rptr)
|
|
cert = d[rptr + cert_off:rptr + cert_off + size]
|
|
# DER: every CN is OID 2.5.4.3 then a short string. Issuer comes
|
|
# before subject, so the last one is the subject.
|
|
cns = [cert[m.end() + 2:m.end() + 2 + cert[m.end() + 1]].decode(errors="replace")
|
|
for m in re.finditer(rb"\x06\x03\x55\x04\x03(?=[\x0c\x13\x16])", cert)]
|
|
return cns[-1] if cns else None
|
|
except (OSError, struct.error, ValueError, IndexError):
|
|
return None
|
|
return None
|
|
|
|
|
|
def 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"\0xhci\0|\0ehci\0", data) else "firmware"
|
|
|
|
|
|
def describe(path):
|
|
if not os.path.isfile(path):
|
|
return None
|
|
st = os.stat(path)
|
|
efi = path.endswith(".efi")
|
|
pairs, error = [], ""
|
|
if efi:
|
|
try:
|
|
pairs = signers(path)
|
|
except (OSError, subprocess.SubprocessError) as ex:
|
|
error = f"sbverify failed: {ex}"
|
|
return {
|
|
"size": st.st_size, "modified": st.st_mtime,
|
|
"sha256": uc.file_hash(path), "version": version_of(path),
|
|
"sbat": sbat_generations(path) if efi else "",
|
|
"signers": pairs, "kind": (signature_kind(pairs) if not error else "error") if efi else "bios",
|
|
"trusts": vendor_cert_cn(path) if efi else None,
|
|
"drivers": drivers_of(path) if efi else "",
|
|
"sig_error": error,
|
|
}
|
|
|
|
|
|
def _installed_log():
|
|
"""{name: {tag, sha256, at}}: which release each installed file came from."""
|
|
log = uc.read_json(os.path.join(_dir(), "installed.json"), {})
|
|
return log if isinstance(log, dict) else {}
|
|
|
|
|
|
def installed():
|
|
log = _installed_log()
|
|
rows = []
|
|
for name in list(MANAGED) + list(READ_ONLY):
|
|
info = describe(os.path.join(config.TFTP_ROOT, name))
|
|
origin = log.get(name) or {}
|
|
if info and origin.get("sha256") == info["sha256"]:
|
|
info["release"] = origin.get("tag")
|
|
rows.append({"name": name, "managed": name in MANAGED, "info": info})
|
|
# The shim only starts an iPXE signed under the CA it embeds.
|
|
by = {r["name"]: r["info"] for r in rows}
|
|
_check_chain(by.get("ipxe-shim.efi") or by.get("snponly-shim.efi"),
|
|
[by.get("ipxe.efi"), by.get("snponly.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 = []
|
|
for f in sorted(os.listdir(folder)):
|
|
info = describe(os.path.join(folder, f))
|
|
if info:
|
|
files.append({"name": f, "info": info})
|
|
by = {f["name"]: f["info"] for f in files}
|
|
_check_chain(next((by[n] for n in SHIMS if n in by), None), [by.get("ipxe.efi")])
|
|
out.append({"mac": d.replace("-", ":"), "dir": d, "files": files})
|
|
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 {}
|
|
for name in ("ipxe.efi", "snponly.efi"):
|
|
ipxe = files.get(name) or {}
|
|
if shim.get("ok") and ipxe.get("ok") and shim.get("trusts"):
|
|
if shim["trusts"] not in {i for _, i in ipxe["signers"]}:
|
|
ipxe.update(ok=False, error=f"not signed under {shim['trusts']}, which this shim trusts")
|
|
record = {"tag": info["tag"], "published": info.get("published"), "folder": folder,
|
|
"tarball_sha256": info["sha256"], "files": files, "staged": uc.now()}
|
|
uc.write_json(os.path.join(folder, "staged.json"), record)
|
|
return record
|
|
|
|
|
|
def _use_dual_signed_shim(folder, files):
|
|
"""Swap the tarball's shim for the ipxe/shim release build when that one
|
|
is signed under the Microsoft UEFI CA 2023 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("kind") == "microsoft-2023":
|
|
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["kind"] != "microsoft-2023":
|
|
raise RuntimeError(f"{SHIM_ASSET} is {d['kind']}, not signed under {MS_CA_2023}")
|
|
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, 2011 + 2023)")
|
|
|
|
|
|
def staged(tag=None):
|
|
root = os.path.join(_dir(), "staged")
|
|
if not os.path.isdir(root):
|
|
return []
|
|
out = []
|
|
for t in sorted(os.listdir(root), reverse=True):
|
|
rec = uc.read_json(os.path.join(root, t, "staged.json"))
|
|
if not (isinstance(rec, dict) and rec.get("tag") and isinstance(rec.get("files"), dict)
|
|
and rec["files"]):
|
|
continue # a hand-edited or half-written record has nothing to install
|
|
if tag is None or rec.get("tag") == tag:
|
|
# Where it is now, not the path recorded at staging time.
|
|
rec["folder"] = os.path.join(root, t)
|
|
out.append(rec)
|
|
out.sort(key=lambda r: r.get("staged") or "", reverse=True)
|
|
return out
|
|
|
|
|
|
# --- background stage job ---------------------------------------------------
|
|
|
|
def job_state():
|
|
"""Last stage job's progress, with 'running' from systemd and, while
|
|
downloading, a percent from the partial file size."""
|
|
data = uc.read_json(_job_file(), {}) or {}
|
|
try:
|
|
running = uc.job_running(JOB_UNIT)
|
|
except OSError:
|
|
running = False
|
|
data["running"] = running
|
|
if data.get("state") in ("queued", "running") and not running:
|
|
# Give systemd-run a moment to bring the unit up before calling it dead.
|
|
try:
|
|
age = datetime.now().timestamp() - os.path.getmtime(_job_file())
|
|
except OSError:
|
|
age = 0
|
|
if age > 20:
|
|
data["state"] = "interrupted"
|
|
if data.get("step") == "download" and data.get("size") and data.get("tarball"):
|
|
try:
|
|
done = os.path.getsize(data["tarball"])
|
|
data["percent"] = min(100, int(100 * done / data["size"]))
|
|
except OSError:
|
|
data["percent"] = 0
|
|
return data
|
|
|
|
|
|
def start_stage():
|
|
"""Queue the stage job (systemd-run). The page polls job_state()."""
|
|
if uc.job_running(JOB_UNIT):
|
|
raise RuntimeError("a download is already running")
|
|
info = last_check()
|
|
if not info:
|
|
raise RuntimeError("check for updates first")
|
|
uc.write_json(_job_file(), {"state": "queued", "tag": info["tag"], "started": uc.now(),
|
|
"updated": uc.now()})
|
|
try:
|
|
uc.start_job(JOB_UNIT, ["services.ipxe_update", "stage"])
|
|
except Exception as ex:
|
|
uc.write_json(_job_file(), {"state": "error", "tag": info["tag"], "error": str(ex),
|
|
"updated": uc.now()})
|
|
raise
|
|
|
|
|
|
def run_stage_job():
|
|
"""Entry point of the background job. Never raises; the outcome goes to job.json."""
|
|
_progress(state="running", step="start", error="", started=uc.now())
|
|
try:
|
|
rec = stage(last_check())
|
|
bad = [n for n, f in rec["files"].items() if not f.get("ok")]
|
|
_progress(state="done", step="done", tag=rec["tag"], finished=uc.now(),
|
|
failed=bad)
|
|
return 0
|
|
except Exception as ex: # job boundary: record any failure for the page
|
|
_progress(state="error", error=str(ex)[:500], finished=uc.now())
|
|
return 1
|
|
|
|
|
|
# --- install / rollback -----------------------------------------------------
|
|
|
|
def _new_folder(root):
|
|
"""A fresh timestamped folder; never reuses one (a rollback in the same
|
|
second as its backup must not overwrite the files it restores)."""
|
|
base = os.path.join(root, datetime.now().strftime("%Y%m%d-%H%M%S"))
|
|
folder, n = base, 1
|
|
while True:
|
|
try:
|
|
os.makedirs(folder)
|
|
return folder
|
|
except FileExistsError:
|
|
folder, n = f"{base}-{n}", n + 1
|
|
|
|
|
|
def _backup(names, reason):
|
|
folder = _new_folder(os.path.join(_dir(), "backups"))
|
|
saved, absent = [], []
|
|
for name in names:
|
|
src = os.path.join(config.TFTP_ROOT, name)
|
|
if os.path.isfile(src):
|
|
shutil.copy2(src, os.path.join(folder, name))
|
|
saved.append(name)
|
|
else:
|
|
absent.append(name)
|
|
uc.write_json(os.path.join(folder, "backup.json"),
|
|
{"reason": reason, "files": saved, "absent": absent, "at": uc.now()})
|
|
return folder
|
|
|
|
|
|
class _Lock:
|
|
"""One install or rollback at a time: two at once would back up each
|
|
other's half-placed files."""
|
|
|
|
def __enter__(self):
|
|
self.f = open(os.path.join(_dir(), ".install.lock"), "w")
|
|
try:
|
|
fcntl.flock(self.f, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except OSError:
|
|
self.f.close()
|
|
raise RuntimeError("another install or rollback is running; try again") from None
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
self.f.close()
|
|
|
|
|
|
def _prepare(src, name, sha256=None):
|
|
"""Copy src next to its TFTP name under a temporary name, fsynced, mode
|
|
0644 and with the current file's owner. With sha256, the copy is checked.
|
|
Returns the temporary path; nothing is live yet."""
|
|
dest = os.path.join(config.TFTP_ROOT, name)
|
|
fd, tmp = tempfile.mkstemp(dir=config.TFTP_ROOT, prefix=f".{name}.", suffix=".new")
|
|
try:
|
|
with open(src, "rb") as s, os.fdopen(fd, "wb") as out:
|
|
shutil.copyfileobj(s, out)
|
|
out.flush()
|
|
os.fsync(out.fileno())
|
|
if sha256 and uc.file_hash(tmp).lower() != sha256.lower():
|
|
raise RuntimeError(f"{name}: copy does not match its sha256, nothing replaced")
|
|
os.chmod(tmp, 0o644)
|
|
try:
|
|
st = os.stat(dest)
|
|
os.chown(tmp, st.st_uid, st.st_gid)
|
|
except OSError:
|
|
pass # no current file, or not root (local tests)
|
|
return tmp
|
|
except BaseException:
|
|
os.remove(tmp)
|
|
raise
|
|
|
|
|
|
def _place_all(items):
|
|
"""[(src, name, sha256)] into the TFTP root. Every copy is made and
|
|
checked first, then each is renamed into place (atomic per file, so a
|
|
booting client never reads half a file). A failed copy leaves all the
|
|
live files as they were."""
|
|
ready = []
|
|
try:
|
|
for src, name, sha256 in items:
|
|
ready.append((_prepare(src, name, sha256), name))
|
|
except BaseException:
|
|
for tmp, _ in ready:
|
|
os.remove(tmp)
|
|
raise
|
|
for tmp, name in ready:
|
|
os.replace(tmp, os.path.join(config.TFTP_ROOT, name))
|
|
|
|
|
|
def install(tag, names=DEFAULT_INSTALL):
|
|
"""Copy staged files for `tag` into the TFTP root after backing up the
|
|
current ones. Refuses a set with a failed signature check, and refuses to
|
|
install ipxe-shim.efi without its matching ipxe.efi (the shim only trusts
|
|
the iPXE CA the same release signs with). Returns the backup folder."""
|
|
rec = next(iter(staged(tag)), None)
|
|
if not rec:
|
|
raise ValueError(f"{tag} is not staged")
|
|
names = [n for n in MANAGED if n in names]
|
|
if not names:
|
|
raise ValueError("nothing selected")
|
|
if 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):
|
|
for n in names:
|
|
f = rec["files"].get(n) or {}
|
|
if not f.get("ok"):
|
|
raise ValueError(f"{n}: {f.get('error') or 'not staged'}")
|
|
path = os.path.join(rec["folder"], n)
|
|
if not os.path.isfile(path) or uc.file_hash(path) != f["sha256"]:
|
|
raise ValueError(f"{n}: staged copy changed since it was verified; download again")
|
|
if 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"))
|
|
for n in ("ipxe.efi", "snponly.efi"):
|
|
if n not in names:
|
|
continue
|
|
issuers = {i for _, i in rec["files"][n].get("signers", [])}
|
|
if trusts and trusts not in issuers:
|
|
raise ValueError(f"the installed shim trusts {trusts}; this {n} is not signed "
|
|
"under it, install the shim from the same release too")
|
|
backup = _backup(names, f"before installing {tag}")
|
|
_place_all([(os.path.join(rec["folder"], n), n, rec["files"][n]["sha256"]) for n in names])
|
|
log = _installed_log()
|
|
for n in names:
|
|
log[n] = {"tag": tag, "sha256": rec["files"][n]["sha256"], "at": uc.now()}
|
|
uc.write_json(os.path.join(_dir(), "installed.json"), log)
|
|
return backup
|
|
|
|
|
|
def backups():
|
|
root = os.path.join(_dir(), "backups")
|
|
if not os.path.isdir(root):
|
|
return []
|
|
out = []
|
|
for b in sorted(os.listdir(root), reverse=True):
|
|
folder = os.path.join(root, b)
|
|
if not os.path.isdir(folder):
|
|
continue
|
|
meta = uc.read_json(os.path.join(folder, "backup.json"), {})
|
|
meta = meta if isinstance(meta, dict) else {}
|
|
files = [f for f in MANAGED if os.path.isfile(os.path.join(folder, f))]
|
|
out.append({"id": b, "reason": meta.get("reason", ""), "at": meta.get("at", ""),
|
|
"files": files, "absent": meta.get("absent", []),
|
|
"versions": {f: version_of(os.path.join(folder, f)) for f in files}})
|
|
return out
|
|
|
|
|
|
def rollback(backup_id):
|
|
"""Put the files of one backup back, after backing up what is there now."""
|
|
bid = os.path.basename(str(backup_id or ""))
|
|
folder = os.path.join(_dir(), "backups", bid)
|
|
if not bid or bid.startswith(".") or not os.path.isdir(folder):
|
|
raise ValueError("no such backup")
|
|
files = [f for f in MANAGED if os.path.isfile(os.path.join(folder, f))]
|
|
if not files:
|
|
raise ValueError("that backup holds no boot files")
|
|
with _Lock():
|
|
_backup(files, f"before rolling back to {bid}")
|
|
_place_all([(os.path.join(folder, f), f, uc.file_hash(os.path.join(folder, f)))
|
|
for f in files])
|
|
log = _installed_log()
|
|
for f in files:
|
|
log.pop(f, None)
|
|
uc.write_json(os.path.join(_dir(), "installed.json"), log)
|
|
return files
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if sys.argv[1:2] == ["stage"]:
|
|
sys.exit(run_stage_job())
|
|
print("usage: python3 -m services.ipxe_update stage", file=sys.stderr)
|
|
sys.exit(2)
|