From 853a503a3fe5c23cc3755e2b0138afeba115d0ba Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Tue, 21 Jul 2026 18:50:55 +0200 Subject: [PATCH] test(e2e): full virtualized build+boot+unlock QEMU harness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- tests/e2e/qemu/README.md | 85 +++++++++++++ tests/e2e/qemu/__init__.py | 0 tests/e2e/qemu/boot_unlock.py | 164 +++++++++++++++++++++++++ tests/e2e/qemu/build_image.sh | 182 ++++++++++++++++++++++++++++ tests/e2e/qemu/config.py | 132 ++++++++++++++++++++ tests/e2e/qemu/harness.py | 180 +++++++++++++++++++++++++++ tests/e2e/qemu/tor_net.py | 146 ++++++++++++++++++++++ tests/e2e/test_qemu_harness_unit.py | 181 +++++++++++++++++++++++++++ tests/e2e/test_qemu_unlock_e2e.py | 72 +++++++++++ 9 files changed, 1142 insertions(+) create mode 100644 tests/e2e/qemu/README.md create mode 100644 tests/e2e/qemu/__init__.py create mode 100644 tests/e2e/qemu/boot_unlock.py create mode 100755 tests/e2e/qemu/build_image.sh create mode 100644 tests/e2e/qemu/config.py create mode 100644 tests/e2e/qemu/harness.py create mode 100644 tests/e2e/qemu/tor_net.py create mode 100644 tests/e2e/test_qemu_harness_unit.py create mode 100644 tests/e2e/test_qemu_unlock_e2e.py diff --git a/tests/e2e/qemu/README.md b/tests/e2e/qemu/README.md new file mode 100644 index 0000000..7c6687e --- /dev/null +++ b/tests/e2e/qemu/README.md @@ -0,0 +1,85 @@ +# Virtualized end-to-end Tor unlock harness + +This harness models the **entire** encrypted-image process in a virtual +machine: it builds a LUKS image carrying the real `lim` Tor-in-initramfs +artifacts, boots it in QEMU, lets the real `netconf → tor → dropbear → +encryptssh` chain run in early userspace, then unlocks the root over Tor and +proves the real system booted. + +It complements the lightweight, always-rootless +[`test_tor_unlock_e2e.py`](../test_tor_unlock_e2e.py): that one round-trips a +passphrase through Tor against a dropbear *stand-in*; this one runs the real +dropbear, real cryptsetup, real mkinitcpio initramfs, and a real tor daemon in +a booted machine. + +## Why a virtio machine and not an emulated Raspberry Pi + +QEMU's `raspi3b`/`raspi4b` machines do not emulate the Pi's USB-gadget +Ethernet — exactly the network path the Pi images use in the initramfs +(`g_ether`/`smsc95xx`/`lan78xx`). Without early networking there is no Tor and +no unlock, so emulating the literal board is pointless here. Instead we +reproduce the same *software stack* (same HOOKS chain, same tor hook, same +torrc and onion keys) on a QEMU-friendly `virt`/`q35` machine with +`virtio-net`. The only deltas from a real Pi image are the NIC driver and the +CPU architecture. + +## Stages + +| Stage | File | Privilege | +|---|---|---| +| Build encrypted image | `build_image.sh` | **root** (loop, cryptsetup, chroot) | +| Tor network | `tor_net.py` | rootless | +| Boot + unlock | `boot_unlock.py` | rootless (QEMU user-mode net) | +| Orchestration | `harness.py` | mixed (uses `sudo` for the build only) | +| Pure command builders | `config.py` | none — unit-tested offline | + +## Keeping the host rootless + +Only `build_image.sh` needs root (there is no rootless dm-crypt). To keep your +host unprivileged, run the whole harness **inside a throwaway VM** that has +`/dev/kvm`, and drive it from there. QEMU itself, the tor client and the unlock +are rootless: user-mode networking (`-netdev user`) needs no tap device, and a +Tor onion service needs no inbound port forward (its rendezvous is outbound +from both ends). + +When run directly on the host, `harness.py` invokes the build via `sudo -E` +and then `chown`s the artifacts back so rootless QEMU can read them. + +## Requirements + +- `qemu-system-x86_64` (or `-aarch64` for `LIM_E2E_ARCH=aarch64`) +- `arch-install-scripts` (`pacstrap`, `arch-chroot`) +- `cryptsetup`, `mkinitcpio`, `tor`, `openssh`, `ncat` (SOCKS5 ProxyCommand) +- Network access to the public Tor network, **or** `chutney` for a private one +- `/dev/kvm` recommended (TCG works but is slow; aarch64-on-x86 is always TCG) + +## Running + +```bash +# Public Tor network (simplest; non-deterministic, needs internet): +LIM_E2E_QEMU=1 pytest tests/e2e/test_qemu_unlock_e2e.py -v -s + +# Private, offline, deterministic Tor network via chutney: +git clone https://gitlab.torproject.org/tpo/core/chutney +CHUTNEY_PATH=$PWD/chutney LIM_E2E_QEMU=1 \ + pytest tests/e2e/test_qemu_unlock_e2e.py -v -s +``` + +Environment knobs: `LIM_E2E_ARCH` (`x86_64` default, or `aarch64`), +`CHUTNEY_PATH` (enables the private network), `CHUTNEY_PATH` unset → public. + +## Determinism note (chutney) + +For a fully private loop the **guest image** must trust the same authorities as +the host client. `ChutneyNetwork.test_network_conf()` distils the +`TestingTorNetwork`/`DirAuthority` lines (rewriting `127.0.0.1` to the QEMU +user-net host alias `10.0.2.2`) into a file, which `build_image.sh` appends to +the baked-in torrc via `TOR_TEST_NETWORK_CONF`. The offline onion key +generation is unaffected — it never touches the network either way. + +## What runs in CI without any of this + +The pure logic (`config.py`, the env parser, the chutney torrc parsing) and the +drift guards that keep `build_image.sh` aligned with what the harness expects +are covered by [`test_qemu_harness_unit.py`](../test_qemu_harness_unit.py), +which runs in the normal `pytest` suite — no QEMU, root, or network. diff --git a/tests/e2e/qemu/__init__.py b/tests/e2e/qemu/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/e2e/qemu/boot_unlock.py b/tests/e2e/qemu/boot_unlock.py new file mode 100644 index 0000000..3ea232f --- /dev/null +++ b/tests/e2e/qemu/boot_unlock.py @@ -0,0 +1,164 @@ +"""Boot a built image in QEMU and unlock its LUKS root over Tor. + +This is the live half of the harness. QEMU runs rootless (user-mode net); +the passphrase is delivered exactly as an operator would: SSH through Tor to +the initramfs dropbear, piping the passphrase into the encryptssh prompt. +Success is observed on the serial console, not inferred. +""" + +import contextlib +import subprocess +import threading +import time +from pathlib import Path + +from tests.e2e.qemu import config +from tests.e2e.qemu.config import QemuSpec + +# One generous budget covers guest boot + tor bootstrap + onion publish + +# unlock. TCG (no KVM) is slow, so keep it roomy. +UNLOCK_TIMEOUT = 420 +DELIVERY_INTERVAL = 8 +_SSH_ATTEMPT_TIMEOUT = 45 +_POLL_INTERVAL = 2 +_PROMPT_WAIT = 4 # let the encryptssh prompt appear before sending + + +def _serial_contains(spec: QemuSpec, needle: str) -> bool: + log = spec.serial_log + return log.is_file() and needle in log.read_text(errors="replace") + + +def _stream_new_serial(spec: QemuSpec, offset: int) -> int: + """Print serial output written since ``offset``; return the new offset. + + Streams the guest's boot messages (netconf, the tor hook, dropbear, + encryptssh) live so the operator sees the VM working, not a black box. + """ + log = spec.serial_log + if not log.is_file(): + return offset + data = log.read_bytes() + if len(data) > offset: + print(data[offset:].decode(errors="replace"), end="", flush=True) + return len(data) + + +def _serial_tail(spec: QemuSpec, lines: int = 15) -> str: + if not spec.serial_log.is_file(): + return "(no serial output captured)" + tail = spec.serial_log.read_text(errors="replace").splitlines()[-lines:] + return "serial tail:\n " + "\n ".join(tail) + + +def _qemu_stderr_path(spec: QemuSpec) -> Path: + return spec.work_dir / "qemu.stderr.log" + + +def _qemu_stderr_tail(spec: QemuSpec, lines: int = 10) -> str: + path = _qemu_stderr_path(spec) + if not path.is_file() or not path.read_text(errors="replace").strip(): + return "" + tail = path.read_text(errors="replace").splitlines()[-lines:] + return "\nqemu stderr:\n " + "\n ".join(tail) + + +def _deliver_worker(spec: QemuSpec) -> None: + """Feed the passphrase over an SSH-through-Tor session, held open. + + Runs in a background thread so the main loop keeps streaming serial. The + channel is deliberately kept OPEN after sending: closing stdin immediately + (as a plain pipe does) tears the session down before cryptsetup finishes + its argon2 KDF, killing the unlock mid-flight. We wait for the remote to + close it once the boot proceeds and dropbear is torn down. + """ + proc = subprocess.Popen( + config.ssh_argv(spec), + stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + text=True, + ) + try: + time.sleep(_PROMPT_WAIT) + if proc.poll() is None and proc.stdin is not None: + try: + proc.stdin.write(spec.passphrase + "\n") + proc.stdin.flush() + except OSError: + pass + proc.wait(timeout=_SSH_ATTEMPT_TIMEOUT) + except subprocess.TimeoutExpired: + proc.kill() + finally: + if proc.stdin is not None: + with contextlib.suppress(OSError): + proc.stdin.close() + + +def boot_and_unlock(spec: QemuSpec) -> None: + """Boot the image, unlock it over Tor, and confirm the real system booted. + + Returns normally only when the boot-ok marker appears on the serial + console; the passphrase is (re)delivered on each poll until then. Raises + with the serial tail on QEMU death or timeout. + """ + if spec.serial_log.exists(): + spec.serial_log.unlink() + print(f"\n========== booting in QEMU ({spec.arch}, accel={spec.accel}) ==========") + print(f"Target onion: {spec.onion_address}:22 (unlock budget {UNLOCK_TIMEOUT}s)") + print("---- guest serial console ----", flush=True) + # QEMU's own diagnostics (bad -machine/-cpu, /dev/kvm denied, image busy) + # go to stderr; capture them so an early exit is debuggable, not opaque. + qemu_stderr = _qemu_stderr_path(spec).open("wb") + qemu = subprocess.Popen( + config.qemu_argv(spec), + stdout=subprocess.DEVNULL, stderr=qemu_stderr, + ) + try: + deadline = time.monotonic() + UNLOCK_TIMEOUT + offset = 0 + next_delivery = 0.0 + delivery: threading.Thread | None = None + while time.monotonic() < deadline: + offset = _stream_new_serial(spec, offset) + if _serial_contains(spec, config.BOOT_OK_MARKER): + print("\n[harness] boot-ok marker seen — LUKS root unlocked and booted.") + return + if qemu.poll() is not None: + raise RuntimeError( + f"QEMU exited early (code {qemu.returncode}).\n" + f"{_serial_tail(spec)}{_qemu_stderr_tail(spec)}" + ) + # One delivery at a time; each holds its SSH session open in the + # background while we keep streaming serial and watching the marker. + idle = delivery is None or not delivery.is_alive() + if idle and time.monotonic() >= next_delivery: + print(f"\n[harness] delivering passphrase over Tor to " + f"{spec.onion_address}:22 (holding session open) ...", flush=True) + delivery = threading.Thread( + target=_deliver_worker, args=(spec,), daemon=True + ) + delivery.start() + next_delivery = time.monotonic() + DELIVERY_INTERVAL + time.sleep(_POLL_INTERVAL) + raise RuntimeError( + f"Boot-ok marker never appeared within {UNLOCK_TIMEOUT}s.\n" + f"{_serial_tail(spec)}{_qemu_stderr_tail(spec)}" + ) + finally: + qemu.terminate() + try: + qemu.wait(timeout=15) + except subprocess.TimeoutExpired: + qemu.kill() + qemu.wait(timeout=10) + qemu_stderr.close() + + +def parse_env_file(path: Path) -> dict[str, str]: + """Read the KEY=VALUE image.env written by build_image.sh into a dict.""" + result: dict[str, str] = {} + for line in path.read_text().splitlines(): + if "=" in line and not line.startswith("#"): + key, value = line.split("=", 1) + result[key.strip()] = value.strip() + return result diff --git a/tests/e2e/qemu/build_image.sh b/tests/e2e/qemu/build_image.sh new file mode 100755 index 0000000..9280a34 --- /dev/null +++ b/tests/e2e/qemu/build_image.sh @@ -0,0 +1,182 @@ +#!/bin/bash +# Build a QEMU-bootable, LUKS-encrypted image that reproduces lim's Tor +# unlock stack: the REAL initcpio files from lim/configuration/initcpio/, +# the tor hook wired between netconf and dropbear, offline onion keys, and +# a boot-ok marker service. Boots on a virtio machine (we model the stack, +# not the Raspberry Pi board, whose USB-gadget net QEMU cannot emulate). +# +# Needs root (loop device, cryptsetup, chroot). To keep a host rootless, +# run this inside a throwaway VM — see tests/e2e/qemu/README.md. +# +# Required environment: +# WORK_DIR writable scratch directory (outputs land here) +# REPO_ROOT linux-image-manager checkout (for the real initcpio files) +# ARCH x86_64 (default) or aarch64 +# PASSPHRASE LUKS passphrase the harness will send over Tor +# SSH_PUBKEY path to the public key authorized for dropbear +# +# Writes $WORK_DIR/image.env with LUKS_UUID / MAPPER_NAME / KERNEL / INITRAMFS +# / IMAGE for the Python harness to consume. + +set -euo pipefail + +: "${WORK_DIR:?set WORK_DIR}" +: "${REPO_ROOT:?set REPO_ROOT}" +: "${PASSPHRASE:?set PASSPHRASE}" +: "${SSH_PUBKEY:?set SSH_PUBKEY}" +ARCH="${ARCH:-x86_64}" +MAPPER_NAME="cryptroot" + +ROOTFS="$WORK_DIR/rootfs" +IMAGE="$WORK_DIR/image.raw" +INITCPIO_SRC="$REPO_ROOT/lim/configuration/initcpio" + +case "$ARCH" in + x86_64) KERNEL_PKG=linux; CONSOLE=ttyS0 ;; + aarch64) KERNEL_PKG=linux; CONSOLE=ttyAMA0 ;; + *) echo "Unsupported ARCH: $ARCH" >&2; exit 2 ;; +esac + +banner() { printf "\n========== %s ==========\n" "$1"; } + +cleanup() { + set +e + mountpoint -q "$WORK_DIR/mnt" && umount -R "$WORK_DIR/mnt" + [ -e "/dev/mapper/$MAPPER_NAME" ] && cryptsetup close "$MAPPER_NAME" + [ -n "${LOOP:-}" ] && losetup -d "$LOOP" +} +trap cleanup EXIT + +banner "bootstrapping rootfs with pacstrap" +mkdir -p "$ROOTFS" +# -c reuses the host package cache (fast). If a cached package is corrupt, +# clear it first: sudo rm -f /var/cache/pacman/pkg/-*.pkg.tar.zst* +# A file of blank lines on stdin takes the default answer for every provider +# and confirmation prompt (e.g. the `linux` virtual package), keeping the +# bootstrap non-interactive without the SIGPIPE/pipefail pitfalls of `yes`. +printf '\n%.0s' $(seq 100) > "$WORK_DIR/pacman-answers" +pacstrap -c "$ROOTFS" \ + base "$KERNEL_PKG" mkinitcpio cryptsetup \ + busybox tor openssh \ + mkinitcpio-utils mkinitcpio-netconf mkinitcpio-dropbear \ + < "$WORK_DIR/pacman-answers" + +banner "configuring mkinitcpio (virtio modules, tor between netconf/dropbear)" +# No `autodetect`: it narrows the initramfs to the *build host's* modules and +# would strip the virtio drivers the guest VM needs for early networking. +cat > "$ROOTFS/etc/mkinitcpio.conf" < "$ROOTFS/etc/initcpio/install/ptsmount" <<'EOF' +#!/bin/bash +build() { add_runscript; } +help() { echo "Mount devpts so dropbear can allocate PTYs for encryptssh."; } +EOF +cat > "$ROOTFS/etc/initcpio/hooks/ptsmount" <<'EOF' +#!/usr/bin/ash +run_hook() { + mkdir -p /dev/pts + mount -t devpts devpts /dev/pts -o mode=620,gid=5,ptmxmode=666 2>/dev/null + [ -e /dev/ptmx ] || ln -s /dev/pts/ptmx /dev/ptmx +} +EOF + +banner "installing the real lim initcpio files" +install -D -m0644 "$INITCPIO_SRC/tor_install" "$ROOTFS/etc/initcpio/install/tor" +install -D -m0644 "$INITCPIO_SRC/tor_hook" "$ROOTFS/etc/initcpio/hooks/tor" +install -D -m0644 "$INITCPIO_SRC/torrc" "$ROOTFS/etc/tor/initramfs-torrc" + +if [ -n "${TOR_TEST_NETWORK_CONF:-}" ]; then + banner "wiring the image to a private (chutney) Tor network" + cat "$TOR_TEST_NETWORK_CONF" >> "$ROOTFS/etc/tor/initramfs-torrc" +fi + +banner "generating onion keys offline" +install -d -m0700 "$ROOTFS/etc/tor/initramfs-onion" +: > "$WORK_DIR/keygen-torrc" +tor -f "$WORK_DIR/keygen-torrc" --DisableNetwork 1 \ + --DataDirectory "$WORK_DIR/keygen-data" \ + --HiddenServiceDir "$ROOTFS/etc/tor/initramfs-onion" \ + --HiddenServicePort "22 127.0.0.1:22" \ + --SocksPort 0 --RunAsDaemon 0 --Log "notice stderr" & +keygen_pid=$! +for _ in $(seq 1 30); do + [ -s "$ROOTFS/etc/tor/initramfs-onion/hostname" ] && break + sleep 1 +done +kill "$keygen_pid" 2>/dev/null || true +ONION_ADDRESS="$(cat "$ROOTFS/etc/tor/initramfs-onion/hostname")" + +banner "authorizing the ssh key for dropbear" +install -D -m0600 "$SSH_PUBKEY" "$ROOTFS/etc/dropbear/root_key" + +banner "baking the boot-ok marker service" +cat > "$ROOTFS/etc/systemd/system/lim-boot-ok.service" <<'EOF' +[Unit] +Description=Signal a successful LUKS-unlock boot to the serial console +After=multi-user.target +[Service] +Type=oneshot +ExecStart=/bin/sh -c 'echo LIM_UNLOCK_BOOT_OK > /dev/console' +[Install] +WantedBy=multi-user.target +EOF +ln -sf /etc/systemd/system/lim-boot-ok.service \ + "$ROOTFS/etc/systemd/system/multi-user.target.wants/lim-boot-ok.service" +ln -sf "/usr/lib/systemd/system/getty@.service" \ + "$ROOTFS/etc/systemd/system/getty.target.wants/getty@$CONSOLE.service" + +banner "generating the initramfs" +# arch-chroot binds /proc, /sys, /dev itself; no pre-mount needed. +arch-chroot "$ROOTFS" mkinitcpio -P + +banner "creating the LUKS image" +truncate -s 4G "$IMAGE" +LOOP="$(losetup -f --show "$IMAGE")" +# Cap the argon2 memory so the unlock fits in the guest's RAM (see QemuSpec). +printf '%s' "$PASSPHRASE" | cryptsetup luksFormat --type luks2 --batch-mode \ + --pbkdf-memory 262144 "$LOOP" - +LUKS_UUID="$(cryptsetup luksUUID "$LOOP")" +printf '%s' "$PASSPHRASE" | cryptsetup open "$LOOP" "$MAPPER_NAME" - +mkfs.ext4 -q "/dev/mapper/$MAPPER_NAME" + +banner "copying the rootfs into the encrypted volume" +mkdir -p "$WORK_DIR/mnt" +mount "/dev/mapper/$MAPPER_NAME" "$WORK_DIR/mnt" +cp -a "$ROOTFS/." "$WORK_DIR/mnt/" +echo "/dev/mapper/$MAPPER_NAME / ext4 defaults,noatime 0 1" > "$WORK_DIR/mnt/etc/fstab" +# Kernel/initramfs filenames differ per distro (Arch: vmlinuz-linux / +# initramfs-linux.img; Manjaro: vmlinuz-6.1-x86_64 / initramfs-6.1-x86_64.img). +# Discover them instead of hard-coding, and skip the larger fallback image. +KERNEL_IMG="$(find "$ROOTFS/boot" -maxdepth 1 -name 'vmlinuz-*' | sort | head -1)" +INITRAMFS_IMG="$(find "$ROOTFS/boot" -maxdepth 1 -name 'initramfs-*.img' \ + ! -name '*fallback*' | sort | head -1)" +[ -n "$KERNEL_IMG" ] || { echo "No kernel image in $ROOTFS/boot" >&2; exit 1; } +[ -n "$INITRAMFS_IMG" ] || { echo "No initramfs in $ROOTFS/boot" >&2; exit 1; } +echo "Using kernel $KERNEL_IMG and initramfs $INITRAMFS_IMG" +cp "$KERNEL_IMG" "$WORK_DIR/kernel" +cp "$INITRAMFS_IMG" "$WORK_DIR/initramfs" +sync +umount -R "$WORK_DIR/mnt" +cryptsetup close "$MAPPER_NAME" +losetup -d "$LOOP" +LOOP="" + +banner "writing image.env" +cat > "$WORK_DIR/image.env" < Path: + return self.work_dir / "serial.log" + + +def qemu_binary(arch: str) -> str: + """Return the qemu-system-* binary for an arch (raises on unsupported).""" + if arch not in _QEMU_BINARY: + raise ValueError(f"Unsupported arch: {arch}") + return _QEMU_BINARY[arch] + + +def kernel_cmdline(spec: QemuSpec) -> str: + """Kernel command line unlocking the LUKS root over the mkinitcpio hooks. + + ``ip=dhcp`` is what the netconf hook consumes to bring up virtio-net in + early userspace; the Pi images use USB-gadget networking instead, which + QEMU cannot emulate — hence virtio here (we model the stack, not the board). + """ + console = _SERIAL_CONSOLE[spec.arch] + # The netconf hook needs the explicit ip= form + # :::::: — the device + # name is mandatory (as lim's own boot config uses it); a bare "ip=dhcp" + # leaves netconf looping on "SIOCGIFFLAGS: No such device". net.ifnames=0 + # keeps the virtio NIC named eth0. + # The LAST console= owns /dev/console, which the boot-ok marker echoes to + # and QEMU captures via -serial file, so the serial console must come last. + return ( + f"cryptdevice=UUID={spec.luks_uuid}:{spec.mapper_name} " + f"root=/dev/mapper/{spec.mapper_name} rw " + f"ip=::::lim-e2e:eth0:dhcp net.ifnames=0 console=tty0 console={console}" + ) + + +def qemu_argv(spec: QemuSpec) -> list[str]: + """Full QEMU command: direct kernel boot, virtio disk/net, serial to log. + + User-mode networking (``-netdev user``) needs no root and no tap device; + it is enough for a Tor onion service, whose rendezvous is outbound from + both ends, so no inbound host port forward is required. + """ + argv = [qemu_binary(spec.arch)] # validates the arch + net_device = "virtio-net-pci" if spec.arch == "x86_64" else "virtio-net-device" + if spec.arch == "x86_64": + argv += ["-machine", "q35"] + else: + argv += ["-machine", "virt", "-cpu", "cortex-a72"] + if spec.accel == "kvm": + argv += ["-cpu", "host", "-accel", "kvm"] + elif spec.arch == "x86_64": + argv += ["-cpu", "max", "-accel", "tcg"] + argv += [ + "-m", str(spec.memory_mb), + "-kernel", str(spec.kernel_path), + "-initrd", str(spec.initramfs_path), + "-append", kernel_cmdline(spec), + "-drive", f"file={spec.image_path},if=virtio,format=raw", + "-netdev", "user,id=net0", + "-device", f"{net_device},netdev=net0", + "-nographic", + "-serial", f"file:{spec.serial_log}", + "-no-reboot", + ] + return argv + + +def ssh_proxy_command(socks_port: int) -> str: + """ncat-based SOCKS5 ProxyCommand routing the SSH stream through Tor.""" + return f"ncat --proxy 127.0.0.1:{socks_port} --proxy-type socks5 %h %p" + + +def ssh_argv(spec: QemuSpec) -> list[str]: + """Non-interactive SSH into the initramfs dropbear via the onion address. + + ``-tt`` forces a pty so the cryptsetup askpass inside the encryptssh + session reads the passphrase we pipe on stdin. Host-key checking is off + because a throwaway onion has no known_hosts entry. + """ + return [ + "ssh", "-tt", + "-o", "StrictHostKeyChecking=no", + "-o", "UserKnownHostsFile=/dev/null", + "-o", "BatchMode=yes", + "-o", f"ProxyCommand={ssh_proxy_command(spec.socks_port)}", + "-i", str(spec.ssh_key_path), + f"root@{spec.onion_address}", + ] diff --git a/tests/e2e/qemu/harness.py b/tests/e2e/qemu/harness.py new file mode 100644 index 0000000..88894e0 --- /dev/null +++ b/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) diff --git a/tests/e2e/qemu/tor_net.py b/tests/e2e/qemu/tor_net.py new file mode 100644 index 0000000..ffc36d0 --- /dev/null +++ b/tests/e2e/qemu/tor_net.py @@ -0,0 +1,146 @@ +"""Tor network providers for the QEMU unlock harness. + +Two interchangeable ways to give the harness a working SOCKS port that can +reach the guest's onion service: + +* PublicTorClient - a local tor client on the public network (default, both + the guest's baked-in torrc and this client use the real Tor authorities). +* ChutneyNetwork - a private, deterministic, offline Tor network via the + official `chutney` tool. Fully private operation also needs the image + built against the same authorities; test_network_conf() emits the file + build_image.sh consumes for that (see README.md). + +Both expose ``socks_port``, ``start()`` and ``stop()``. +""" + +import subprocess +import time +from pathlib import Path + +BOOTSTRAP_TIMEOUT = 180 +_GUEST_HOST_ALIAS = "10.0.2.2" # QEMU user-net alias for the host's 127.0.0.1 + + +class PublicTorClient: + """A throwaway tor client bootstrapped against the public network.""" + + def __init__(self, work_dir: Path, socks_port: int) -> None: + self.socks_port = socks_port + self.test_network_conf: Path | None = None + self._work_dir = work_dir + self._log = work_dir / "tor-client.log" + self._data = work_dir / "tor-client-data" + self._process: subprocess.Popen | None = None + + def start(self) -> None: + self._data.mkdir(parents=True, exist_ok=True) + self._data.chmod(0o700) + # An empty -f keeps the system /etc/tor/torrc (User tor, ...) out of + # this rootless client, which would otherwise refuse to start. + empty_torrc = self._work_dir / "tor-client-torrc" + empty_torrc.write_text("") + self._process = subprocess.Popen( + [ + "tor", "-f", str(empty_torrc), + "--SocksPort", str(self.socks_port), + "--DataDirectory", str(self._data), + "--Log", f"notice file {self._log}", + ], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + ) + _wait_for_bootstrap(self._log, self._process) + + def stop(self) -> None: + if self._process is None: + return + self._process.terminate() + try: + self._process.wait(timeout=15) + except subprocess.TimeoutExpired: + # Never leak a tor holding the fixed SOCKS port into the next run. + self._process.kill() + self._process.wait(timeout=5) + + +class ChutneyNetwork: + """A private Tor network managed by the chutney tool.""" + + def __init__(self, chutney_path: Path, work_dir: Path, network: str) -> None: + self._chutney = chutney_path + self._network = network + self._work_dir = work_dir + self.socks_port = 0 + self.test_network_conf: Path | None = None + + def _run(self, *args: str) -> None: + subprocess.run( + [str(self._chutney / "chutney"), *args, self._network], + cwd=self._chutney, check=True, + ) + + def start(self) -> None: + self._run("configure") + self._run("start") + self._run("wait_for_bootstrap") + client_torrc = _find_client_torrc(self._chutney / "net" / "nodes") + self.socks_port = _parse_socks_port(client_torrc) + self.test_network_conf = self._emit_network_conf(client_torrc) + + def _emit_network_conf(self, client_torrc: Path) -> Path: + """Distil the authority/test-network lines the guest image needs. + + 127.0.0.1 authority addresses are rewritten to the QEMU user-net host + alias so the guest's initramfs tor can reach them through user-mode + networking. + """ + wanted = ("TestingTorNetwork", "DirAuthority", "DirServer", "AlternateDirAuthority") + lines = [ + line.replace("127.0.0.1", _GUEST_HOST_ALIAS) + for line in client_torrc.read_text().splitlines() + if line.strip().startswith(wanted) + ] + conf = self._work_dir / "tor-test-network.conf" + conf.write_text("\n".join(lines) + "\n") + return conf + + def stop(self) -> None: + self._run("stop") + + +def _wait_for_bootstrap(log_path: Path, process: subprocess.Popen) -> None: + deadline = time.monotonic() + BOOTSTRAP_TIMEOUT + while time.monotonic() < deadline: + if process.poll() is not None: + raise RuntimeError( + f"tor exited early (code {process.returncode}).\n{_log_tail(log_path)}" + ) + if log_path.is_file() and "Bootstrapped 100%" in log_path.read_text(): + return + time.sleep(1) + raise RuntimeError(f"tor client did not bootstrap in time.\n{_log_tail(log_path)}") + + +def _log_tail(log_path: Path, lines: int = 8) -> str: + if not log_path.is_file(): + return "(no tor log written)" + tail = log_path.read_text().splitlines()[-lines:] + return "tor log:\n " + "\n ".join(tail) + + +def _find_client_torrc(nodes_dir: Path) -> Path: + for torrc in sorted(nodes_dir.glob("*/torrc")): + if _parse_socks_port(torrc): + return torrc + raise RuntimeError(f"No chutney client with a SocksPort under {nodes_dir}.") + + +def _parse_socks_port(torrc: Path) -> int: + if not torrc.is_file(): + return 0 + for line in torrc.read_text().splitlines(): + parts = line.split() + if len(parts) >= 2 and parts[0] == "SocksPort" and parts[1].isdigit(): + port = int(parts[1]) + if port: + return port + return 0 diff --git a/tests/e2e/test_qemu_harness_unit.py b/tests/e2e/test_qemu_harness_unit.py new file mode 100644 index 0000000..0fca3f8 --- /dev/null +++ b/tests/e2e/test_qemu_harness_unit.py @@ -0,0 +1,181 @@ +"""Always-on unit tests for the QEMU harness logic. + +No QEMU, root, or network: they exercise the pure command builders and guard +the root build script against drifting away from what the harness expects. +Run as part of the normal suite. +""" + +from pathlib import Path + +import pytest + +from lim import config as lim_config +from tests.e2e.qemu import boot_unlock, config, harness, tor_net +from tests.e2e.qemu.config import EXPECTED_HOOKS_ORDER, QemuSpec + +BUILD_SCRIPT = Path(__file__).resolve().parents[2] / "tests/e2e/qemu/build_image.sh" + + +def _spec(**overrides) -> QemuSpec: + base = { + "arch": "x86_64", + "work_dir": Path("/work"), + "image_path": Path("/work/image.raw"), + "kernel_path": Path("/work/kernel"), + "initramfs_path": Path("/work/initramfs"), + "luks_uuid": "1111-UUID", + "mapper_name": "cryptroot", + "passphrase": "secret", + "ssh_key_path": Path("/work/unlock_key"), + "onion_address": "abcd.onion", + "socks_port": 9052, + } + base.update(overrides) + return QemuSpec(**base) + + +class TestKernelCmdline: + def test_unlocks_the_luks_root_over_dhcp(self): + line = config.kernel_cmdline(_spec()) + assert "cryptdevice=UUID=1111-UUID:cryptroot" in line + assert "root=/dev/mapper/cryptroot" in line + # Explicit device in the ip= form, else netconf loops on "No such device". + assert ":eth0:dhcp" in line + assert "console=ttyS0" in line + + def test_serial_console_is_last_so_marker_reaches_serial_log(self): + line = config.kernel_cmdline(_spec()) + assert line.index("console=tty0") < line.index("console=ttyS0") + + def test_aarch64_uses_amba_serial(self): + assert "console=ttyAMA0" in config.kernel_cmdline(_spec(arch="aarch64")) + + +class TestQemuArgv: + def test_x86_uses_virtio_net_and_direct_kernel_boot(self): + argv = config.qemu_argv(_spec()) + assert argv[0] == "qemu-system-x86_64" + assert "virtio-net-pci,netdev=net0" in argv + assert "-kernel" in argv + assert "/work/kernel" in argv + assert "-initrd" in argv + assert "/work/initramfs" in argv + assert "file=/work/image.raw,if=virtio,format=raw" in argv + assert f"file:{_spec().serial_log}" in argv + assert "user,id=net0" in argv + + def test_aarch64_uses_virt_machine_and_net_device(self): + argv = config.qemu_argv(_spec(arch="aarch64")) + assert argv[0] == "qemu-system-aarch64" + assert "virt" in argv + assert "virtio-net-device,netdev=net0" in argv + + def test_kvm_accel_is_emitted_when_requested(self): + assert "kvm" in config.qemu_argv(_spec(accel="kvm")) + assert "kvm" not in config.qemu_argv(_spec(accel="tcg")) + + def test_unknown_arch_raises(self): + with pytest.raises(ValueError, match="Unsupported arch"): + config.qemu_argv(_spec(arch="riscv64")) + + +class TestQemuBinary: + def test_derives_binary_per_arch(self): + assert config.qemu_binary("x86_64") == "qemu-system-x86_64" + assert config.qemu_binary("aarch64") == "qemu-system-aarch64" + + def test_unknown_arch_raises(self): + with pytest.raises(ValueError, match="Unsupported arch"): + config.qemu_binary("riscv64") + + +class TestSshInvocation: + def test_proxy_command_routes_through_socks5(self): + proxy = config.ssh_proxy_command(9052) + assert "--proxy 127.0.0.1:9052" in proxy + assert "--proxy-type socks5" in proxy + + def test_ssh_argv_targets_onion_via_key_and_proxy(self): + argv = config.ssh_argv(_spec()) + assert "root@abcd.onion" in argv + assert "-i" in argv + assert "/work/unlock_key" in argv + assert any("ProxyCommand=" in part for part in argv) + assert "StrictHostKeyChecking=no" in argv + + +class TestBuildScriptStaysAligned: + """Guards: the root build script must match what the harness assumes.""" + + def _script(self) -> str: + return BUILD_SCRIPT.read_text() + + def test_hooks_place_tor_between_netconf_and_dropbear(self): + hooks_line = next( + line for line in self._script().splitlines() if line.startswith("HOOKS=") + ) + positions = [hooks_line.index(hook) for hook in EXPECTED_HOOKS_ORDER] + assert positions == sorted(positions), hooks_line + + def test_installs_the_real_lim_initcpio_files(self): + script = self._script() + for name in ("tor_install", "tor_hook", "torrc"): + assert f"$INITCPIO_SRC/{name}" in script + assert "etc/dropbear/root_key" in script + + def test_uses_virtio_net_and_emits_boot_marker(self): + script = self._script() + assert "virtio_net" in script + assert config.BOOT_OK_MARKER in script + + def test_mounts_devpts_before_dropbear_for_pty(self): + hooks_line = next( + line for line in self._script().splitlines() if line.startswith("HOOKS=") + ) + assert "ptsmount" in hooks_line + assert hooks_line.index("ptsmount") < hooks_line.index("dropbear") + assert "mount -t devpts" in self._script() + + def test_real_initcpio_source_directory_exists(self): + assert (lim_config.CONFIGURATION_PATH / "initcpio" / "tor_hook").is_file() + + +class TestOrchestratorGlue: + def test_load_spec_reads_image_env(self, tmp_path): + (tmp_path / "image.env").write_text( + "LUKS_UUID=abc-123\n" + "MAPPER_NAME=cryptroot\n" + "ONION_ADDRESS=xyz.onion\n" + "IMAGE=/w/image.raw\n" + "KERNEL=/w/kernel\n" + "INITRAMFS=/w/initramfs\n" + ) + cfg = harness.HarnessConfig(repo_root=Path("/repo"), work_dir=tmp_path) + spec = harness._load_spec(cfg, ssh_key=tmp_path / "k", socks_port=9052) + assert spec.luks_uuid == "abc-123" + assert spec.onion_address == "xyz.onion" + assert spec.image_path == Path("/w/image.raw") + assert spec.socks_port == 9052 + + def test_env_parser_ignores_comments_and_blanks(self, tmp_path): + env = tmp_path / "image.env" + env.write_text("# a comment\n\nKEY=value\n") + assert boot_unlock.parse_env_file(env) == {"KEY": "value"} + + def test_choose_accel_falls_back_to_tcg_for_foreign_arch(self): + assert harness.choose_accel("riscv64") == "tcg" + + +class TestChutneyParsing: + def test_parse_socks_port_picks_first_nonzero(self, tmp_path): + torrc = tmp_path / "torrc" + torrc.write_text("SocksPort 0\nSocksPort 9008\n") + assert tor_net._parse_socks_port(torrc) == 9008 + + def test_find_client_torrc_skips_relays(self, tmp_path): + (tmp_path / "000a").mkdir() + (tmp_path / "000a" / "torrc").write_text("SocksPort 0\n") + (tmp_path / "001c").mkdir() + (tmp_path / "001c" / "torrc").write_text("SocksPort 9010\n") + found = tor_net._find_client_torrc(tmp_path) + assert found == tmp_path / "001c" / "torrc" diff --git a/tests/e2e/test_qemu_unlock_e2e.py b/tests/e2e/test_qemu_unlock_e2e.py new file mode 100644 index 0000000..d13f5b5 --- /dev/null +++ b/tests/e2e/test_qemu_unlock_e2e.py @@ -0,0 +1,72 @@ +"""Full virtualized end-to-end test: build -> boot in QEMU -> unlock over Tor. + +Models the whole process the way lim runs it, on a QEMU-bootable virtio +machine (the Raspberry Pi's USB-gadget net can't be emulated, so we model +the software stack, not the board): + + 1. build a LUKS image carrying the REAL lim initcpio tor artifacts, + 2. boot it in QEMU (rootless, user-mode net), + 3. its initramfs runs the real netconf/tor/dropbear/encryptssh chain and + publishes the onion service, + 4. deliver the LUKS passphrase by SSHing to that onion through Tor, + 5. assert the boot-ok marker appears on the serial console — i.e. the real + root actually unlocked and booted. + +Heavy and privileged (QEMU + a tor network; the build needs root, ideally in +a throwaway VM). Opt-in and skipped unless LIM_E2E_QEMU=1: + + LIM_E2E_QEMU=1 pytest tests/e2e/test_qemu_unlock_e2e.py -v -s + +Environment knobs: LIM_E2E_ARCH (x86_64|aarch64), CHUTNEY_PATH (use a private +chutney network instead of the public one). See tests/e2e/qemu/README.md. +""" + +import os +import shutil +from pathlib import Path + +import pytest + +from tests.e2e.qemu import config as qemu_config +from tests.e2e.qemu.harness import HarnessConfig, run + +REPO_ROOT = Path(__file__).resolve().parents[2] + +# The QEMU binary is arch-specific, so derive it from LIM_E2E_ARCH instead of +# hardcoding x86_64 — otherwise the documented aarch64 run is never reachable. +_ARCH = os.environ.get("LIM_E2E_ARCH", "x86_64") +_REQUIRED = ( + qemu_config.qemu_binary(_ARCH), + "cryptsetup", "pacstrap", "tor", "ssh", "ncat", +) + + +def _missing_tools() -> list[str]: + return [tool for tool in _REQUIRED if shutil.which(tool) is None] + + +pytestmark = pytest.mark.skipif( + os.environ.get("LIM_E2E_QEMU") != "1" or bool(_missing_tools()), + reason=( + "needs LIM_E2E_QEMU=1 and " + + ", ".join(_REQUIRED) + + " (build stage needs root; slow)." + ), +) + + +def test_full_build_boot_and_tor_unlock(tmp_path): + chutney_path = os.environ.get("CHUTNEY_PATH") + cfg = HarnessConfig( + repo_root=REPO_ROOT, + work_dir=tmp_path / "qemu-e2e", + arch=_ARCH, + chutney_path=Path(chutney_path) if chutney_path else None, + ) + + spec = run(cfg) + + # run() only returns after the marker was observed; re-assert for clarity. + assert spec.onion_address.endswith(".onion") + serial = spec.serial_log.read_text(errors="replace") + assert qemu_config.BOOT_OK_MARKER in serial