Release version 2.2.0

Split the rootless Tor e2e skip so the deterministic offline checks (the
real offline onion keygen and the production-flag guard) run whenever the
tor binary is present, while only the live onion round-trip stays gated on
LIM_E2E_TOR=1. The pytest job now installs tor so the keygen — which
validates the exact production keygen path — runs on every push. A separate
continue-on-error tor-network-e2e job exercises the full round-trip without
making the public Tor network a blocking merge gate.

The full QEMU build/boot/unlock e2e still needs root + KVM + Arch tooling
and stays local/opt-in.

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

.github/workflows/test.yml

@@ -18,5 +18,22 @@ jobs:
       - uses: actions/setup-python@v5
         with:
           python-version: "3.12"
+      # tor lets the deterministic offline onion-keygen test run (it validates
+      # the real production keygen path); that test needs no network.
+      - run: sudo apt-get update && sudo apt-get install -y tor
       - run: pip install pytest pyyaml
       - run: make test
+
+  tor-network-e2e:
+    # The live onion round-trip depends on the public Tor network, so it is
+    # informational (continue-on-error) rather than a blocking merge gate.
+    runs-on: ubuntu-latest
+    continue-on-error: true
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - run: sudo apt-get update && sudo apt-get install -y tor
+      - run: pip install pytest pyyaml
+      - run: LIM_E2E_TOR=1 make test-tor PYTHON=python

.gitignore

@@ -5,3 +5,6 @@ log.txt
 __pycache__/
 *.pyc
 .pytest_cache/
+dist/
+build/
+*.egg-info/

CHANGELOG.md

@@ -1,5 +1,23 @@
 # Changelog
 
+## [2.2.0] - 2026-07-21
+
+- Optional remote LUKS unlock via a Tor onion service in the initramfs.
+  Encrypted image setups can bake *tor* into the initramfs (mkinitcpio hooks
+  ordered *netconf tor dropbear encryptssh*), generate the v3 onion keys
+  offline in the image chroot, and print the stable *.onion* address. Unlock
+  with *torsocks ssh root@<onion-address>*. Useful behind NAT/CGNAT or a
+  dynamic IP, where the direct *ssh root@<ip>* unlock cannot reach the host.
+- End-to-end tests: a rootless one (*LIM_E2E_TOR=1*) that round-trips a
+  passphrase through a real Tor onion service, and a full virtualized one
+  (*LIM_E2E_QEMU=1*) that builds a LUKS image, boots it in QEMU, and unlocks
+  the root over Tor. The offline onion keygen runs in CI; *make test-tor* /
+  *make test-qemu* / *make test-all* run the opt-in stages.
+
+## [2.1.0] - 2026-07-14
+
+Initial PyPI release 🥳
+
 ## [2.0.0] - 2026-07-14
 
 All bash scripts were ported to the *lim/* Python package with a unified CLI, a mocked test suite, and CI.

Makefile

@@ -1,6 +1,9 @@
-.PHONY: install test
+.PHONY: install test test-tor test-qemu test-all
 
 PREFIX ?= $(HOME)/.local
+# Interpreter used for the tests. Override if your active venv lacks pytest,
+# e.g.  make test PYTHON=/usr/bin/python3
+PYTHON ?= python3
 
 install:
 	chmod +x main.py
@@ -8,5 +11,20 @@ install:
 	ln -sf $(CURDIR)/main.py $(PREFIX)/bin/lim
 	@echo "Installed lim to $(PREFIX)/bin/lim"
 
+# Default suite: fully mocked, no root/QEMU/network. The opt-in e2e tests skip.
 test:
-	python3 -m pytest
+	$(PYTHON) -m pytest
+
+# Rootless Tor onion unlock e2e (needs the `tor` binary and network access).
+test-tor:
+	LIM_E2E_TOR=1 $(PYTHON) -m pytest tests/e2e/test_tor_unlock_e2e.py -v
+
+# Full virtualized build+boot+unlock e2e. Needs qemu, arch-install-scripts,
+# cryptsetup, tor and ncat; the build stage runs under sudo. Set CHUTNEY_PATH
+# to use a private, offline Tor network instead of the public one.
+test-qemu:
+	LIM_E2E_QEMU=1 $(PYTHON) -m pytest tests/e2e/test_qemu_unlock_e2e.py -v -s
+
+# Everything: mocked suite plus both opt-in e2e stages, in one run.
+test-all:
+	LIM_E2E_TOR=1 LIM_E2E_QEMU=1 $(PYTHON) -m pytest -v -s

README.md

@@ -10,6 +10,7 @@ Linux Image Manager (lim) is a Python tool for downloading, configuring, and man
 
 - **Image Download & Setup:** Automatically download, verify (checksum + GPG signature) and prepare Linux distributions.
 - **Encrypted Storage:** Configure LUKS encryption for secure image management.
+- **Tor Onion Unlock:** Optionally bake a Tor onion service into the initramfs so the dropbear LUKS unlock shell stays reachable behind NAT or dynamic IPs.
 - **Virtual RAID1:** Easily set up virtual Btrfs RAID1 for data redundancy.
 - **Backup & Restore:** Create image backups from devices using dd.
 - **Chroot Environment:** Easily enter a chroot shell to maintain or modify Linux images.
@@ -26,7 +27,7 @@ package-manager install lim
 
 This command makes Linux Image Manager globally available as `lim` in your terminal. The `lim` alias points to the **main.py** entry point.
 
-There are no Python dependencies beyond the standard library. The commands call the usual system tools (`cryptsetup`, `fdisk`, `dd`, `rsync`, `wget`, `gpg`, `encfs`, `pv`, `bsdtar`, ...), so those need to be installed for the command you use.
+The only Python dependency is **PyYAML** (`>=6`), used to read the image catalog. The pip/wheel install pulls it in automatically; on the `package-manager`/symlink install path make sure it is present (`python-yaml` on Arch). The commands also call the usual system tools (`cryptsetup`, `fdisk`, `dd`, `rsync`, `wget`, `gpg`, `encfs`, `pv`, `bsdtar`, ...), so those need to be installed for the command you use.
 
 ## Usage ⚙️
 
@@ -82,7 +83,6 @@ lim --type chroot
 
 ```
 main.py               # entry point (the `lim` alias)
-distributions.yml     # single point of truth for the image catalog
 lim/
   cli.py              # argument parsing and command dispatch
   catalog.py          # read-only access to distributions.yml
@@ -92,25 +92,96 @@ lim/
   storage/            # single drive and RAID1 encryption setups
   image/              # image setup, backup, chroot, verification
   data/               # encfs lock/unlock and data import/export
+  distributions.yml   # single point of truth for the image catalog
   configuration/      # package collections used during image setup
 tests/unit/           # unit tests (all external commands mocked)
+tests/e2e/            # Tor onion unlock end-to-end tests (rootless + QEMU, opt-in)
+tests/e2e/qemu/       # full virtualized build+boot+unlock harness
 tests/lint/           # architecture guards (e.g. max file length)
 ```
 
 ## Configuration & Customization 🔧
 
-Customize your environment in the `configuration/` folder:
+Customize your environment in the `lim/configuration/` folder:
 - **General Packages:** Contains common packages for all setup scripts.
 - **Server LUKS Packages:** Contains packages needed for setting up LUKS encryption on servers.
+- **initcpio Hooks:** The mkinitcpio install/runtime hooks and torrc baked into images for the Tor onion unlock.
+
+## Remote LUKS Unlock via Tor 🧅
+
+When you answer yes to *"Should the system be remotely unlockable via a Tor
+onion service?"* during an encrypted `lim --type image` setup, the image gets:
+
+- `tor` and `busybox` installed, plus a custom `tor` mkinitcpio hook ordered
+  between `netconf` and `dropbear`.
+- A v3 onion service key generated **offline** inside the image chroot; the
+  resulting `.onion` address is printed at the end of the setup and stays
+  stable across reboots and IP changes.
+- An NTP clock sync in early userspace (boards like the Raspberry Pi have no
+  RTC and would otherwise boot with a clock Tor rejects). Override the server
+  with the kernel parameter `tor_ntp=<host>`.
+
+After booting the device, unlock it from any machine with a running Tor
+client:
+
+```bash
+torsocks ssh root@<onion-address>
+```
+
+Typing the LUKS passphrase into that shell resumes the boot. The direct
+`ssh root@<ip>` unlock keeps working as before; Tor is an additional path,
+useful behind NAT/CGNAT where no port forwarding is possible.
+
+**Security note:** the onion private key is stored in the initramfs on the
+unencrypted boot partition. Anyone with physical access to the SD card can
+read it and impersonate the onion service — treat the address as
+non-secret. Access control remains the SSH public key
+(`/etc/dropbear/root_key`), exactly as with the direct unlock.
 
 ## Development & Tests 🧪
 
-The test suite mocks all external commands, so it runs safely on any machine:
+The unit suite mocks all external commands, so it runs safely on any machine:
 
 ```bash
 pytest
 ```
 
+### Tor unlock end-to-end test
+
+`tests/e2e/test_tor_unlock_e2e.py` reproduces the operator-visible half of the
+Tor decryption process **rootless** — no block devices, no `cryptsetup`, no
+`chroot`. It generates the v3 onion keys offline (exactly as `lim.image.tor`
+does inside the image), starts a real Tor onion service from a torrc mirroring
+the baked-in one, then connects to the `.onion` address through Tor and
+delivers a passphrase to a dropbear stand-in, asserting it arrives and the
+endpoint "unlocks".
+
+It needs the real `tor` binary and live Tor network access (onion bootstrap is
+slow), so it is opt-in and skipped otherwise:
+
+```bash
+LIM_E2E_TOR=1 pytest tests/e2e/test_tor_unlock_e2e.py -v
+```
+
+The offline-only checks in that file (onion keygen, production-flag guard) run
+without network. The physical flashing half (`dd`, `cryptsetup luksFormat`,
+`mount`, `chroot`, `mkinitcpio`) requires root and a matching CPU/qemu setup and
+is not covered by this rootless test.
+
+### Full virtualized unlock (QEMU)
+
+For the *whole* process — build an encrypted image, boot it, run the real
+initramfs Tor+dropbear chain, and unlock it — `tests/e2e/qemu/` drives a QEMU
+virtual machine end to end and asserts the real system booted only after the
+passphrase was delivered over Tor. It is opt-in (`LIM_E2E_QEMU=1`) and needs
+QEMU, `arch-install-scripts`, `cryptsetup`, `tor` and `ncat`; the build stage
+needs root (ideally inside a throwaway VM to keep the host rootless). A private,
+offline Tor network via [chutney](https://gitlab.torproject.org/tpo/core/chutney)
+is supported. See [tests/e2e/qemu/README.md](tests/e2e/qemu/README.md).
+
+The harness's pure logic and drift guards run in the normal suite
+(`test_qemu_harness_unit.py`) — no QEMU, root, or network needed.
+
 ## License 📜
 
 This project is licensed under the GNU General Public License Version 3. See the [LICENSE.txt](./LICENSE.txt) file for details.

lim/catalog.py

@@ -1,4 +1,4 @@
-"""Read-only access to the image catalog (distributions.yml in the repo root)."""
+"""Read-only access to the image catalog (distributions.yml in the package)."""
 
 from functools import cache
 
@@ -6,7 +6,7 @@ import yaml
 
 from lim import config
 
-CATALOG_PATH = config.REPOSITORY_PATH / "distributions.yml"
+CATALOG_PATH = config.CATALOG_PATH
 
 
 @cache

lim/config.py

@@ -1,10 +1,17 @@
-"""Repository paths shared by all modules."""
+"""Package and repository paths shared by all modules.
+
+Configuration and the image catalog ship inside the package so that
+wheel installs work; the encfs data store stays repository-relative.
+"""
 
 from pathlib import Path
 
-REPOSITORY_PATH = Path(__file__).resolve().parent.parent
-CONFIGURATION_PATH = REPOSITORY_PATH / "configuration"
+PACKAGE_ROOT = Path(__file__).resolve().parent
+CONFIGURATION_PATH = PACKAGE_ROOT / "configuration"
 PACKAGE_PATH = CONFIGURATION_PATH / "packages"
+CATALOG_PATH = PACKAGE_ROOT / "distributions.yml"
+
+REPOSITORY_PATH = PACKAGE_ROOT.parent
 ENCRYPTED_PATH = REPOSITORY_PATH / ".encrypted"
 DECRYPTED_PATH = REPOSITORY_PATH / "decrypted"
 DATA_PATH = DECRYPTED_PATH / "data"

lim/configuration/initcpio/tor_hook

@@ -0,0 +1,47 @@
+#!/usr/bin/ash
+# mkinitcpio runtime hook: start Tor so the dropbear unlock shell is
+# reachable as an onion service while encryptssh waits for the passphrase.
+# Installed to /etc/initcpio/hooks/tor by linux-image-manager.
+
+# netconf's ipconfig drops its DHCP lease data (incl. DNS) into
+# /tmp/net-*.conf; busybox's resolver only reads /etc/resolv.conf.
+_tor_write_resolv_conf() {
+    [ -s /etc/resolv.conf ] && return 0
+    local conf dns
+    for conf in /tmp/net-*.conf; do
+        [ -f "$conf" ] || continue
+        # Extract ONLY the DNS fields with sed; never source these files —
+        # they hold attacker-controllable DHCP option strings (hostname,
+        # domain, rootpath), and sourcing would run them as root pre-boot.
+        for dns in $(sed -n 's/^IPV4DNS[01]=//p' "$conf"); do
+            [ -n "$dns" ] && [ "$dns" != "0.0.0.0" ] \
+                && echo "nameserver $dns" >> /etc/resolv.conf
+        done
+    done
+    [ -s /etc/resolv.conf ]
+}
+
+run_hook() {
+    # Tor rejects consensus documents when the clock is far off; boards
+    # without an RTC boot in 1970, so sync before starting Tor. Bounded:
+    # a failed sync must never block the boot.
+    msg "tor: syncing clock via NTP..."
+    # Best-effort DNS; a tor_ntp IP literal needs none, so never gate on it.
+    _tor_write_resolv_conf || msg "tor: no DNS from DHCP (fine if tor_ntp is an IP)"
+    /usr/local/bin/busybox timeout 30 \
+        /usr/local/bin/busybox ntpd -n -q -p "${tor_ntp:-pool.ntp.org}" \
+        || msg "tor: NTP sync failed, keeping current clock"
+
+    msg "tor: starting onion service for remote unlock..."
+    mkdir -p /var/lib/tor
+    chmod 0700 /var/lib/tor /etc/tor/onion
+    chmod 0600 /etc/tor/onion/hs_ed25519_secret_key
+    tor -f /etc/tor/torrc --RunAsDaemon 1 --Log "notice file /tmp/tor.log" \
+        || msg "tor: failed to start, unlock stays reachable via direct IP"
+}
+
+run_cleanuphook() {
+    # Nothing from the initramfs may keep running after the pivot.
+    /usr/local/bin/busybox killall tor 2>/dev/null
+    return 0
+}

lim/configuration/initcpio/tor_install

@@ -0,0 +1,37 @@
+#!/bin/bash
+# mkinitcpio install hook: Tor onion service for remote LUKS unlock.
+# Installed to /etc/initcpio/install/tor by linux-image-manager.
+
+build() {
+    add_binary /usr/bin/tor
+    # Full busybox for the ntpd applet: boards without an RTC (e.g. most
+    # Raspberry Pis) wake up in 1970, which Tor's consensus checks reject.
+    # A different target path keeps mkinitcpio's own busybox untouched.
+    add_binary /usr/bin/busybox /usr/local/bin/busybox
+    # glibc resolves the NTP server hostname via getaddrinfo(), which dlopen()s
+    # these NSS modules at runtime — add_binary only follows NEEDED libs, so
+    # without them DNS silently fails, ntpd never syncs, and the clock stays
+    # at 1970. glibc's built-in default (no nsswitch.conf) is "dns ... files".
+    add_binary /usr/lib/libnss_dns.so.2
+    add_binary /usr/lib/libnss_files.so.2
+    add_file /etc/tor/initramfs-torrc /etc/tor/torrc
+    add_file /etc/tor/initramfs-onion/hostname /etc/tor/onion/hostname
+    add_file /etc/tor/initramfs-onion/hs_ed25519_public_key /etc/tor/onion/hs_ed25519_public_key
+    add_file /etc/tor/initramfs-onion/hs_ed25519_secret_key /etc/tor/onion/hs_ed25519_secret_key
+    add_runscript
+}
+
+help() {
+    cat <<HELPEOF
+Starts a Tor onion service in early userspace so the dropbear unlock
+shell stays reachable under the .onion address baked into the image,
+even behind NAT or a dynamic IP.
+
+Place it between netconf and dropbear:
+  HOOKS=(... netconf tor dropbear encryptssh ...)
+
+Optional kernel parameter:
+  tor_ntp=<server>  NTP server used to set the clock before Tor starts
+                    (default: pool.ntp.org)
+HELPEOF
+}

lim/configuration/initcpio/torrc

@@ -0,0 +1,7 @@
+# Tor configuration for the initramfs onion unlock service.
+# Baked into the initramfs as /etc/tor/torrc by the "tor" mkinitcpio hook;
+# the onion keys come from /etc/tor/initramfs-onion on the system.
+DataDirectory /var/lib/tor
+HiddenServiceDir /etc/tor/onion
+HiddenServicePort 22 127.0.0.1:22
+SocksPort 0

lim/configuration/packages/server/tor.txt

@@ -0,0 +1,3 @@
+# Packages for the Tor onion unlock service in the initramfs
+tor
+busybox  # full applet set: ntpd for clock sync on RTC-less boards

lim/image/encryption.py

@@ -9,6 +9,7 @@ from pathlib import Path
 from lim import catalog, fsutil, packages, runner, ui
 from lim.image.plan import ImagePlan
 from lim.image.session import ImageSession, chroot_bash, install_packages
+from lim.image.tor import configure_tor_unlock
 
 MKINITCPIO_HOOKS_PREFIX = (
     "base udev autodetect microcode modconf kms keyboard keymap consolefont block"
@@ -32,9 +33,10 @@ def _configure_mkinitcpio(plan: ImagePlan, root: Path) -> None:
     fsutil.replace_in_file(
         "BINARIES=()", "BINARIES=(/usr/lib/libgcc_s.so.1)", mkinitcpio_path
     )
+    tor_hook = "tor " if plan.tor_unlock else ""
     fsutil.replace_in_file(
         f"HOOKS=({MKINITCPIO_HOOKS_PREFIX} {MKINITCPIO_HOOKS_SUFFIX})",
-        f"HOOKS=({MKINITCPIO_HOOKS_PREFIX} sleep netconf dropbear encryptssh "
+        f"HOOKS=({MKINITCPIO_HOOKS_PREFIX} sleep netconf {tor_hook}dropbear encryptssh "
         f"{MKINITCPIO_HOOKS_SUFFIX})",
         mkinitcpio_path,
     )
@@ -90,9 +92,15 @@ def _configure_bootloader(plan: ImagePlan, session: ImageSession, root: Path) ->
     else:
         cmdline_txt_path = session.boot_mount_path / "cmdline.txt"
         ui.info(f"Configuring {cmdline_txt_path}...")
+        # Firmware-boot boards (e.g. RPi4 cmdline.txt) need the same early
+        # networking as the boot.txt branch, or the initramfs netconf hook
+        # brings up no interface and remote unlock (dropbear + tor) is
+        # unreachable — the onion never even publishes.
+        hostname = (root / "etc/hostname").read_text().strip()
         fsutil.replace_in_file(
             "root=/dev/mmcblk0p2",
-            f"{cryptdevice} rootfstype={plan.root_filesystem}",
+            f"{cryptdevice} rootfstype={plan.root_filesystem} "
+            f"ip=::::{hostname}:eth0:dhcp net.ifnames=0 biosdevname=0",
             cmdline_txt_path,
         )
         ui.info(f"Content of {cmdline_txt_path}:{cmdline_txt_path.read_text()}")
@@ -112,6 +120,10 @@ def configure_encryption(
     ui.info(f"Adding {authorized_keys} to dropbear...")
     runner.run(["cp", "-v", str(authorized_keys), str(dropbear_root_key_path)], sudo=True)
 
+    if plan.tor_unlock:
+        # Hook files and onion keys must exist before mkinitcpio bakes the image.
+        configure_tor_unlock(plan, root)
+
     _configure_mkinitcpio(plan, root)
     _register_encrypted_root(plan, session, root)
     _configure_bootloader(plan, session, root)

lim/image/plan.py

@@ -17,6 +17,7 @@ class ImagePlan:
     luks_memory_cost: str | None = None
     raspberry_pi_version: str | None = None
     encrypt_system: bool = False
+    tor_unlock: bool = False
     root_filesystem: str = ""
     image_folder: Path = field(default_factory=Path)
 

lim/image/setup.py

@@ -49,6 +49,10 @@ def _choose_and_verify_image(plan: ImagePlan) -> None:
     if plan.operation_system == "linux":
         choosers.choose_linux_image(plan)
         plan.encrypt_system = ui.confirm("Should the system be encrypted?")
+        if plan.encrypt_system:
+            plan.tor_unlock = ui.confirm(
+                "Should the system be remotely unlockable via a Tor onion service?"
+            )
         ui.info("Generating os-image...")
         transfer.download_image(plan)
     else:

lim/image/tor.py

@@ -0,0 +1,79 @@
+"""Tor onion service in the initramfs for remote LUKS unlock.
+
+The onion keys are generated offline inside the image chroot
+(``tor --DisableNetwork 1`` writes them without touching the network)
+and baked into the initramfs by the "tor" mkinitcpio hook, so the
+dropbear unlock shell stays reachable under a stable .onion address
+even behind NAT or a dynamic IP.
+"""
+
+from pathlib import Path
+
+from lim import config, packages, runner, ui
+from lim.errors import LimError
+from lim.image.plan import ImagePlan
+from lim.image.session import chroot_bash, install_packages
+
+# Paths inside the image (relative to the mounted root).
+ONION_DIR = "etc/tor/initramfs-onion"
+TORRC_PATH = "etc/tor/initramfs-torrc"
+
+# An empty -f torrc keeps the image's /etc/tor/torrc (User tor, ...) out of
+# the keygen run; with DisableNetwork the keys appear within a second, the
+# loop only cushions slow qemu-emulated chroots.
+_KEYGEN_SCRIPT = f"""
+mkdir -p /{ONION_DIR}
+chmod 0700 /{ONION_DIR}
+: > /tmp/tor-keygen-torrc
+tor -f /tmp/tor-keygen-torrc --DisableNetwork 1 \\
+    --DataDirectory /tmp/tor-keygen-data \\
+    --HiddenServiceDir /{ONION_DIR} \\
+    --HiddenServicePort "22 127.0.0.1:22" \\
+    --SocksPort 0 --RunAsDaemon 0 --Log "notice stderr" &
+tor_pid=$!
+for _ in $(seq 1 30); do
+    [ -s /{ONION_DIR}/hostname ] && break
+    sleep 1
+done
+kill "$tor_pid" 2>/dev/null || true
+rm -rf /tmp/tor-keygen-data /tmp/tor-keygen-torrc
+[ -s /{ONION_DIR}/hostname ]
+"""
+
+
+def _install_initcpio_files(root: Path) -> None:
+    source_dir = config.CONFIGURATION_PATH / "initcpio"
+    for source, target in (
+        (source_dir / "tor_install", root / "etc/initcpio/install/tor"),
+        (source_dir / "tor_hook", root / "etc/initcpio/hooks/tor"),
+        (source_dir / "torrc", root / TORRC_PATH),
+    ):
+        ui.info(f"Installing {target}...")
+        runner.run(
+            ["install", "-D", "-m", "0644", str(source), str(target)], sudo=True
+        )
+
+
+def _generate_onion_keys(root: Path) -> str:
+    hostname_path = root / ONION_DIR / "hostname"
+    if hostname_path.is_file():
+        ui.info("Onion keys already exist, keeping the existing address.")
+    else:
+        ui.info("Generating onion service keys (offline, inside the chroot)...")
+        chroot_bash(root, _KEYGEN_SCRIPT, error_msg="Onion key generation failed.")
+        if not hostname_path.is_file():
+            raise LimError(f"Onion key generation produced no {hostname_path}.")
+    return hostname_path.read_text().strip()
+
+
+def configure_tor_unlock(plan: ImagePlan, root: Path) -> str:
+    """Install everything the "tor" mkinitcpio hook bakes in; return the onion address."""
+    ui.info("Setting up remote unlock via Tor onion service...")
+    install_packages(
+        plan.distribution, root, " ".join(packages.get_packages("server/tor"))
+    )
+    _install_initcpio_files(root)
+    onion_address = _generate_onion_keys(root)
+    ui.success(f"Onion unlock address: {onion_address}")
+    ui.info(f"Unlock later with: torsocks ssh root@{onion_address}")
+    return onion_address

pyproject.toml

@@ -1,11 +1,39 @@
+[build-system]
+# 77+ for PEP 639 SPDX `license = "..."` + `license-files`.
+requires = ["setuptools>=77"]
+build-backend = "setuptools.build_meta"
+
 [project]
 name = "linux-image-manager"
-version = "2.0.0"
-description = "Linux Image Manager — manages Linux images and encrypted storage"
+version = "2.2.0"
+description = "CLI (`lim`) for downloading, verifying and flashing Linux images, LUKS/Btrfs encrypted storage (single drive and RAID1), image backups and chroot maintenance"
+readme = "README.md"
 requires-python = ">=3.10"
-license = { file = "LICENSE.txt" }
+authors = [{ name = "Kevin Veen-Birkenbach", email = "kevin@veen.world" }]
+maintainers = [{ name = "Kevin Veen-Birkenbach", email = "kevin@veen.world" }]
+license = "GPL-3.0-only"
+license-files = ["LICENSE.txt"]
+urls = { Homepage = "https://veen.world", Repository = "https://github.com/kevinveenbirkenbach/linux-image-manager" }
+classifiers = [
+  "Environment :: Console",
+  "Operating System :: POSIX :: Linux",
+  "Programming Language :: Python :: 3",
+]
 dependencies = ["PyYAML>=6"]
 
+[project.scripts]
+lim = "lim.cli:main"
+
+[tool.setuptools.packages.find]
+include = ["lim*"]
+
+[tool.setuptools.package-data]
+lim = [
+  "distributions.yml",
+  "configuration/packages/**/*.txt",
+  "configuration/initcpio/*",
+]
+
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 

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.

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

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>-*.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" <<EOF
+MODULES=(virtio_pci virtio_blk virtio_net)
+BINARIES=(/usr/lib/libgcc_s.so.1)
+FILES=()
+HOOKS=(base udev modconf keyboard block ptsmount netconf tor dropbear encryptssh filesystems fsck)
+EOF
+
+banner "adding a devpts-mount hook (dropbear needs a PTY for encryptssh)"
+# Without a mounted devpts, dropbear cannot allocate /dev/pts/0, so the
+# interactive encryptssh passphrase prompt never runs and the SSH session
+# just exits ("TIOCSCTTY: Input/output error", "/dev/pts/0: No such file").
+cat > "$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" <<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"

tests/e2e/qemu/config.py

@@ -0,0 +1,132 @@
+"""Pure, side-effect-free builders for the QEMU unlock harness.
+
+Everything here is deterministic string/list construction so it can be
+unit-tested without QEMU, root, or a network (see test_qemu_harness_unit.py).
+The stages that actually touch the system live in build_image.sh,
+chutney.py and boot_unlock.py.
+"""
+
+from dataclasses import dataclass
+from pathlib import Path
+
+# Printed to /dev/console by a oneshot service baked into the image; its
+# appearance on the serial log is the proof that the LUKS root actually
+# unlocked and the real system finished booting.
+BOOT_OK_MARKER = "LIM_UNLOCK_BOOT_OK"
+
+# The HOOKS the built image must carry — the tor hook sits between netconf
+# and dropbear, exactly as lim.image.encryption wires it for real images.
+EXPECTED_HOOKS_ORDER = ("netconf", "tor", "dropbear", "encryptssh")
+
+# Serial console device per architecture (kernel console= and getty).
+_SERIAL_CONSOLE = {"x86_64": "ttyS0", "aarch64": "ttyAMA0"}
+_QEMU_BINARY = {"x86_64": "qemu-system-x86_64", "aarch64": "qemu-system-aarch64"}
+
+
+@dataclass
+class QemuSpec:
+    """Everything needed to boot one built image and unlock it."""
+
+    arch: str
+    work_dir: Path
+    image_path: Path
+    kernel_path: Path
+    initramfs_path: Path
+    luks_uuid: str
+    mapper_name: str
+    passphrase: str
+    ssh_key_path: Path
+    onion_address: str = ""
+    socks_port: int = 9050
+    # 2 GiB so the LUKS argon2 KDF has enough memory to unlock in the guest
+    # (build_image.sh also caps --pbkdf-memory to keep this headroom).
+    memory_mb: int = 2048
+    accel: str = "tcg"  # "kvm" when the host arch matches and /dev/kvm exists
+
+    @property
+    def serial_log(self) -> 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
+    # <client>:<server>:<gw>:<netmask>:<host>:<device>:<proto> — 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}",
+    ]

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)

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

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"

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

tests/e2e/test_tor_unlock_e2e.py

@@ -0,0 +1,114 @@
+"""End-to-end test of the Tor onion LUKS-unlock path (rootless).
+
+It reproduces the operator-visible half of the decryption process that the
+"tor" mkinitcpio hook drives at boot:
+
+  1. generate the v3 onion keys offline (as lim.image.tor does in the chroot),
+  2. start a real Tor onion service from a torrc mirroring the baked-in one,
+     forwarding the virtual port 22 to a local dropbear stand-in,
+  3. connect to the .onion address through Tor and deliver the passphrase,
+  4. assert the passphrase reached the unlock endpoint and it "unlocked".
+
+The physical flashing half (loop device, cryptsetup, mount, chroot,
+mkinitcpio) needs root and is out of scope here by design.
+
+Requires the real `tor` binary and live Tor network access, so it is
+opt-in and skipped unless LIM_E2E_TOR=1:
+
+    LIM_E2E_TOR=1 pytest tests/e2e/test_tor_unlock_e2e.py -v
+"""
+
+import os
+import shutil
+import time
+
+import pytest
+
+from lim.image import tor as tor_module
+from tests.e2e import tor_harness
+
+# The offline checks only need the tor binary (no network) and are
+# deterministic, so they run in CI whenever tor is installed. Only the live
+# onion round-trip needs the public Tor network, so it stays opt-in behind
+# LIM_E2E_TOR=1 to keep an external, occasionally-flaky dependency out of the
+# blocking gate.
+_needs_tor = pytest.mark.skipif(
+    shutil.which("tor") is None, reason="needs the tor binary"
+)
+_needs_tor_network = pytest.mark.skipif(
+    shutil.which("tor") is None or os.environ.get("LIM_E2E_TOR") != "1",
+    reason="needs the tor binary and LIM_E2E_TOR=1 (live Tor network, slow)",
+)
+
+PASSPHRASE = b"correct horse battery staple"
+
+
+@pytest.fixture
+def workdir(tmp_path):
+    (tmp_path / "onion").mkdir()
+    (tmp_path / "data").mkdir()
+    return tmp_path
+
+
+def test_offline_keygen_matches_production_flags():
+    """Guard: the harness keygen mirrors the flags production actually runs."""
+    script = tor_module._KEYGEN_SCRIPT
+    assert "--DisableNetwork 1" in script
+    assert "HiddenServicePort" in script
+    assert "22 127.0.0.1:22" in script
+    assert "--SocksPort 0" in script
+
+
+@_needs_tor
+def test_onion_keygen_is_deterministic_and_offline(workdir):
+    """Keys generate without network and the .onion address is stable."""
+    address = tor_harness.generate_onion_keys(
+        workdir / "onion", workdir / "data"
+    )
+    assert address.endswith(".onion")
+    assert len(address) == len("v" * 56) + len(".onion")  # v3 = 56 base32 chars
+    for name in ("hs_ed25519_secret_key", "hs_ed25519_public_key", "hostname"):
+        assert (workdir / "onion" / name).is_file()
+
+
+@_needs_tor_network
+def test_unlock_passphrase_travels_over_onion(workdir):
+    """Full rootless round-trip: client -> Tor -> onion -> dropbear stand-in."""
+    onion_dir = workdir / "onion"
+    data_dir = workdir / "data"
+    onion_address = tor_harness.generate_onion_keys(onion_dir, data_dir)
+
+    backend_port = tor_harness.free_port()
+    socks_port = tor_harness.free_port()
+    torrc = workdir / "torrc"
+    log_path = workdir / "tor.log"
+    tor_harness.write_test_torrc(torrc, data_dir, onion_dir, backend_port)
+
+    dropbear = tor_harness.FakeDropbear(backend_port)
+    dropbear.start()
+    service = tor_harness.start_onion_service(torrc, socks_port, log_path)
+    try:
+        tor_harness.wait_bootstrapped(log_path, service)
+
+        # Onion descriptors need a moment to publish after bootstrap; retry.
+        last_error = None
+        for _ in range(5):
+            try:
+                stream = tor_harness.socks5_connect(socks_port, onion_address, 22)
+                break
+            except (OSError, RuntimeError) as exc:
+                last_error = exc
+                time.sleep(2)
+        else:
+            pytest.fail(f"Could not reach {onion_address} via Tor: {last_error}")
+
+        with stream:
+            stream.sendall(PASSPHRASE + b"\n")
+            reply = stream.recv(64)
+    finally:
+        service.terminate()
+        service.wait(timeout=15)
+        dropbear.stop()
+
+    assert reply.strip() == b"UNLOCKED"
+    assert dropbear.received == PASSPHRASE

tests/e2e/tor_harness.py

@@ -0,0 +1,178 @@
+"""Rootless helpers to stand up a real Tor onion service and reach it.
+
+These reproduce, without root/block-devices/chroot, what the "tor"
+mkinitcpio hook does at LUKS-decryption time: generate v3 onion keys
+offline, run a real Tor onion service that forwards the virtual port 22
+to a local "dropbear", and let a client reach that service through Tor.
+"""
+
+import socket
+import struct
+import subprocess
+import threading
+import time
+from pathlib import Path
+
+# Matches lim.image.tor._KEYGEN_SCRIPT: an empty -f torrc keeps the host's
+# /etc/tor/torrc out of the run, --DisableNetwork 1 makes the keys appear
+# without ever touching the network.
+KEYGEN_TIMEOUT = 30
+BOOTSTRAP_TIMEOUT = 180
+ONION_CONNECT_TIMEOUT = 90
+
+
+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]
+
+
+def generate_onion_keys(onion_dir: Path, data_dir: Path) -> str:
+    """Create v3 onion keys offline; return the .onion hostname.
+
+    Mirrors lim.image.tor._KEYGEN_SCRIPT, run directly on the host instead
+    of inside the image chroot (the tor invocation is identical).
+    """
+    onion_dir.mkdir(parents=True, exist_ok=True)
+    onion_dir.chmod(0o700)
+    empty_torrc = data_dir.parent / "keygen-torrc"
+    empty_torrc.write_text("")
+    process = subprocess.Popen(
+        [
+            "tor", "-f", str(empty_torrc), "--DisableNetwork", "1",
+            "--DataDirectory", str(data_dir),
+            "--HiddenServiceDir", str(onion_dir),
+            "--HiddenServicePort", "22 127.0.0.1:22",
+            "--SocksPort", "0", "--RunAsDaemon", "0", "--Log", "notice stderr",
+        ],
+        stderr=subprocess.DEVNULL,
+    )
+    hostname = onion_dir / "hostname"
+    try:
+        deadline = time.monotonic() + KEYGEN_TIMEOUT
+        while time.monotonic() < deadline:
+            if hostname.is_file() and hostname.read_text().strip():
+                break
+            time.sleep(0.5)
+    finally:
+        process.terminate()
+        process.wait(timeout=10)
+    if not (hostname.is_file() and hostname.read_text().strip()):
+        raise RuntimeError("Offline onion key generation produced no hostname.")
+    return hostname.read_text().strip()
+
+
+def start_onion_service(
+    torrc_path: Path, socks_port: int, log_path: Path
+) -> subprocess.Popen:
+    """Start Tor with the given torrc (service side) plus a SOCKS port.
+
+    The torrc carries the HiddenServiceDir/HiddenServicePort just like the
+    baked-in image torrc; --SocksPort lets the same instance also act as the
+    client's entry point so the test needs only one bootstrap.
+    """
+    return subprocess.Popen(
+        [
+            "tor", "-f", str(torrc_path),
+            "--SocksPort", str(socks_port),
+            "--Log", f"notice file {log_path}",
+        ],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+
+
+def wait_bootstrapped(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}).")
+        if log_path.is_file() and "Bootstrapped 100%" in log_path.read_text():
+            return
+        time.sleep(1)
+    raise RuntimeError("Tor did not bootstrap in time.")
+
+
+def _socks5_handshake(sock: socket.socket, onion_host: str, onion_port: int) -> None:
+    sock.sendall(b"\x05\x01\x00")  # VER, 1 method, no-auth
+    if sock.recv(2) != b"\x05\x00":
+        raise RuntimeError("SOCKS5 handshake rejected.")
+    host = onion_host.encode()
+    request = (
+        b"\x05\x01\x00\x03"
+        + bytes([len(host)]) + host
+        + struct.pack(">H", onion_port)
+    )
+    sock.settimeout(ONION_CONNECT_TIMEOUT)
+    sock.sendall(request)
+    header = sock.recv(4)
+    if len(header) < 2 or header[1] != 0x00:
+        code = header[1] if len(header) >= 2 else -1
+        raise RuntimeError(f"SOCKS5 CONNECT failed (reply code {code}).")
+    # Drain the bound-address tail so the stream starts at real payload.
+    atyp = header[3]
+    tail = {0x01: 4, 0x04: 16}.get(atyp)
+    if tail is None:  # domain name
+        tail = sock.recv(1)[0]
+    sock.recv(tail + 2)
+
+
+def socks5_connect(socks_port: int, onion_host: str, onion_port: int) -> socket.socket:
+    """Open a stream to onion_host:onion_port through Tor's SOCKS5 proxy."""
+    sock = socket.create_connection(("127.0.0.1", socks_port), timeout=30)
+    try:
+        _socks5_handshake(sock, onion_host, onion_port)
+    except (OSError, RuntimeError):
+        sock.close()
+        raise
+    return sock
+
+
+class FakeDropbear:
+    r"""Stand-in for the initramfs dropbear+cryptsetup unlock endpoint.
+
+    Accepts one connection on 127.0.0.1:<port>, reads a passphrase line, and
+    replies "UNLOCKED\n" — the observable proof that a passphrase delivered
+    over the onion connection reached the unlock shell.
+    """
+
+    def __init__(self, port: int) -> None:
+        self.port = port
+        self.received: bytes | None = None
+        self._server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self._server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self._server.bind(("127.0.0.1", port))
+        self._server.listen(1)
+        self._thread = threading.Thread(target=self._serve, daemon=True)
+
+    def start(self) -> None:
+        self._thread.start()
+
+    def _serve(self) -> None:
+        try:
+            self._server.settimeout(BOOTSTRAP_TIMEOUT + ONION_CONNECT_TIMEOUT)
+            conn, _ = self._server.accept()
+            with conn:
+                conn.settimeout(ONION_CONNECT_TIMEOUT)
+                self.received = conn.recv(4096).strip()
+                conn.sendall(b"UNLOCKED\n")
+        except OSError:
+            pass
+
+    def stop(self) -> None:
+        self._server.close()
+        self._thread.join(timeout=5)
+
+
+def write_test_torrc(path: Path, data_dir: Path, onion_dir: Path, backend_port: int) -> None:
+    """Write a torrc mirroring the baked-in one, with test-writable paths.
+
+    The virtual onion port stays 22 (as in production); only the loopback
+    backend and the data/key directories move to rootless temp locations.
+    """
+    path.write_text(
+        f"DataDirectory {data_dir}\n"
+        f"HiddenServiceDir {onion_dir}\n"
+        f"HiddenServicePort 22 127.0.0.1:{backend_port}\n"
+        "SocksPort 0\n"
+    )

tests/unit/test_tor.py

@@ -0,0 +1,137 @@
+import pytest
+
+from lim import config
+from lim.device import Device
+from lim.errors import LimError
+from lim.image import encryption, tor
+from lim.image.plan import ImagePlan
+from lim.image.session import ImageSession
+
+
+@pytest.fixture
+def plan():
+    return ImagePlan(distribution="arch", raspberry_pi_version="4", tor_unlock=True)
+
+
+def _write_onion_hostname(root, address="abcdefghijklmnop.onion"):
+    onion_dir = root / tor.ONION_DIR
+    onion_dir.mkdir(parents=True)
+    (onion_dir / "hostname").write_text(f"{address}\n")
+
+
+class TestConfigureTorUnlock:
+    def test_installs_hooks_and_returns_address(self, tmp_path, plan, fake_runner):
+        _write_onion_hostname(tmp_path, "stableaddress.onion")
+
+        address = tor.configure_tor_unlock(plan, tmp_path)
+
+        assert address == "stableaddress.onion"
+        package_installs = [
+            input_text
+            for _, _, input_text in fake_runner.calls
+            if input_text and "pacman" in input_text and "tor busybox" in input_text
+        ]
+        assert len(package_installs) == 1
+        assert len(fake_runner.find("install", "tor_install", "etc/initcpio/install/tor")) == 1
+        assert len(fake_runner.find("install", "tor_hook", "etc/initcpio/hooks/tor")) == 1
+        assert len(fake_runner.find("install", "torrc", "etc/tor/initramfs-torrc")) == 1
+        # Existing keys must be kept: no keygen chroot run.
+        assert fake_runner.find("chroot", "DisableNetwork") == []
+
+    def test_generates_keys_when_missing(self, tmp_path, plan, fake_runner):
+        with pytest.raises(LimError, match="produced no"):
+            tor.configure_tor_unlock(plan, tmp_path)
+        # FakeRunner executes nothing, so the hostname file never appears —
+        # but the keygen chroot script must have been issued exactly once.
+        keygen_calls = [
+            (kind, cmd, input_text)
+            for kind, cmd, input_text in fake_runner.calls
+            if input_text and "DisableNetwork" in input_text
+        ]
+        assert len(keygen_calls) == 1
+        assert "HiddenServiceDir" in keygen_calls[0][2]
+
+    def test_hook_resources_exist(self):
+        source_dir = config.CONFIGURATION_PATH / "initcpio"
+        for name in ("tor_install", "tor_hook", "torrc"):
+            assert (source_dir / name).is_file()
+
+
+class TestInitcpioHookHardening:
+    """Guards for review findings in the shipped initramfs hooks."""
+
+    def _read(self, name):
+        return (config.CONFIGURATION_PATH / "initcpio" / name).read_text()
+
+    def test_install_bakes_the_dns_resolver(self):
+        # Without the NSS module the NTP hostname never resolves and the clock
+        # stays at 1970, so Tor rejects the consensus and never publishes.
+        assert "libnss_dns.so.2" in self._read("tor_install")
+
+    def test_hook_never_sources_dhcp_lease_files(self):
+        # Sourcing /tmp/net-*.conf would run attacker-controlled DHCP option
+        # strings as root before the LUKS root is unlocked.
+        hook = self._read("tor_hook")
+        assert '. "$conf"' not in hook
+        assert "sed -n 's/^IPV4DNS" in hook
+
+    def test_hook_attempts_ntp_without_gating_on_dhcp_dns(self):
+        # NTP must run even when tor_ntp is an IP literal (no DNS needed).
+        hook = self._read("tor_hook")
+        assert "skipping NTP sync" not in hook
+
+
+class TestBootloaderNetworking:
+    """The cmdline.txt boot path (RPi4-class) must set ip= for remote unlock."""
+
+    def _session(self, tmp_path):
+        session = ImageSession(Device("mmcblk0"))
+        session.root_partition_uuid = "ROOT-UUID"
+        session.root_mapper_name = "cryptroot"
+        session.root_mapper_path = "/dev/mapper/cryptroot"
+        session.boot_mount_path = tmp_path / "boot"
+        session.boot_mount_path.mkdir()
+        return session
+
+    def test_cmdline_txt_gets_network_params(self, tmp_path, plan):
+        root = tmp_path / "root"
+        (root / "etc").mkdir(parents=True)
+        (root / "etc" / "hostname").write_text("myhost\n")
+        session = self._session(tmp_path)
+        (session.boot_mount_path / "cmdline.txt").write_text(
+            "root=/dev/mmcblk0p2 rw rootwait\n"
+        )
+
+        encryption._configure_bootloader(plan, session, root)
+
+        content = (session.boot_mount_path / "cmdline.txt").read_text()
+        assert "ip=::::myhost:eth0:dhcp" in content
+        assert "net.ifnames=0" in content
+        assert "cryptdevice=UUID=ROOT-UUID:cryptroot" in content
+        assert "root=/dev/mmcblk0p2" not in content
+
+
+class TestMkinitcpioHooksLine:
+    def _mkinitcpio_conf(self, root):
+        path = root / "etc/mkinitcpio.conf"
+        path.parent.mkdir(parents=True)
+        path.write_text(
+            "MODULES=()\n"
+            "BINARIES=()\n"
+            f"HOOKS=({encryption.MKINITCPIO_HOOKS_PREFIX} "
+            f"{encryption.MKINITCPIO_HOOKS_SUFFIX})\n"
+        )
+        return path
+
+    def test_tor_hook_between_netconf_and_dropbear(self, tmp_path, plan, fake_runner):
+        path = self._mkinitcpio_conf(tmp_path)
+        encryption._configure_mkinitcpio(plan, tmp_path)
+        assert "netconf tor dropbear encryptssh" in path.read_text()
+
+    def test_no_tor_hook_when_disabled(self, tmp_path, plan, fake_runner):
+        plan.tor_unlock = False
+        path = self._mkinitcpio_conf(tmp_path)
+        encryption._configure_mkinitcpio(plan, tmp_path)
+        content = path.read_text()
+        assert "netconf dropbear encryptssh" in content
+        assert " tor " not in content