test(e2e): Debian QEMU build+boot+unlock harness and deterministic direct transport

Extend the QEMU e2e to cover the initramfs-tools backend and add a
deterministic unlock transport that avoids the flaky public-Tor onion
round-trip inside QEMU.

- build_image_debian.sh: debootstrap Bookworm, install the real
  lim/configuration/initramfs-tools hooks, LUKS + cryptsetup-initramfs +
  dropbear-initramfs, offline onion keys, boot-ok marker; same image.env
  contract as build_image.sh.
- config.py: QemuSpec gains os_family / unlock_command / direct_ssh_port;
  Debian cmdline uses root=/dev/mapper (crypttab-baked, no cryptdevice=);
  direct_ssh_port adds hostfwd to guest dropbear and a plain-SSH target.
- harness.py: unlock_transport="direct" default, _NullNet, per-OS build
  script + unlock command, up-front sudo priming with keepalive.
- boot_unlock.py: background delivery worker holds the SSH session open;
  direct vs tor target and initial delay.
- test_qemu_harness_unit.py: always-on guards for the Debian/direct branches;
  test_qemu_unlock_e2e.py parameterized by ARCH/OS/TRANSPORT env.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Kevin Veen-Birkenbach
2026-07-22 17:03:16 +02:00
parent 9b7a34989d
commit e7712880f1
7 changed files with 427 additions and 69 deletions

View File

@@ -17,8 +17,13 @@ from tests.e2e.qemu.config import QemuSpec
# One generous budget covers guest boot + tor bootstrap + onion publish + # One generous budget covers guest boot + tor bootstrap + onion publish +
# unlock. TCG (no KVM) is slow, so keep it roomy. # unlock. TCG (no KVM) is slow, so keep it roomy.
UNLOCK_TIMEOUT = 420 UNLOCK_TIMEOUT = 480
DELIVERY_INTERVAL = 8 DELIVERY_INTERVAL = 8
# Hold off the first delivery until the guest onion has had time to publish.
# A connect to a not-yet-published onion makes the client Tor cache the failed
# descriptor fetch and back off, so early hammering can lock us out for
# minutes; waiting until the onion is up means the first fetch succeeds.
INITIAL_DELIVERY_DELAY = 100
_SSH_ATTEMPT_TIMEOUT = 45 _SSH_ATTEMPT_TIMEOUT = 45
_POLL_INTERVAL = 2 _POLL_INTERVAL = 2
_PROMPT_WAIT = 4 # let the encryptssh prompt appear before sending _PROMPT_WAIT = 4 # let the encryptssh prompt appear before sending
@@ -63,18 +68,49 @@ def _qemu_stderr_tail(spec: QemuSpec, lines: int = 10) -> str:
return "\nqemu stderr:\n " + "\n ".join(tail) return "\nqemu stderr:\n " + "\n ".join(tail)
def _ssh_log_path(spec: QemuSpec) -> Path:
return spec.work_dir / "ssh-delivery.log"
def _ssh_log_tail(spec: QemuSpec, lines: int = 15) -> str:
path = _ssh_log_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 "\nssh delivery log:\n " + "\n ".join(tail)
def _client_tor_tail(spec: QemuSpec, lines: int = 20) -> str:
"""Return the host Tor client's HS log — onion descriptor fetch/rendezvous."""
path = spec.work_dir / "tor-client-hs.log"
if not path.is_file() or not path.read_text(errors="replace").strip():
return ""
lines_out = [
line
for line in path.read_text(errors="replace").splitlines()
if any(k in line.lower() for k in ("desc", "rend", "intro", "hs_", "onion"))
][-lines:]
return "\nclient tor HS log:\n " + "\n ".join(lines_out)
def _deliver_worker(spec: QemuSpec) -> None: def _deliver_worker(spec: QemuSpec) -> None:
"""Feed the passphrase over an SSH-through-Tor session, held open. """Feed the passphrase over an SSH-through-Tor session, held open.
Runs in a background thread so the main loop keeps streaming serial. The Runs in a background thread so the main loop keeps streaming serial. The
channel is deliberately kept OPEN after sending: closing stdin immediately channel is deliberately kept OPEN after sending: closing stdin immediately
(as a plain pipe does) tears the session down before cryptsetup finishes (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 its argon2 KDF, killing the unlock mid-flight. Session output is appended to
close it once the boot proceeds and dropbear is torn down. ssh-delivery.log so a failed unlock (onion unreachable, cryptroot-unlock
error) is diagnosable.
""" """
log_file = _ssh_log_path(spec).open("a")
log_file.write("--- delivery attempt ---\n")
log_file.flush()
proc = subprocess.Popen( proc = subprocess.Popen(
config.ssh_argv(spec), config.ssh_argv(spec),
stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, stdin=subprocess.PIPE,
stdout=log_file,
stderr=log_file,
text=True, text=True,
) )
try: try:
@@ -92,6 +128,7 @@ def _deliver_worker(spec: QemuSpec) -> None:
if proc.stdin is not None: if proc.stdin is not None:
with contextlib.suppress(OSError): with contextlib.suppress(OSError):
proc.stdin.close() proc.stdin.close()
log_file.close()
def boot_and_unlock(spec: QemuSpec) -> None: def boot_and_unlock(spec: QemuSpec) -> None:
@@ -103,20 +140,33 @@ def boot_and_unlock(spec: QemuSpec) -> None:
""" """
if spec.serial_log.exists(): if spec.serial_log.exists():
spec.serial_log.unlink() spec.serial_log.unlink()
if spec.direct_ssh_port:
target = f"127.0.0.1:{spec.direct_ssh_port} -> guest dropbear (direct)"
# dropbear comes up within seconds; no need to wait for an onion.
initial_delay = 15
else:
target = f"{spec.onion_address}:22 (Tor onion)"
initial_delay = INITIAL_DELIVERY_DELAY
print(f"\n========== booting in QEMU ({spec.arch}, accel={spec.accel}) ==========") 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(f"Unlock target: {target} (budget {UNLOCK_TIMEOUT}s)")
print("---- guest serial console ----", flush=True) print("---- guest serial console ----", flush=True)
# QEMU's own diagnostics (bad -machine/-cpu, /dev/kvm denied, image busy) # 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. # go to stderr; capture them so an early exit is debuggable, not opaque.
qemu_stderr = _qemu_stderr_path(spec).open("wb") qemu_stderr = _qemu_stderr_path(spec).open("wb")
qemu = subprocess.Popen( qemu = subprocess.Popen(
config.qemu_argv(spec), config.qemu_argv(spec),
stdout=subprocess.DEVNULL, stderr=qemu_stderr, stdout=subprocess.DEVNULL,
stderr=qemu_stderr,
) )
try: try:
deadline = time.monotonic() + UNLOCK_TIMEOUT deadline = time.monotonic() + UNLOCK_TIMEOUT
offset = 0 offset = 0
next_delivery = 0.0 # Hold off the first delivery until the unlock endpoint is up (see above).
next_delivery = time.monotonic() + initial_delay
print(
f"[harness] waiting {initial_delay}s for the unlock endpoint before delivering...",
flush=True,
)
delivery: threading.Thread | None = None delivery: threading.Thread | None = None
while time.monotonic() < deadline: while time.monotonic() < deadline:
offset = _stream_new_serial(spec, offset) offset = _stream_new_serial(spec, offset)
@@ -132,17 +182,15 @@ def boot_and_unlock(spec: QemuSpec) -> None:
# background while we keep streaming serial and watching the marker. # background while we keep streaming serial and watching the marker.
idle = delivery is None or not delivery.is_alive() idle = delivery is None or not delivery.is_alive()
if idle and time.monotonic() >= next_delivery: if idle and time.monotonic() >= next_delivery:
print(f"\n[harness] delivering passphrase over Tor to " print("\n[harness] delivering passphrase (holding session open) ...", flush=True)
f"{spec.onion_address}:22 (holding session open) ...", flush=True) delivery = threading.Thread(target=_deliver_worker, args=(spec,), daemon=True)
delivery = threading.Thread(
target=_deliver_worker, args=(spec,), daemon=True
)
delivery.start() delivery.start()
next_delivery = time.monotonic() + DELIVERY_INTERVAL next_delivery = time.monotonic() + DELIVERY_INTERVAL
time.sleep(_POLL_INTERVAL) time.sleep(_POLL_INTERVAL)
raise RuntimeError( raise RuntimeError(
f"Boot-ok marker never appeared within {UNLOCK_TIMEOUT}s.\n" f"Boot-ok marker never appeared within {UNLOCK_TIMEOUT}s.\n"
f"{_serial_tail(spec)}{_qemu_stderr_tail(spec)}" f"{_serial_tail(spec, lines=45)}{_qemu_stderr_tail(spec)}"
f"{_ssh_log_tail(spec)}{_client_tor_tail(spec)}"
) )
finally: finally:
qemu.terminate() qemu.terminate()

View File

@@ -0,0 +1,152 @@
#!/bin/bash
# Build a QEMU-bootable, LUKS-encrypted Debian image that reproduces lim's Tor
# unlock stack for the initramfs-tools backend: the REAL hooks from
# lim/configuration/initramfs-tools/, cryptsetup-initramfs + dropbear-initramfs,
# offline onion keys, and a boot-ok marker service. Boots on a virtio machine
# (models the software stack, not the Raspberry Pi board).
#
# Debian amd64 on an x86_64 host — no qemu-user needed. Needs root (debootstrap,
# loop device, cryptsetup, chroot). See tests/e2e/qemu/README.md.
#
# Same env contract and image.env output as build_image.sh.
set -euo pipefail
: "${WORK_DIR:?set WORK_DIR}"
: "${REPO_ROOT:?set REPO_ROOT}"
: "${PASSPHRASE:?set PASSPHRASE}"
: "${SSH_PUBKEY:?set SSH_PUBKEY}"
SUITE="${DEBIAN_SUITE:-bookworm}"
MIRROR="${DEBIAN_MIRROR:-http://deb.debian.org/debian}"
MAPPER_NAME="cryptroot"
ROOTFS="$WORK_DIR/rootfs"
IMAGE="$WORK_DIR/image.raw"
HOOK_SRC="$REPO_ROOT/lim/configuration/initramfs-tools"
banner() { printf "\n========== %s ==========\n" "$1"; }
# Debian keeps tools in /usr/sbin, which an Arch host's PATH (merged /usr/bin)
# omits, so chroot can't find update-initramfs. Force a Debian-style PATH.
in_chroot() {
chroot "$ROOTFS" /usr/bin/env \
PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin "$@"
}
cleanup() {
set +e
for m in dev/pts dev proc sys; do
mountpoint -q "$ROOTFS/$m" && umount -l "$ROOTFS/$m"
done
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 Debian rootfs with debootstrap ($SUITE)"
mkdir -p "$ROOTFS"
debootstrap --arch=amd64 --variant=minbase \
--include=linux-image-amd64,cryptsetup,cryptsetup-initramfs,dropbear-initramfs,tor,busybox,systemd-sysv,udev,ifupdown,isc-dhcp-client \
"$SUITE" "$ROOTFS" "$MIRROR"
banner "binding chroot filesystems"
cp /etc/resolv.conf "$ROOTFS/etc/resolv.conf"
for m in proc sys dev dev/pts; do
mkdir -p "$ROOTFS/$m"
mount --bind "/$m" "$ROOTFS/$m"
done
banner "creating the LUKS image (UUID must exist before update-initramfs)"
truncate -s 6G "$IMAGE"
LOOP="$(losetup -f --show "$IMAGE")"
printf '%s' "$PASSPHRASE" | cryptsetup luksFormat --type luks2 --batch-mode \
--pbkdf-memory 262144 "$LOOP" -
LUKS_UUID="$(cryptsetup luksUUID "$LOOP")"
banner "configuring crypttab / fstab / networking"
# `initramfs` bakes the mapping in so cryptsetup-initramfs unlocks before pivot.
echo "$MAPPER_NAME UUID=$LUKS_UUID none luks,initramfs" > "$ROOTFS/etc/crypttab"
echo "/dev/mapper/$MAPPER_NAME / ext4 defaults,noatime 0 1" > "$ROOTFS/etc/fstab"
echo "lim-e2e" > "$ROOTFS/etc/hostname"
# initramfs-tools includes virtio via MODULES=most (default); be explicit.
sed -i 's/^MODULES=.*/MODULES=most/' "$ROOTFS/etc/initramfs-tools/initramfs.conf"
banner "authorizing the ssh key for dropbear-initramfs"
install -D -m0600 "$SSH_PUBKEY" "$ROOTFS/etc/dropbear/initramfs/authorized_keys"
banner "installing the real lim initramfs-tools tor hooks"
install -D -m0755 "$HOOK_SRC/tor_hook" "$ROOTFS/etc/initramfs-tools/hooks/tor"
install -D -m0755 "$HOOK_SRC/tor_premount" "$ROOTFS/etc/initramfs-tools/scripts/init-premount/tor"
install -D -m0755 "$HOOK_SRC/tor_bottom" "$ROOTFS/etc/initramfs-tools/scripts/init-bottom/tor"
install -D -m0644 "$HOOK_SRC/torrc" "$ROOTFS/etc/tor/initramfs-torrc"
if [ -n "${TOR_TEST_NETWORK_CONF:-}" ]; then
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 "baking the boot-ok marker service + serial getty"
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
in_chroot systemctl enable lim-boot-ok.service serial-getty@ttyS0.service
banner "generating the initramfs (update-initramfs)"
in_chroot update-initramfs -c -k all || in_chroot update-initramfs -u -k all
banner "exporting kernel + initramfs"
KERNEL_IMG="$(find "$ROOTFS/boot" -maxdepth 1 -name 'vmlinuz-*' | sort | tail -1)"
INITRAMFS_IMG="$(find "$ROOTFS/boot" -maxdepth 1 -name 'initrd.img-*' | sort | tail -1)"
[ -n "$KERNEL_IMG" ] || { echo "No kernel 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"
banner "unbinding chroot filesystems"
for m in dev/pts dev proc sys; do umount -l "$ROOTFS/$m"; done
banner "copying the rootfs into the encrypted volume"
printf '%s' "$PASSPHRASE" | cryptsetup open "$LOOP" "$MAPPER_NAME" -
mkfs.ext4 -q "/dev/mapper/$MAPPER_NAME"
mkdir -p "$WORK_DIR/mnt"
mount "/dev/mapper/$MAPPER_NAME" "$WORK_DIR/mnt"
cp -a "$ROOTFS/." "$WORK_DIR/mnt/"
sync
umount -R "$WORK_DIR/mnt"
cryptsetup close "$MAPPER_NAME"
losetup -d "$LOOP"
LOOP=""
banner "writing image.env"
cat > "$WORK_DIR/image.env" <<EOF
LUKS_UUID=$LUKS_UUID
MAPPER_NAME=$MAPPER_NAME
ONION_ADDRESS=$ONION_ADDRESS
IMAGE=$IMAGE
KERNEL=$WORK_DIR/kernel
INITRAMFS=$WORK_DIR/initramfs
EOF
echo "Build complete. Onion address: $ONION_ADDRESS"

View File

@@ -42,6 +42,15 @@ class QemuSpec:
# (build_image.sh also caps --pbkdf-memory to keep this headroom). # (build_image.sh also caps --pbkdf-memory to keep this headroom).
memory_mb: int = 2048 memory_mb: int = 2048
accel: str = "tcg" # "kvm" when the host arch matches and /dev/kvm exists accel: str = "tcg" # "kvm" when the host arch matches and /dev/kvm exists
# "arch" (mkinitcpio/encryptssh) or "debian" (initramfs-tools/cryptroot-unlock).
os_family: str = "arch"
# Remote command the unlock SSH session runs; empty means the session
# presents the passphrase prompt itself (Arch encryptssh).
unlock_command: str = ""
# >0 selects the deterministic direct transport: QEMU forwards this host
# port to the guest's dropbear (:22) and the passphrase is delivered over a
# plain SSH (no Tor). 0 uses the Tor onion transport (onion_address).
direct_ssh_port: int = 0
@property @property
def serial_log(self) -> Path: def serial_log(self) -> Path:
@@ -63,18 +72,23 @@ def kernel_cmdline(spec: QemuSpec) -> str:
QEMU cannot emulate — hence virtio here (we model the stack, not the board). QEMU cannot emulate — hence virtio here (we model the stack, not the board).
""" """
console = _SERIAL_CONSOLE[spec.arch] console = _SERIAL_CONSOLE[spec.arch]
# The netconf hook needs the explicit ip= form # Arch's mkinitcpio encrypt/encryptssh finds the LUKS device from the
# cryptdevice= param; Debian's cryptsetup-initramfs reads /etc/crypttab
# baked into the initramfs, so it needs only root=/dev/mapper/<name>.
if spec.os_family == "debian":
crypt = f"root=/dev/mapper/{spec.mapper_name} rw"
else:
crypt = (
f"cryptdevice=UUID={spec.luks_uuid}:{spec.mapper_name} "
f"root=/dev/mapper/{spec.mapper_name} rw"
)
# The netconf/dropbear-initramfs hooks need the explicit ip= form
# <client>:<server>:<gw>:<netmask>:<host>:<device>:<proto> — the device # <client>:<server>:<gw>:<netmask>:<host>:<device>:<proto> — the device
# name is mandatory (as lim's own boot config uses it); a bare "ip=dhcp" # name is mandatory; a bare "ip=dhcp" leaves netconf looping on
# leaves netconf looping on "SIOCGIFFLAGS: No such device". net.ifnames=0 # "SIOCGIFFLAGS: No such device". net.ifnames=0 keeps the NIC named eth0.
# keeps the virtio NIC named eth0.
# The LAST console= owns /dev/console, which the boot-ok marker echoes to # 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. # and QEMU captures via -serial file, so the serial console must come last.
return ( return f"{crypt} ip=::::lim-e2e:eth0:dhcp net.ifnames=0 console=tty0 console={console}"
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]: def qemu_argv(spec: QemuSpec) -> list[str]:
@@ -94,16 +108,28 @@ def qemu_argv(spec: QemuSpec) -> list[str]:
argv += ["-cpu", "host", "-accel", "kvm"] argv += ["-cpu", "host", "-accel", "kvm"]
elif spec.arch == "x86_64": elif spec.arch == "x86_64":
argv += ["-cpu", "max", "-accel", "tcg"] argv += ["-cpu", "max", "-accel", "tcg"]
netdev = "user,id=net0"
if spec.direct_ssh_port:
# Forward a host port to the guest's dropbear for the direct transport.
netdev += f",hostfwd=tcp:127.0.0.1:{spec.direct_ssh_port}-:22"
argv += [ argv += [
"-m", str(spec.memory_mb), "-m",
"-kernel", str(spec.kernel_path), str(spec.memory_mb),
"-initrd", str(spec.initramfs_path), "-kernel",
"-append", kernel_cmdline(spec), str(spec.kernel_path),
"-drive", f"file={spec.image_path},if=virtio,format=raw", "-initrd",
"-netdev", "user,id=net0", str(spec.initramfs_path),
"-device", f"{net_device},netdev=net0", "-append",
kernel_cmdline(spec),
"-drive",
f"file={spec.image_path},if=virtio,format=raw",
"-netdev",
netdev,
"-device",
f"{net_device},netdev=net0",
"-nographic", "-nographic",
"-serial", f"file:{spec.serial_log}", "-serial",
f"file:{spec.serial_log}",
"-no-reboot", "-no-reboot",
] ]
return argv return argv
@@ -115,18 +141,34 @@ def ssh_proxy_command(socks_port: int) -> str:
def ssh_argv(spec: QemuSpec) -> list[str]: def ssh_argv(spec: QemuSpec) -> list[str]:
"""Non-interactive SSH into the initramfs dropbear via the onion address. """Non-interactive SSH into the initramfs dropbear to feed the passphrase.
``-tt`` forces a pty so the cryptsetup askpass inside the encryptssh ``-tt`` forces a pty so the askpass inside the session reads the passphrase
session reads the passphrase we pipe on stdin. Host-key checking is off we pipe on stdin. Host-key checking is off (throwaway host key). On Debian
because a throwaway onion has no known_hosts entry. the session runs ``cryptroot-unlock``; on Arch encryptssh presents the
prompt on login. The direct transport connects to a forwarded host port;
the Tor transport routes through the onion via a SOCKS ProxyCommand.
""" """
return [ argv = [
"ssh", "-tt", "ssh",
"-o", "StrictHostKeyChecking=no", "-tt",
"-o", "UserKnownHostsFile=/dev/null", "-o",
"-o", "BatchMode=yes", "StrictHostKeyChecking=no",
"-o", f"ProxyCommand={ssh_proxy_command(spec.socks_port)}", "-o",
"-i", str(spec.ssh_key_path), "UserKnownHostsFile=/dev/null",
"-o",
"BatchMode=yes",
"-i",
str(spec.ssh_key_path),
]
if spec.direct_ssh_port:
argv += ["-p", str(spec.direct_ssh_port), "root@127.0.0.1"]
else:
argv += [
"-o",
f"ProxyCommand={ssh_proxy_command(spec.socks_port)}",
f"root@{spec.onion_address}", f"root@{spec.onion_address}",
] ]
if spec.unlock_command:
argv.append(spec.unlock_command)
return argv

View File

@@ -7,6 +7,7 @@ config.py and is unit-tested without any of this running.
import os import os
import platform import platform
import socket
import subprocess import subprocess
import threading import threading
from dataclasses import dataclass from dataclasses import dataclass
@@ -31,6 +32,34 @@ class HarnessConfig:
passphrase: str = DEFAULT_PASSPHRASE passphrase: str = DEFAULT_PASSPHRASE
chutney_path: Path | None = None chutney_path: Path | None = None
chutney_network: str = "networks/basic" chutney_network: str = "networks/basic"
os_family: str = "arch" # "arch" (pacstrap) or "debian" (debootstrap)
# "direct" (deterministic; QEMU port-forward to dropbear) or "tor" (full
# onion transport, opt-in — flaky over public Tor inside QEMU).
unlock_transport: str = "direct"
class _NullNet:
"""Stand-in tor network for the direct transport: no Tor needed."""
socks_port = 0
test_network_conf: Path | None = None
def start(self) -> None:
pass
def stop(self) -> None:
pass
def _free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.bind(("127.0.0.1", 0))
return sock.getsockname()[1]
# Per-OS build script and the remote command that delivers the passphrase.
_BUILD_SCRIPT = {"arch": "build_image.sh", "debian": "build_image_debian.sh"}
_UNLOCK_COMMAND = {"arch": "", "debian": "cryptroot-unlock"}
def choose_accel(arch: str) -> str: def choose_accel(arch: str) -> str:
@@ -85,7 +114,7 @@ def _acquire_sudo() -> _SudoKeepalive | None:
def _build_image(cfg: HarnessConfig, ssh_key: Path, test_network_conf: Path | None) -> None: def _build_image(cfg: HarnessConfig, ssh_key: Path, test_network_conf: Path | None) -> None:
script = cfg.repo_root / "tests/e2e/qemu/build_image.sh" script = cfg.repo_root / "tests/e2e/qemu" / _BUILD_SCRIPT[cfg.os_family]
env = { env = {
**os.environ, **os.environ,
"WORK_DIR": str(cfg.work_dir), "WORK_DIR": str(cfg.work_dir),
@@ -119,6 +148,8 @@ def _reclaim_work_dir(cfg: HarnessConfig, *, required: bool = False) -> None:
def _make_tor_net(cfg: HarnessConfig): def _make_tor_net(cfg: HarnessConfig):
if cfg.unlock_transport == "direct":
return _NullNet()
if cfg.chutney_path is not None: if cfg.chutney_path is not None:
return tor_net.ChutneyNetwork(cfg.chutney_path, cfg.work_dir, cfg.chutney_network) return tor_net.ChutneyNetwork(cfg.chutney_path, cfg.work_dir, cfg.chutney_network)
return tor_net.PublicTorClient(cfg.work_dir, DEFAULT_SOCKS_PORT) return tor_net.PublicTorClient(cfg.work_dir, DEFAULT_SOCKS_PORT)
@@ -126,6 +157,7 @@ def _make_tor_net(cfg: HarnessConfig):
def _load_spec(cfg: HarnessConfig, ssh_key: Path, socks_port: int) -> QemuSpec: def _load_spec(cfg: HarnessConfig, ssh_key: Path, socks_port: int) -> QemuSpec:
env = boot_unlock.parse_env_file(cfg.work_dir / IMAGE_ENV_NAME) env = boot_unlock.parse_env_file(cfg.work_dir / IMAGE_ENV_NAME)
direct_port = _free_port() if cfg.unlock_transport == "direct" else 0
return QemuSpec( return QemuSpec(
arch=cfg.arch, arch=cfg.arch,
work_dir=cfg.work_dir, work_dir=cfg.work_dir,
@@ -139,6 +171,9 @@ def _load_spec(cfg: HarnessConfig, ssh_key: Path, socks_port: int) -> QemuSpec:
ssh_key_path=ssh_key, ssh_key_path=ssh_key,
socks_port=socks_port, socks_port=socks_port,
accel=choose_accel(cfg.arch), accel=choose_accel(cfg.arch),
os_family=cfg.os_family,
unlock_command=_UNLOCK_COMMAND[cfg.os_family],
direct_ssh_port=direct_port,
) )
@@ -155,11 +190,14 @@ def run(cfg: HarnessConfig) -> QemuSpec:
sudo = _acquire_sudo() sudo = _acquire_sudo()
net = _make_tor_net(cfg) net = _make_tor_net(cfg)
try: try:
if cfg.unlock_transport != "direct":
print("\n[harness] bootstrapping Tor client (~30-60s)...", flush=True) print("\n[harness] bootstrapping Tor client (~30-60s)...", flush=True)
net.start() net.start()
print(f"[harness] Tor client ready on socks port {net.socks_port}.", flush=True) print(
print("[harness] building encrypted image " f"[harness] transport={cfg.unlock_transport}, building encrypted "
"(pacstrap + mkinitcpio, several minutes)...", flush=True) "image (several minutes)...",
flush=True,
)
_build_image(cfg, ssh_key, net.test_network_conf) _build_image(cfg, ssh_key, net.test_network_conf)
spec = _load_spec(cfg, ssh_key, net.socks_port) spec = _load_spec(cfg, ssh_key, net.socks_port)
boot_unlock.boot_and_unlock(spec) boot_unlock.boot_and_unlock(spec)

View File

@@ -17,7 +17,7 @@ import subprocess
import time import time
from pathlib import Path from pathlib import Path
BOOTSTRAP_TIMEOUT = 180 BOOTSTRAP_TIMEOUT = 300 # public Tor bootstrap can stall on slow guards
_GUEST_HOST_ALIAS = "10.0.2.2" # QEMU user-net alias for the host's 127.0.0.1 _GUEST_HOST_ALIAS = "10.0.2.2" # QEMU user-net alias for the host's 127.0.0.1
@@ -41,12 +41,21 @@ class PublicTorClient:
empty_torrc.write_text("") empty_torrc.write_text("")
self._process = subprocess.Popen( self._process = subprocess.Popen(
[ [
"tor", "-f", str(empty_torrc), "tor",
"--SocksPort", str(self.socks_port), "-f",
"--DataDirectory", str(self._data), str(empty_torrc),
"--Log", f"notice file {self._log}", "--SocksPort",
str(self.socks_port),
"--DataDirectory",
str(self._data),
"--Log",
f"notice file {self._log}",
# HS-level log so onion descriptor fetch / rendezvous is visible.
"--Log",
f"[rend]info file {self._work_dir / 'tor-client-hs.log'}",
], ],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
) )
_wait_for_bootstrap(self._log, self._process) _wait_for_bootstrap(self._log, self._process)
@@ -75,7 +84,8 @@ class ChutneyNetwork:
def _run(self, *args: str) -> None: def _run(self, *args: str) -> None:
subprocess.run( subprocess.run(
[str(self._chutney / "chutney"), *args, self._network], [str(self._chutney / "chutney"), *args, self._network],
cwd=self._chutney, check=True, cwd=self._chutney,
check=True,
) )
def start(self) -> None: def start(self) -> None:

View File

@@ -50,6 +50,15 @@ class TestKernelCmdline:
def test_aarch64_uses_amba_serial(self): def test_aarch64_uses_amba_serial(self):
assert "console=ttyAMA0" in config.kernel_cmdline(_spec(arch="aarch64")) assert "console=ttyAMA0" in config.kernel_cmdline(_spec(arch="aarch64"))
def test_debian_uses_crypttab_not_cryptdevice(self):
# Debian's cryptsetup-initramfs reads /etc/crypttab, so the cmdline
# carries only root=/dev/mapper/<name>, no cryptdevice= param.
line = config.kernel_cmdline(_spec(os_family="debian"))
assert "cryptdevice=" not in line
assert "root=/dev/mapper/cryptroot" in line
assert ":eth0:dhcp" in line
assert line.index("console=tty0") < line.index("console=ttyS0")
class TestQemuArgv: class TestQemuArgv:
def test_x86_uses_virtio_net_and_direct_kernel_boot(self): def test_x86_uses_virtio_net_and_direct_kernel_boot(self):
@@ -103,6 +112,31 @@ class TestSshInvocation:
assert any("ProxyCommand=" in part for part in argv) assert any("ProxyCommand=" in part for part in argv)
assert "StrictHostKeyChecking=no" in argv assert "StrictHostKeyChecking=no" in argv
def test_arch_session_runs_no_remote_command(self):
assert config.ssh_argv(_spec())[-1] == "root@abcd.onion"
def test_debian_session_runs_cryptroot_unlock(self):
argv = config.ssh_argv(_spec(unlock_command="cryptroot-unlock"))
assert argv[-1] == "cryptroot-unlock"
class TestDirectTransport:
def test_qemu_forwards_a_host_port_to_dropbear(self):
argv = config.qemu_argv(_spec(direct_ssh_port=2222))
assert "user,id=net0,hostfwd=tcp:127.0.0.1:2222-:22" in argv
def test_tor_transport_adds_no_hostfwd(self):
argv = config.qemu_argv(_spec())
assert "user,id=net0" in argv
assert not any("hostfwd" in part for part in argv)
def test_ssh_argv_direct_connects_to_forwarded_port(self):
argv = config.ssh_argv(_spec(direct_ssh_port=2222))
assert "root@127.0.0.1" in argv
assert "-p" in argv
assert "2222" in argv
assert not any("ProxyCommand" in part for part in argv)
class TestBuildScriptStaysAligned: class TestBuildScriptStaysAligned:
"""Guards: the root build script must match what the harness assumes.""" """Guards: the root build script must match what the harness assumes."""
@@ -111,9 +145,7 @@ class TestBuildScriptStaysAligned:
return BUILD_SCRIPT.read_text() return BUILD_SCRIPT.read_text()
def test_hooks_place_tor_between_netconf_and_dropbear(self): def test_hooks_place_tor_between_netconf_and_dropbear(self):
hooks_line = next( hooks_line = next(line for line in self._script().splitlines() if line.startswith("HOOKS="))
line for line in self._script().splitlines() if line.startswith("HOOKS=")
)
positions = [hooks_line.index(hook) for hook in EXPECTED_HOOKS_ORDER] positions = [hooks_line.index(hook) for hook in EXPECTED_HOOKS_ORDER]
assert positions == sorted(positions), hooks_line assert positions == sorted(positions), hooks_line
@@ -129,9 +161,7 @@ class TestBuildScriptStaysAligned:
assert config.BOOT_OK_MARKER in script assert config.BOOT_OK_MARKER in script
def test_mounts_devpts_before_dropbear_for_pty(self): def test_mounts_devpts_before_dropbear_for_pty(self):
hooks_line = next( hooks_line = next(line for line in self._script().splitlines() if line.startswith("HOOKS="))
line for line in self._script().splitlines() if line.startswith("HOOKS=")
)
assert "ptsmount" in hooks_line assert "ptsmount" in hooks_line
assert hooks_line.index("ptsmount") < hooks_line.index("dropbear") assert hooks_line.index("ptsmount") < hooks_line.index("dropbear")
assert "mount -t devpts" in self._script() assert "mount -t devpts" in self._script()
@@ -140,6 +170,35 @@ class TestBuildScriptStaysAligned:
assert (lim_config.CONFIGURATION_PATH / "initcpio" / "tor_hook").is_file() assert (lim_config.CONFIGURATION_PATH / "initcpio" / "tor_hook").is_file()
class TestDebianBuildScript:
"""Guards for the debootstrap build script and the OS dispatch."""
def _script(self) -> str:
path = Path(__file__).resolve().parents[2] / "tests/e2e/qemu/build_image_debian.sh"
return path.read_text()
def test_installs_the_real_initramfs_tools_hooks(self):
script = self._script()
for name in ("tor_hook", "tor_premount", "tor_bottom", "torrc"):
assert f"$HOOK_SRC/{name}" in script
assert "etc/dropbear/initramfs/authorized_keys" in script
def test_crypttab_uses_initramfs_option_and_emits_marker(self):
script = self._script()
assert "none luks,initramfs" in script
assert config.BOOT_OK_MARKER in script
def test_harness_maps_os_family_to_script_and_unlock_command(self):
assert harness._BUILD_SCRIPT["debian"] == "build_image_debian.sh"
assert harness._BUILD_SCRIPT["arch"] == "build_image.sh"
assert harness._UNLOCK_COMMAND["debian"] == "cryptroot-unlock"
assert harness._UNLOCK_COMMAND["arch"] == ""
def test_real_initramfs_tools_source_directory_exists(self):
hook = lim_config.CONFIGURATION_PATH / "initramfs-tools" / "tor_premount"
assert hook.is_file()
class TestOrchestratorGlue: class TestOrchestratorGlue:
def test_load_spec_reads_image_env(self, tmp_path): def test_load_spec_reads_image_env(self, tmp_path):
(tmp_path / "image.env").write_text( (tmp_path / "image.env").write_text(

View File

@@ -35,10 +35,20 @@ REPO_ROOT = Path(__file__).resolve().parents[2]
# The QEMU binary is arch-specific, so derive it from LIM_E2E_ARCH instead of # 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. # hardcoding x86_64 — otherwise the documented aarch64 run is never reachable.
_ARCH = os.environ.get("LIM_E2E_ARCH", "x86_64") _ARCH = os.environ.get("LIM_E2E_ARCH", "x86_64")
_REQUIRED = ( # The OS family selects the build tool: pacstrap (Arch) or debootstrap (Debian).
_OS = os.environ.get("LIM_E2E_OS", "arch")
# "direct" (deterministic port-forward) is the default; "tor" runs the full,
# opt-in onion transport (flaky over public Tor inside QEMU).
_TRANSPORT = os.environ.get("LIM_E2E_TRANSPORT", "direct")
_BUILD_TOOL = {"arch": "pacstrap", "debian": "debootstrap"}
_REQUIRED = [
qemu_config.qemu_binary(_ARCH), qemu_config.qemu_binary(_ARCH),
"cryptsetup", "pacstrap", "tor", "ssh", "ncat", _BUILD_TOOL.get(_OS, "pacstrap"),
) "cryptsetup",
"ssh",
]
if _TRANSPORT == "tor": # the onion transport also needs a Tor client + ncat
_REQUIRED += ["tor", "ncat"]
def _missing_tools() -> list[str]: def _missing_tools() -> list[str]:
@@ -48,25 +58,24 @@ def _missing_tools() -> list[str]:
pytestmark = pytest.mark.skipif( pytestmark = pytest.mark.skipif(
os.environ.get("LIM_E2E_QEMU") != "1" or bool(_missing_tools()), os.environ.get("LIM_E2E_QEMU") != "1" or bool(_missing_tools()),
reason=( reason=(
"needs LIM_E2E_QEMU=1 and " "needs LIM_E2E_QEMU=1 and " + ", ".join(_REQUIRED) + " (build stage needs root; slow)."
+ ", ".join(_REQUIRED)
+ " (build stage needs root; slow)."
), ),
) )
def test_full_build_boot_and_tor_unlock(tmp_path): def test_full_build_boot_and_unlock(tmp_path):
chutney_path = os.environ.get("CHUTNEY_PATH") chutney_path = os.environ.get("CHUTNEY_PATH")
cfg = HarnessConfig( cfg = HarnessConfig(
repo_root=REPO_ROOT, repo_root=REPO_ROOT,
work_dir=tmp_path / "qemu-e2e", work_dir=tmp_path / "qemu-e2e",
arch=_ARCH, arch=_ARCH,
os_family=_OS,
unlock_transport=_TRANSPORT,
chutney_path=Path(chutney_path) if chutney_path else None, chutney_path=Path(chutney_path) if chutney_path else None,
) )
spec = run(cfg) spec = run(cfg)
# run() only returns after the marker was observed; re-assert for clarity. # run() only returns after the boot-ok marker was observed on the console.
assert spec.onion_address.endswith(".onion")
serial = spec.serial_log.read_text(errors="replace") serial = spec.serial_log.read_text(errors="replace")
assert qemu_config.BOOT_OK_MARKER in serial assert qemu_config.BOOT_OK_MARKER in serial