"""Install Blancco Drive Eraser releases from the webapp, test before switch.

A release is a vendor ISO (or the vendor's USB image, .img or .img.gz, with
the ISO inside its FAT32 partition). What it replaces in WEB_ROOT/blancco:

  * arch/x86_64/airootfs.sfs   - the rootfs; both boot paths fetch it over HTTP
  * vmlinuz-bde-linux,         - Blancco's own kernel + initramfs, used by the
    initramfs-bde-linux.img      BIOS (native) path; the initramfs is patched
                                 so copytoram keeps the network up
  * intel-ucode.img, amd-ucode.img
  * config.img                 - vendor cpio (report key, licence container,
                                 certificates) with OUR preferences written in
  * preferences.xml            - our preferences, served for the UEFI path
  * config-clean.xml           - config.xml from config.img, NULs stripped
                                 (the licence container the UEFI path fetches)
  * blancco-shim.efi           - Blancco's Microsoft-signed shim from the ISO
                                 (EFI/BOOT/bootx64.efi), when the ISO has one

The UEFI path's Ubuntu kernel (vmlinuz-ubuntu*), its initrd (kexec-initrd*)
and kmod tarballs, grubx64.efi, shimx64.efi and blancco-progress.py are NOT
part of a release and are never touched here.

The transformations are the ones scripts/prepare-boot-tools.sh does (same
checks, same results); keep the two in step. The preferences guard is the
important one: our preferences are a merge onto one vendor release's
preferences, and a release whose <version> generation.major differs would
silently lose settings the new schema added or renamed. Staging STOPS there
until a human merges playbook/blancco-preferences.xml onto the new vendor
file and uploads the result.

Flow:
  stage    (background job, python3 -m services.blancco_update stage FILE):
           extract into WEB_ROOT/blancco/staged/<version>/, HTTP-served so a
           test menu entry can boot it;
  test     a menu entry in GetPxeScript.aspx boots the staged files (BIOS:
           Blancco's kernel with archiso_http_srv at the staged dir; UEFI:
           the live Ubuntu kernel line plus blancco.base=, which needs the
           init from playbook/blancco-init.sh as of 2026-10-13);
  activate back up the current release files to
           WEB_ROOT/blancco/backup-<version>-<ts>/ (hard links) and swap each
           staged file in with an atomic rename;
  rollback the same swap from any backup, after backing up what is live.
"""
from __future__ import annotations

import gzip
import lzma
import os
import re
import shutil
import signal
import struct
import subprocess
import sys
import tempfile
import threading
import time
import xml.etree.ElementTree as ET
from datetime import datetime

import config
from services import updates_common as uc

JOB_UNIT = "pxe-webapp-blancco-stage"
# The PXE server address the preferences must point the report share at
# (blancco-init.sh greps for it too).
SERVER = config.BLANCCO_SERVER

# Release path (relative to WEB_ROOT/blancco) <- member of the vendor ISO.
ISO_MEMBERS = {
    "vmlinuz-bde-linux": "arch/boot/x86_64/vmlinuz-bde-linux",
    "initramfs-bde-linux.img": "arch/boot/x86_64/initramfs-bde-linux.img",
    "intel-ucode.img": "arch/boot/intel-ucode.img",
    "amd-ucode.img": "arch/boot/amd-ucode.img",
    "config.img": "arch/boot/config.img",
    "arch/x86_64/airootfs.sfs": "arch/x86_64/airootfs.sfs",
}
ISO_OPTIONAL = {
    "blancco-shim.efi": "EFI/BOOT/bootx64.efi",
}
ISO_VERSION = "arch/version"
ISO_SHA512 = "arch/x86_64/airootfs.sha512"
# Every file a release must have, in swap order: the rootfs first, the
# small files the boot reads first last, so a boot during the swap mostly
# sees one release.
RELEASE_FILES = (
    "arch/x86_64/airootfs.sfs",
    "vmlinuz-bde-linux",
    "initramfs-bde-linux.img",
    "intel-ucode.img",
    "amd-ucode.img",
    "config-clean.xml",
    "preferences.xml",
    "config.img",
)
OPTIONAL_FILES = ("blancco-shim.efi",)
# Kernel file names in the boot menu that a test entry rewrites to the staged
# copy (the Ubuntu kernel and its initrd stay where they are).
_MENU_RELEASE_NAMES = ("vmlinuz-bde-linux", "initramfs-bde-linux.img",
                       "intel-ucode.img", "amd-ucode.img", "config.img")
SOURCE_SUFFIXES = (".iso", ".img", ".img.gz", ".iso.gz")
_SAFE_VERSION = re.compile(r"^[0-9A-Za-z][0-9A-Za-z._-]{0,39}$")
_SAFE_NAME = re.compile(r"^[0-9A-Za-z][0-9A-Za-z._ ()+-]{0,199}$")

TEST_LABEL = "blancco_staged"
TEST_BEGIN = "# BEGIN blancco-staged test entry (PXE webapp, Updates > Blancco)"
TEST_END = "# END blancco-staged test entry"

# Serialises the inline swaps (activate, rollback, menu edits, prefs merge).
_lock = threading.Lock()


class StageBlocked(Exception):
    """Staging cannot finish until a human acts (preferences merge)."""


# --- paths ------------------------------------------------------------------

def blancco_root():
    return os.path.join(config.WEB_ROOT, "blancco")


def staged_root():
    return os.path.join(blancco_root(), "staged")


def _dir(*parts):
    return uc.state_dir("blancco", *parts)


def upload_dir():
    return _dir("uploads")


def job_file():
    return os.path.join(_dir(), "job.json")


def menu_path():
    return os.path.join(config.WEB_ROOT, "Altiris", "iPXE", "GetPxeScript.aspx")


def _vendor_dir(version):
    return os.path.join(_dir("vendor"), version)


def merged_prefs_path(version):
    return os.path.join(_dir("prefs"), f"{version}.xml")


def active_prefs_path():
    """Our preferences as served now (the playbook deploys
    playbook/blancco-preferences.xml here)."""
    return os.path.join(blancco_root(), "preferences.xml")


def _check_version(version):
    if not _SAFE_VERSION.match(str(version or "")):
        raise ValueError(f"bad version name: {version!r}")
    return version


def _ts():
    return datetime.now().strftime("%Y%m%d-%H%M%S")


# --- reading release files --------------------------------------------------

def kernel_version(path):
    """Version string from a bzImage header ('6.12.93-1bde-linux'), or None."""
    try:
        with open(path, "rb") as f:
            head = f.read(0x10000)
    except OSError:
        return None
    if head[0x202:0x206] != b"HdrS":
        return None
    off = struct.unpack_from("<H", head, 0x20E)[0] + 0x200
    text = head[off:off + 200].split(b"\0")[0].decode("ascii", "replace")
    return text.split(" ")[0] or None


def cpio_entries(data):
    """[{name, header, data_off, size}, ...] of a newc cpio, to TRAILER!!!."""
    out, off = [], 0
    while data.startswith((b"070701", b"070702"), off):
        size = int(data[off + 54:off + 62], 16)
        namesize = int(data[off + 94:off + 102], 16)
        name = data[off + 110:off + 110 + namesize - 1].decode("utf-8", "replace")
        data_off = (off + 110 + namesize + 3) & ~3
        if name == "TRAILER!!!":
            break
        out.append({"name": name, "header": off, "data_off": data_off, "size": size})
        off = (data_off + size + 3) & ~3
    return out


def cpio_member(data, name):
    for e in cpio_entries(data):
        if e["name"].lstrip("./") == name:
            return data[e["data_off"]:e["data_off"] + e["size"]]
    return None


def prefs_version(xml):
    """'7.20.0' from <version><generation>/<major>/<minor>, or None."""
    m = re.search(rb"<version>\s*<generation>(\d+)</generation>\s*<major>(\d+)</major>"
                  rb"(?:\s*<minor>(\d+)</minor>)?", xml or b"")
    if not m:
        return None
    return ".".join(x.decode() for x in m.groups() if x is not None)


def _guard_version(xml):
    # Exactly what prepare-boot-tools.sh compares: generation + major.
    m = re.search(rb"<version>\s*<generation>(\d+)</generation>\s*"
                  rb"<major>(\d+)</major>", xml)
    return m.groups() if m else None


def _read(path, limit=None):
    try:
        with open(path, "rb") as f:
            return f.read() if limit is None else f.read(limit)
    except OSError:
        return None


def release_info(folder):
    """What is in a release folder (the live one, a staged one or a backup).
    release.json, written by this module, wins; anything missing is read from
    the files themselves. A release.json whose file fingerprint no longer
    matches (files copied over by hand or by a playbook run) is ignored."""
    meta = uc.read_json(os.path.join(folder, "release.json"), {}) or {}
    if meta.get("fingerprint") and meta["fingerprint"] != _fingerprint(folder):
        meta = {"changed_by_hand": True}
    info = dict(meta)
    files, total, missing = {}, 0, []
    for rel in RELEASE_FILES + OPTIONAL_FILES:
        path = _find_release_file(folder, rel)
        if not path:
            if rel in RELEASE_FILES:
                missing.append(rel)
            continue
        st = os.stat(path)
        files[rel] = {"size": st.st_size, "mtime": st.st_mtime}
        total += st.st_size
    info["files"] = files
    info["missing"] = missing
    info["size"] = total
    if not info.get("kernel"):
        info["kernel"] = kernel_version(os.path.join(folder, "vmlinuz-bde-linux"))
    if not info.get("prefs_version"):
        cfg = _read(os.path.join(folder, "config.img"))
        info["prefs_version"] = prefs_version(cpio_member(cfg or b"", "preferences.xml"))
    if not info.get("version"):
        # A release installed by hand: the vendor's preferences version is
        # the Blancco version (7.20.0 = Drive Eraser 7.20), and ours is
        # merged onto it, so it still says which release is in config.img.
        info["version"] = info.get("prefs_version") or "unknown"
        info["inferred"] = True
    rootfs = files.get("arch/x86_64/airootfs.sfs")
    info["rootfs_mtime"] = rootfs["mtime"] if rootfs else None
    return info


def _find_release_file(folder, rel):
    path = os.path.join(folder, rel)
    if os.path.isfile(path):
        return path
    # Hand-made backups keep airootfs.sfs flat.
    flat = os.path.join(folder, os.path.basename(rel))
    if rel != os.path.basename(rel) and os.path.isfile(flat):
        return flat
    return None


def _fingerprint(folder):
    """Size + mtime of each release file: tells release.json whether the
    files are still the ones it describes (an in-place cp changes both)."""
    out = {}
    for rel in RELEASE_FILES + OPTIONAL_FILES:
        path = _find_release_file(folder, rel)
        if path:
            st = os.stat(path)
            out[rel] = [st.st_size, int(st.st_mtime)]
    return out


def current():
    info = release_info(blancco_root())
    info["path"] = blancco_root()
    return info


def staged():
    out = []
    root = staged_root()
    if not os.path.isdir(root):
        return out
    test = test_entry()
    for name in sorted(os.listdir(root)):
        path = os.path.join(root, name)
        if name.startswith(".") or not os.path.isdir(path) or not _SAFE_VERSION.match(name):
            continue
        info = release_info(path)
        info["name"] = name
        info["status"] = info.get("status") or "unknown"
        info["in_menu"] = bool(test and test.get("version") == name)
        info["merged_prefs"] = os.path.isfile(merged_prefs_path(name))
        out.append(info)
    return out


def backups():
    out = []
    root = blancco_root()
    if not os.path.isdir(root):
        return out
    for name in os.listdir(root):
        path = os.path.join(root, name)
        if not name.startswith("backup-") or not os.path.isdir(path):
            continue
        info = release_info(path)
        info["name"] = name
        info["mtime"] = os.stat(path).st_mtime
        info["created"] = info.get("backed_up") or info["mtime"]
        info["complete"] = not info["missing"]
        out.append(info)
    out.sort(key=lambda b: b["mtime"], reverse=True)
    return out


def source_dirs():
    """Folders a release file may be picked from: our upload folder, and the
    image upload folder (files copied in over SMB or scp)."""
    dirs = [("uploads", upload_dir())]
    if getattr(config, "UPLOAD_DIR", None):
        dirs.append(("image-upload", config.UPLOAD_DIR))
    return dirs


def sources():
    out = []
    for key, folder in source_dirs():
        try:
            names = os.listdir(folder)
        except OSError:
            continue
        for name in sorted(names):
            path = os.path.join(folder, name)
            if not name.lower().endswith(SOURCE_SUFFIXES) or not os.path.isfile(path):
                continue
            st = os.stat(path)
            out.append({"key": key, "name": name, "path": path, "size": st.st_size,
                        "modified": st.st_mtime, "deletable": key == "uploads"})
    return out


def resolve_source(key, name):
    for k, folder in source_dirs():
        if k == key:
            if name != os.path.basename(name) or not name.lower().endswith(SOURCE_SUFFIXES):
                break
            path = os.path.join(folder, name)
            if os.path.isfile(path):
                return path
    raise ValueError("no such release file")


def clean_upload_name(name):
    name = os.path.basename(str(name or "").replace("\\", "/")).strip()
    if not _SAFE_NAME.match(name) or not name.lower().endswith(SOURCE_SUFFIXES):
        raise ValueError("upload a Blancco .iso, .img or .img.gz (letters, digits, . _ - only)")
    return name


def save_upload(stream, name, length=None):
    """Stream an upload to the upload folder (never held in memory). Written
    to a .part file and renamed when complete."""
    name = clean_upload_name(name)
    folder = upload_dir()
    dest = os.path.join(folder, name)
    if length:
        free = shutil.disk_usage(folder).free
        if length > free - (1 << 30):
            raise RuntimeError(f"not enough disk space for {length // (1 << 20)} MB "
                               f"({free // (1 << 20)} MB free)")
    # A part file of its own, so two uploads of the same name cannot
    # interleave their blocks into one file.
    fd, part = tempfile.mkstemp(dir=folder, prefix=f".{name}.", suffix=".part")
    written = 0
    try:
        with os.fdopen(fd, "wb") as f:
            while True:
                block = stream.read(1 << 20)
                if not block:
                    break
                f.write(block)
                written += len(block)
        if length and written != length:
            raise RuntimeError(f"upload incomplete: {written} of {length} bytes")
        if not written:
            raise RuntimeError("empty upload")
        os.chmod(part, 0o644)  # mkstemp makes 0600
        os.replace(part, dest)
    finally:
        if os.path.exists(part):
            os.remove(part)
    return name, written


def delete_upload(name):
    path = os.path.join(upload_dir(), clean_upload_name(name))
    if not os.path.isfile(path):
        raise ValueError("no such upload")
    os.remove(path)


# --- transformations (same as scripts/prepare-boot-tools.sh) ----------------

def check_prefs(ours, vendor, server=SERVER):
    """Raise StageBlocked when our preferences are not a merge onto this
    release's preferences, ValueError when they lack the report share."""
    if _guard_version(ours) != _guard_version(vendor):
        raise StageBlocked(
            f"Our preferences are for Blancco {prefs_version(ours) or '?'}, this release's "
            f"vendor preferences are {prefs_version(vendor) or '?'}. The preferences format "
            "can change between releases, so playbook/blancco-preferences.xml must be merged "
            "onto the new vendor preferences.xml first (download it below), then upload the "
            "merged file to finish staging.")
    for need in (f"<hostname>{server}</hostname>".encode(), b"<path>blancco-reports</path>",
                 b"<auto_backup>true</auto_backup>"):
        if need not in ours:
            raise ValueError(f"our preferences are missing {need.decode()}")


def patch_preferences(vendor, ours, server=SERVER):
    """Vendor preferences.xml (as stored in config.img) with our XML in place
    of theirs, at exactly the vendor file's size. The vendor file is
    <xml> [NUL padding] <!-- filler comment -->: the gap is refilled, or the
    filler comment gives up bytes when ours is longer."""
    check_prefs(ours, vendor, server)
    orig_size = len(vendor)
    xml = ours.strip() + b"\n"
    end = vendor.find(b"</root>")
    cstart = vendor.rfind(b"<!--")
    if end < 0 or cstart < end:
        raise ValueError("vendor preferences.xml layout not recognised")
    gap, tail = vendor[end + 7:cstart], vendor[cstart:]
    room = orig_size - len(xml) - len(tail)
    if room >= 0:
        gap = (b"\x00" if b"\x00" in gap else b"\n") * room
    else:
        cut = -room
        if len(tail) - cut < len(b"<!--  -->\n"):
            raise ValueError(f"preferences need {cut} more bytes than the padding and "
                             "filler comment can give")
        tail = tail[:5] + tail[5 + cut:]
        gap = b""
    out = xml + gap + tail
    if len(out) != orig_size:
        raise ValueError(f"preferences.xml size changed {orig_size} to {len(out)}")
    return out


def patch_config_img(data, ours, server=SERVER):
    """config.img with our preferences written in. The file keeps its size,
    so its bytes are replaced in place and every other member (report key,
    licence container, certificates) and header stays the vendor's."""
    for e in cpio_entries(data):
        if e["name"].lstrip("./") != "preferences.xml":
            continue
        start, size = e["data_off"], e["size"]
        new = patch_preferences(data[start:start + size], ours, server)
        out = bytearray(data)
        out[start:start + size] = new
        if data.startswith(b"070702", e["header"]):
            # crc format: the header carries a byte sum of the file data.
            check = b"%08X" % (sum(new) & 0xFFFFFFFF)
            out[e["header"] + 102:e["header"] + 110] = check
        return bytes(out)
    raise ValueError("config.img has no preferences.xml")


def config_clean(config_img):
    """config.xml from config.img with the NUL padding stripped (what the
    playbook serves as config-clean.xml)."""
    xml = cpio_member(config_img, "config.xml")
    if xml is None:
        raise ValueError("config.img has no config.xml")
    return xml.replace(b"\x00", b"")


def patch_initramfs(data):
    """Disable archiso_pxe_common's network flush after copytoram (BIOS path).
    7.20 prepends an uncompressed early cpio (NIC firmware) to the XZ
    archive: walk its headers to TRAILER!!! and keep it byte for byte."""
    off = 0
    while data.startswith(b"070701", off):
        namesize = int(data[off + 94:off + 102], 16)
        filesize = int(data[off + 54:off + 62], 16)
        name = data[off + 110:off + 110 + namesize - 1]
        off = (((off + 110 + namesize + 3) & ~3) + filesize + 3) & ~3
        if name == b"TRAILER!!!":
            while off < len(data) and data[off] == 0:
                off += 1
    if not data.startswith(b"\xfd7zXZ\x00", off):
        raise ValueError(f"initramfs layout not recognised (no XZ archive at {off})")
    prefix, compressed = data[:off], data[off:]
    raw = lzma.decompress(compressed)
    old = b'"y" ]; then\n            for curif in /sys/class/net'
    new = b'"N" ]; then\n            for curif in /sys/class/net'
    count = raw.count(old)
    if count != 1:
        raise ValueError(f"copytoram network-flush pattern found {count} times, expected 1; "
                         "BIOS PXE boot would lose the network after copytoram")
    raw = raw.replace(old, new, 1)
    return prefix + lzma.compress(raw, format=lzma.FORMAT_XZ, check=lzma.CHECK_CRC32, preset=6)


# --- getting the ISO out of the upload ---------------------------------------

def _open_source(path):
    with open(path, "rb") as f:
        magic = f.read(2)
    return gzip.open(path, "rb") if magic == b"\x1f\x8b" else open(path, "rb")


def _is_iso(f):
    f.seek(0x8001)
    return f.read(5) == b"CD001"


class _Fat:
    """Just enough FAT16/FAT32 to find one file and where its clusters are.
    Works on a gzip stream: reads go forward, the file data in order."""

    def __init__(self, f, base):
        self.f, self.base = f, base
        f.seek(base)
        bs = f.read(512)
        if len(bs) < 512 or bs[510:512] != b"\x55\xaa" or bs[0] not in (0xEB, 0xE9):
            raise ValueError("no FAT boot sector")
        (bps, spc, rsv, nfats, rootents, tot16, _media,
         fsz16) = struct.unpack_from("<HBHBHHBH", bs, 11)
        if bps not in (512, 1024, 2048, 4096) or not spc or spc & (spc - 1) or not nfats:
            raise ValueError("no FAT boot sector")
        fsz = fsz16 or struct.unpack_from("<I", bs, 36)[0]
        tot = tot16 or struct.unpack_from("<I", bs, 32)[0]
        root_secs = (rootents * 32 + bps - 1) // bps
        self.data_off = base + (rsv + nfats * fsz + root_secs) * bps
        self.csize = spc * bps
        clusters = (tot - (rsv + nfats * fsz + root_secs)) // spc
        if clusters < 4085:
            raise ValueError("FAT12 is not supported")
        self.fat32 = clusters >= 65525
        self.root_cluster = struct.unpack_from("<I", bs, 44)[0] if self.fat32 else None
        self.root_region = (base + (rsv + nfats * fsz) * bps, rootents * 32)
        f.seek(base + rsv * bps)
        self.fat = f.read(fsz * bps)
        self.max_cluster = clusters + 1

    def chain(self, cluster):
        out = []
        while 2 <= cluster <= self.max_cluster and len(out) <= self.max_cluster:
            out.append(cluster)
            if self.fat32:
                cluster = struct.unpack_from("<I", self.fat, cluster * 4)[0] & 0x0FFFFFFF
            else:
                cluster = struct.unpack_from("<H", self.fat, cluster * 2)[0]
        return out

    def runs(self, cluster, size):
        """[(offset, length), ...] of a file's data, contiguous clusters merged."""
        out = []
        for c in self.chain(cluster):
            off = self.data_off + (c - 2) * self.csize
            if out and out[-1][0] + out[-1][1] == off:
                out[-1] = (out[-1][0], out[-1][1] + self.csize)
            else:
                out.append((off, self.csize))
        left, trimmed = size, []
        for off, length in out:
            if left <= 0:
                break
            trimmed.append((off, min(length, left)))
            left -= length
        if left > 0:
            raise ValueError("FAT cluster chain shorter than the file")
        return trimmed

    def listdir(self, cluster=None):
        if cluster is None and not self.fat32:
            self.f.seek(self.root_region[0])
            raw = self.f.read(self.root_region[1])
        else:
            raw = b""
            for off, length in self.runs(cluster or self.root_cluster,
                                         len(self.chain(cluster or self.root_cluster)) * self.csize):
                self.f.seek(off)
                raw += self.f.read(length)
        out, lfn = [], {}
        for i in range(0, len(raw) - 31, 32):
            e = raw[i:i + 32]
            if e[0] == 0:
                break
            attr = e[11]
            if e[0] == 0xE5:
                lfn = {}
                continue
            if attr == 0x0F:
                part = (e[1:11] + e[14:26] + e[28:32]).decode("utf-16-le", "ignore")
                lfn[e[0] & 0x1F] = part.split("\x00")[0].replace("\uffff", "")
                continue
            if attr & 0x08:
                lfn = {}
                continue
            if lfn:
                name = "".join(lfn[k] for k in sorted(lfn))
            else:
                base = e[0:8].decode("ascii", "replace").rstrip()
                ext = e[8:11].decode("ascii", "replace").rstrip()
                name = base + ("." + ext if ext else "")
            lfn = {}
            if name in (".", ".."):
                continue
            cluster = (struct.unpack_from("<H", e, 20)[0] << 16) | struct.unpack_from("<H", e, 26)[0]
            out.append({"name": name, "dir": bool(attr & 0x10), "cluster": cluster,
                        "size": struct.unpack_from("<I", e, 28)[0]})
        return out

    def find(self, suffix, depth=3):
        """Files whose name ends with suffix: [(path, entry), ...]."""
        found, todo = [], [("", None, 0)]
        while todo:
            prefix, cluster, level = todo.pop(0)
            for e in self.listdir(cluster):
                path = prefix + e["name"]
                if e["dir"] and level < depth and e["cluster"]:
                    todo.append((path + "/", e["cluster"], level + 1))
                elif not e["dir"] and e["name"].lower().endswith(suffix) and e["size"]:
                    found.append((path, e))
        return found


def _fat_volumes(f):
    """FAT volumes on a disk image: the whole image, else each MBR partition."""
    try:
        yield _Fat(f, 0)
        return
    except ValueError:
        pass
    f.seek(0)
    mbr = f.read(512)
    if len(mbr) < 512 or mbr[510:512] != b"\x55\xaa":
        return
    for i in range(4):
        entry = mbr[446 + 16 * i:462 + 16 * i]
        ptype, start = entry[4], struct.unpack_from("<I", entry, 8)[0]
        if not ptype or not start or ptype == 0xEE:
            continue
        try:
            yield _Fat(f, start * 512)
        except ValueError:
            continue


def _copy(f, dest, runs, total, progress):
    done = 0
    with open(dest, "wb") as out:
        for off, length in runs:
            f.seek(off)
            left = length
            while left:
                block = f.read(min(left, 4 << 20))
                if not block:
                    raise ValueError("source ended early")
                out.write(block)
                left -= len(block)
                done += len(block)
                progress(done, total)
    return done


def iso_from_source(path, work, progress):
    """Path of the vendor ISO for an upload: the file itself, or the ISO
    pulled out of a .gz / USB disk image into work/. Streams; nothing is
    read into memory whole."""
    f = _open_source(path)
    try:
        if _is_iso(f):
            if not isinstance(f, gzip.GzipFile):
                return path, os.path.basename(path)
            dest = os.path.join(work, "source.iso")
            f.seek(0)
            total, done = None, 0
            with open(dest, "wb") as out:
                for block in iter(lambda: f.read(4 << 20), b""):
                    out.write(block)
                    done += len(block)
                    progress(done, total)
            return dest, os.path.basename(path)
        for vol in _fat_volumes(f):
            isos = vol.find(".iso")
            if not isos:
                continue
            # The vendor's USB image carries one, under images/.
            isos.sort(key=lambda x: x[1]["size"], reverse=True)
            name, entry = isos[0]
            free = shutil.disk_usage(work).free
            if entry["size"] > free - (1 << 30):
                raise RuntimeError(f"not enough disk space in {work} for the "
                                   f"{entry['size'] // (1 << 20)} MB ISO")
            dest = os.path.join(work, "source.iso")
            _copy(f, dest, vol.runs(entry["cluster"], entry["size"]), entry["size"], progress)
            return dest, name
        raise ValueError("not an ISO, and no FAT partition with an .iso file in it found "
                         "(extract the ISO with 7z and upload the .iso instead)")
    finally:
        f.close()


def extract_iso(iso, dest):
    """Pull the release members out of the ISO with 7z (as the script does)."""
    members = list(ISO_MEMBERS.values()) + list(ISO_OPTIONAL.values()) + [ISO_VERSION, ISO_SHA512]
    r = subprocess.run(["7z", "x", "-y", "-bd", f"-o{dest}", iso] + members,
                       capture_output=True, text=True)
    if r.returncode not in (0, 1):
        raise RuntimeError(f"7z failed ({r.returncode}): {(r.stderr or r.stdout).strip()[-300:]}")
    missing = [m for m in ISO_MEMBERS.values() if not os.path.isfile(os.path.join(dest, m))]
    if missing:
        raise ValueError("not a Blancco Drive Eraser ISO: missing " + ", ".join(missing))


# --- staging (background job) -----------------------------------------------

class _Progress:
    def __init__(self, source):
        self.data = {"state": "running", "source": os.path.basename(source), "step": "",
                     "percent": None, "message": "", "log": [], "started": uc.now(),
                     "pid": os.getpid()}
        self._last = 0
        self.save()

    def save(self):
        uc.write_json(job_file(), self.data)
        self._last = time.monotonic()

    def step(self, text):
        self.data.update(step=text, percent=None)
        self.log(text)

    def log(self, text):
        self.data["log"] = (self.data["log"] + [f"{datetime.now():%H:%M:%S} {text}"])[-60:]
        self.save()

    def bytes(self, done, total):
        if total:
            self.data["percent"] = round(100.0 * done / total, 1)
        self.data["message"] = f"{done // (1 << 20)} MB" + (f" of {total // (1 << 20)} MB" if total else "")
        if time.monotonic() - self._last > 1:
            self.save()

    def finish(self, state, message, **extra):
        self.data.update(state=state, message=message, finished=uc.now(), percent=None, **extra)
        self.data.pop("pid", None)
        self.log(message)


def job_state():
    data = uc.read_json(job_file(), {}) or {}
    if data.get("state") == "running" and not job_alive(data):
        data["state"] = "failed"
        data["message"] = "the staging job stopped without finishing"
    return data


def job_alive(data=None):
    data = data if data is not None else (uc.read_json(job_file(), {}) or {})
    if data.get("state") != "running":
        return False
    try:
        if uc.job_running(JOB_UNIT):
            return True
    except OSError:
        pass
    pid = data.get("pid")
    if pid and pid == os.getpid():
        return True
    return False


def _refuse_while_staging(what):
    """The job renames whole folders in staged/ (re-staging a version
    replaces its folder), so nothing may read or change them meanwhile."""
    if job_alive():
        raise RuntimeError(f"a staging job is running; {what} when it has finished")


def start_stage(path):
    if job_alive():
        raise RuntimeError("a staging job is already running")
    uc.write_json(job_file(), {"state": "running", "source": os.path.basename(path),
                               "step": "starting", "log": [], "started": uc.now()})
    try:
        uc.start_job(JOB_UNIT, ["services.blancco_update", "stage", path])
    except Exception:
        uc.write_json(job_file(), {})
        raise


def cancel_stage():
    was_running = (uc.read_json(job_file(), {}) or {}).get("state") == "running"
    # systemctl stop waits for the job, which records itself as failed on SIGTERM.
    uc.stop_job(JOB_UNIT)
    data = uc.read_json(job_file(), {}) or {}
    if was_running:
        data.update(state="cancelled", message="cancelled", finished=uc.now())
        uc.write_json(job_file(), data)
    cleanup_partials()


def cleanup_partials():
    """Remove what a stopped job left: staged/.tmp-* and the work folder."""
    root = staged_root()
    if os.path.isdir(root):
        for name in os.listdir(root):
            if name.startswith((".tmp-", ".old-")):
                shutil.rmtree(os.path.join(root, name), ignore_errors=True)
    shutil.rmtree(os.path.join(_dir(), "work"), ignore_errors=True)


def stage(path):
    """Background job: vendor ISO / USB image to WEB_ROOT/blancco/staged/<v>/."""
    progress = _Progress(path)
    root = staged_root()
    os.makedirs(root, exist_ok=True)
    tmp = os.path.join(root, f".tmp-{_ts()}")
    work = os.path.join(_dir(), "work")
    shutil.rmtree(work, ignore_errors=True)
    os.makedirs(work)
    try:
        progress.step("Finding the ISO")
        iso, iso_name = iso_from_source(path, work, progress.bytes)
        progress.log(f"ISO: {iso_name} ({os.path.getsize(iso) // (1 << 20)} MB)")

        progress.step("Extracting release files with 7z")
        os.makedirs(tmp)
        extract_iso(iso, tmp)
        if iso != path:
            os.remove(iso)

        version = (_read(os.path.join(tmp, ISO_VERSION)) or b"").decode("ascii", "replace").strip()
        cfg = _read(os.path.join(tmp, "arch/boot/config.img"))
        vendor_prefs = cpio_member(cfg, "preferences.xml")
        if vendor_prefs is None:
            raise ValueError("config.img has no preferences.xml")
        if not _SAFE_VERSION.match(version):
            version = prefs_version(vendor_prefs) or f"unknown-{_ts()}"
        progress.data["version"] = version
        progress.log(f"Blancco {version}, preferences schema {prefs_version(vendor_prefs)}")

        sha_line = (_read(os.path.join(tmp, ISO_SHA512)) or b"").split()
        if sha_line:
            progress.step("Checking airootfs.sfs against the ISO's sha512")
            got = uc.file_hash(os.path.join(tmp, ISO_MEMBERS["arch/x86_64/airootfs.sfs"]), "sha512")
            if got.lower() != sha_line[0].decode("ascii", "replace").lower():
                raise ValueError("airootfs.sfs does not match airootfs.sha512 (corrupt ISO?)")
            progress.log("airootfs.sfs sha512 OK")

        progress.step("Laying out the release")
        dest = os.path.join(root, f".tmp-{version}-{_ts()}")
        os.makedirs(os.path.join(dest, "arch", "x86_64"))
        for rel, member in list(ISO_MEMBERS.items()) + list(ISO_OPTIONAL.items()):
            src = os.path.join(tmp, member)
            if os.path.isfile(src):
                os.replace(src, os.path.join(dest, rel))
        shutil.rmtree(tmp, ignore_errors=True)
        tmp = dest

        progress.step("Patching the initramfs (keep the network after copytoram)")
        initramfs = os.path.join(dest, "initramfs-bde-linux.img")
        _atomic_write(initramfs, patch_initramfs(_read(initramfs)))

        # Keep the vendor's config.img + preferences for a later merge.
        vdir = _vendor_dir(version)
        os.makedirs(vdir, exist_ok=True)
        shutil.copyfile(os.path.join(dest, "config.img"), os.path.join(vdir, "config.img"))
        _atomic_write(os.path.join(vdir, "preferences.xml"), _vendor_prefs_text(vendor_prefs))

        meta = {"version": version, "kernel": kernel_version(os.path.join(dest, "vmlinuz-bde-linux")),
                "vendor_prefs_version": prefs_version(vendor_prefs), "source": os.path.basename(path),
                "iso": os.path.basename(iso_name), "staged": uc.now(),
                "rootfs_sha512": sha_line[0].decode() if sha_line else None,
                "status": "extracting"}
        uc.write_json(os.path.join(dest, "release.json"), meta)

        final = os.path.join(root, version)
        if os.path.exists(final):
            old = os.path.join(root, f".old-{version}-{_ts()}")
            os.replace(final, old)
            os.replace(dest, final)
            shutil.rmtree(old, ignore_errors=True)
            progress.log(f"Replaced the earlier staged {version}")
        else:
            os.replace(dest, final)
        tmp = None
        _fix_modes(final)

        progress.step("Writing our preferences into config.img")
        try:
            apply_prefs(version)
        except StageBlocked as e:
            progress.finish("blocked", str(e), version=version)
            return 2
        progress.finish("done", f"Blancco {version} staged and ready to test", version=version)
        return 0
    except (Exception, SystemExit) as e:
        msg = str(e) if not isinstance(e, SystemExit) else "stopped"
        progress.finish("failed", f"Staging failed: {msg}")
        return 1
    finally:
        if tmp:
            shutil.rmtree(tmp, ignore_errors=True)
        shutil.rmtree(work, ignore_errors=True)


def _vendor_prefs_text(vendor):
    """The vendor preferences without the NUL padding and filler comment,
    as a file a human can merge against."""
    end = vendor.find(b"</root>")
    return (vendor[:end + 7] if end >= 0 else vendor.replace(b"\x00", b"")).rstrip() + b"\n"


def vendor_prefs_file(version):
    path = os.path.join(_vendor_dir(_check_version(version)), "preferences.xml")
    return path if os.path.isfile(path) else None


def apply_prefs(version):
    """Write our preferences into a staged release: config.img (vendor
    copy + ours), preferences.xml, config-clean.xml. Our preferences are the
    merged file uploaded for this version, else the ones served now.
    Raises StageBlocked when they fail the version guard."""
    _check_version(version)
    folder = os.path.join(staged_root(), version)
    meta_path = os.path.join(folder, "release.json")
    meta = uc.read_json(meta_path, {}) or {}
    vendor_cfg = _read(os.path.join(_vendor_dir(version), "config.img"))
    if vendor_cfg is None:
        raise ValueError(f"no vendor config.img kept for {version}; stage it again")
    merged = merged_prefs_path(version)
    src = merged if os.path.isfile(merged) else active_prefs_path()
    ours = _read(src)
    if not ours:
        raise ValueError(f"no preferences at {src}")
    # The site's business location and Wi-Fi (Updates > Blancco, site
    # settings) go into every release, so an update keeps them.
    from services import blancco_prefs
    try:
        ours = blancco_prefs.apply(ours)
    except ValueError as e:
        meta.update(status="failed", error=f"our preferences ({src}): {e}", prefs_source=src)
        uc.write_json(meta_path, meta)
        raise
    try:
        cfg = patch_config_img(vendor_cfg, ours)
    except StageBlocked as e:
        meta.update(status="blocked", blocked=str(e), prefs_source=src)
        uc.write_json(meta_path, meta)
        raise
    except ValueError as e:
        # Not a merge problem (no report share, too long for the vendor
        # file): the row says why instead of sitting at "extracting".
        meta.update(status="failed", error=f"our preferences ({src}): {e}", prefs_source=src)
        uc.write_json(meta_path, meta)
        raise
    _atomic_write(os.path.join(folder, "config.img"), cfg)
    _atomic_write(os.path.join(folder, "preferences.xml"), ours)
    _atomic_write(os.path.join(folder, "config-clean.xml"), config_clean(cfg))
    meta.pop("blocked", None)
    meta.pop("error", None)
    meta.update(status="ready", prefs_version=prefs_version(ours), prefs_source=src,
                ready=uc.now())
    uc.write_json(meta_path, meta)
    _fix_modes(folder)
    return meta


def save_merged_prefs(version, data):
    """A human-merged preferences.xml for a staged version: check it, keep it,
    and finish staging with it."""
    _check_version(version)
    _refuse_while_staging("upload the merged preferences")
    if not os.path.isdir(os.path.join(staged_root(), version)):
        raise ValueError(f"{version} is not staged")
    try:
        ET.fromstring(data.replace(b"\x00", b""))
    except ET.ParseError as e:
        raise ValueError(f"not well-formed XML: {e}")
    vendor = _read(os.path.join(_vendor_dir(version), "preferences.xml")) or b""
    if _guard_version(data) != _guard_version(vendor):
        raise ValueError(f"the merged file says {prefs_version(data) or 'no version'}, the "
                         f"release's preferences are {prefs_version(vendor)}: merge onto "
                         "the new vendor file, including its <version> block")
    vendor_cfg = _read(os.path.join(_vendor_dir(version), "config.img"))
    if vendor_cfg is None:
        raise ValueError(f"no vendor config.img kept for {version}; stage it again")
    # Every other check (report share, fits the vendor file) before the file
    # is kept: a stored merged file wins over the live preferences from then
    # on, also for a later re-stage of this version.
    patch_config_img(vendor_cfg, data)
    with _lock:
        _atomic_write(merged_prefs_path(version), data)
        return apply_prefs(version)


def delete_staged(version):
    _check_version(version)
    _refuse_while_staging("delete it")
    test = test_entry()
    if test and test.get("version") == version:
        raise RuntimeError(f"the test menu entry boots {version}; remove it first")
    path = os.path.join(staged_root(), version)
    if not os.path.isdir(path):
        raise ValueError(f"{version} is not staged")
    shutil.rmtree(path)
    shutil.rmtree(_vendor_dir(version), ignore_errors=True)


# --- file helpers -----------------------------------------------------------

def _atomic_write(path, data):
    tmp = f"{path}.tmp-{os.getpid()}"
    with open(tmp, "wb") as f:
        f.write(data)
    os.chmod(tmp, 0o644)
    os.replace(tmp, path)


def _fix_modes(folder):
    """Apache must read everything: dirs 755, files 644."""
    for dirpath, dirnames, filenames in os.walk(folder):
        os.chmod(dirpath, 0o755)
        for name in filenames:
            os.chmod(os.path.join(dirpath, name), 0o644)


def _link_or_copy(src, dst):
    """Hard link src at dst (same filesystem: instant, no space), else copy."""
    os.makedirs(os.path.dirname(dst), exist_ok=True)
    try:
        os.link(src, dst)
    except OSError:
        shutil.copy2(src, dst)


def _place(src, dst):
    """Put src's content at dst atomically: link to a temp name, rename."""
    tmp = os.path.join(os.path.dirname(dst), f".{os.path.basename(dst)}.new-{os.getpid()}")
    if os.path.lexists(tmp):
        os.remove(tmp)
    _link_or_copy(src, tmp)
    os.replace(tmp, dst)


def _backup_live(reason, files):
    """Hard-link the live copies of files into backup-<version>-<ts>/."""
    live = current()
    label = re.sub(r"[^0-9A-Za-z._-]", "_", str(live.get("version") or "unknown"))
    base = name = f"backup-{label}-{_ts()}"
    n = 1
    while os.path.exists(os.path.join(blancco_root(), name)):
        n += 1
        name = f"{base}-{n}"
    folder = os.path.join(blancco_root(), name)
    os.makedirs(folder)
    saved = []
    for rel in files:
        src = os.path.join(blancco_root(), rel)
        if os.path.isfile(src):
            _link_or_copy(src, os.path.join(folder, rel))
            saved.append(rel)
    meta = {k: live.get(k) for k in ("version", "kernel", "prefs_version", "source", "staged",
                                     "activated", "inferred") if live.get(k) is not None}
    meta.update(backed_up=uc.now(), reason=reason, files_saved=saved)
    uc.write_json(os.path.join(folder, "release.json"), meta)
    _fix_modes(folder)
    return name


def _swap_in(folder, files):
    """Place each file from folder into the live release. The callers undo a
    failure by swapping the backup they just took back in."""
    for rel in files:
        _place(_find_release_file(folder, rel), os.path.join(blancco_root(), rel))


def _write_live_meta(meta):
    meta["fingerprint"] = _fingerprint(blancco_root())
    path = os.path.join(blancco_root(), "release.json")
    uc.write_json(path, meta)
    os.chmod(path, 0o644)


def activate(version):
    """Make a staged release live. Returns (backup name, files swapped,
    test entry note): the note is True when the release's test entry was
    removed from the menu, or the error text when that edit failed (the
    release is live either way)."""
    _check_version(version)
    _refuse_while_staging("activate")
    folder = os.path.join(staged_root(), version)
    with _lock:
        meta = uc.read_json(os.path.join(folder, "release.json"), {}) or {}
        if meta.get("status") != "ready":
            raise RuntimeError(f"{version} is not ready to activate ({meta.get('status') or 'unknown'})")
        info = release_info(folder)
        if info["missing"]:
            raise RuntimeError(f"{version} is missing " + ", ".join(info["missing"]))
        files = list(RELEASE_FILES) + [r for r in OPTIONAL_FILES if r in info["files"]]
        backup = _backup_live(f"before activating {version}", files)
        try:
            _swap_in(folder, files)
        except Exception as e:
            _swap_in(os.path.join(blancco_root(), backup),
                     [r for r in files if _find_release_file(os.path.join(blancco_root(), backup), r)])
            raise RuntimeError(f"activation failed and was undone: {e}")
        meta.update(activated=uc.now(), backup=backup)
        _write_live_meta(meta)
        # The staged copies are hard links of what is live now; the folder
        # goes so a later edit there can never reach the live files.
        shutil.rmtree(folder, ignore_errors=True)
        removed_entry = False
        test = test_entry()
        if test and test.get("version") == version:
            try:
                _remove_test_entry_locked()
                removed_entry = True
            except (OSError, RuntimeError) as e:
                removed_entry = f"the test entry is still in the menu ({e}); remove it"
        return backup, files, removed_entry


def rollback(name):
    """Put a backup's files back live (after backing up what is live now)."""
    if not re.match(r"^backup-[0-9A-Za-z._-]+$", name or ""):
        raise ValueError("bad backup name")
    folder = os.path.join(blancco_root(), name)
    with _lock:
        if not os.path.isdir(folder):
            raise ValueError("no such backup")
        info = release_info(folder)
        if info["missing"]:
            raise RuntimeError(f"{name} is incomplete, missing " + ", ".join(info["missing"]))
        files = list(RELEASE_FILES) + [r for r in OPTIONAL_FILES if r in info["files"]]
        backup = _backup_live(f"before rolling back to {name}", files)
        try:
            _swap_in(folder, files)
        except Exception as e:
            _swap_in(os.path.join(blancco_root(), backup),
                     [r for r in files if _find_release_file(os.path.join(blancco_root(), backup), r)])
            raise RuntimeError(f"rollback failed and was undone: {e}")
        meta = {k: v for k, v in info.items() if k in ("version", "kernel", "prefs_version",
                                                         "source", "staged", "inferred")}
        meta.update(restored_from=name, activated=uc.now(), backup=backup)
        _write_live_meta(meta)
        try:
            _release_backup(folder, files)
        except OSError:
            pass  # the rollback itself is done; a backup left whole is harmless
        return backup, files


def _release_backup(folder, files):
    """After a rollback the backup's files ARE the live files (hard links).
    Drop the backup's names for them, so an in-place write to a live file
    (a playbook run's cp -r boot-tools/blancco/*, or a cp by hand) cannot
    also change a backup. Nothing is lost: what is live is backed up again
    the next time something replaces it."""
    for rel in files:
        path = _find_release_file(folder, rel)
        if path:
            os.remove(path)
    meta_path = os.path.join(folder, "release.json")
    left = [os.path.join(d, n) for d, _, ns in os.walk(folder) for n in ns]
    if all(p == meta_path for p in left):
        shutil.rmtree(folder, ignore_errors=True)
    else:
        # A hand-made backup with extra files (preferences.save, ...): keep
        # those, say what happened to the rest.
        uc.write_json(meta_path, {"reason": f"restored to live {uc.now()}; only the files "
                                            "that are not part of a release remain"})


def delete_backup(name):
    if not re.match(r"^backup-[0-9A-Za-z._-]+$", name or ""):
        raise ValueError("bad backup name")
    folder = os.path.join(blancco_root(), name)
    if not os.path.isdir(folder):
        raise ValueError("no such backup")
    shutil.rmtree(folder)


# --- test entry in the iPXE menu --------------------------------------------

def _read_menu():
    path = menu_path()
    # surrogateescape: any byte a hand edit left in the menu round-trips.
    with open(path, newline="", encoding="utf-8", errors="surrogateescape") as f:
        text = f.read()
    nl = "\r\n" if "\r\n" in text else "\n"
    return path, text, nl


def _write_menu(path, text, tag):
    """Back up the menu next to itself, then replace it atomically with the
    same owner and mode."""
    st = os.stat(path)
    backup = f"{path}.bak-{_ts()}-{tag}"
    shutil.copy2(path, backup)
    tmp = f"{path}.tmp-{os.getpid()}"
    try:
        with open(tmp, "w", newline="", encoding="utf-8", errors="surrogateescape") as f:
            f.write(text)
        os.chmod(tmp, st.st_mode & 0o7777)
        try:
            os.chown(tmp, st.st_uid, st.st_gid)
        except PermissionError:
            pass
        os.replace(tmp, path)
    finally:
        if os.path.exists(tmp):
            os.remove(tmp)
    return backup


def _block(lines, label):
    """The kernel line and the initrd lines right after it, in a menu label's
    section (up to the next label)."""
    try:
        i = lines.index(f":{label}")
    except ValueError:
        return None
    kernel, initrds = None, []
    for line in lines[i + 1:]:
        if line.startswith(":"):
            break
        if kernel is None:
            if line.startswith("kernel "):
                kernel = line
            continue
        if line.startswith("initrd "):
            initrds.append(line)
        else:
            break
    return (kernel, initrds) if kernel else None


def _retarget(line, version, goto):
    """A menu line pointed at the staged release files, failing to goto."""
    base = f"blancco/staged/{version}/"
    names = "|".join(re.escape(n) for n in _MENU_RELEASE_NAMES)
    line = re.sub(r"(http://[^/\s]+/)blancco/(?=(?:%s)(?:\s|$))" % names, r"\g<1>" + base, line)
    line = re.sub(r"(archiso_http_srv=http://[^/\s]+/)blancco/", r"\g<1>" + base, line)
    line = re.sub(r"\s*\|\|\s*goto\s+\S+\s*$", "", line)
    return f"{line} || goto {goto}"


def test_entry():
    """{version, item, efi} of the staged test entry in the menu, or None.
    efi is False when the entry was added with its UEFI branch off."""
    try:
        _, text, _ = _read_menu()
    except OSError:
        return None
    m = re.search(r"^item\s+%s\s+.*$" % TEST_LABEL, text, re.M)
    if not m and TEST_BEGIN not in text:
        return None
    v = re.search(r"blancco\.base=/blancco/staged/([0-9A-Za-z._-]+)", text) or \
        re.search(r"/blancco/staged/([0-9A-Za-z._-]+)/", text)
    return {"version": v.group(1) if v else None, "item": m.group(0).strip() if m else None,
            "efi": _EFI_OFF not in text}


_EFI_OFF = "echo UEFI test boot is off:"


def build_test_entry(text, version, efi_ok=True):
    """(menu text with the test entry added, the entry's lines). efi_ok
    False: the UEFI initrd's init does not read blancco.base=, so the UEFI
    branch would boot the LIVE release while the screen says STAGED; it
    says so and goes back to the menu instead."""
    nl = "\r\n" if "\r\n" in text else "\n"
    lines = text.split(nl)
    efi = _block(lines, "blancco_efi")
    bios = _block(lines, "blancco_bios")
    if not efi:
        raise RuntimeError("the menu has no :blancco_efi kernel line to copy")
    items = [i for i, l in enumerate(lines) if re.match(r"^item\s+blancco\s", l)]
    if len(items) != 1:
        raise RuntimeError("cannot find the menu's 'item blancco' line")
    fail = f"{TEST_LABEL}_failed"
    kernel = _retarget(efi[0], version, fail)
    kernel = kernel.replace(f" || goto {fail}", f" blancco.base=/blancco/staged/{version} || goto {fail}")
    block = [
        TEST_BEGIN,
        f":{TEST_LABEL}",
        f"echo Loading STAGED Blancco {version} for testing (not the active release)...",
        f"iseq ${{platform}} pcbios && goto {TEST_LABEL}_bios ||",
    ]
    if efi_ok:
        block += [kernel] + [_retarget(l, version, fail) for l in efi[1]] + [f"boot || goto {fail}"]
    else:
        block += [
            f"{_EFI_OFF} the UEFI initrd predates blancco.base= and would boot the LIVE release.",
            "echo Rebuild the initrd from playbook/blancco-init.sh and add the test entry again,",
            "echo or test this release on a legacy BIOS boot.",
            "prompt --timeout 30000 Press any key to return to the menu... ||",
            "goto menu",
        ]
    block += [f":{TEST_LABEL}_bios"]
    if bios:
        block += [_retarget(bios[0], version, fail)] + [_retarget(l, version, fail) for l in bios[1]]
        block += [f"boot || goto {fail}"]
    else:
        block += ["echo No legacy BIOS Blancco entry in this menu to copy."]
    block += [
        f":{fail}",
        "echo",
        f"echo Staged Blancco {version} boot FAILED: ${{errno}}",
        "prompt --timeout 30000 Press any key to return to the menu... ||",
        "goto menu",
        TEST_END,
    ]
    item = f"item {TEST_LABEL:<20}Blancco Drive Eraser {version} (STAGED - TEST)"
    lines.insert(items[0] + 1, item)
    # At the end of the script, where nothing falls into it (a guard jump
    # inside the markers when the script's last line would fall through).
    while lines and not lines[-1].strip():
        lines.pop()
    last = lines[-1].strip() if lines else ""
    if not re.match(r"^(exit|goto\s|reboot|shell|poweroff)", last):
        block.insert(1, "goto menu")
    lines += [""] + block + [""]
    return nl.join(lines), block


def add_test_entry(version):
    _check_version(version)
    folder = os.path.join(staged_root(), version)
    meta = uc.read_json(os.path.join(folder, "release.json"), {}) or {}
    _refuse_while_staging("add the test entry")
    if meta.get("status") != "ready":
        raise RuntimeError(f"{version} is not ready to test ({meta.get('status') or 'not staged'})")
    efi_ok = initrd_supports_base()["state"] != "no"
    with _lock:
        path, text, _ = _read_menu()
        if test_entry():
            text = _strip_test_entry(text)
        new, _ = build_test_entry(text, version, efi_ok)
        return _write_menu(path, new, "pre-blancco-staged")


def _strip_test_entry(text):
    nl = "\r\n" if "\r\n" in text else "\n"
    lines = text.split(nl)
    out, skip = [], False
    for line in lines:
        if line == TEST_BEGIN:
            skip = True
            # The blank line added before the block.
            if out and not out[-1].strip():
                out.pop()
            continue
        if skip:
            if line == TEST_END:
                skip = False
            continue
        if re.match(r"^item\s+%s\s" % TEST_LABEL, line):
            continue
        out.append(line)
    if skip:
        raise RuntimeError("test entry has no end marker; edit the menu by hand")
    return nl.join(out)


def _remove_test_entry_locked():
    path, text, _ = _read_menu()
    new = _strip_test_entry(text)
    if new == text:
        return None
    return _write_menu(path, new, "pre-blancco-staged-remove")


def remove_test_entry():
    with _lock:
        return _remove_test_entry_locked()


def menu_efi_initrd():
    """File name of the UEFI path's init (kexec-initrd*.img) in the menu."""
    try:
        _, text, nl = _read_menu()
    except OSError:
        return None
    block = _block(text.split(nl), "blancco_efi")
    for line in (block[1] if block else []):
        m = re.search(r"/blancco/(kexec-initrd[^\s/]*\.img)", line)
        if m:
            return m.group(1)
    return None


def initrd_supports_base():
    """Whether the UEFI initrd's init knows blancco.base= (else the UEFI test
    boot gets the ACTIVE release). Cached by file size + mtime."""
    name = menu_efi_initrd()
    if not name:
        return {"state": "unknown", "file": None, "detail": "no :blancco_efi initrd in the menu"}
    path = os.path.join(blancco_root(), name)
    try:
        st = os.stat(path)
    except OSError:
        return {"state": "unknown", "file": name, "detail": "file not found"}
    cache_path = os.path.join(_dir(), "initrd-check.json")
    key = f"{name}:{st.st_size}:{int(st.st_mtime)}"
    cache = uc.read_json(cache_path, {}) or {}
    if cache.get("key") == key:
        return cache["result"]
    result = {"state": "unknown", "file": name, "detail": "not gzip"}
    try:
        with open(path, "rb") as f:
            magic = f.read(2)
        if magic == b"\x1f\x8b":
            found, tail = False, b""
            with gzip.open(path, "rb") as f:
                for block in iter(lambda: f.read(1 << 20), b""):
                    if b"blancco.base=" in tail + block:
                        found = True
                        break
                    tail = block[-32:]
            result = {"state": "yes" if found else "no", "file": name,
                      "detail": "init reads blancco.base=" if found else
                      "init predates blancco.base= (rebuild the initrd from playbook/blancco-init.sh)"}
    except (OSError, EOFError, gzip.BadGzipFile) as e:
        result = {"state": "unknown", "file": name, "detail": str(e)}
    uc.write_json(cache_path, {"key": key, "result": result})
    return result


# --- entry point for the background job -------------------------------------

def _on_term(signum, frame):
    raise SystemExit(143)


def main(argv):
    if len(argv) == 2 and argv[0] == "stage":
        signal.signal(signal.SIGTERM, _on_term)
        return stage(argv[1])
    print("usage: python3 -m services.blancco_update stage FILE", file=sys.stderr)
    return 64


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))