test(e2e): full virtualized build+boot+unlock QEMU harness

Opt-in via LIM_E2E_QEMU=1. Models the whole process on a virtio VM (the
Pi's USB-gadget net can't be emulated, so it models the software stack,
not the board): builds a LUKS image carrying the real lim initcpio Tor
artifacts, boots it in QEMU rootless, lets the real netconf/tor/dropbear/
encryptssh chain publish the onion, delivers the passphrase over Tor, and
asserts the boot-ok marker on the serial console. Supports a private
offline Tor network via chutney.

The pure command builders (config.qemu_argv/kernel_cmdline/ssh_argv,
qemu_binary), the env parser, and drift guards that keep build_image.sh
aligned with the harness run in the normal suite — no QEMU/root/network.

The build stage needs root; harness.py primes sudo up front, keeps the
credential warm, and reclaims work-dir ownership on every exit path so
the pytest tmp cleanup never trips on root-owned files. QEMU stderr is
captured so an early exit is debuggable, and each teardown step is
fault-isolated so none masks the real error.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

tests/e2e/qemu/harness.py

@@ -0,0 +1,180 @@
+"""Orchestrate the full virtualized unlock: build -> tor net -> boot -> unlock.
+
+Wires the stages together and owns their teardown. Kept thin so each stage
+stays independently testable; the pure command construction lives in
+config.py and is unit-tested without any of this running.
+"""
+
+import os
+import platform
+import subprocess
+import threading
+from dataclasses import dataclass
+from pathlib import Path
+
+from tests.e2e.qemu import boot_unlock, tor_net
+from tests.e2e.qemu.config import QemuSpec
+
+IMAGE_ENV_NAME = "image.env"
+_SUDO_REFRESH_INTERVAL = 60
+
+DEFAULT_PASSPHRASE = "lim-e2e-luks-passphrase"  # noqa: S105 (throwaway test secret)
+DEFAULT_ARCH = "x86_64"
+DEFAULT_SOCKS_PORT = 9052
+
+
+@dataclass
+class HarnessConfig:
+    repo_root: Path
+    work_dir: Path
+    arch: str = DEFAULT_ARCH
+    passphrase: str = DEFAULT_PASSPHRASE
+    chutney_path: Path | None = None
+    chutney_network: str = "networks/basic"
+
+
+def choose_accel(arch: str) -> str:
+    """Use KVM only when the guest arch matches the host and /dev/kvm exists."""
+    host = "x86_64" if platform.machine() in ("x86_64", "AMD64") else platform.machine()
+    if arch == host and Path("/dev/kvm").exists():
+        return "kvm"
+    return "tcg"
+
+
+def _generate_ssh_key(work_dir: Path) -> Path:
+    key_path = work_dir / "unlock_key"
+    subprocess.run(
+        ["ssh-keygen", "-t", "ed25519", "-N", "", "-f", str(key_path), "-q"],
+        check=True,
+    )
+    return key_path
+
+
+class _SudoKeepalive:
+    """Refresh the cached sudo credential so the long build never re-prompts.
+
+    The password is entered once, up front (before the slow tor bootstrap),
+    then a background thread keeps the timestamp warm until stop().
+    """
+
+    def __init__(self) -> None:
+        self._stop = threading.Event()
+        self._thread = threading.Thread(target=self._loop, daemon=True)
+
+    def _loop(self) -> None:
+        while not self._stop.wait(_SUDO_REFRESH_INTERVAL):
+            subprocess.run(["sudo", "-n", "-v"], check=False)
+
+    def start(self) -> None:
+        self._thread.start()
+
+    def stop(self) -> None:
+        self._stop.set()
+        self._thread.join(timeout=5)
+
+
+def _acquire_sudo() -> _SudoKeepalive | None:
+    """Prompt for sudo once now; return a keepalive (None when already root)."""
+    if os.geteuid() == 0:
+        return None
+    print("\n[harness] The build stage needs root — enter your sudo password now.")
+    subprocess.run(["sudo", "-v"], check=True)
+    keepalive = _SudoKeepalive()
+    keepalive.start()
+    return keepalive
+
+
+def _build_image(cfg: HarnessConfig, ssh_key: Path, test_network_conf: Path | None) -> None:
+    script = cfg.repo_root / "tests/e2e/qemu/build_image.sh"
+    env = {
+        **os.environ,
+        "WORK_DIR": str(cfg.work_dir),
+        "REPO_ROOT": str(cfg.repo_root),
+        "ARCH": cfg.arch,
+        "PASSPHRASE": cfg.passphrase,
+        "SSH_PUBKEY": f"{ssh_key}.pub",
+    }
+    if test_network_conf is not None:
+        env["TOR_TEST_NETWORK_CONF"] = str(test_network_conf)
+    # -n: never prompt here; the credential was primed by _acquire_sudo().
+    prefix = [] if os.geteuid() == 0 else ["sudo", "-n", "-E"]
+    subprocess.run([*prefix, "bash", str(script)], env=env, check=True)
+    # Hand the root-built artifacts back so rootless QEMU can read them.
+    _reclaim_work_dir(cfg, required=True)
+
+
+def _reclaim_work_dir(cfg: HarnessConfig, *, required: bool = False) -> None:
+    """Reclaim the root-built work dir for the invoking user via chown.
+
+    Needed before boot so rootless QEMU can read the image, and again in the
+    finally so a failed build never leaves root-owned files that the pytest
+    tmp cleanup then cannot remove.
+    """
+    if os.geteuid() == 0 or not cfg.work_dir.exists():
+        return
+    subprocess.run(
+        ["sudo", "-n", "chown", "-R", str(os.getuid()), str(cfg.work_dir)],
+        check=required,
+    )
+
+
+def _make_tor_net(cfg: HarnessConfig):
+    if cfg.chutney_path is not None:
+        return tor_net.ChutneyNetwork(cfg.chutney_path, cfg.work_dir, cfg.chutney_network)
+    return tor_net.PublicTorClient(cfg.work_dir, DEFAULT_SOCKS_PORT)
+
+
+def _load_spec(cfg: HarnessConfig, ssh_key: Path, socks_port: int) -> QemuSpec:
+    env = boot_unlock.parse_env_file(cfg.work_dir / IMAGE_ENV_NAME)
+    return QemuSpec(
+        arch=cfg.arch,
+        work_dir=cfg.work_dir,
+        image_path=Path(env["IMAGE"]),
+        kernel_path=Path(env["KERNEL"]),
+        initramfs_path=Path(env["INITRAMFS"]),
+        luks_uuid=env["LUKS_UUID"],
+        mapper_name=env["MAPPER_NAME"],
+        onion_address=env["ONION_ADDRESS"],
+        passphrase=cfg.passphrase,
+        ssh_key_path=ssh_key,
+        socks_port=socks_port,
+        accel=choose_accel(cfg.arch),
+    )
+
+
+def run(cfg: HarnessConfig) -> QemuSpec:
+    """Execute every stage; return the QemuSpec on a fully verified unlock.
+
+    A chutney network must exist before the build so the image can be wired
+    to its authorities; a public client only needs to be up before boot.
+    """
+    cfg.work_dir.mkdir(parents=True, exist_ok=True)
+    ssh_key = _generate_ssh_key(cfg.work_dir)
+    # Prime sudo before the slow tor bootstrap so the prompt is the first thing
+    # the operator sees, not a surprise five minutes in.
+    sudo = _acquire_sudo()
+    net = _make_tor_net(cfg)
+    try:
+        print("\n[harness] bootstrapping Tor client (~30-60s)...", flush=True)
+        net.start()
+        print(f"[harness] Tor client ready on socks port {net.socks_port}.", flush=True)
+        print("[harness] building encrypted image "
+              "(pacstrap + mkinitcpio, several minutes)...", flush=True)
+        _build_image(cfg, ssh_key, net.test_network_conf)
+        spec = _load_spec(cfg, ssh_key, net.socks_port)
+        boot_unlock.boot_and_unlock(spec)
+        return spec
+    finally:
+        # Reclaim ownership FIRST so a slow/failing teardown can never leave
+        # root-owned files behind, and isolate each step so none masks the
+        # real error propagating out of the try or skips the others.
+        _reclaim_work_dir(cfg)
+        try:
+            net.stop()
+        except Exception as exc:  # noqa: BLE001 - teardown must not mask the real failure
+            print(f"[harness] tor network teardown failed: {exc}", flush=True)
+        if sudo is not None:
+            try:
+                sudo.stop()
+            except Exception as exc:  # noqa: BLE001 - teardown must not mask the real failure
+                print(f"[harness] sudo keepalive teardown failed: {exc}", flush=True)