feat(cli): numbered selection menus for distributions and versions

Replace the free-text distribution/version prompts with numbered menus via a
shared _choose helper. Name-first resolution keeps typing the name working, so
numeric catalog keys (e.g. arch "4") still select by name, not by position.
Applied to the distribution, arch, manjaro, retropie and raspios choosers.

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

.github/workflows/test.yml

@@ -3,6 +3,7 @@ name: tests
 on:
   push:
   pull_request:
+  workflow_dispatch:
 
 jobs:
   lint:
@@ -18,5 +19,37 @@ 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
+
+  qemu-e2e-debian:
+    # Full build -> boot -> LUKS-unlock in QEMU, deterministic via the direct
+    # transport (debootstrap works on Ubuntu; no Tor). Heavy (root, loop,
+    # cryptsetup, and TCG when the runner has no /dev/kvm).
+    runs-on: ubuntu-latest
+    timeout-minutes: 45
+    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 qemu-system-x86 debootstrap cryptsetup
+      - run: pip install pytest pyyaml
+      - run: make test-qemu-debian PYTHON=python

CHANGELOG.md

@@ -1,5 +1,19 @@
 # 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 🥳

Makefile

@@ -1,6 +1,9 @@
-.PHONY: install test
+.PHONY: install test test-tor test-qemu test-qemu-debian 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,27 @@ 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 and ssh; the build stage runs under sudo. The default "direct"
+# transport delivers the passphrase deterministically via a QEMU port-forward;
+# set LIM_E2E_TRANSPORT=tor for the full onion path (also needs tor + ncat, and
+# is flaky over public Tor inside QEMU).
+test-qemu:
+	LIM_E2E_QEMU=1 $(PYTHON) -m pytest tests/e2e/test_qemu_unlock_e2e.py -v -s
+
+# Same, but builds a Debian image with debootstrap (initramfs-tools backend)
+# instead of Arch. Needs debootstrap + cryptsetup + ssh; runs under sudo.
+test-qemu-debian:
+	LIM_E2E_QEMU=1 LIM_E2E_OS=debian $(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 ⚙️
 
@@ -94,6 +95,8 @@ lim/
   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)
 ```
 
@@ -102,15 +105,83 @@ tests/lint/           # architecture guards (e.g. max file length)
 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

@@ -27,5 +27,9 @@ def retropie_images() -> dict[str, dict[str, str]]:
     return _load()["retropie_images"]
 
 
+def raspios_images() -> dict[str, dict[str, str]]:
+    return _load()["raspios_images"]
+
+
 def mkinitcpio_modules_by_rpi() -> dict[str, str]:
     return _load()["mkinitcpio_modules_by_rpi"]

lim/cli.py

@@ -10,6 +10,8 @@ from lim.data import crypt, sync
 from lim.errors import LimError
 from lim.image import backup, chroot
 from lim.image import setup as image_setup
+from lim.image import unlock as image_unlock
+from lim.image import wizard as image_wizard
 from lim.storage import raid1, single_drive
 
 
@@ -29,6 +31,20 @@ COMMANDS: dict[str, Command] = {
         "  - Configures boot and root partitions.",
         needs_root=True,
     ),
+    "guided": Command(
+        image_wizard.run_guided,
+        "Guided Encrypted Linux Image Setup (default command):\n"
+        "  - Walks through flashing a Linux image (Arch, Raspberry Pi OS, ...) onto a device.\n"
+        "  - Encrypts the root with LUKS and enables remote unlock over Tor.",
+        needs_root=True,
+    ),
+    "remote-unlock": Command(
+        image_unlock.remote_unlock,
+        "Remote LUKS Boot Unlock:\n"
+        "  - Reaches a waiting initramfs over Tor (onion) or plain SSH.\n"
+        "  - Runs cryptroot-unlock or presents the passphrase prompt.",
+        needs_root=False,
+    ),
     "single": Command(
         single_drive.setup,
         "Single Drive Encryption Setup:\n"
@@ -59,14 +75,12 @@ COMMANDS: dict[str, Command] = {
     ),
     "mount": Command(
         single_drive.mount,
-        "Mount Encrypted Storage:\n"
+        "Mount Encrypted Storage:\n  - Unlocks a LUKS partition and mounts it.",
-        "  - Unlocks a LUKS partition and mounts it.",
         needs_root=True,
     ),
     "umount": Command(
         single_drive.umount,
-        "Unmount Encrypted Storage:\n"
+        "Unmount Encrypted Storage:\n  - Unmounts a LUKS partition and closes the mapper.",
-        "  - Unmounts a LUKS partition and closes the mapper.",
         needs_root=True,
     ),
     "single-boot": Command(
@@ -83,26 +97,22 @@ COMMANDS: dict[str, Command] = {
     ),
     "unlock": Command(
         crypt.unlock,
-        "Unlock Data:\n"
+        "Unlock Data:\n  - Decrypts the encfs data store into the decrypted folder.",
-        "  - Decrypts the encfs data store into the decrypted folder.",
         needs_root=False,
     ),
     "lock": Command(
         crypt.lock,
-        "Lock Data:\n"
+        "Lock Data:\n  - Unmounts the decrypted encfs folder.",
-        "  - Unmounts the decrypted encfs folder.",
         needs_root=False,
     ),
     "import": Command(
         sync.import_from_system,
-        "Import Data:\n"
+        "Import Data:\n  - Copies personal data from the system into the encrypted store.",
-        "  - Copies personal data from the system into the encrypted store.",
         needs_root=False,
     ),
     "export": Command(
         sync.export_to_system,
-        "Export Data:\n"
+        "Export Data:\n  - Copies personal data from the encrypted store back to the system.",
-        "  - Copies personal data from the encrypted store back to the system.",
         needs_root=False,
     ),
 }
@@ -122,21 +132,15 @@ def build_parser() -> argparse.ArgumentParser:
     )
     parser.add_argument(
         "--type",
-        required=True,
+        default="guided",
         choices=list(COMMANDS.keys()),
-        help="Select the command to execute.",
+        help="Command to execute (default: guided).",
     )
     parser.add_argument(
         "--auto-confirm",
         action="store_true",
         help="Automatically confirm execution without prompting the user.",
     )
-    parser.add_argument(
-        "--extra",
-        nargs=argparse.REMAINDER,
-        default=[],
-        help="Deprecated, ignored. Former extra parameters for the shell scripts.",
-    )
     return parser
 
 
@@ -144,8 +148,6 @@ def main(argv: list[str] | None = None) -> None:
     args = build_parser().parse_args(argv)
     command = COMMANDS[args.type]
 
-    if args.extra:
-        ui.warning("--extra is deprecated and ignored.")
     if command.needs_root:
         system.ensure_root()
 

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/initramfs-tools/tor_bottom

@@ -0,0 +1,19 @@
+#!/bin/sh
+# initramfs-tools cleanup hook: stop Tor before the pivot to the real root.
+# Installed to /etc/initramfs-tools/scripts/init-bottom/tor by lim.
+# Nothing from the initramfs may keep running once run-init replaces it.
+PREREQ=""
+prereqs() { echo "$PREREQ"; }
+case "$1" in
+    prereqs) prereqs; exit 0 ;;
+esac
+
+# pidof/killall may be absent; scan /proc for the tor binary (no extra tools).
+for pid in $(ls /proc 2>/dev/null); do
+    case "$pid" in
+        *[!0-9]*) continue ;;
+    esac
+    [ "$(readlink "/proc/$pid/exe" 2>/dev/null)" = "/usr/bin/tor" ] \
+        && kill "$pid" 2>/dev/null
+done
+exit 0

lim/configuration/initramfs-tools/tor_hook

@@ -0,0 +1,37 @@
+#!/bin/sh
+# initramfs-tools build hook: bake the Tor onion service into the initramfs.
+# Installed to /etc/initramfs-tools/hooks/tor by linux-image-manager.
+PREREQ="dropbear"
+prereqs() { echo "$PREREQ"; }
+case "$1" in
+    prereqs) prereqs; exit 0 ;;
+esac
+
+. /usr/share/initramfs-tools/hook-functions
+
+copy_exec /usr/bin/tor /usr/bin
+
+# glibc resolves the NTP server hostname via getaddrinfo(), which dlopen()s
+# these NSS modules at runtime; without them DNS silently fails and the clock
+# (RTC-less boards) stays at 1970, so Tor rejects the consensus.
+for nss in /usr/lib/*/libnss_dns.so.2 /usr/lib/*/libnss_files.so.2 \
+           /lib/*/libnss_dns.so.2 /lib/*/libnss_files.so.2; do
+    [ -e "$nss" ] && copy_exec "$nss"
+done
+
+# Full busybox for the ntpd applet (the initramfs busybox may lack it); a
+# distinct path keeps the initramfs's own busybox untouched.
+for bb in /bin/busybox /usr/bin/busybox; do
+    [ -x "$bb" ] && { copy_exec "$bb" /usr/local/bin/busybox; break; }
+done
+
+date -u +%s > "${DESTDIR}/etc/tor-clock-floor"
+
+# Onion keys + torrc, staged on the system by lim.
+mkdir -p "${DESTDIR}/etc/tor/onion"
+for key in hostname hs_ed25519_public_key hs_ed25519_secret_key; do
+    cp -a "/etc/tor/initramfs-onion/${key}" "${DESTDIR}/etc/tor/onion/${key}"
+done
+chmod 0700 "${DESTDIR}/etc/tor/onion"
+chmod 0600 "${DESTDIR}/etc/tor/onion/hs_ed25519_secret_key"
+cp -a /etc/tor/initramfs-torrc "${DESTDIR}/etc/tor/torrc"

lim/configuration/initramfs-tools/tor_premount

@@ -0,0 +1,55 @@
+#!/bin/sh
+# initramfs-tools runtime hook: bring up networking, sync the clock and start
+# the Tor onion service so the dropbear unlock shell is reachable while
+# cryptsetup waits. Installed to /etc/initramfs-tools/scripts/init-premount/tor.
+PREREQ=""
+prereqs() { echo "$PREREQ"; }
+case "$1" in
+    prereqs) prereqs; exit 0 ;;
+esac
+
+. /scripts/functions
+
+log_begin_msg "tor: bringing up networking and starting the onion service"
+
+# init-premount runs BEFORE the cryptroot/dropbear networking, so Tor would
+# start with no network and never publish the onion. Bring the interface up
+# ourselves from the ip= cmdline (idempotent; the later dropbear setup reuses
+# the lease in /run/net-*.conf).
+configure_networking
+
+# initramfs-tools does not export arbitrary cmdline params as shell vars.
+tor_ntp=$(sed -n 's/.*\btor_ntp=\([^ ]*\).*/\1/p' /proc/cmdline)
+
+# Best-effort DNS from the DHCP lease; extract ONLY the DNS fields with sed,
+# never source the files (attacker-controllable DHCP option strings would run
+# as root pre-boot).
+if [ ! -s /etc/resolv.conf ]; then
+    for conf in /run/net-*.conf /tmp/net-*.conf; do
+        [ -f "$conf" ] || continue
+        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
+fi
+
+floor=$(cat /etc/tor-clock-floor 2>/dev/null || echo 0)
+if [ "$(date +%s)" -lt "$floor" ]; then
+    /usr/local/bin/busybox date -s "@$floor" >/dev/null 2>&1 \
+        || date -s "@$floor" >/dev/null 2>&1 || true
+fi
+
+if [ -x /usr/local/bin/busybox ]; then
+    /usr/local/bin/busybox timeout 30 \
+        /usr/local/bin/busybox ntpd -n -q -p "${tor_ntp:-pool.ntp.org}" \
+        || log_warning_msg "tor: NTP sync failed, keeping current clock"
+fi
+
+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 /run/tor.log" \
+    || log_warning_msg "tor: failed to start, unlock stays reachable via direct IP"
+
+log_end_msg

lim/configuration/initramfs-tools/torrc

@@ -0,0 +1,16 @@
+# Tor configuration for the initramfs onion unlock service (initramfs-tools).
+# Baked into the initramfs as /etc/tor/torrc by the "tor" 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
+# Single-hop, non-anonymous onion service. A boot-unlock onion needs
+# reachability (behind NAT / a dynamic IP), NOT server-location anonymity, so
+# it can skip the multi-hop, vanguards-restricted circuits that a fresh Tor
+# with a limited relay view fails to build ("Giving up on launching a
+# rendezvous circuit"). Single-hop builds trivial 1-hop circuits that publish
+# and rendezvous reliably. Entry guards are likewise unnecessary here.
+HiddenServiceNonAnonymousMode 1
+HiddenServiceSingleHopMode 1
+UseEntryGuards 0

lim/configuration/packages/server/luks-debian.txt

@@ -0,0 +1,5 @@
+# Packages for LUKS remote unlock on Debian / Raspberry Pi OS (initramfs-tools)
+cryptsetup
+cryptsetup-initramfs
+dropbear-initramfs
+busybox

lim/configuration/packages/server/tor-debian.txt

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

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/data/sync.py

@@ -90,31 +90,22 @@ def execute_sync_plan(operations: list[SyncOperation]) -> None:
         if not operation.is_directory and Path(operation.destination).is_file():
             ui.info("The destination file already exists!")
             ui.info("Difference:")
-            runner.run(
+            runner.run(["diff", operation.destination, operation.source], check=False)
-                ["diff", operation.destination, operation.source], check=False
-            )
         runner.run(rsync_command(operation))
 
 
 def ensure_unlocked() -> None:
     mounts = runner.output(["mount"], check=False)
     if str(config.DECRYPTED_PATH) not in mounts:
-        ui.info(
+        ui.info(f"The decrypted folder {config.DECRYPTED_PATH} is locked. You need to unlock it!")
-            f"The decrypted folder {config.DECRYPTED_PATH} is locked. "
-            "You need to unlock it!"
-        )
         crypt.unlock()
 
 
 def import_from_system(mode: str = "import") -> None:
     ensure_unlocked()
-    backup_folder = (
+    backup_folder = config.BACKUP_PATH / mode / time.strftime("%Y%m%d%H%M%S")
-        config.BACKUP_PATH / mode / time.strftime("%Y%m%d%H%M%S")
-    )
     backup_folder.mkdir(parents=True, exist_ok=True)
-    operations = build_sync_plan(
+    operations = build_sync_plan(mode, system.real_home(), config.DATA_PATH, backup_folder)
-        mode, system.real_home(), config.DATA_PATH, backup_folder
-    )
     execute_sync_plan(operations)
 
 

lim/device.py

@@ -119,9 +119,7 @@ def overwrite_device(device: Device) -> None:
 
 def blkid_value(path: str, tag: str) -> str:
     """Return a blkid tag value ("TYPE", "UUID", ...) or "" when unavailable."""
-    return runner.output(
+    return runner.output(["blkid", path, "-s", tag, "-o", "value"], sudo=True, check=False)
-        ["blkid", path, "-s", tag, "-o", "value"], sudo=True, check=False
-    )
 
 
 def is_mounted(path_fragment: str) -> bool:

lim/distributions.yml

@@ -57,6 +57,19 @@ retropie_images:
     checksum: b5daa6e7660a99c246966f3f09b4014b
     image: retropie-buster-4.8-rpi4_400.img.gz
 
+# Raspberry Pi OS. The "_latest" endpoints 302-redirect to the current release;
+# image is only the local filename (must end .img.xz for decompress_command).
+raspios_images:
+  lite64:
+    source_url: https://downloads.raspberrypi.com/raspios_lite_arm64_latest
+    image: raspios_lite_arm64_latest.img.xz
+  desktop64:
+    source_url: https://downloads.raspberrypi.com/raspios_arm64_latest
+    image: raspios_arm64_latest.img.xz
+  lite32:
+    source_url: https://downloads.raspberrypi.com/raspios_lite_armhf_latest
+    image: raspios_lite_armhf_latest.img.xz
+
 # Extra kernel modules the initramfs needs for early network access,
 # see https://raspberrypi.stackexchange.com/questions/67051
 mkinitcpio_modules_by_rpi:

lim/fsutil.py

@@ -16,6 +16,23 @@ def replace_in_file(search: str, replace: str, path: str | Path) -> None:
     path.write_text(new_text)
 
 
+def drop_fstab_mount(mount_point: str, path: str | Path) -> None:
+    """Remove any (non-comment) fstab line whose mount point field equals mount_point.
+
+    Lets a fresh mapper/boot entry replace a stock image's existing one instead
+    of colliding with it. No-op when the file or a matching line is absent.
+    """
+    path = Path(path)
+    if not path.exists():
+        return
+    kept = [
+        line
+        for line in path.read_text().splitlines()
+        if line.lstrip().startswith("#") or line.split()[1:2] != [mount_point]
+    ]
+    path.write_text("".join(f"{line}\n" for line in kept))
+
+
 def ensure_line_in_file(line: str, path: str | Path) -> bool:
     """Append ``line`` unless already present. Returns True when appended."""
     path = Path(path)

lim/image/choosers.py

@@ -5,8 +5,21 @@ from lim.errors import LimError
 from lim.image.plan import ImagePlan
 
 
+def _choose(prompt: str, options: list[str]) -> str:
+    """List options numbered; accept either a number or the name itself."""
+    ui.info(f"{prompt}:")
+    for index, name in enumerate(options, 1):
+        ui.info(f"  {index}) {name}")
+    answer = ui.ask("Pick a number or name:").strip()
+    if answer in options:  # name wins, so numeric names (e.g. arch "4") still work
+        return answer
+    if answer.isdigit() and 1 <= int(answer) <= len(options):
+        return options[int(answer) - 1]
+    return answer
+
+
 def choose_arch(plan: ImagePlan) -> None:
-    version = ui.ask("Which Raspberry Pi will be used (e.g.: 1, 2, 3b, 3b+, 4...):")
+    version = _choose("Which Raspberry Pi version", list(catalog.arch_rpi_images()))
     entry = catalog.arch_rpi_images().get(version)
     if entry is None:
         raise LimError(f"Version {version} isn't supported.")
@@ -19,7 +32,7 @@ def choose_arch(plan: ImagePlan) -> None:
 
 def choose_manjaro(plan: ImagePlan) -> None:
     plan.boot_size = "+500M"
-    flavour = ui.ask("Which version(e.g.:architect,gnome) should be used:")
+    flavour = _choose("Which Manjaro version", ["architect", "gnome"])
     if flavour == "architect":
         plan.image_checksum = "6b1c2fce12f244c1e32212767a9d3af2cf8263b2"
         plan.base_download_url = (
@@ -28,7 +41,7 @@ def choose_manjaro(plan: ImagePlan) -> None:
         )
         plan.image_name = "manjaro-architect-20.0-200426-linux56.iso"
     elif flavour == "gnome":
-        release = ui.ask("Which release(e.g.:20,21,raspberrypi) should be used:")
+        release = _choose("Which Gnome release", list(catalog.manjaro_gnome_releases()))
         entry = catalog.manjaro_gnome_releases().get(release)
         if entry is None:
             raise LimError(f"Gnome release {release} isn't supported.")
@@ -44,32 +57,37 @@ def choose_manjaro(plan: ImagePlan) -> None:
 def choose_moode(plan: ImagePlan) -> None:
     plan.boot_size = "+200M"
     plan.image_checksum = "185cbc9a4994534bb7a4bc2744c78197"
-    plan.base_download_url = (
+    plan.base_download_url = "https://github.com/moode-player/moode/releases/download/r651prod/"
-        "https://github.com/moode-player/moode/releases/download/r651prod/"
-    )
     plan.image_name = "moode-r651-iso.zip"
 
 
 def choose_retropie(plan: ImagePlan) -> None:
     plan.boot_size = "+500M"
-    version = ui.ask("Which version(e.g.:1,2,3,4) should be used:")
+    version = _choose("Which RetroPie version", list(catalog.retropie_images()))
     entry = catalog.retropie_images().get(version)
     if entry is None:
         raise LimError(f"Version {version} isn't supported.")
     plan.raspberry_pi_version = version
-    plan.base_download_url = (
+    plan.base_download_url = "https://github.com/RetroPie/RetroPie-Setup/releases/download/4.8/"
-        "https://github.com/RetroPie/RetroPie-Setup/releases/download/4.8/"
-    )
     plan.image_checksum = entry["checksum"]
     plan.image_name = entry["image"]
 
 
+def choose_raspios(plan: ImagePlan) -> None:
+    plan.boot_size = "+512M"
+    plan.root_filesystem = "ext4"
+    variant = _choose("Which Raspberry Pi OS image", list(catalog.raspios_images()))
+    entry = catalog.raspios_images().get(variant)
+    if entry is None:
+        raise LimError(f"Raspberry Pi OS variant {variant} isn't supported.")
+    plan.source_url = entry["source_url"]
+    plan.image_name = entry["image"]
+
+
 def choose_torbox(plan: ImagePlan) -> None:
     plan.base_download_url = "https://www.torbox.ch/data/"
     plan.image_name = "torbox-20220102-v050.gz"
-    plan.image_checksum = (
+    plan.image_checksum = "0E1BA7FFD14AAAE5F0462C8293D95B62C3BF1D9E726E26977BD04772C55680D3"
-        "0E1BA7FFD14AAAE5F0462C8293D95B62C3BF1D9E726E26977BD04772C55680D3"
-    )
     plan.boot_size = "+200M"
 
 
@@ -78,9 +96,7 @@ def choose_android_x86(plan: ImagePlan) -> None:
         "https://www.fosshub.com/Android-x86.html?dwl=android-x86_64-9.0-r2.iso"
     )
     plan.image_name = "android-x86_64-9.0-r2.iso"
-    plan.image_checksum = (
+    plan.image_checksum = "f7eb8fc56f29ad5432335dc054183acf086c539f3990f0b6e9ff58bd6df4604e"
-        "f7eb8fc56f29ad5432335dc054183acf086c539f3990f0b6e9ff58bd6df4604e"
-    )
     plan.boot_size = "+500M"
 
 
@@ -91,13 +107,12 @@ DISTRIBUTION_CHOOSERS = {
     "manjaro": choose_manjaro,
     "moode": choose_moode,
     "retropie": choose_retropie,
+    "raspios": choose_raspios,
 }
 
 
 def choose_linux_image(plan: ImagePlan) -> None:
-    distribution = ui.ask(
+    distribution = _choose("Which distribution", list(DISTRIBUTION_CHOOSERS))
-        "Which distribution should be used [arch,moode,retropie,manjaro,torbox...]?"
-    )
     chooser = DISTRIBUTION_CHOOSERS.get(distribution)
     if chooser is None:
         raise LimError(f"Distribution {distribution} isn't supported.")

lim/image/crossarch.py

@@ -0,0 +1,71 @@
+"""Ensure the host can chroot into a foreign-architecture image (qemu binfmt)."""
+
+import platform
+import shutil
+from pathlib import Path
+
+from lim import runner, ui
+from lim.errors import LimError
+
+# Building natively needs no qemu; on these the chroot runs directly.
+_NATIVE_ARM = ("aarch64", "armv7l", "armv6l", "arm64")
+
+# Repo-based installs only; AUR helpers refuse to run from this root process.
+_BINFMT_INSTALLERS = {
+    "apt-get": ["apt-get", "install", "-y", "qemu-user-static", "binfmt-support"],
+    "dnf": ["dnf", "install", "-y", "qemu-user-static"],
+    "zypper": ["zypper", "--non-interactive", "install", "qemu-linux-user"],
+    "pacman": ["pacman", "-S", "--needed", "--noconfirm", "qemu-user-static-binfmt"],
+}
+
+
+def _entry_enabled(entry: Path) -> bool:
+    try:
+        return "enabled" in entry.read_text()
+    except OSError:
+        return False
+
+
+def qemu_binfmt_enabled() -> bool:
+    binfmt = Path("/proc/sys/fs/binfmt_misc")
+    return binfmt.is_dir() and any(_entry_enabled(e) for e in binfmt.glob("qemu-*"))
+
+
+def auto_enable() -> bool:
+    """Install qemu-user-static via the host package manager and register binfmt."""
+    command = next((cmd for tool, cmd in _BINFMT_INSTALLERS.items() if shutil.which(tool)), None)
+    if command is None:
+        ui.warning("No supported host package manager found (apt-get/dnf/zypper/pacman).")
+        return False
+    runner.run(command, sudo=True, check=False)
+    runner.run(["systemctl", "restart", "systemd-binfmt"], sudo=True, check=False)
+    return qemu_binfmt_enabled()
+
+
+def ensure_ready() -> None:
+    """Make foreign-arch chroot work, or fail with actionable guidance.
+
+    Called before the target is erased, so a host that cannot run the image's
+    binaries aborts early instead of mid-build with a cryptic "Exec format error".
+    """
+    if platform.machine() in _NATIVE_ARM:
+        return
+    if qemu_binfmt_enabled():
+        ui.info("Cross-architecture build: qemu binfmt handlers are registered.")
+        return
+    ui.warning(
+        f"Cross-architecture build: this host ({platform.machine()}) cannot run the "
+        "image's ARM binaries in chroot yet."
+    )
+    if ui.confirm("Install qemu-user-static and register binfmt now?"):
+        if auto_enable():
+            ui.success("qemu binfmt handlers registered.")
+            return
+        ui.warning("Automatic setup did not register a handler.")
+    raise LimError(
+        "Cross-architecture chroot is not available. Install qemu-user-static + binfmt "
+        "manually, or build on the target itself (e.g. the Pi booted from USB).\n"
+        "  Arch/Manjaro : qemu-user-static + qemu-user-static-binfmt (AUR), "
+        "then: sudo systemctl restart systemd-binfmt\n"
+        "  Debian/Ubuntu: sudo apt install qemu-user-static binfmt-support"
+    )

lim/image/encryption.py

@@ -1,117 +1,42 @@
-"""Remote-unlockable LUKS boot configuration (dropbear + mkinitcpio).
+"""Remote-unlockable LUKS boot configuration.
 
-See https://gist.github.com/gea0/4fc2be0cb7a74d0e7cc4322aed710d38 and
+Distro-agnostic orchestration; the init-system-specific steps live in the
-https://gist.github.com/EnigmaCurry/2f9bed46073da8e38057fe78a61e7994
+initramfs backends (mkinitcpio for Arch/Manjaro, initramfs-tools for
+Debian/Raspberry Pi OS). See lim/image/initramfs/.
 """
 
 from pathlib import Path
 
-from lim import catalog, fsutil, packages, runner, ui
+from lim import packages, ui
+from lim.image.initramfs import get_backend
 from lim.image.plan import ImagePlan
-from lim.image.session import ImageSession, chroot_bash, install_packages
+from lim.image.session import ImageSession, install_packages
-
-MKINITCPIO_HOOKS_PREFIX = (
-    "base udev autodetect microcode modconf kms keyboard keymap consolefont block"
-)
-MKINITCPIO_HOOKS_SUFFIX = "filesystems fsck"
-
-
-def _configure_mkinitcpio(plan: ImagePlan, root: Path) -> None:
-    # Concerning mkinitcpio warnings, see
-    # https://gist.github.com/imrvelj/c65cd5ca7f5505a65e59204f5a3f7a6d
-    mkinitcpio_path = root / "etc/mkinitcpio.conf"
-    ui.info(f"Configuring {mkinitcpio_path}...")
-    additional_modules = catalog.mkinitcpio_modules_by_rpi().get(
-        plan.raspberry_pi_version
-    )
-    if additional_modules is None:
-        ui.warning(f"Version {plan.raspberry_pi_version} isn't supported.")
-        additional_modules = ""
-    modules = f"g_cdc usb_f_acm usb_f_ecm {additional_modules} g_ether".replace("  ", " ")
-    fsutil.replace_in_file("MODULES=()", f"MODULES=({modules})", mkinitcpio_path)
-    fsutil.replace_in_file(
-        "BINARIES=()", "BINARIES=(/usr/lib/libgcc_s.so.1)", mkinitcpio_path
-    )
-    fsutil.replace_in_file(
-        f"HOOKS=({MKINITCPIO_HOOKS_PREFIX} {MKINITCPIO_HOOKS_SUFFIX})",
-        f"HOOKS=({MKINITCPIO_HOOKS_PREFIX} sleep netconf dropbear encryptssh "
-        f"{MKINITCPIO_HOOKS_SUFFIX})",
-        mkinitcpio_path,
-    )
-    ui.info(f"Content of {mkinitcpio_path}:{mkinitcpio_path.read_text()}")
-    ui.info("Generating mkinitcpio...")
-    chroot_bash(root, "mkinitcpio -vP")
-
-
-def _register_encrypted_root(plan: ImagePlan, session: ImageSession, root: Path) -> None:
-    fstab_path = root / "etc/fstab"
-    fstab_line = (
-        f"UUID={session.root_partition_uuid}  / {plan.root_filesystem}"
-        "    defaults,noatime  0  1"
-    )
-    ui.info(f"Configuring {fstab_path}...")
-    fsutil.ensure_line_in_file(fstab_line, fstab_path)
-    ui.info(f"Content of {fstab_path}:{fstab_path.read_text()}")
-
-    crypttab_path = root / "etc/crypttab"
-    crypttab_line = (
-        f"{session.root_mapper_name} UUID={session.root_partition_uuid} none luks"
-    )
-    ui.info(f"Configuring {crypttab_path}...")
-    fsutil.ensure_line_in_file(crypttab_line, crypttab_path)
-    ui.info(f"Content of {crypttab_path}:{crypttab_path.read_text()}")
-
-
-def _configure_bootloader(plan: ImagePlan, session: ImageSession, root: Path) -> None:
-    cryptdevice = (
-        f"cryptdevice=UUID={session.root_partition_uuid}:{session.root_mapper_name} "
-        f"root={session.root_mapper_path}"
-    )
-    boot_txt_path = session.boot_mount_path / "boot.txt"
-    if boot_txt_path.is_file():
-        ui.info(f"Configuring {boot_txt_path}...")
-        hostname = (root / "etc/hostname").read_text().strip()
-        fsutil.replace_in_file(
-            "part uuid ${devtype} ${devnum}:2 uuid", "", boot_txt_path
-        )
-        fsutil.replace_in_file(
-            "setenv bootargs console=ttyS1,115200 console=tty0 root=PARTUUID=${uuid} "
-            'rw rootwait smsc95xx.macaddr="${usbethaddr}"',
-            f"setenv bootargs console=ttyS1,115200 console=tty0 "
-            f"ip=::::{hostname}:eth0:dhcp {cryptdevice} rw rootwait "
-            # Concerning issues with network adapter names, see
-            # https://forum.iobroker.net/topic/40542/raspberry-pi4-kein-eth0-mehr/16
-            'smsc95xx.macaddr="${usbethaddr}" net.ifnames=0 biosdevname=0',
-            boot_txt_path,
-        )
-        ui.info(f"Content of {boot_txt_path}:{boot_txt_path.read_text()}")
-        ui.info("Generating...")
-        chroot_bash(root, "cd /boot/ && ./mkscr || exit 1")
-    else:
-        cmdline_txt_path = session.boot_mount_path / "cmdline.txt"
-        ui.info(f"Configuring {cmdline_txt_path}...")
-        fsutil.replace_in_file(
-            "root=/dev/mmcblk0p2",
-            f"{cryptdevice} rootfstype={plan.root_filesystem}",
-            cmdline_txt_path,
-        )
-        ui.info(f"Content of {cmdline_txt_path}:{cmdline_txt_path.read_text()}")
 
 
 def configure_encryption(
     plan: ImagePlan, session: ImageSession, authorized_keys: Path
-) -> None:
+) -> str | None:
+    """Configure the remote-unlock LUKS stack; return the onion address, if any."""
     root = session.root_mount_path
+    backend = get_backend(plan.distribution)
     ui.info("Setup encryption...")
     ui.info("Installing neccessary software...")
     install_packages(
-        plan.distribution, root, " ".join(packages.get_packages("server/luks"))
+        plan.distribution,
+        root,
+        " ".join(packages.get_packages(backend.luks_package_collection())),
     )
 
-    dropbear_root_key_path = root / "etc/dropbear/root_key"
+    backend.install_authorized_key(root, authorized_keys)
-    ui.info(f"Adding {authorized_keys} to dropbear...")
-    runner.run(["cp", "-v", str(authorized_keys), str(dropbear_root_key_path)], sudo=True)
 
-    _configure_mkinitcpio(plan, root)
+    onion_address = None
-    _register_encrypted_root(plan, session, root)
+    if plan.tor_unlock:
-    _configure_bootloader(plan, session, root)
+        # Hook files and onion keys must exist before the initramfs is baked.
+        onion_address = backend.install_tor_unlock(plan, root)
+
+    # crypttab must be written before the initramfs is (re)generated so the
+    # Debian cryptsetup hook can bake the mapping in.
+    backend.register_encrypted_root(plan, session, root)
+    backend.configure_initramfs(plan, root)
+    backend.configure_bootloader(plan, session, root)
+    return onion_address

lim/image/initramfs/__init__.py

@@ -0,0 +1,21 @@
+"""Distro-specific initramfs backends for encrypted remote unlock."""
+
+from lim.errors import LimError
+from lim.image.initramfs.base import InitramfsBackend
+from lim.image.initramfs.initramfs_tools import InitramfsToolsBackend
+from lim.image.initramfs.mkinitcpio import MkinitcpioBackend
+
+_MKINITCPIO = frozenset({"arch", "manjaro"})
+_INITRAMFS_TOOLS = frozenset({"raspios", "moode", "retropie"})
+
+
+def get_backend(distribution: str | None) -> InitramfsBackend:
+    """Pick the initramfs backend for a distribution's init system."""
+    if distribution in _MKINITCPIO:
+        return MkinitcpioBackend()
+    if distribution in _INITRAMFS_TOOLS:
+        return InitramfsToolsBackend()
+    raise LimError(f"No initramfs backend for distribution {distribution!r}.")
+
+
+__all__ = ["InitramfsBackend", "get_backend"]

lim/image/initramfs/base.py

@@ -0,0 +1,42 @@
+"""The initramfs backend interface.
+
+Encrypted remote-unlock images differ by init system: Arch/Manjaro use
+mkinitcpio, Debian/Raspberry Pi OS use initramfs-tools. Each backend owns the
+distro-specific bits (package set, crypttab syntax, initramfs generation,
+bootloader cmdline, and how the Tor hook is baked in); the encryption flow
+orchestrates them the same way for every distro.
+"""
+
+from abc import ABC, abstractmethod
+from pathlib import Path
+
+from lim.image.plan import ImagePlan
+from lim.image.session import ImageSession
+
+
+class InitramfsBackend(ABC):
+    """Distro-specific steps to build an encrypted, remote-unlockable image."""
+
+    @abstractmethod
+    def luks_package_collection(self) -> str:
+        """Name of the package collection providing the LUKS/unlock stack."""
+
+    @abstractmethod
+    def install_authorized_key(self, root: Path, authorized_keys: Path) -> None:
+        """Place the SSH public key where the initramfs dropbear reads it."""
+
+    @abstractmethod
+    def install_tor_unlock(self, plan: ImagePlan, root: Path) -> str:
+        """Bake the Tor onion service into the initramfs; return the address."""
+
+    @abstractmethod
+    def register_encrypted_root(self, plan: ImagePlan, session: ImageSession, root: Path) -> None:
+        """Write fstab and crypttab in the distro's syntax."""
+
+    @abstractmethod
+    def configure_initramfs(self, plan: ImagePlan, root: Path) -> None:
+        """Configure the initramfs modules/hooks and (re)generate it."""
+
+    @abstractmethod
+    def configure_bootloader(self, plan: ImagePlan, session: ImageSession, root: Path) -> None:
+        """Point the bootloader at the encrypted root with early networking."""

lim/image/initramfs/initramfs_tools.py

@@ -0,0 +1,117 @@
+"""initramfs-tools backend (Debian / Raspberry Pi OS).
+
+Debian does not use mkinitcpio; the LUKS unlock stack is cryptsetup-initramfs
+(reads /etc/crypttab, option ``initramfs``) + dropbear-initramfs (serves the
+passphrase prompt over SSH). Raspberry Pi OS ships without an initramfs, so we
+enable one in config.txt and regenerate with update-initramfs.
+
+NOTE: the exact dropbear-initramfs key path, the boot-partition mount point
+(/boot vs /boot/firmware) and the initramfs enable knob vary across Raspberry
+Pi OS releases; verify on the target release. This path is not exercised by the
+QEMU harness yet.
+"""
+
+import re
+from pathlib import Path
+
+from lim import config, fsutil, packages, runner, ui
+from lim.image.initramfs import keygen
+from lim.image.initramfs.base import InitramfsBackend
+from lim.image.plan import ImagePlan
+from lim.image.session import ImageSession, chroot_bash, install_packages
+
+# Numeric IP, not a hostname: the initramfs has no DNS resolver.
+_DEFAULT_NTP = "162.159.200.123"
+
+
+class InitramfsToolsBackend(InitramfsBackend):
+    def luks_package_collection(self) -> str:
+        return "server/luks-debian"
+
+    def install_authorized_key(self, root: Path, authorized_keys: Path) -> None:
+        # dropbear-initramfs (bookworm) reads /etc/dropbear/initramfs/authorized_keys.
+        target = root / "etc/dropbear/initramfs/authorized_keys"
+        ui.info(f"Adding {authorized_keys} to {target}...")
+        runner.run(
+            ["install", "-D", "-m", "0600", str(authorized_keys), str(target)],
+            sudo=True,
+        )
+
+    def install_tor_unlock(self, plan: ImagePlan, root: Path) -> str:
+        ui.info("Setting up remote unlock via Tor onion service (initramfs-tools)...")
+        install_packages(
+            plan.distribution, root, " ".join(packages.get_packages("server/tor-debian"))
+        )
+        source_dir = config.CONFIGURATION_PATH / "initramfs-tools"
+        for source, target, mode in (
+            (source_dir / "tor_hook", root / "etc/initramfs-tools/hooks/tor", "0755"),
+            (
+                source_dir / "tor_premount",
+                root / "etc/initramfs-tools/scripts/init-premount/tor",
+                "0755",
+            ),
+            (
+                source_dir / "tor_bottom",
+                root / "etc/initramfs-tools/scripts/init-bottom/tor",
+                "0755",
+            ),
+            (source_dir / "torrc", root / keygen.TORRC_STAGING_PATH, "0644"),
+        ):
+            ui.info(f"Installing {target}...")
+            runner.run(["install", "-D", "-m", mode, str(source), str(target)], sudo=True)
+        onion_address = keygen.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
+
+    def register_encrypted_root(self, plan: ImagePlan, session: ImageSession, root: Path) -> None:
+        fstab_path = root / "etc/fstab"
+        fstab_line = (
+            f"/dev/mapper/{session.root_mapper_name}  /  {plan.root_filesystem}"
+            "  defaults,noatime  0  1"
+        )
+        ui.info(f"Configuring {fstab_path}...")
+        # A stock image already mounts / by PARTUUID; drop it so the mapper wins.
+        fsutil.drop_fstab_mount("/", fstab_path)
+        fsutil.ensure_line_in_file(fstab_line, fstab_path)
+
+        crypttab_path = root / "etc/crypttab"
+        # `initramfs` bakes the mapping into the initramfs so cryptsetup-initramfs
+        # unlocks the root before the pivot.
+        crypttab_line = (
+            f"{session.root_mapper_name} UUID={session.root_partition_uuid} none luks,initramfs"
+        )
+        ui.info(f"Configuring {crypttab_path}...")
+        fsutil.ensure_line_in_file(crypttab_line, crypttab_path)
+
+    def configure_initramfs(self, plan: ImagePlan, root: Path) -> None:  # noqa: ARG002
+        ui.info("Regenerating the Debian initramfs (update-initramfs)...")
+        # Create for all installed kernels when none exists yet (Raspberry Pi OS
+        # ships without one), else update the existing one. `-k all` avoids the
+        # build host's `uname -r` leaking into the chroot.
+        chroot_bash(
+            root,
+            "update-initramfs -c -k all 2>/dev/null || update-initramfs -u -k all",
+        )
+
+    def configure_bootloader(self, plan: ImagePlan, session: ImageSession, root: Path) -> None:
+        hostname = (root / "etc/hostname").read_text().strip()
+        boot = session.boot_mount_path
+
+        config_txt = boot / "config.txt"
+        if config_txt.is_file():
+            ui.info(f"Enabling an initramfs in {config_txt}...")
+            fsutil.ensure_line_in_file("auto_initramfs=1", config_txt)
+
+        cmdline_txt = boot / "cmdline.txt"
+        ui.info(f"Configuring {cmdline_txt}...")
+        text = cmdline_txt.read_text().strip()
+        text = re.sub(r"root=\S+", f"root=/dev/mapper/{session.root_mapper_name}", text)
+        if "ip=" not in text:
+            text = f"{text} ip=::::{hostname}:eth0:dhcp"
+        if "net.ifnames=0" not in text:
+            text = f"{text} net.ifnames=0"
+        if plan.tor_unlock and "tor_ntp=" not in text:
+            text = f"{text} tor_ntp={_DEFAULT_NTP}"
+        cmdline_txt.write_text(text + "\n")
+        ui.info(f"Content of {cmdline_txt}:{cmdline_txt.read_text()}")

lim/image/initramfs/keygen.py

@@ -0,0 +1,54 @@
+"""Offline Tor onion key generation, shared by every initramfs backend.
+
+``tor --DisableNetwork 1`` writes a v3 onion service's keys without touching
+the network, so the same stable ``.onion`` address can be baked into the
+initramfs whether the distro uses mkinitcpio or initramfs-tools.
+"""
+
+from pathlib import Path
+
+from lim import ui
+from lim.errors import LimError
+from lim.image.session import chroot_bash
+
+# Staged relative to the mounted image root.
+ONION_STAGING_DIR = "etc/tor/initramfs-onion"
+TORRC_STAGING_PATH = "etc/tor/initramfs-torrc"
+
+# An empty -f torrc keeps the image's /etc/tor/torrc (User tor, ...) out of the
+# keygen run, which would otherwise fail as non-root inside the chroot.
+_KEYGEN_SCRIPT = f"""
+mkdir -p /{ONION_STAGING_DIR}
+chmod 0700 /{ONION_STAGING_DIR}
+: > /tmp/tor-keygen-torrc
+tor -f /tmp/tor-keygen-torrc --DisableNetwork 1 \\
+    --DataDirectory /tmp/tor-keygen-data \\
+    --HiddenServiceDir /{ONION_STAGING_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_STAGING_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_STAGING_DIR}/hostname ]
+"""
+
+
+def generate_onion_keys(root: Path) -> str:
+    """Generate the onion keys offline in the chroot; return the .onion address.
+
+    Idempotent: existing keys are kept so the address stays stable across
+    re-runs.
+    """
+    hostname_path = root / ONION_STAGING_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()

lim/image/initramfs/mkinitcpio.py

@@ -0,0 +1,122 @@
+"""mkinitcpio backend (Arch Linux ARM / Manjaro ARM).
+
+See https://gist.github.com/gea0/4fc2be0cb7a74d0e7cc4322aed710d38 and
+https://gist.github.com/EnigmaCurry/2f9bed46073da8e38057fe78a61e7994
+"""
+
+from pathlib import Path
+
+from lim import catalog, config, fsutil, packages, runner, ui
+from lim.image.initramfs import keygen
+from lim.image.initramfs.base import InitramfsBackend
+from lim.image.plan import ImagePlan
+from lim.image.session import ImageSession, chroot_bash, install_packages
+
+MKINITCPIO_HOOKS_PREFIX = (
+    "base udev autodetect microcode modconf kms keyboard keymap consolefont block"
+)
+MKINITCPIO_HOOKS_SUFFIX = "filesystems fsck"
+
+
+class MkinitcpioBackend(InitramfsBackend):
+    def luks_package_collection(self) -> str:
+        return "server/luks"
+
+    def install_authorized_key(self, root: Path, authorized_keys: Path) -> None:
+        dropbear_root_key_path = root / "etc/dropbear/root_key"
+        ui.info(f"Adding {authorized_keys} to dropbear...")
+        runner.run(["cp", "-v", str(authorized_keys), str(dropbear_root_key_path)], sudo=True)
+
+    def install_tor_unlock(self, plan: ImagePlan, root: Path) -> str:
+        ui.info("Setting up remote unlock via Tor onion service...")
+        install_packages(plan.distribution, root, " ".join(packages.get_packages("server/tor")))
+        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 / keygen.TORRC_STAGING_PATH),
+        ):
+            ui.info(f"Installing {target}...")
+            runner.run(["install", "-D", "-m", "0644", str(source), str(target)], sudo=True)
+        onion_address = keygen.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
+
+    def register_encrypted_root(self, plan: ImagePlan, session: ImageSession, root: Path) -> None:
+        fstab_path = root / "etc/fstab"
+        fstab_line = (
+            f"UUID={session.root_partition_uuid}  / {plan.root_filesystem}"
+            "    defaults,noatime  0  1"
+        )
+        ui.info(f"Configuring {fstab_path}...")
+        fsutil.ensure_line_in_file(fstab_line, fstab_path)
+        ui.info(f"Content of {fstab_path}:{fstab_path.read_text()}")
+
+        crypttab_path = root / "etc/crypttab"
+        crypttab_line = f"{session.root_mapper_name} UUID={session.root_partition_uuid} none luks"
+        ui.info(f"Configuring {crypttab_path}...")
+        fsutil.ensure_line_in_file(crypttab_line, crypttab_path)
+        ui.info(f"Content of {crypttab_path}:{crypttab_path.read_text()}")
+
+    def configure_initramfs(self, plan: ImagePlan, root: Path) -> None:
+        # Concerning mkinitcpio warnings, see
+        # https://gist.github.com/imrvelj/c65cd5ca7f5505a65e59204f5a3f7a6d
+        mkinitcpio_path = root / "etc/mkinitcpio.conf"
+        ui.info(f"Configuring {mkinitcpio_path}...")
+        additional_modules = catalog.mkinitcpio_modules_by_rpi().get(plan.raspberry_pi_version)
+        if additional_modules is None:
+            ui.warning(f"Version {plan.raspberry_pi_version} isn't supported.")
+            additional_modules = ""
+        modules = f"g_cdc usb_f_acm usb_f_ecm {additional_modules} g_ether".replace("  ", " ")
+        fsutil.replace_in_file("MODULES=()", f"MODULES=({modules})", mkinitcpio_path)
+        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 {tor_hook}dropbear "
+            f"encryptssh {MKINITCPIO_HOOKS_SUFFIX})",
+            mkinitcpio_path,
+        )
+        ui.info(f"Content of {mkinitcpio_path}:{mkinitcpio_path.read_text()}")
+        ui.info("Generating mkinitcpio...")
+        chroot_bash(root, "mkinitcpio -vP")
+
+    def configure_bootloader(self, plan: ImagePlan, session: ImageSession, root: Path) -> None:
+        cryptdevice = (
+            f"cryptdevice=UUID={session.root_partition_uuid}:{session.root_mapper_name} "
+            f"root={session.root_mapper_path}"
+        )
+        boot_txt_path = session.boot_mount_path / "boot.txt"
+        if boot_txt_path.is_file():
+            ui.info(f"Configuring {boot_txt_path}...")
+            hostname = (root / "etc/hostname").read_text().strip()
+            fsutil.replace_in_file("part uuid ${devtype} ${devnum}:2 uuid", "", boot_txt_path)
+            fsutil.replace_in_file(
+                "setenv bootargs console=ttyS1,115200 console=tty0 root=PARTUUID=${uuid} "
+                'rw rootwait smsc95xx.macaddr="${usbethaddr}"',
+                f"setenv bootargs console=ttyS1,115200 console=tty0 "
+                f"ip=::::{hostname}:eth0:dhcp {cryptdevice} rw rootwait "
+                # Concerning issues with network adapter names, see
+                # https://forum.iobroker.net/topic/40542/raspberry-pi4-kein-eth0-mehr/16
+                'smsc95xx.macaddr="${usbethaddr}" net.ifnames=0 biosdevname=0',
+                boot_txt_path,
+            )
+            ui.info(f"Content of {boot_txt_path}:{boot_txt_path.read_text()}")
+            ui.info("Generating...")
+            chroot_bash(root, "cd /boot/ && ./mkscr || exit 1")
+        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"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()}")

lim/image/loopimg.py

@@ -0,0 +1,28 @@
+"""Attach a stock disk image as a loop device to read its partitions."""
+
+from pathlib import Path
+
+from lim import runner, ui
+from lim.errors import LimError
+
+
+def attach(image_path: Path) -> str:
+    """Attach the image via losetup with partition scanning; return /dev/loopN."""
+    ui.info(f"Attaching {image_path} as a loop device...")
+    loop = runner.output(
+        ["losetup", "-Pf", "--show", str(image_path)],
+        sudo=True,
+        error_msg=f"Attaching {image_path} as a loop device failed.",
+    ).strip()
+    if not loop:
+        raise LimError(f"losetup returned no device for {image_path}.")
+    return loop
+
+
+def detach(loop: str) -> None:
+    runner.run(["losetup", "-d", loop], sudo=True, check=False)
+
+
+def partition(loop: str, number: int) -> str:
+    """/dev/loop0 -> /dev/loop0p1 (losetup -P names partitions with a p suffix)."""
+    return f"{loop}p{number}"

lim/image/plan.py

@@ -12,16 +12,23 @@ class ImagePlan:
     distribution: str | None = None
     base_download_url: str | None = None
     image_name: str | None = None
+    # Direct download URL when it cannot be derived as base_download_url + image_name
+    # (Raspberry Pi OS ships behind a "_latest" redirect but must land under a
+    # .img.xz name so decompress_command recognises it).
+    source_url: str | None = None
     image_checksum: str | None = None
     boot_size: str = ""
     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)
 
     @property
     def download_url(self) -> str | None:
+        if self.source_url is not None:
+            return self.source_url
         if self.base_download_url is None or self.image_name is None:
             return None
         return f"{self.base_download_url}{self.image_name}"

lim/image/raspberry.py

@@ -23,9 +23,7 @@ def configure_sudoers(root_mount_path: Path, username: str) -> None:
         raise LimError(f"Failed to create sudoers file for {username}: {exc}") from exc
 
 
-def configure_ssh_key(
+def configure_ssh_key(public_key_path: str, target_ssh_folder: Path, authorized_keys: Path) -> None:
-    public_key_path: str, target_ssh_folder: Path, authorized_keys: Path
-) -> None:
     source = Path(public_key_path)
     if not source.is_file():
         raise LimError(
@@ -64,9 +62,7 @@ def _seed_boot_uuid(session: ImageSession) -> None:
         ui.info(f"{fstab_path} does not reference /dev/mmcblk0p1. Skipping UUID seeding.")
         return
     ui.info(f"Seeding UUID to {fstab_path} to avoid path conflicts...")
-    fsutil.replace_in_file(
+    fsutil.replace_in_file("/dev/mmcblk0p1", f"UUID={session.boot_partition_uuid}", fstab_path)
-        "/dev/mmcblk0p1", f"UUID={session.boot_partition_uuid}", fstab_path
-    )
     ui.info(f"Content of {fstab_path}:{fstab_path.read_text()}")
 
 
@@ -135,7 +131,7 @@ def _update_system(plan: ImagePlan, root: Path) -> None:
     ui.info("Updating system...")
     if plan.distribution in ("arch", "manjaro"):
         chroot_bash(root, "pacman --noconfirm -Syyu")
-    elif plan.distribution in ("moode", "retropie"):
+    elif plan.distribution in ("moode", "retropie", "raspios"):
         chroot_bash(root, "yes | apt update\nyes | apt upgrade\n")
     else:
         ui.warning(

lim/image/session.py

@@ -49,12 +49,8 @@ class ImageSession:
 
     def read_partition_uuids(self) -> None:
         """Fetch partition UUIDs via blkid; derive the LUKS mapper name if unset."""
-        self.root_partition_uuid = device_module.blkid_value(
+        self.root_partition_uuid = device_module.blkid_value(self.root_partition_path, "UUID")
-            self.root_partition_path, "UUID"
+        self.boot_partition_uuid = device_module.blkid_value(self.boot_partition_path, "UUID")
-        )
-        self.boot_partition_uuid = device_module.blkid_value(
-            self.boot_partition_path, "UUID"
-        )
         if self.root_mapper_name is None and self.root_is_luks():
             # Same deterministic name decrypt_root() would have used.
             self.root_mapper_name = f"linux-image-manager-{self.root_partition_uuid}"
@@ -63,9 +59,7 @@ class ImageSession:
     def decrypt_root(self) -> None:
         if not self.root_is_luks():
             return
-        self.root_partition_uuid = device_module.blkid_value(
+        self.root_partition_uuid = device_module.blkid_value(self.root_partition_path, "UUID")
-            self.root_partition_path, "UUID"
-        )
         self.root_mapper_name = f"linux-image-manager-{self.root_partition_uuid}"
         self.root_mapper_path = f"/dev/mapper/{self.root_mapper_name}"
         ui.info(f"Decrypting of {self.root_partition_path} is neccessary...")
@@ -88,18 +82,30 @@ class ImageSession:
             ["mount", "-v", self.boot_partition_path, str(self.boot_mount_path)],
             sudo=True,
         )
-        runner.run(
+        runner.run(["mount", "-v", self.root_mapper_path, str(self.root_mount_path)], sudo=True)
-            ["mount", "-v", self.root_mapper_path, str(self.root_mount_path)], sudo=True
-        )
         ui.info("Setting uuid variables...")
         self.read_partition_uuids()
         ui.info("The following mounts refering this setup exist:")
         runner.run(["findmnt", "-R", str(self.working_folder)], check=False)
 
+    def boot_bind_target(self) -> str:
+        """Where the FAT partition belongs inside the chroot.
+
+        Raspberry Pi OS Bookworm keeps it at /boot/firmware (the raspi-firmware
+        post-update hook syncs the initramfs there); older layouts use /boot.
+        Binding it wrong sends update-initramfs's output to the ext4 root, so the
+        firmware boots a stock initramfs without cryptsetup -> initramfs shell.
+        """
+        if (self.root_mount_path / "boot/firmware").is_dir():
+            return f"{self.root_mount_path}/boot/firmware"
+        return f"{self.root_mount_path}/boot"
+
     def mount_chroot_binds(self) -> None:
         ui.info("Mount chroot environments...")
         root = self.root_mount_path
-        runner.run(["mount", "--bind", str(self.boot_mount_path), f"{root}/boot"], sudo=True)
+        runner.run(
+            ["mount", "--bind", str(self.boot_mount_path), self.boot_bind_target()], sudo=True
+        )
         runner.run(["mount", "--bind", "/dev", f"{root}/dev"], sudo=True)
         runner.run(["mount", "--bind", "/sys", f"{root}/sys"], sudo=True)
         runner.run(["mount", "--bind", "/proc", f"{root}/proc"], sudo=True)
@@ -150,7 +156,7 @@ class ImageSession:
             self._umount(f"{root}/dev", lazy=True)
             self._umount(f"{root}/proc")
             self._umount(f"{root}/sys")
-            self._umount(f"{root}/boot")
+            self._umount(self.boot_bind_target())
             self._umount(str(root))
         if self.boot_mount_path is not None:
             self._umount(str(self.boot_mount_path))
@@ -169,11 +175,16 @@ class ImageSession:
                 ui.warning("Failed.")
 
 
+# The host PATH leaks into the chroot; force one that includes the sbin dirs
+# where Debian keeps update-initramfs, useradd, chpasswd, ... (else code 127).
+_CHROOT_PATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
+
+
 def chroot_bash(root_mount_path: Path | str, script: str, error_msg: str | None = None) -> None:
-    """Run a bash script inside the image via chroot."""
+    """Run a bash script inside the image via chroot (with a Debian-safe PATH)."""
     runner.run(
         ["chroot", str(root_mount_path), "/bin/bash"],
-        input_text=script,
+        input_text=f"export PATH={_CHROOT_PATH}\n{script}",
         sudo=True,
         error_msg=error_msg,
     )
@@ -183,8 +194,13 @@ def install_packages(distribution: str, root_mount_path: Path, package_names: st
     """Install packages inside the image with the distribution's package manager."""
     ui.info(f"Installing {package_names}...")
     if distribution in ("arch", "manjaro"):
-        chroot_bash(root_mount_path, f"pacman --noconfirm -S --needed {package_names}")
+        chroot_bash(root_mount_path, f"pacman --noconfirm -Sy --needed {package_names}")
-    elif distribution in ("moode", "retropie"):
+    elif distribution in ("moode", "retropie", "raspios"):
-        chroot_bash(root_mount_path, f"yes | apt install {package_names}")
+        # apt-get update first: a stock image ships stale lists whose old .deb
+        # URLs 404 once the mirror has superseded them.
+        chroot_bash(
+            root_mount_path,
+            f"apt-get update\nDEBIAN_FRONTEND=noninteractive apt-get install -y {package_names}",
+        )
     else:
         raise LimError("Package manager not supported.")

lim/image/setup.py

@@ -36,8 +36,7 @@ def _select_unmounted_device() -> device_module.Device:
     device = device_module.select_device()
     if device_module.is_mounted(device.path):
         raise LimError(
-            f'Device {device.path} is allready mounted. '
+            f'Device {device.path} is allready mounted. Umount with "umount {device.path}*".'
-            f'Umount with "umount {device.path}*".'
         )
     return device
 
@@ -49,6 +48,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:
@@ -79,9 +82,7 @@ def run_setup() -> None:
         session.make_working_folder()
         session.make_mount_folders()
 
-        plan.root_filesystem = ui.ask(
+        plan.root_filesystem = ui.ask("Which filesystem should be used? E.g.:btrfs,ext4... (none):")
-            "Which filesystem should be used? E.g.:btrfs,ext4... (none):"
-        )
 
         if ui.confirm(f"Should the image be transfered to {device.path}?"):
             transfer.transfer_image(plan, session)

lim/image/transfer.py

@@ -1,24 +1,23 @@
 """Download the selected image and transfer it onto the target device."""
 
+import shutil
 from pathlib import Path
 
 from lim import device as device_module
 from lim import runner, ui
 from lim.errors import LimError
+from lim.image import loopimg
 from lim.image.plan import DEFAULT_BOOT_SIZE, ImagePlan
 from lim.image.session import ImageSession
 
 
-def download_image(plan: ImagePlan) -> None:
+def download_image(plan: ImagePlan, *, force_prompt: bool = True) -> None:
-    if ui.confirm("Should the image download be forced?"):
+    if force_prompt and ui.confirm("Should the image download be forced?"):
         if plan.image_path.is_file():
             ui.info(f"Removing image {plan.image_path}.")
             plan.image_path.unlink()
         else:
-            ui.info(
+            ui.info(f"Forcing download wasn't neccessary. File {plan.image_path} doesn't exist.")
-                "Forcing download wasn't neccessary. "
-                f"File {plan.image_path} doesn't exist."
-            )
 
     ui.info("Start Download procedure...")
     if plan.image_path.is_file():
@@ -61,9 +60,18 @@ def decompress_command(image_path: Path) -> list[str]:
 
 def _luks_format_root(plan: ImagePlan, session: ImageSession) -> None:
     luks_format = [
-        "cryptsetup", "-v", "luksFormat",
+        "cryptsetup",
-        "-c", "aes-xts-plain64", "-s", "512", "-h", "sha512",
+        "-v",
-        "--use-random", "-i", "1000",
+        "luksFormat",
+        "-c",
+        "aes-xts-plain64",
+        "-s",
+        "512",
+        "-h",
+        "sha512",
+        "--use-random",
+        "-i",
+        "1000",
     ]
     if plan.luks_memory_cost:
         ui.info(
@@ -110,19 +118,107 @@ def transfer_arch_image(plan: ImagePlan, session: ImageSession) -> None:
         runner.run(["mv", "-v", str(entry), str(session.boot_mount_path)], sudo=True)
 
 
-def transfer_image(plan: ImagePlan, session: ImageSession) -> None:
+def _create_encrypted_layout(plan: ImagePlan, session: ImageSession) -> None:
+    """Partition the target, LUKS-format the root, format and mount both."""
+    boot_size = plan.boot_size or DEFAULT_BOOT_SIZE
+    ui.info(f"Creating partitions (boot {boot_size})...")
+    runner.run(
+        ["fdisk", session.device.path],
+        input_text=arch_partition_input(boot_size),
+        sudo=True,
+    )
+    runner.run(["mkfs.vfat", session.boot_partition_path], sudo=True)
+    _luks_format_root(plan, session)
+    session.decrypt_root()
+    runner.run([f"mkfs.{plan.root_filesystem}", "-F", session.root_mapper_path], sudo=True)
+    session.mount_partitions()
+
+
+def _copy_tree(source: str, target: str, flags: list[str]) -> None:
+    """Copy a directory's contents, showing progress when rsync is available."""
+    if shutil.which("rsync"):
+        runner.run(["rsync", *flags, "--info=progress2", f"{source}/", f"{target}/"], sudo=True)
+    else:
+        runner.run(["cp", "-a", f"{source}/.", target], sudo=True)
+
+
+def _fix_boot_fstab(session: ImageSession) -> None:
+    """Repoint the stock boot fstab line at the freshly created boot partition.
+
+    The copied image mounts /boot(/firmware) by the stock PARTUUID, which no
+    longer exists after repartitioning; swap in the new boot UUID, mount point
+    unchanged. No-op when the image carries no boot line.
+    """
+    fstab = session.root_mount_path / "etc/fstab"
+    if not fstab.is_file():
+        return
+    lines = []
+    for line in fstab.read_text().splitlines():
+        fields = line.split()
+        if not line.lstrip().startswith("#") and fields[1:2] and fields[1].startswith("/boot"):
+            fields[0] = f"UUID={session.boot_partition_uuid}"
+            lines.append("  ".join(fields))
+        else:
+            lines.append(line)
+    fstab.write_text("".join(f"{entry}\n" for entry in lines))
+
+
+def transfer_disk_image(plan: ImagePlan, session: ImageSession) -> None:
+    """Copy a full disk image (boot + root partitions) into a fresh encrypted layout.
+
+    Distributions that ship a partitioned .img rather than a rootfs tarball
+    (Raspberry Pi OS, moode, RetroPie, Manjaro ARM) are loop-mounted and their
+    first two partitions copied into a new LUKS container.
+    """
+    raw = plan.image_path.with_suffix("")
+    ui.info(f"Decompressing {plan.image_path} to {raw}...")
+    runner.pipeline(
+        decompress_command(plan.image_path),
+        ["dd", f"of={raw}", "bs=4M", "conv=fsync", "status=progress"],
+        error_msg=f"Decompressing {plan.image_path} failed.",
+    )
+    loop = loopimg.attach(raw)
+    source_boot = session.working_folder / "source-boot"
+    source_root = session.working_folder / "source-root"
+    try:
+        source_boot.mkdir()
+        source_root.mkdir()
+        runner.run(["mount", "-o", "ro", loopimg.partition(loop, 1), str(source_boot)], sudo=True)
+        runner.run(["mount", "-o", "ro", loopimg.partition(loop, 2), str(source_root)], sudo=True)
+
+        _create_encrypted_layout(plan, session)
+
+        ui.info("Copying the root filesystem into the encrypted container...")
+        _copy_tree(str(source_root), str(session.root_mount_path), ["-aHAX"])
+        ui.info("Copying the boot partition...")
+        _copy_tree(str(source_boot), str(session.boot_mount_path), ["-a"])
+        runner.sync_disks()
+        _fix_boot_fstab(session)
+    finally:
+        runner.run(["umount", "-v", str(source_boot)], sudo=True, check=False)
+        runner.run(["umount", "-v", str(source_root)], sudo=True, check=False)
+        loopimg.detach(loop)
+        for path in (source_boot, source_root):
+            runner.run(["rmdir", str(path)], sudo=True, check=False)
+
+
+def transfer_image(plan: ImagePlan, session: ImageSession, *, interactive: bool = True) -> None:
+    if interactive:
         if ui.confirm(f"Should the partition table of {session.device.path} be deleted?"):
             ui.info("Deleting...")
             runner.run(["wipefs", "-a", session.device.path], sudo=True)
         else:
             ui.info("Skipping partition table deletion...")
-
         device_module.overwrite_device(session.device)
 
     ui.info("Starting image transfer...")
     if plan.distribution == "arch":
         transfer_arch_image(plan, session)
         return
+    if plan.encrypt_system:
+        # Non-tarball distros ship a full .img; copy it into a LUKS container.
+        transfer_disk_image(plan, session)
+        return
     blocksize = session.device.optimal_blocksize
     ui.info(f"Transfering {plan.image_path.suffix} file...")
     runner.pipeline(

lim/image/unlock.py

@@ -0,0 +1,100 @@
+"""Remote LUKS boot unlock over the initramfs SSH/dropbear session.
+
+Reaches the waiting initramfs either through a Tor onion address (wrapped in
+torsocks) or a plain host/IP, then either runs cryptroot-unlock (initramfs-tools
+targets) or drops into the passphrase prompt (mkinitcpio encryptssh).
+"""
+
+import json
+import os
+from pathlib import Path
+
+from lim import runner, ui
+from lim.errors import LimError
+
+RECORDS_SUBPATH = ".config/lim/unlocks"
+
+
+def records_dir(home: Path | None = None) -> Path:
+    return (home or Path.home()) / RECORDS_SUBPATH
+
+
+def build_unlock_argv(target: str, key: str, unlock_command: str) -> list[str]:
+    """SSH invocation for the unlock; torsocks-wrapped for .onion targets.
+
+    The initramfs dropbear key differs from the booted host key, so host-key
+    checking is disabled; a v3 onion address authenticates the endpoint itself.
+    """
+    ssh = ["ssh", "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null"]
+    if key:
+        ssh += ["-i", str(Path(key).expanduser())]
+    if unlock_command:
+        ssh += ["-t"]  # PTY so the remote passphrase prompt is interactive
+    ssh += [f"root@{target}"]
+    if unlock_command:
+        ssh += [unlock_command]
+    if target.endswith(".onion"):
+        return ["torsocks", *ssh]
+    return ssh
+
+
+def save_record(home: Path, uid: int, gid: int, name: str, record: dict) -> None:
+    """Persist an unlock target under the target user's home, owned by them."""
+    directory = records_dir(home)
+    directory.mkdir(parents=True, exist_ok=True)
+    path = directory / f"{name}.json"
+    path.write_text(json.dumps(record, indent=2) + "\n")
+    try:
+        for entry in (path, directory, directory.parent, directory.parent.parent):
+            os.chown(entry, uid, gid)
+    except OSError:
+        pass
+
+
+def _read_record(path: Path) -> dict | None:
+    try:
+        return json.loads(path.read_text())
+    except (OSError, ValueError):
+        return None
+
+
+def _load_records() -> list[tuple[str, dict]]:
+    directory = records_dir()
+    if not directory.is_dir():
+        return []
+    loaded = ((path.stem, _read_record(path)) for path in sorted(directory.glob("*.json")))
+    return [(name, record) for name, record in loaded if record is not None]
+
+
+def _select_record() -> dict | None:
+    records = _load_records()
+    if not records:
+        return None
+    ui.info("Saved unlock targets:")
+    for index, (name, record) in enumerate(records, 1):
+        ui.info(f"  {index}) {name} -> {record.get('target')}")
+    answer = ui.ask("Pick a number, or Enter for manual entry:")
+    if answer.isdigit() and 1 <= int(answer) <= len(records):
+        return records[int(answer) - 1][1]
+    return None
+
+
+def remote_unlock() -> None:
+    record = _select_record()
+    if record:
+        target = record["target"]
+        key = record.get("key", "")
+        unlock_command = record.get("unlock_command", "")
+    else:
+        target = ui.ask("Onion address or host/IP to unlock:")
+        if not target:
+            raise LimError("No target given.")
+        key = ui.ask("Path to the SSH private key (empty for ssh defaults):")
+        unlock_command = (
+            "cryptroot-unlock"
+            if ui.confirm("Debian / Raspberry Pi OS target (run cryptroot-unlock)?")
+            else ""
+        )
+    if target.endswith(".onion"):
+        ui.info("Routing through Tor (torsocks); a local tor must be running.")
+    runner.run(build_unlock_argv(target, key, unlock_command), check=False)

lim/image/verify.py

@@ -24,8 +24,7 @@ def resolve_checksum(download_url: str) -> str | None:
     for extension in ("sha1", "sha512", "md5"):
         checksum_url = f"{download_url}.{extension}"
         ui.info(
-            "Image Checksum is not defined. "
+            f"Image Checksum is not defined. Try to download image signature from {checksum_url}."
-            f"Try to download image signature from {checksum_url}."
         )
         if url_exists(checksum_url):
             content = runner.output(["wget", checksum_url, "-q", "-O", "-"])
@@ -62,8 +61,7 @@ def verify_signature(download_url: str, image_path: str | Path, image_folder: Pa
     """Best-effort GPG signature verification of the downloaded image."""
     ui.info("Note: Checksums verify integrity but do not confirm authenticity.")
     ui.info(
-        "Proceeding to signature verification, "
+        "Proceeding to signature verification, which ensures the file comes from a trusted source."
-        "which ensures the file comes from a trusted source."
     )
     signature_url = f"{download_url}.sig"
     ui.info(f"Attempting to download the image signature from: {signature_url}")

lim/image/wizard.py

@@ -0,0 +1,206 @@
+"""Guided setup: an encrypted, Tor-remote-unlockable Linux image on a device.
+
+All questions are asked up front (``_collect``); the build then runs unattended
+(``_execute``): pick a distribution and target block device (the internal SSD of
+a USB-booted Pi, or a spare stick to clone later), download the image, copy it
+into a LUKS container, create the login user, and bake the initramfs Tor onion
+unlock. Works for any distribution with an initramfs backend (Arch/Manjaro via
+mkinitcpio, Raspberry Pi OS/moode/RetroPie via initramfs-tools).
+"""
+
+import os
+import pwd
+import shlex
+from dataclasses import dataclass
+from pathlib import Path
+
+from lim import device as device_module
+from lim import ui
+from lim.errors import LimError
+from lim.image import choosers, crossarch, transfer, unlock
+from lim.image.encryption import configure_encryption
+from lim.image.plan import ImagePlan
+from lim.image.session import ImageSession, chroot_bash
+
+
+@dataclass
+class _Answers:
+    host_user: str
+    device: device_module.Device
+    pubkey: Path
+    hostname: str
+    login_user: str
+    login_password: str
+
+
+def _intro() -> None:
+    ui.info("Guided encrypted Linux image setup with Tor remote unlock.")
+    ui.info("The TARGET must be an attached block device:")
+    ui.info("  - internal SSD: boot the Pi from USB, then target /dev/sda or /dev/nvme0n1")
+    ui.info("  - golden image: target a spare USB stick, clone it onto the SSD afterwards")
+    ui.warning("The target device will be COMPLETELY ERASED.")
+
+
+def _prepare_image_folder(plan: ImagePlan) -> str:
+    """Cache downloads under the invoking (sudo) user's home; return that user."""
+    host_user = os.environ.get("SUDO_USER") or "root"
+    try:
+        home = Path(pwd.getpwnam(host_user).pw_dir)
+    except KeyError:
+        host_user, home = "root", Path("/root")
+    plan.image_folder = home / "Software/Images"
+    plan.image_folder.mkdir(parents=True, exist_ok=True)
+    ui.info(f"Images are cached in {plan.image_folder}.")
+    return host_user
+
+
+def _select_target() -> device_module.Device:
+    device = device_module.select_device()
+    if device_module.is_mounted(device.path):
+        raise LimError(f"{device.path} is mounted. Unmount it first (umount {device.path}*).")
+    ui.warning(f"Everything on {device.path} will be destroyed.")
+    answer = ui.ask(f"Retype the device name ({device.name}) to confirm ERASE of {device.path}:")
+    if answer.strip().removeprefix("/dev/") != device.name:
+        raise LimError("Confirmation did not match. Aborting.")
+    return device
+
+
+def _ask_pubkey() -> Path:
+    answer = ui.ask("Path to the SSH PUBLIC key (unlocks AND logs in):").strip()
+    if not answer:
+        raise LimError("No SSH public key path given.")
+    pubkey = Path(answer).expanduser()
+    if not pubkey.is_file():
+        raise LimError(f"SSH public key {pubkey} not found.")
+    return pubkey
+
+
+def _collect(plan: ImagePlan) -> _Answers | None:
+    """Ask everything up front so ``_execute`` can run without prompts."""
+    _intro()
+    if not ui.confirm("Continue?"):
+        return None
+    choosers.choose_linux_image(plan)
+    plan.root_filesystem = plan.root_filesystem or "ext4"
+    device = _select_target()
+    hostname = ui.ask("Hostname for the system:").strip()
+    login_user = ui.ask("Login username to create on the system (default pi):").strip() or "pi"
+    login_password = ui.ask(f"Login password for {login_user} (empty for key-only login):")
+    pubkey = _ask_pubkey()
+    host_user = _prepare_image_folder(plan)
+    crossarch.ensure_ready()  # may install qemu; done here, before the erase in _execute
+    ui.success("All questions answered — the build now runs unattended.")
+    return _Answers(host_user, device, pubkey, hostname, login_user, login_password)
+
+
+def _build_authorized_keys(session: ImageSession, pubkey: Path) -> Path:
+    authorized_keys = session.working_folder / "authorized_keys"
+    authorized_keys.write_text(pubkey.read_text())
+    return authorized_keys
+
+
+def _apply_system_config(session: ImageSession, hostname: str) -> str:
+    """Write the hostname and enable the post-boot SSH server; return the hostname."""
+    root = session.root_mount_path
+    if hostname:
+        (root / "etc/hostname").write_text(hostname + "\n")
+    # RPi OS enables sshd on first boot when this marker exists on the boot fs.
+    (session.boot_mount_path / "ssh").write_text("")
+    return (root / "etc/hostname").read_text().strip()
+
+
+def _configure_login_user(session: ImageSession, user: str, password: str, pubkey: Path) -> None:
+    """Rename a stock default user to the chosen one (or create it); add sudo + key.
+
+    Renaming pi/alarm reuses the account (uid, sudoers) and leaves no lingering
+    default; a fresh useradd covers images without one (e.g. RPi OS Bookworm).
+    """
+    key = shlex.quote(pubkey.read_text().strip())
+    ui.info(f"Configuring login user {user}...")
+    script = f"""\
+target={shlex.quote(user)}
+for old in pi alarm; do
+  id "$old" >/dev/null 2>&1 || continue
+  [ "$old" = "$target" ] && continue
+  id "$target" >/dev/null 2>&1 && continue
+  usermod -l "$target" "$old"
+  usermod -d "/home/$target" -m "$target"
+  groupmod -n "$target" "$old" 2>/dev/null || true
+  break
+done
+id "$target" >/dev/null 2>&1 || useradd -m -s /bin/bash "$target"
+usermod -aG sudo "$target" 2>/dev/null || true
+install -d -m 700 "/home/$target/.ssh"
+printf '%s\\n' {key} > "/home/$target/.ssh/authorized_keys"
+chmod 600 "/home/$target/.ssh/authorized_keys"
+chown -R "$target:$target" "/home/$target/.ssh"
+"""
+    if password:
+        script += f"printf '%s' {shlex.quote(f'{user}:{password}')} | chpasswd\n"
+    chroot_bash(session.root_mount_path, script)
+
+
+def _save_unlock_record(
+    plan: ImagePlan, host_user: str, hostname: str, onion: str | None, pubkey: Path
+) -> None:
+    if not onion:
+        return
+    private_key = str(pubkey).removesuffix(".pub")
+    unlock_command = "" if plan.distribution in ("arch", "manjaro") else "cryptroot-unlock"
+    record = {
+        "target": onion,
+        "key": private_key,
+        "unlock_command": unlock_command,
+        "hostname": hostname,
+    }
+    entry = pwd.getpwnam(host_user)
+    try:
+        unlock.save_record(
+            Path(entry.pw_dir), entry.pw_uid, entry.pw_gid, hostname or "device", record
+        )
+        ui.info("Saved unlock target; run later with: lim --type remote-unlock")
+    except OSError as exc:
+        ui.warning(f"Could not save unlock record: {exc}")
+
+
+def _print_summary(answers: _Answers, hostname: str, onion: str | None) -> None:
+    ui.success("Encrypted Linux image is ready.")
+    ui.info(f"Target device : {answers.device.path}  (hostname: {hostname})")
+    if onion:
+        ui.info(f"Onion address : {onion}")
+        ui.info(f"Unlock later  : lim --type remote-unlock   (or: torsocks ssh root@{onion})")
+    ui.info(f"After unlock, log in: ssh {answers.login_user}@<the box>")
+    ui.info("Golden image? Clone this device onto the SSD, then grow it:")
+    ui.info("  parted <ssd> resizepart 2 100% && cryptsetup resize <mapper> && resize2fs <mapper>")
+
+
+def _execute(plan: ImagePlan, answers: _Answers) -> None:
+    transfer.download_image(plan, force_prompt=False)
+    session = ImageSession(answers.device)
+    authorized_keys = None
+    try:
+        session.make_working_folder()
+        session.make_mount_folders()
+        transfer.transfer_image(plan, session, interactive=False)
+        authorized_keys = _build_authorized_keys(session, answers.pubkey)
+        hostname = _apply_system_config(session, answers.hostname)
+        session.mount_chroot_binds()
+        session.copy_resolv_conf()
+        onion = configure_encryption(plan, session, authorized_keys)
+        _configure_login_user(session, answers.login_user, answers.login_password, answers.pubkey)
+        _save_unlock_record(plan, answers.host_user, hostname, onion, answers.pubkey)
+        _print_summary(answers, hostname, onion)
+    finally:
+        if authorized_keys is not None:
+            authorized_keys.unlink(missing_ok=True)
+        session.destructor()
+
+
+def run_guided() -> None:
+    ui.header()
+    plan = ImagePlan(encrypt_system=True, tor_unlock=True)
+    answers = _collect(plan)
+    if answers is None:
+        ui.info("Aborted.")
+        return
+    _execute(plan, answers)

lim/luks.py

@@ -40,14 +40,10 @@ def create_luks_key_and_update_crypttab(
     runner.sync_disks()
 
     ui.info("Opening and closing device to verify that everything works fine...")
-    closed = runner.run(
+    closed = runner.run(["cryptsetup", "-v", "luksClose", mapper_name], sudo=True, check=False)
-        ["cryptsetup", "-v", "luksClose", mapper_name], sudo=True, check=False
-    )
     if closed.returncode != 0:
         ui.info(f"No need to luksClose {mapper_name}. Device isn't open.")
-    runner.run(
+    runner.run(["cryptsetup", "luksAddKey", partition_path, str(secret_key_path)], sudo=True)
-        ["cryptsetup", "luksAddKey", partition_path, str(secret_key_path)], sudo=True
-    )
     runner.run(
         [
             "cryptsetup",
@@ -70,9 +66,7 @@ def create_luks_key_and_update_crypttab(
     print(crypttab_path.read_text())
 
 
-def update_fstab(
+def update_fstab(mapper_path: str, mount_path: str, *, fstab_path: Path = FSTAB_PATH) -> None:
-    mapper_path: str, mount_path: str, *, fstab_path: Path = FSTAB_PATH
-) -> None:
     entry = f"{mapper_path} {mount_path} btrfs   defaults   0       2"
     ui.info("Adding fstab entry...")
     fsutil.ensure_line_in_file(entry, fstab_path)

lim/runner.py

@@ -50,8 +50,7 @@ def run(
         return subprocess.CompletedProcess(cmd, COMMAND_NOT_FOUND)
     if check and result.returncode != 0:
         raise LimError(
-            error_msg
+            error_msg or f"Command failed with code {result.returncode}: {shlex.join(cmd)}"
-            or f"Command failed with code {result.returncode}: {shlex.join(cmd)}"
         )
     return result
 
@@ -98,8 +97,7 @@ def pipeline(
 ) -> None:
     """Run commands as a shell-style pipeline (cmd1 | cmd2 | ...)."""
     prepared = [
-        _with_sudo(cmd, sudo=sudo_last and index == len(cmds) - 1)
+        _with_sudo(cmd, sudo=sudo_last and index == len(cmds) - 1) for index, cmd in enumerate(cmds)
-        for index, cmd in enumerate(cmds)
     ]
     pretty = " | ".join(shlex.join(cmd) for cmd in prepared)
     ui.info(f"Running: {pretty}")

lim/storage/raid1.py

@@ -25,9 +25,7 @@ def setup() -> None:
     runner.run(["cryptsetup", "luksFormat", second.device.path], sudo=True)
 
     runner.run(["cryptsetup", "luksOpen", first.device.path, first.mapper_name], sudo=True)
-    runner.run(
+    runner.run(["cryptsetup", "luksOpen", second.device.path, second.mapper_name], sudo=True)
-        ["cryptsetup", "luksOpen", second.device.path, second.mapper_name], sudo=True
-    )
     runner.run(["cryptsetup", "status", first.mapper_path], sudo=True)
     runner.run(["cryptsetup", "status", second.mapper_path], sudo=True)
 

lim/storage/single_drive.py

@@ -32,9 +32,7 @@ def setup() -> None:
     runner.run(["cryptsetup", "-v", "-y", "luksFormat", target.partition_path], sudo=True)
 
     ui.info("Unlock partition...")
-    runner.run(
+    runner.run(["cryptsetup", "luksOpen", target.partition_path, target.mapper_name], sudo=True)
-        ["cryptsetup", "luksOpen", target.partition_path, target.mapper_name], sudo=True
-    )
 
     ui.info("Create btrfs file system...")
     runner.run(["mkfs.btrfs", target.mapper_path], sudo=True)
@@ -57,9 +55,7 @@ def mount() -> None:
     target = select_storage_target()
 
     ui.info("Unlock partition...")
-    runner.run(
+    runner.run(["cryptsetup", "luksOpen", target.partition_path, target.mapper_name], sudo=True)
-        ["cryptsetup", "luksOpen", target.partition_path, target.mapper_name], sudo=True
-    )
 
     ui.info("Mount partition...")
     runner.run(["mount", target.mapper_path, target.mount_path], sudo=True)

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "linux-image-manager"
-version = "2.1.0"
+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"
@@ -28,7 +28,12 @@ lim = "lim.cli:main"
 include = ["lim*"]
 
 [tool.setuptools.package-data]
-lim = ["distributions.yml", "configuration/packages/**/*.txt"]
+lim = [
+  "distributions.yml",
+  "configuration/packages/**/*.txt",
+  "configuration/initcpio/*",
+  "configuration/initramfs-tools/*",
+]
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]

tests/e2e/qemu/README.md

@@ -0,0 +1,109 @@
+# 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 (Arch) | `build_image.sh` | **root** (loop, cryptsetup, chroot) |
+| Build encrypted image (Debian) | `build_image_debian.sh` | **root** (debootstrap, 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)
+
+For the Debian build (`LIM_E2E_OS=debian`), additionally: `debootstrap` and
+network access to a Debian mirror. The guest uses the initramfs-tools backend
+(cryptsetup-initramfs + dropbear-initramfs), so the unlock SSH session runs
+`cryptroot-unlock` instead of Arch's login-time encryptssh prompt.
+
+## Unlock transport: direct vs Tor
+
+`LIM_E2E_TRANSPORT` (default `direct`) selects how the passphrase reaches the
+guest's dropbear:
+
+- **direct** — QEMU forwards a host port to the guest's dropbear (`:22`) and the
+  passphrase is delivered over a plain SSH. Deterministic; it verifies the whole
+  LUKS-unlock stack (cryptsetup-initramfs + dropbear-initramfs + cryptroot-unlock
+  + LUKS + boot). This is what CI/`make` runs.
+- **tor** — the full onion path: the guest publishes its onion and the host
+  reaches it through Tor. Representative of production but **flaky over public
+  Tor inside QEMU** (fresh guest Tor + user-mode net make the onion rendezvous
+  unreliable), so it is opt-in. The onion transport itself is covered
+  deterministically by the rootless `test_tor_unlock_e2e.py`.
+
+## Running
+
+```bash
+# Arch (mkinitcpio), direct transport (deterministic, default):
+LIM_E2E_QEMU=1 pytest tests/e2e/test_qemu_unlock_e2e.py -v -s
+
+# Debian (initramfs-tools) via debootstrap, direct transport:
+LIM_E2E_QEMU=1 LIM_E2E_OS=debian pytest tests/e2e/test_qemu_unlock_e2e.py -v -s
+
+# Full onion transport (opt-in, flaky over public Tor):
+LIM_E2E_QEMU=1 LIM_E2E_TRANSPORT=tor pytest tests/e2e/test_qemu_unlock_e2e.py -v -s
+```
+
+`make test-qemu` / `make test-qemu-debian` wrap the direct-transport runs.
+
+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,212 @@
+"""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 = 480
+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
+_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 _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:
+    """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. Session output is appended to
+    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(
+        config.ssh_argv(spec),
+        stdin=subprocess.PIPE,
+        stdout=log_file,
+        stderr=log_file,
+        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()
+        log_file.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()
+    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"Unlock target: {target}  (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
+        # 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
+        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("\n[harness] delivering passphrase (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, lines=45)}{_qemu_stderr_tail(spec)}"
+            f"{_ssh_log_tail(spec)}{_client_tor_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/build_image_debian.sh

@@ -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"

tests/e2e/qemu/config.py

@@ -0,0 +1,174 @@
+"""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
+    # "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
+    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]
+    # 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
+    # name is mandatory; a bare "ip=dhcp" leaves netconf looping on
+    # "SIOCGIFFLAGS: No such device". net.ifnames=0 keeps the 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"{crypt} 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"]
+    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 += [
+        "-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",
+        netdev,
+        "-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 to feed the passphrase.
+
+    ``-tt`` forces a pty so the askpass inside the session reads the passphrase
+    we pipe on stdin. Host-key checking is off (throwaway host key). On Debian
+    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.
+    """
+    argv = [
+        "ssh",
+        "-tt",
+        "-o",
+        "StrictHostKeyChecking=no",
+        "-o",
+        "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}",
+        ]
+    if spec.unlock_command:
+        argv.append(spec.unlock_command)
+    return argv

tests/e2e/qemu/harness.py

@@ -0,0 +1,218 @@
+"""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 socket
+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"
+    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:
+    """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_SCRIPT[cfg.os_family]
+    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.unlock_transport == "direct":
+        return _NullNet()
+    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)
+    direct_port = _free_port() if cfg.unlock_transport == "direct" else 0
+    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),
+        os_family=cfg.os_family,
+        unlock_command=_UNLOCK_COMMAND[cfg.os_family],
+        direct_ssh_port=direct_port,
+    )
+
+
+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:
+        if cfg.unlock_transport != "direct":
+            print("\n[harness] bootstrapping Tor client (~30-60s)...", flush=True)
+        net.start()
+        print(
+            f"[harness] transport={cfg.unlock_transport}, building encrypted "
+            "image (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,156 @@
+"""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 = 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
+
+
+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}",
+                # 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,
+        )
+        _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,240 @@
+"""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"))
+
+    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:
+    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
+
+    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:
+    """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 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:
+    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,81 @@
+"""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")
+# 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),
+    _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]:
+    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_unlock(tmp_path):
+    chutney_path = os.environ.get("CHUTNEY_PATH")
+    cfg = HarnessConfig(
+        repo_root=REPO_ROOT,
+        work_dir=tmp_path / "qemu-e2e",
+        arch=_ARCH,
+        os_family=_OS,
+        unlock_transport=_TRANSPORT,
+        chutney_path=Path(chutney_path) if chutney_path else None,
+    )
+
+    spec = run(cfg)
+
+    # run() only returns after the boot-ok marker was observed on the console.
+    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,110 @@
+"""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.initramfs.keygen 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.initramfs import keygen
+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 = keygen._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,188 @@
+"""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.initramfs.keygen._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.initramfs.keygen._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_cli.py

@@ -24,9 +24,7 @@ def test_every_command_dispatches_to_its_function(monkeypatch):
 def test_needs_root_command_requests_root(monkeypatch):
     escalated = []
     monkeypatch.setattr(system, "ensure_root", lambda: escalated.append(True))
-    monkeypatch.setitem(
+    monkeypatch.setitem(cli.COMMANDS, "backup", Command(lambda: None, "d", needs_root=True))
-        cli.COMMANDS, "backup", Command(lambda: None, "d", needs_root=True)
-    )
     cli.main(["--type", "backup", "--auto-confirm"])
     assert escalated == [True]
 
@@ -59,6 +57,10 @@ def test_unknown_type_is_rejected_by_argparse():
     assert excinfo.value.code == 2
 
 
+def test_type_defaults_to_guided():
+    assert cli.build_parser().parse_args([]).type == "guided"
+
+
 def test_all_registered_commands_have_descriptions():
     for name, command in cli.COMMANDS.items():
         assert command.description, name

tests/unit/test_cross_arch.py

@@ -0,0 +1,50 @@
+import pytest
+
+from lim.errors import LimError
+from lim.image import crossarch
+
+
+class TestCrossArch:
+    def test_native_arm_is_ok(self, monkeypatch):
+        monkeypatch.setattr(crossarch.platform, "machine", lambda: "aarch64")
+        crossarch.ensure_ready()
+
+    def test_cross_arch_with_binfmt_ok(self, monkeypatch):
+        monkeypatch.setattr(crossarch.platform, "machine", lambda: "x86_64")
+        monkeypatch.setattr(crossarch, "qemu_binfmt_enabled", lambda: True)
+        crossarch.ensure_ready()
+
+    def test_declined_auto_install_raises(self, monkeypatch, answers):
+        monkeypatch.setattr(crossarch.platform, "machine", lambda: "x86_64")
+        monkeypatch.setattr(crossarch, "qemu_binfmt_enabled", lambda: False)
+        answers("n")
+        with pytest.raises(LimError, match="Cross-architecture"):
+            crossarch.ensure_ready()
+
+    def test_auto_install_success_proceeds(self, monkeypatch, answers):
+        monkeypatch.setattr(crossarch.platform, "machine", lambda: "x86_64")
+        monkeypatch.setattr(crossarch, "qemu_binfmt_enabled", lambda: False)
+        monkeypatch.setattr(crossarch, "auto_enable", lambda: True)
+        answers("y")
+        crossarch.ensure_ready()
+
+    def test_auto_install_failure_raises(self, monkeypatch, answers):
+        monkeypatch.setattr(crossarch.platform, "machine", lambda: "x86_64")
+        monkeypatch.setattr(crossarch, "qemu_binfmt_enabled", lambda: False)
+        monkeypatch.setattr(crossarch, "auto_enable", lambda: False)
+        answers("y")
+        with pytest.raises(LimError, match="Cross-architecture"):
+            crossarch.ensure_ready()
+
+    def test_auto_enable_uses_detected_manager_and_registers(self, monkeypatch, fake_runner):
+        monkeypatch.setattr(
+            crossarch.shutil, "which", lambda tool: "/bin/pacman" if tool == "pacman" else None
+        )
+        monkeypatch.setattr(crossarch, "qemu_binfmt_enabled", lambda: True)
+        assert crossarch.auto_enable() is True
+        assert fake_runner.find("pacman", "-S", "qemu-user-static-binfmt")
+        assert fake_runner.find("systemctl", "restart", "systemd-binfmt")
+
+    def test_auto_enable_without_manager_returns_false(self, monkeypatch):
+        monkeypatch.setattr(crossarch.shutil, "which", lambda _tool: None)
+        assert crossarch.auto_enable() is False

tests/unit/test_fsutil.py

@@ -33,3 +33,25 @@ def test_ensure_line_creates_file_and_handles_missing_newline(tmp_path):
     target.write_text("no newline at end")
     assert fsutil.ensure_line_in_file("second", target) is True
     assert target.read_text() == "no newline at end\nsecond\n"
+
+
+def test_drop_fstab_mount_removes_matching_mount_only(tmp_path):
+    target = tmp_path / "fstab"
+    target.write_text(
+        "# comment / stays\n"
+        "PARTUUID=aa-02  /  ext4  defaults  0  1\n"
+        "PARTUUID=aa-01  /boot/firmware  vfat  defaults  0  2\n"
+    )
+    fsutil.drop_fstab_mount("/", target)
+    remaining = target.read_text()
+    assert "PARTUUID=aa-02" not in remaining
+    assert "# comment / stays" in remaining
+    assert "/boot/firmware" in remaining
+
+
+def test_drop_fstab_mount_is_noop_when_absent(tmp_path):
+    target = tmp_path / "fstab"
+    target.write_text("PARTUUID=aa-01  /boot  vfat  defaults  0  2\n")
+    fsutil.drop_fstab_mount("/", target)
+    assert target.read_text() == "PARTUUID=aa-01  /boot  vfat  defaults  0  2\n"
+    fsutil.drop_fstab_mount("/", tmp_path / "missing")  # no crash

tests/unit/test_image_session.py

@@ -60,3 +60,17 @@ def test_destructor_closes_luks_mapper(tmp_path, fake_runner):
     session.decrypt_root()
     session.destructor()
     assert len(fake_runner.find("luksClose", "linux-image-manager-uuid-1")) == 1
+
+
+def test_boot_bind_target_bookworm_uses_boot_firmware(tmp_path):
+    session = ImageSession(Device("sda"))
+    session.root_mount_path = tmp_path
+    (tmp_path / "boot/firmware").mkdir(parents=True)
+    assert session.boot_bind_target() == f"{tmp_path}/boot/firmware"
+
+
+def test_boot_bind_target_legacy_uses_boot(tmp_path):
+    session = ImageSession(Device("sda"))
+    session.root_mount_path = tmp_path
+    (tmp_path / "boot").mkdir()
+    assert session.boot_bind_target() == f"{tmp_path}/boot"

tests/unit/test_image_setup.py

@@ -63,6 +63,17 @@ class TestDistributionChoosers:
         with pytest.raises(LimError):
             choosers.choose_linux_image(plan)
 
+    def test_numeric_selection_maps_to_distribution(self, plan, answers):
+        number = str(list(choosers.DISTRIBUTION_CHOOSERS).index("raspios") + 1)
+        answers(number, "lite64")
+        choosers.choose_linux_image(plan)
+        assert plan.distribution == "raspios"
+
+    def test_name_selection_still_works(self, plan, answers):
+        answers("arch", "1")
+        choosers.choose_linux_image(plan)
+        assert plan.distribution == "arch"
+
 
 class TestPartitionInput:
     def test_contains_boot_size_and_writes_table(self):
@@ -101,7 +112,8 @@ class TestInstallPackages:
             for kind, cmd, input_text in fake_runner.calls
             if kind == "run" and cmd[0] == "chroot"
         ]
-        assert chroot_calls == ["pacman --noconfirm -S --needed btrfs-progs"]
+        assert len(chroot_calls) == 1
+        assert "pacman --noconfirm -Sy --needed btrfs-progs" in chroot_calls[0]
 
     def test_apt_for_retropie(self, tmp_path, fake_runner):
         install_packages("retropie", tmp_path, "btrfs-progs")
@@ -110,7 +122,9 @@ class TestInstallPackages:
             for kind, cmd, input_text in fake_runner.calls
             if kind == "run" and cmd[0] == "chroot"
         ]
-        assert chroot_calls == ["yes | apt install btrfs-progs"]
+        assert len(chroot_calls) == 1
+        assert "apt-get update" in chroot_calls[0]
+        assert "apt-get install -y btrfs-progs" in chroot_calls[0]
 
     def test_unsupported_distribution_raises(self, tmp_path, fake_runner):
         with pytest.raises(LimError):

tests/unit/test_storage.py

@@ -20,9 +20,7 @@ def test_storage_target_derives_all_paths():
     assert target.partition_path == "/dev/sdb1"
 
 
-def test_single_drive_setup_command_sequence(
+def test_single_drive_setup_command_sequence(fake_runner, answers, block_devices, monkeypatch):
-    fake_runner, answers, block_devices, monkeypatch
-):
     monkeypatch.setattr("lim.system.real_user", lambda: "kevin")
     answers("sdb", "N")  # device name, skip overwrite
 

tests/unit/test_sync.py

@@ -58,10 +58,19 @@ def test_rsync_command_uses_delete_only_for_directories():
     directory_op = sync.SyncOperation("src/", "dst/", "backup", is_directory=True)
     file_op = sync.SyncOperation("src", "dst", "backup", is_directory=False)
     assert sync.rsync_command(directory_op) == [
-        "rsync", "-abcEPuvW", "--delete", "--backup-dir=backup", "src/", "dst/"
+        "rsync",
+        "-abcEPuvW",
+        "--delete",
+        "--backup-dir=backup",
+        "src/",
+        "dst/",
     ]
     assert sync.rsync_command(file_op) == [
-        "rsync", "-abcEPuvW", "--backup-dir=backup", "src", "dst"
+        "rsync",
+        "-abcEPuvW",
+        "--backup-dir=backup",
+        "src",
+        "dst",
     ]
 
 
@@ -95,18 +104,12 @@ def test_export_chowns_real_home_without_sudo(tmp_path, fake_runner, monkeypatch
 
     sync.export_to_system()
 
-    chown_calls = [
+    chown_calls = [(cmd, sudo) for cmd, sudo in fake_runner.sudo_log if cmd[0] == "chown"]
-        (cmd, sudo) for cmd, sudo in fake_runner.sudo_log if cmd[0] == "chown"
+    assert chown_calls == [(["chown", "-R", "kevin:kevin", str(real_home)], False)]
-    ]
-    assert chown_calls == [
-        (["chown", "-R", "kevin:kevin", str(real_home)], False)
-    ]
     assert (real_home / ".ssh/id_rsa").stat().st_mode & 0o777 == 0o600
 
 
-def test_execute_sync_plan_creates_folders_and_runs_rsync(
+def test_execute_sync_plan_creates_folders_and_runs_rsync(tmp_path, fake_runner):
-    tmp_path, fake_runner
-):
     operation = sync.SyncOperation(
         source=str(tmp_path / "src.txt"),
         destination=str(tmp_path / "deep/nested/dst.txt"),

tests/unit/test_tor.py

@@ -0,0 +1,220 @@
+import pytest
+
+from lim import config
+from lim.device import Device
+from lim.errors import LimError
+from lim.image.initramfs import get_backend, keygen, mkinitcpio
+from lim.image.initramfs.initramfs_tools import InitramfsToolsBackend
+from lim.image.initramfs.mkinitcpio import MkinitcpioBackend
+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)
+
+
+@pytest.fixture
+def debian_plan():
+    return ImagePlan(distribution="raspios", root_filesystem="ext4", tor_unlock=True)
+
+
+def _write_onion_hostname(root, address="abcdefghijklmnop.onion"):
+    onion_dir = root / keygen.ONION_STAGING_DIR
+    onion_dir.mkdir(parents=True)
+    (onion_dir / "hostname").write_text(f"{address}\n")
+
+
+def _session(tmp_path, mapper="cryptroot"):
+    session = ImageSession(Device("mmcblk0"))
+    session.root_partition_uuid = "ROOT-UUID"
+    session.root_mapper_name = mapper
+    session.root_mapper_path = f"/dev/mapper/{mapper}"
+    session.boot_mount_path = tmp_path / "boot"
+    session.boot_mount_path.mkdir()
+    return session
+
+
+class TestBackendDispatch:
+    def test_arch_and_manjaro_use_mkinitcpio(self):
+        assert isinstance(get_backend("arch"), MkinitcpioBackend)
+        assert isinstance(get_backend("manjaro"), MkinitcpioBackend)
+
+    def test_raspios_uses_initramfs_tools(self):
+        assert isinstance(get_backend("raspios"), InitramfsToolsBackend)
+
+    def test_unknown_distribution_raises(self):
+        with pytest.raises(LimError, match="No initramfs backend"):
+            get_backend("gentoo")
+
+
+class TestMkinitcpioTorUnlock:
+    def test_installs_hooks_and_returns_address(self, tmp_path, plan, fake_runner):
+        _write_onion_hostname(tmp_path, "stableaddress.onion")
+
+        address = MkinitcpioBackend().install_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
+        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"):
+            MkinitcpioBackend().install_tor_unlock(plan, tmp_path)
+        keygen_calls = [
+            input_text
+            for _, _, 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]
+
+
+class TestInitcpioHookHardening:
+    """Guards for review findings in the shipped mkinitcpio hooks."""
+
+    def _read(self, name):
+        return (config.CONFIGURATION_PATH / "initcpio" / name).read_text()
+
+    def test_hook_resources_exist(self):
+        for name in ("tor_install", "tor_hook", "torrc"):
+            assert (config.CONFIGURATION_PATH / "initcpio" / name).is_file()
+
+    def test_install_bakes_the_dns_resolver(self):
+        assert "libnss_dns.so.2" in self._read("tor_install")
+
+    def test_hook_never_sources_dhcp_lease_files(self):
+        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):
+        assert "skipping NTP sync" not in self._read("tor_hook")
+
+
+class TestMkinitcpioBootloader:
+    """The cmdline.txt boot path (RPi4-class) must set ip= for remote unlock."""
+
+    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 = _session(tmp_path)
+        (session.boot_mount_path / "cmdline.txt").write_text("root=/dev/mmcblk0p2 rw rootwait\n")
+
+        MkinitcpioBackend().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=({mkinitcpio.MKINITCPIO_HOOKS_PREFIX} "
+            f"{mkinitcpio.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)
+        MkinitcpioBackend().configure_initramfs(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)
+        MkinitcpioBackend().configure_initramfs(plan, tmp_path)
+        content = path.read_text()
+        assert "netconf dropbear encryptssh" in content
+        assert " tor " not in content
+
+
+class TestInitramfsToolsBackend:
+    """The Debian / Raspberry Pi OS backend."""
+
+    def test_crypttab_uses_initramfs_option(self, tmp_path, debian_plan):
+        root = tmp_path / "root"
+        (root / "etc").mkdir(parents=True)
+        session = _session(tmp_path)
+        InitramfsToolsBackend().register_encrypted_root(debian_plan, session, root)
+        crypttab = (root / "etc/crypttab").read_text()
+        assert "cryptroot UUID=ROOT-UUID none luks,initramfs" in crypttab
+        assert "/dev/mapper/cryptroot" in (root / "etc/fstab").read_text()
+
+    def test_cmdline_points_at_mapper_with_network(self, tmp_path, debian_plan):
+        root = tmp_path / "root"
+        (root / "etc").mkdir(parents=True)
+        (root / "etc" / "hostname").write_text("pi\n")
+        session = _session(tmp_path)
+        (session.boot_mount_path / "cmdline.txt").write_text(
+            "console=serial0,115200 root=PARTUUID=abcd-02 rootfstype=ext4 rootwait\n"
+        )
+        (session.boot_mount_path / "config.txt").write_text("dtparam=audio=on\n")
+
+        InitramfsToolsBackend().configure_bootloader(debian_plan, session, root)
+
+        cmdline = (session.boot_mount_path / "cmdline.txt").read_text()
+        assert "root=/dev/mapper/cryptroot" in cmdline
+        assert "root=PARTUUID=abcd-02" not in cmdline
+        assert "ip=::::pi:eth0:dhcp" in cmdline
+        assert "tor_ntp=" in cmdline
+        assert cmdline.count("\n") == 1  # cmdline must stay a single line
+        assert "auto_initramfs=1" in (session.boot_mount_path / "config.txt").read_text()
+
+    def test_install_tor_unlock_places_initramfs_tools_scripts(
+        self, tmp_path, debian_plan, fake_runner
+    ):
+        _write_onion_hostname(tmp_path, "debianonion.onion")
+        address = InitramfsToolsBackend().install_tor_unlock(debian_plan, tmp_path)
+        assert address == "debianonion.onion"
+        assert len(fake_runner.find("install", "tor_hook", "hooks/tor")) == 1
+        assert len(fake_runner.find("tor_premount", "init-premount/tor")) == 1
+        assert len(fake_runner.find("tor_bottom", "init-bottom/tor")) == 1
+        apt = [
+            text
+            for _, _, text in fake_runner.calls
+            if text and "apt-get install" in text and "tor busybox" in text
+        ]
+        assert len(apt) == 1
+
+
+class TestInitramfsToolsHookHardening:
+    def _read(self, name):
+        return (config.CONFIGURATION_PATH / "initramfs-tools" / name).read_text()
+
+    def test_hook_files_exist(self):
+        for name in ("tor_hook", "tor_premount", "tor_bottom", "torrc"):
+            assert (config.CONFIGURATION_PATH / "initramfs-tools" / name).is_file()
+
+    def test_hook_bakes_dns_resolver(self):
+        assert "libnss_dns.so.2" in self._read("tor_hook")
+
+    def test_premount_does_not_source_lease_files(self):
+        premount = self._read("tor_premount")
+        assert '. "$conf"' not in premount
+        assert "sed -n 's/^IPV4DNS" in premount
+
+    def test_hook_stamps_a_clock_floor(self):
+        assert "/etc/tor-clock-floor" in self._read("tor_hook")
+
+    def test_premount_raises_clock_to_floor_before_ntp(self):
+        premount = self._read("tor_premount")
+        assert "/etc/tor-clock-floor" in premount
+        assert premount.index("tor-clock-floor") < premount.index("ntpd")

tests/unit/test_unlock.py

@@ -0,0 +1,42 @@
+import os
+
+from lim import cli
+from lim.image import unlock
+
+
+class TestBuildUnlockArgv:
+    def test_onion_wraps_torsocks_and_runs_cryptroot_unlock(self):
+        argv = unlock.build_unlock_argv("abc.onion", "/k/id", "cryptroot-unlock")
+        assert argv[0] == "torsocks"
+        assert "root@abc.onion" in argv
+        assert argv[argv.index("-i") + 1] == "/k/id"
+        assert argv[-1] == "cryptroot-unlock"
+        assert "-t" in argv
+
+    def test_direct_target_is_plain_ssh(self):
+        argv = unlock.build_unlock_argv("192.168.1.5", "", "")
+        assert "torsocks" not in argv
+        assert argv[-1] == "root@192.168.1.5"
+        assert "-t" not in argv
+        assert "-i" not in argv
+
+    def test_host_key_checking_disabled(self):
+        assert "StrictHostKeyChecking=no" in unlock.build_unlock_argv("a.onion", "", "")
+
+
+class TestRecords:
+    def test_save_and_load_record_roundtrip(self, tmp_path, monkeypatch):
+        record = {"target": "xyz.onion", "key": "/k/id", "unlock_command": "cryptroot-unlock"}
+        unlock.save_record(tmp_path, os.getuid(), os.getgid(), "myhost", record)
+        assert (tmp_path / unlock.RECORDS_SUBPATH / "myhost.json").is_file()
+        monkeypatch.setattr(
+            unlock, "records_dir", lambda home=None: tmp_path / unlock.RECORDS_SUBPATH
+        )
+        assert unlock._load_records() == [("myhost", record)]
+
+
+class TestCliRegistration:
+    def test_remote_unlock_registered_without_root(self):
+        command = cli.COMMANDS["remote-unlock"]
+        assert command.func is unlock.remote_unlock
+        assert command.needs_root is False

tests/unit/test_wizard.py

@@ -0,0 +1,235 @@
+import pytest
+
+from lim import catalog, cli
+from lim.device import Device
+from lim.errors import LimError
+from lim.image import choosers, transfer, wizard
+from lim.image.initramfs.initramfs_tools import InitramfsToolsBackend
+from lim.image.plan import ImagePlan
+from lim.image.session import ImageSession
+
+
+def _session(tmp_path):
+    session = ImageSession(Device("sda"))
+    session.working_folder = tmp_path
+    session.boot_mount_path = tmp_path / "boot"
+    session.root_mount_path = tmp_path / "root"
+    session.boot_mount_path.mkdir()
+    session.root_mount_path.mkdir()
+    return session
+
+
+class TestCatalogAndChooser:
+    def test_catalog_exposes_raspios_latest_endpoints(self):
+        images = catalog.raspios_images()
+        assert images["lite64"]["source_url"].endswith("raspios_lite_arm64_latest")
+        assert images["lite64"]["image"].endswith(".img.xz")
+
+    def test_choose_raspios_sets_source_and_defaults(self, answers):
+        answers("lite64")
+        plan = ImagePlan(distribution="raspios")
+        choosers.choose_raspios(plan)
+        assert plan.source_url.endswith("raspios_lite_arm64_latest")
+        assert plan.image_name.endswith(".img.xz")
+        assert plan.root_filesystem == "ext4"
+        assert plan.boot_size == "+512M"
+
+    def test_unknown_variant_raises(self, answers):
+        answers("lite999")
+        with pytest.raises(LimError, match="isn't supported"):
+            choosers.choose_raspios(ImagePlan(distribution="raspios"))
+
+    def test_raspios_is_registered(self):
+        assert choosers.DISTRIBUTION_CHOOSERS["raspios"] is choosers.choose_raspios
+
+
+class TestSourceUrlOverride:
+    def test_source_url_wins_over_derived(self):
+        plan = ImagePlan(
+            base_download_url="http://x/", image_name="a.img.xz", source_url="http://latest"
+        )
+        assert plan.download_url == "http://latest"
+
+    def test_falls_back_to_base_plus_name(self):
+        plan = ImagePlan(base_download_url="http://x/", image_name="a.img.xz")
+        assert plan.download_url == "http://x/a.img.xz"
+
+
+class TestTransferDiskImage:
+    def test_builds_encrypted_layout_and_copies_partitions(
+        self, tmp_path, fake_runner, monkeypatch
+    ):
+        monkeypatch.setattr(transfer.shutil, "which", lambda _name: None)  # force cp branch
+        fake_runner.outputs = {
+            "losetup -Pf": "/dev/loop9",
+            "-s TYPE": "crypto_LUKS",
+            "-s UUID": "ROOTUUID",
+        }
+        session = _session(tmp_path)
+        plan = ImagePlan(
+            distribution="raspios",
+            image_name="raspios_lite_arm64_latest.img.xz",
+            root_filesystem="ext4",
+            boot_size="+512M",
+        )
+        plan.image_folder = tmp_path
+
+        transfer.transfer_disk_image(plan, session)
+
+        assert fake_runner.find("losetup", "-Pf", "--show")
+        assert fake_runner.find("mount", "-o", "ro", "/dev/loop9p1")
+        assert fake_runner.find("mount", "-o", "ro", "/dev/loop9p2")
+        assert fake_runner.find("fdisk", "/dev/sda")
+        assert fake_runner.find("mkfs.vfat", "/dev/sda1")
+        assert fake_runner.find("cryptsetup", "luksFormat")
+        assert fake_runner.find("cryptsetup", "luksOpen")
+        assert fake_runner.find("mkfs.ext4")
+        assert fake_runner.find("cp", "-a", "source-root")
+        assert fake_runner.find("cp", "-a", "source-boot")
+        assert fake_runner.find("losetup", "-d", "/dev/loop9")
+
+    def test_copy_tree_uses_rsync_with_progress(self, fake_runner, monkeypatch):
+        monkeypatch.setattr(transfer.shutil, "which", lambda _name: "/usr/bin/rsync")
+        transfer._copy_tree("/src", "/dst", ["-aHAX"])
+        assert fake_runner.find("rsync", "-aHAX", "--info=progress2", "/src/", "/dst/")
+
+    def test_copy_tree_falls_back_to_cp(self, fake_runner, monkeypatch):
+        monkeypatch.setattr(transfer.shutil, "which", lambda _name: None)
+        transfer._copy_tree("/src", "/dst", ["-a"])
+        assert fake_runner.find("cp", "-a", "/src/.", "/dst")
+
+    def test_encrypted_non_arch_dispatches_to_disk_image(
+        self, tmp_path, fake_runner, answers, monkeypatch
+    ):
+        calls = []
+        monkeypatch.setattr(
+            transfer, "transfer_disk_image", lambda plan, session: calls.append(plan.distribution)
+        )
+        answers("N", "N")  # keep partition table, skip zero-overwrite
+        plan = ImagePlan(distribution="moode", encrypt_system=True)
+        transfer.transfer_image(plan, _session(tmp_path))
+        assert calls == ["moode"]
+
+    def test_arch_uses_tarball_path_not_disk_image(
+        self, tmp_path, fake_runner, answers, monkeypatch
+    ):
+        disk, arch = [], []
+        monkeypatch.setattr(transfer, "transfer_disk_image", lambda p, s: disk.append(p))
+        monkeypatch.setattr(transfer, "transfer_arch_image", lambda p, s: arch.append(p))
+        answers("N", "N")
+        plan = ImagePlan(distribution="arch", encrypt_system=True)
+        transfer.transfer_image(plan, _session(tmp_path))
+        assert len(arch) == 1
+        assert disk == []
+
+    def test_non_interactive_transfer_skips_erase_prompts(self, tmp_path, fake_runner, monkeypatch):
+        monkeypatch.setattr(transfer, "transfer_disk_image", lambda p, s: None)
+        plan = ImagePlan(distribution="raspios", encrypt_system=True)
+        transfer.transfer_image(plan, _session(tmp_path), interactive=False)
+        assert not fake_runner.find("wipefs")  # no prompt, no destructive pre-step
+
+    def test_download_without_force_prompt_does_not_ask(self, tmp_path, fake_runner):
+        plan = ImagePlan(distribution="raspios", image_name="x.img.xz", source_url="http://h/x")
+        plan.image_folder = tmp_path
+        transfer.download_image(plan, force_prompt=False)  # no answers fixture => must not prompt
+        assert fake_runner.find("wget", "http://h/x")
+
+
+class TestBootFstabFix:
+    def test_repoints_boot_line_to_new_uuid(self, tmp_path):
+        session = _session(tmp_path)
+        session.boot_partition_uuid = "BOOTUUID"
+        (session.root_mount_path / "etc").mkdir()
+        (session.root_mount_path / "etc/fstab").write_text(
+            "PARTUUID=aa-02  /  ext4  defaults  0  1\n"
+            "PARTUUID=aa-01  /boot/firmware  vfat  defaults  0  2\n"
+        )
+        transfer._fix_boot_fstab(session)
+        fstab = (session.root_mount_path / "etc/fstab").read_text()
+        assert "UUID=BOOTUUID  /boot/firmware" in fstab
+        assert "PARTUUID=aa-01" not in fstab
+        assert "PARTUUID=aa-02  /  ext4" in fstab  # root line untouched here
+
+
+class TestRegisterEncryptedRootStockFstab:
+    def test_existing_root_line_is_replaced_not_duplicated(self, tmp_path):
+        session = ImageSession(Device("sda"))
+        session.root_mapper_name = "cryptroot"
+        session.root_partition_uuid = "ROOT-UUID"
+        root = tmp_path / "root"
+        (root / "etc").mkdir(parents=True)
+        (root / "etc/fstab").write_text("PARTUUID=aa-02  /  ext4  defaults  0  1\n")
+        plan = ImagePlan(distribution="raspios", root_filesystem="ext4")
+
+        InitramfsToolsBackend().register_encrypted_root(plan, session, root)
+
+        fstab = (root / "etc/fstab").read_text()
+        assert fstab.count("  /  ") == 1
+        assert "/dev/mapper/cryptroot  /  ext4" in fstab
+        assert "PARTUUID=aa-02" not in fstab
+
+
+class TestWizardSteps:
+    def test_build_authorized_keys_copies_pubkey(self, tmp_path):
+        pubkey = tmp_path / "id.pub"
+        pubkey.write_text("ssh-ed25519 AAAA user@host\n")
+        result = wizard._build_authorized_keys(_session(tmp_path), pubkey)
+        assert result.read_text() == "ssh-ed25519 AAAA user@host\n"
+
+    def test_ask_pubkey_rejects_missing(self, tmp_path, answers):
+        answers(str(tmp_path / "nope.pub"))
+        with pytest.raises(LimError, match="not found"):
+            wizard._ask_pubkey()
+
+    def test_ask_pubkey_rejects_empty(self, answers):
+        answers("")
+        with pytest.raises(LimError, match="No SSH public key"):
+            wizard._ask_pubkey()
+
+    def test_apply_system_config_sets_hostname_and_enables_ssh(self, tmp_path):
+        session = _session(tmp_path)
+        (session.root_mount_path / "etc").mkdir()
+        hostname = wizard._apply_system_config(session, "pi-tor")
+        assert hostname == "pi-tor"
+        assert (session.root_mount_path / "etc/hostname").read_text() == "pi-tor\n"
+        assert (session.boot_mount_path / "ssh").is_file()
+
+    def test_configure_login_user_renames_default_and_installs_key(self, tmp_path, fake_runner):
+        pubkey = tmp_path / "id.pub"
+        pubkey.write_text("ssh-ed25519 AAAA user@host\n")
+        wizard._configure_login_user(_session(tmp_path), "kevin", "secret", pubkey)
+        script = next(text for _, cmd, text in fake_runner.calls if cmd[0] == "chroot" and text)
+        assert "for old in pi alarm" in script
+        assert 'usermod -l "$target" "$old"' in script
+        assert "target=kevin" in script
+        assert 'useradd -m -s /bin/bash "$target"' in script
+        assert "ssh-ed25519 AAAA" in script
+        assert "chpasswd" in script
+
+    def test_configure_login_user_without_password_skips_chpasswd(self, tmp_path, fake_runner):
+        pubkey = tmp_path / "id.pub"
+        pubkey.write_text("ssh-ed25519 AAAA user@host\n")
+        wizard._configure_login_user(_session(tmp_path), "pi", "", pubkey)
+        script = next(text for _, cmd, text in fake_runner.calls if cmd[0] == "chroot" and text)
+        assert "chpasswd" not in script
+
+    def test_select_target_aborts_on_mismatch(self, answers, monkeypatch):
+        monkeypatch.setattr(wizard.device_module, "select_device", lambda: Device("sda"))
+        monkeypatch.setattr(wizard.device_module, "is_mounted", lambda _p: False)
+        answers("typo")
+        with pytest.raises(LimError, match="did not match"):
+            wizard._select_target()
+
+    @pytest.mark.parametrize("confirmation", ["sda", "/dev/sda"])
+    def test_select_target_accepts_name_or_path(self, confirmation, answers, monkeypatch):
+        monkeypatch.setattr(wizard.device_module, "select_device", lambda: Device("sda"))
+        monkeypatch.setattr(wizard.device_module, "is_mounted", lambda _p: False)
+        answers(confirmation)
+        assert wizard._select_target().name == "sda"
+
+
+class TestCliRegistration:
+    def test_guided_command_registered(self):
+        command = cli.COMMANDS["guided"]
+        assert command.func is wizard.run_guided
+        assert command.needs_root is True