Release version 2.2.0

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

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

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

.github/workflows/test.yml

@@ -3,7 +3,6 @@ name: tests
 on:
   push:
   pull_request:
-  workflow_dispatch:
 
 jobs:
   lint:
@@ -38,18 +37,3 @@ jobs:
       - 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

Makefile

@@ -1,4 +1,4 @@
-.PHONY: install test test-tor test-qemu test-qemu-debian test-all
+.PHONY: install test test-tor test-qemu test-all
 
 PREFIX ?= $(HOME)/.local
 # Interpreter used for the tests. Override if your active venv lacks pytest,
@@ -20,18 +20,11 @@ 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).
+# cryptsetup, tor and ncat; the build stage runs under sudo. Set CHUTNEY_PATH
+# to use a private, offline Tor network instead of the public one.
 test-qemu:
 	LIM_E2E_QEMU=1 $(PYTHON) -m pytest tests/e2e/test_qemu_unlock_e2e.py -v -s
 
-# 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

lim/catalog.py

@@ -27,9 +27,5 @@ 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,8 +10,6 @@ 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
 
 
@@ -31,20 +29,6 @@ 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"
@@ -75,12 +59,14 @@ COMMANDS: dict[str, Command] = {
     ),
     "mount": Command(
         single_drive.mount,
-        "Mount Encrypted Storage:\n  - Unlocks a LUKS partition and mounts it.",
+        "Mount Encrypted Storage:\n"
+        "  - Unlocks a LUKS partition and mounts it.",
         needs_root=True,
     ),
     "umount": Command(
         single_drive.umount,
-        "Unmount Encrypted Storage:\n  - Unmounts a LUKS partition and closes the mapper.",
+        "Unmount Encrypted Storage:\n"
+        "  - Unmounts a LUKS partition and closes the mapper.",
         needs_root=True,
     ),
     "single-boot": Command(
@@ -97,22 +83,26 @@ COMMANDS: dict[str, Command] = {
     ),
     "unlock": Command(
         crypt.unlock,
-        "Unlock Data:\n  - Decrypts the encfs data store into the decrypted folder.",
+        "Unlock Data:\n"
+        "  - Decrypts the encfs data store into the decrypted folder.",
         needs_root=False,
     ),
     "lock": Command(
         crypt.lock,
-        "Lock Data:\n  - Unmounts the decrypted encfs folder.",
+        "Lock Data:\n"
+        "  - Unmounts the decrypted encfs folder.",
         needs_root=False,
     ),
     "import": Command(
         sync.import_from_system,
-        "Import Data:\n  - Copies personal data from the system into the encrypted store.",
+        "Import Data:\n"
+        "  - Copies personal data from the system into the encrypted store.",
         needs_root=False,
     ),
     "export": Command(
         sync.export_to_system,
-        "Export Data:\n  - Copies personal data from the encrypted store back to the system.",
+        "Export Data:\n"
+        "  - Copies personal data from the encrypted store back to the system.",
         needs_root=False,
     ),
 }
@@ -132,15 +122,21 @@ def build_parser() -> argparse.ArgumentParser:
     )
     parser.add_argument(
         "--type",
-        default="guided",
+        required=True,
         choices=list(COMMANDS.keys()),
-        help="Command to execute (default: guided).",
+        help="Select the command to execute.",
     )
     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
 
 
@@ -148,6 +144,8 @@ 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/initramfs-tools/tor_bottom

@@ -1,19 +0,0 @@
-#!/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

@@ -1,37 +0,0 @@
-#!/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

@@ -1,55 +0,0 @@
-#!/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

@@ -1,16 +0,0 @@
-# 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

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

@@ -1,3 +0,0 @@
-# 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/data/sync.py

@@ -90,22 +90,31 @@ 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(["diff", operation.destination, operation.source], check=False)
+            runner.run(
+                ["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(f"The decrypted folder {config.DECRYPTED_PATH} is locked. You need to unlock it!")
+        ui.info(
+            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 = config.BACKUP_PATH / mode / time.strftime("%Y%m%d%H%M%S")
+    backup_folder = (
+        config.BACKUP_PATH / mode / time.strftime("%Y%m%d%H%M%S")
+    )
     backup_folder.mkdir(parents=True, exist_ok=True)
-    operations = build_sync_plan(mode, system.real_home(), config.DATA_PATH, backup_folder)
+    operations = build_sync_plan(
+        mode, system.real_home(), config.DATA_PATH, backup_folder
+    )
     execute_sync_plan(operations)
 
 

lim/device.py

@@ -119,7 +119,9 @@ 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(["blkid", path, "-s", tag, "-o", "value"], sudo=True, check=False)
+    return runner.output(
+        ["blkid", path, "-s", tag, "-o", "value"], sudo=True, check=False
+    )
 
 
 def is_mounted(path_fragment: str) -> bool:

lim/distributions.yml

@@ -57,19 +57,6 @@ 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,23 +16,6 @@ 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,21 +5,8 @@ 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 = _choose("Which Raspberry Pi version", list(catalog.arch_rpi_images()))
+    version = ui.ask("Which Raspberry Pi will be used (e.g.: 1, 2, 3b, 3b+, 4...):")
     entry = catalog.arch_rpi_images().get(version)
     if entry is None:
         raise LimError(f"Version {version} isn't supported.")
@@ -32,7 +19,7 @@ def choose_arch(plan: ImagePlan) -> None:
 
 def choose_manjaro(plan: ImagePlan) -> None:
     plan.boot_size = "+500M"
-    flavour = _choose("Which Manjaro version", ["architect", "gnome"])
+    flavour = ui.ask("Which version(e.g.:architect,gnome) should be used:")
     if flavour == "architect":
         plan.image_checksum = "6b1c2fce12f244c1e32212767a9d3af2cf8263b2"
         plan.base_download_url = (
@@ -41,7 +28,7 @@ def choose_manjaro(plan: ImagePlan) -> None:
         )
         plan.image_name = "manjaro-architect-20.0-200426-linux56.iso"
     elif flavour == "gnome":
-        release = _choose("Which Gnome release", list(catalog.manjaro_gnome_releases()))
+        release = ui.ask("Which release(e.g.:20,21,raspberrypi) should be used:")
         entry = catalog.manjaro_gnome_releases().get(release)
         if entry is None:
             raise LimError(f"Gnome release {release} isn't supported.")
@@ -57,37 +44,32 @@ def choose_manjaro(plan: ImagePlan) -> None:
 def choose_moode(plan: ImagePlan) -> None:
     plan.boot_size = "+200M"
     plan.image_checksum = "185cbc9a4994534bb7a4bc2744c78197"
-    plan.base_download_url = "https://github.com/moode-player/moode/releases/download/r651prod/"
+    plan.base_download_url = (
+        "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 = _choose("Which RetroPie version", list(catalog.retropie_images()))
+    version = ui.ask("Which version(e.g.:1,2,3,4) should be used:")
     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 = "https://github.com/RetroPie/RetroPie-Setup/releases/download/4.8/"
+    plan.base_download_url = (
+        "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 = "0E1BA7FFD14AAAE5F0462C8293D95B62C3BF1D9E726E26977BD04772C55680D3"
+    plan.image_checksum = (
+        "0E1BA7FFD14AAAE5F0462C8293D95B62C3BF1D9E726E26977BD04772C55680D3"
+    )
     plan.boot_size = "+200M"
 
 
@@ -96,7 +78,9 @@ 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 = "f7eb8fc56f29ad5432335dc054183acf086c539f3990f0b6e9ff58bd6df4604e"
+    plan.image_checksum = (
+        "f7eb8fc56f29ad5432335dc054183acf086c539f3990f0b6e9ff58bd6df4604e"
+    )
     plan.boot_size = "+500M"
 
 
@@ -107,12 +91,13 @@ DISTRIBUTION_CHOOSERS = {
     "manjaro": choose_manjaro,
     "moode": choose_moode,
     "retropie": choose_retropie,
-    "raspios": choose_raspios,
 }
 
 
 def choose_linux_image(plan: ImagePlan) -> None:
-    distribution = _choose("Which distribution", list(DISTRIBUTION_CHOOSERS))
+    distribution = ui.ask(
+        "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

@@ -1,71 +0,0 @@
-"""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,42 +1,129 @@
-"""Remote-unlockable LUKS boot configuration.
+"""Remote-unlockable LUKS boot configuration (dropbear + mkinitcpio).
 
-Distro-agnostic orchestration; the init-system-specific steps live in the
-initramfs backends (mkinitcpio for Arch/Manjaro, initramfs-tools for
-Debian/Raspberry Pi OS). See lim/image/initramfs/.
+See https://gist.github.com/gea0/4fc2be0cb7a74d0e7cc4322aed710d38 and
+https://gist.github.com/EnigmaCurry/2f9bed46073da8e38057fe78a61e7994
 """
 
 from pathlib import Path
 
-from lim import packages, ui
-from lim.image.initramfs import get_backend
+from lim import catalog, fsutil, packages, runner, ui
 from lim.image.plan import ImagePlan
-from lim.image.session import ImageSession, install_packages
+from lim.image.session import ImageSession, chroot_bash, install_packages
+from lim.image.tor import configure_tor_unlock
+
+MKINITCPIO_HOOKS_PREFIX = (
+    "base udev autodetect microcode modconf kms keyboard keymap consolefont block"
+)
+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
+    )
+    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 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}...")
+        # 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()}")
 
 
 def configure_encryption(
     plan: ImagePlan, session: ImageSession, authorized_keys: Path
-) -> str | None:
-    """Configure the remote-unlock LUKS stack; return the onion address, if any."""
+) -> None:
     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(backend.luks_package_collection())),
+        plan.distribution, root, " ".join(packages.get_packages("server/luks"))
     )
 
-    backend.install_authorized_key(root, authorized_keys)
+    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)
 
-    onion_address = None
     if plan.tor_unlock:
-        # Hook files and onion keys must exist before the initramfs is baked.
-        onion_address = backend.install_tor_unlock(plan, root)
+        # Hook files and onion keys must exist before mkinitcpio bakes the image.
+        configure_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
+    _configure_mkinitcpio(plan, root)
+    _register_encrypted_root(plan, session, root)
+    _configure_bootloader(plan, session, root)

lim/image/initramfs/__init__.py

@@ -1,21 +0,0 @@
-"""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

@@ -1,42 +0,0 @@
-"""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

@@ -1,117 +0,0 @@
-"""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

@@ -1,54 +0,0 @@
-"""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

@@ -1,122 +0,0 @@
-"""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

@@ -1,28 +0,0 @@
-"""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,10 +12,6 @@ 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
@@ -27,8 +23,6 @@ class ImagePlan:
 
     @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,7 +23,9 @@ 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(public_key_path: str, target_ssh_folder: Path, authorized_keys: Path) -> None:
+def configure_ssh_key(
+    public_key_path: str, target_ssh_folder: Path, authorized_keys: Path
+) -> None:
     source = Path(public_key_path)
     if not source.is_file():
         raise LimError(
@@ -62,7 +64,9 @@ 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("/dev/mmcblk0p1", f"UUID={session.boot_partition_uuid}", fstab_path)
+    fsutil.replace_in_file(
+        "/dev/mmcblk0p1", f"UUID={session.boot_partition_uuid}", fstab_path
+    )
     ui.info(f"Content of {fstab_path}:{fstab_path.read_text()}")
 
 
@@ -131,7 +135,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", "raspios"):
+    elif plan.distribution in ("moode", "retropie"):
         chroot_bash(root, "yes | apt update\nyes | apt upgrade\n")
     else:
         ui.warning(

lim/image/session.py

@@ -49,8 +49,12 @@ 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_path, "UUID")
-        self.boot_partition_uuid = device_module.blkid_value(self.boot_partition_path, "UUID")
+        self.root_partition_uuid = device_module.blkid_value(
+            self.root_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}"
@@ -59,7 +63,9 @@ 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_path, "UUID")
+        self.root_partition_uuid = device_module.blkid_value(
+            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...")
@@ -82,30 +88,18 @@ class ImageSession:
             ["mount", "-v", self.boot_partition_path, str(self.boot_mount_path)],
             sudo=True,
         )
-        runner.run(["mount", "-v", self.root_mapper_path, str(self.root_mount_path)], sudo=True)
+        runner.run(
+            ["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), self.boot_bind_target()], sudo=True
-        )
+        runner.run(["mount", "--bind", str(self.boot_mount_path), f"{root}/boot"], 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)
@@ -156,7 +150,7 @@ class ImageSession:
             self._umount(f"{root}/dev", lazy=True)
             self._umount(f"{root}/proc")
             self._umount(f"{root}/sys")
-            self._umount(self.boot_bind_target())
+            self._umount(f"{root}/boot")
             self._umount(str(root))
         if self.boot_mount_path is not None:
             self._umount(str(self.boot_mount_path))
@@ -175,16 +169,11 @@ 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 (with a Debian-safe PATH)."""
+    """Run a bash script inside the image via chroot."""
     runner.run(
         ["chroot", str(root_mount_path), "/bin/bash"],
-        input_text=f"export PATH={_CHROOT_PATH}\n{script}",
+        input_text=script,
         sudo=True,
         error_msg=error_msg,
     )
@@ -194,13 +183,8 @@ 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 -Sy --needed {package_names}")
-    elif distribution in ("moode", "retropie", "raspios"):
-        # 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}",
-        )
+        chroot_bash(root_mount_path, f"pacman --noconfirm -S --needed {package_names}")
+    elif distribution in ("moode", "retropie"):
+        chroot_bash(root_mount_path, f"yes | apt install {package_names}")
     else:
         raise LimError("Package manager not supported.")

lim/image/setup.py

@@ -36,7 +36,8 @@ 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. Umount with "umount {device.path}*".'
+            f'Device {device.path} is allready mounted. '
+            f'Umount with "umount {device.path}*".'
         )
     return device
 
@@ -82,7 +83,9 @@ def run_setup() -> None:
         session.make_working_folder()
         session.make_mount_folders()
 
-        plan.root_filesystem = ui.ask("Which filesystem should be used? E.g.:btrfs,ext4... (none):")
+        plan.root_filesystem = ui.ask(
+            "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/tor.py

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

lim/image/transfer.py

@@ -1,23 +1,24 @@
 """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, *, force_prompt: bool = True) -> None:
-    if force_prompt and ui.confirm("Should the image download be forced?"):
+def download_image(plan: ImagePlan) -> None:
+    if 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(f"Forcing download wasn't neccessary. File {plan.image_path} doesn't exist.")
+            ui.info(
+                "Forcing download wasn't neccessary. "
+                f"File {plan.image_path} doesn't exist."
+            )
 
     ui.info("Start Download procedure...")
     if plan.image_path.is_file():
@@ -34,13 +35,13 @@ def download_image(plan: ImagePlan, *, force_prompt: bool = True) -> None:
 def arch_partition_input(boot_size: str) -> str:
     """Fdisk answers: FAT32 boot partition of boot_size plus root partition."""
     return (
-        "o\n"  # clear out any partitions on the drive
-        "p\n"  # list partitions (should be empty)
+        "o\n"          # clear out any partitions on the drive
+        "p\n"          # list partitions (should be empty)
         "n\np\n1\n\n"  # new primary partition 1, default start sector
         f"{boot_size}\n"
-        "t\nc\n"  # set partition 1 to type W95 FAT32 (LBA)
+        "t\nc\n"       # set partition 1 to type W95 FAT32 (LBA)
         "n\np\n2\n\n\n"  # new primary partition 2 over the remaining space
-        "w\n"  # write partition table
+        "w\n"          # write partition table
     )
 
 
@@ -60,18 +61,9 @@ def decompress_command(image_path: Path) -> list[str]:
 
 def _luks_format_root(plan: ImagePlan, session: ImageSession) -> None:
     luks_format = [
-        "cryptsetup",
-        "-v",
-        "luksFormat",
-        "-c",
-        "aes-xts-plain64",
-        "-s",
-        "512",
-        "-h",
-        "sha512",
-        "--use-random",
-        "-i",
-        "1000",
+        "cryptsetup", "-v", "luksFormat",
+        "-c", "aes-xts-plain64", "-s", "512", "-h", "sha512",
+        "--use-random", "-i", "1000",
     ]
     if plan.luks_memory_cost:
         ui.info(
@@ -118,107 +110,19 @@ def transfer_arch_image(plan: ImagePlan, session: ImageSession) -> None:
         runner.run(["mv", "-v", str(entry), str(session.boot_mount_path)], sudo=True)
 
 
-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)
+def transfer_image(plan: ImagePlan, session: ImageSession) -> None:
+    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:
-        runner.run(["cp", "-a", f"{source}/.", target], sudo=True)
+        ui.info("Skipping partition table deletion...")
 
-
-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)
+    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

@@ -1,100 +0,0 @@
-"""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,7 +24,8 @@ def resolve_checksum(download_url: str) -> str | None:
     for extension in ("sha1", "sha512", "md5"):
         checksum_url = f"{download_url}.{extension}"
         ui.info(
-            f"Image Checksum is not defined. Try to download image signature from {checksum_url}."
+            "Image Checksum is not defined. "
+            f"Try to download image signature from {checksum_url}."
         )
         if url_exists(checksum_url):
             content = runner.output(["wget", checksum_url, "-q", "-O", "-"])
@@ -61,7 +62,8 @@ 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, which ensures the file comes from a trusted source."
+        "Proceeding to signature verification, "
+        "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

@@ -1,206 +0,0 @@
-"""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,10 +40,14 @@ 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(["cryptsetup", "-v", "luksClose", mapper_name], sudo=True, check=False)
+    closed = runner.run(
+        ["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(["cryptsetup", "luksAddKey", partition_path, str(secret_key_path)], sudo=True)
+    runner.run(
+        ["cryptsetup", "luksAddKey", partition_path, str(secret_key_path)], sudo=True
+    )
     runner.run(
         [
             "cryptsetup",
@@ -66,7 +70,9 @@ def create_luks_key_and_update_crypttab(
     print(crypttab_path.read_text())
 
 
-def update_fstab(mapper_path: str, mount_path: str, *, fstab_path: Path = FSTAB_PATH) -> None:
+def update_fstab(
+    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,7 +50,8 @@ def run(
         return subprocess.CompletedProcess(cmd, COMMAND_NOT_FOUND)
     if check and result.returncode != 0:
         raise LimError(
-            error_msg or f"Command failed with code {result.returncode}: {shlex.join(cmd)}"
+            error_msg
+            or f"Command failed with code {result.returncode}: {shlex.join(cmd)}"
         )
     return result
 
@@ -97,7 +98,8 @@ def pipeline(
 ) -> None:
     """Run commands as a shell-style pipeline (cmd1 | cmd2 | ...)."""
     prepared = [
-        _with_sudo(cmd, sudo=sudo_last and index == len(cmds) - 1) for index, cmd in enumerate(cmds)
+        _with_sudo(cmd, sudo=sudo_last and index == len(cmds) - 1)
+        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,7 +25,9 @@ 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(["cryptsetup", "luksOpen", second.device.path, second.mapper_name], sudo=True)
+    runner.run(
+        ["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,7 +32,9 @@ def setup() -> None:
     runner.run(["cryptsetup", "-v", "-y", "luksFormat", target.partition_path], sudo=True)
 
     ui.info("Unlock partition...")
-    runner.run(["cryptsetup", "luksOpen", target.partition_path, target.mapper_name], sudo=True)
+    runner.run(
+        ["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)
@@ -55,7 +57,9 @@ def mount() -> None:
     target = select_storage_target()
 
     ui.info("Unlock partition...")
-    runner.run(["cryptsetup", "luksOpen", target.partition_path, target.mapper_name], sudo=True)
+    runner.run(
+        ["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

@@ -32,7 +32,6 @@ lim = [
   "distributions.yml",
   "configuration/packages/**/*.txt",
   "configuration/initcpio/*",
-  "configuration/initramfs-tools/*",
 ]
 
 [tool.pytest.ini_options]

tests/e2e/qemu/README.md

@@ -27,8 +27,7 @@ CPU architecture.
 
 | 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) |
+| Build encrypted image | `build_image.sh` | **root** (loop, cryptsetup, chroot) |
 | Tor network | `tor_net.py` | rootless |
 | Boot + unlock | `boot_unlock.py` | rootless (QEMU user-mode net) |
 | Orchestration | `harness.py` | mixed (uses `sudo` for the build only) |
@@ -54,41 +53,18 @@ and then `chown`s the artifacts back so rootless QEMU can read them.
 - 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):
+# Public Tor network (simplest; non-deterministic, needs internet):
 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
+# Private, offline, deterministic Tor network via chutney:
+git clone https://gitlab.torproject.org/tpo/core/chutney
+CHUTNEY_PATH=$PWD/chutney LIM_E2E_QEMU=1 \
+  pytest tests/e2e/test_qemu_unlock_e2e.py -v -s
 ```
 
-`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.
 

tests/e2e/qemu/boot_unlock.py

@@ -17,13 +17,8 @@ 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
+UNLOCK_TIMEOUT = 420
 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
@@ -68,49 +63,18 @@ def _qemu_stderr_tail(spec: QemuSpec, lines: int = 10) -> str:
     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.
+    its argon2 KDF, killing the unlock mid-flight. We wait for the remote to
+    close it once the boot proceeds and dropbear is torn down.
     """
-    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,
+        stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
         text=True,
     )
     try:
@@ -128,7 +92,6 @@ def _deliver_worker(spec: QemuSpec) -> None:
         if proc.stdin is not None:
             with contextlib.suppress(OSError):
                 proc.stdin.close()
-        log_file.close()
 
 
 def boot_and_unlock(spec: QemuSpec) -> None:
@@ -140,33 +103,20 @@ def boot_and_unlock(spec: QemuSpec) -> None:
     """
     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(f"Target onion: {spec.onion_address}:22  (unlock budget {UNLOCK_TIMEOUT}s)")
     print("---- guest serial console ----", flush=True)
     # QEMU's own diagnostics (bad -machine/-cpu, /dev/kvm denied, image busy)
     # go to stderr; capture them so an early exit is debuggable, not opaque.
     qemu_stderr = _qemu_stderr_path(spec).open("wb")
     qemu = subprocess.Popen(
         config.qemu_argv(spec),
-        stdout=subprocess.DEVNULL,
-        stderr=qemu_stderr,
+        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,
-        )
+        next_delivery = 0.0
         delivery: threading.Thread | None = None
         while time.monotonic() < deadline:
             offset = _stream_new_serial(spec, offset)
@@ -182,15 +132,17 @@ def boot_and_unlock(spec: QemuSpec) -> None:
             # 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)
+                print(f"\n[harness] delivering passphrase over Tor to "
+                      f"{spec.onion_address}:22 (holding session open) ...", flush=True)
+                delivery = threading.Thread(
+                    target=_deliver_worker, args=(spec,), daemon=True
+                )
                 delivery.start()
                 next_delivery = time.monotonic() + DELIVERY_INTERVAL
             time.sleep(_POLL_INTERVAL)
         raise RuntimeError(
             f"Boot-ok marker never appeared within {UNLOCK_TIMEOUT}s.\n"
-            f"{_serial_tail(spec, lines=45)}{_qemu_stderr_tail(spec)}"
-            f"{_ssh_log_tail(spec)}{_client_tor_tail(spec)}"
+            f"{_serial_tail(spec)}{_qemu_stderr_tail(spec)}"
         )
     finally:
         qemu.terminate()

tests/e2e/qemu/build_image_debian.sh

@@ -1,152 +0,0 @@
-#!/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

@@ -42,15 +42,6 @@ class QemuSpec:
     # (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:
@@ -72,23 +63,18 @@ def kernel_cmdline(spec: QemuSpec) -> str:
     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
+    # The netconf hook needs 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.
+    # name is mandatory (as lim's own boot config uses it); a bare "ip=dhcp"
+    # leaves netconf looping on "SIOCGIFFLAGS: No such device". net.ifnames=0
+    # keeps the virtio NIC named eth0.
     # The LAST console= owns /dev/console, which the boot-ok marker echoes to
     # and QEMU captures via -serial file, so the serial console must come last.
-    return f"{crypt} ip=::::lim-e2e:eth0:dhcp net.ifnames=0 console=tty0 console={console}"
+    return (
+        f"cryptdevice=UUID={spec.luks_uuid}:{spec.mapper_name} "
+        f"root=/dev/mapper/{spec.mapper_name} rw "
+        f"ip=::::lim-e2e:eth0:dhcp net.ifnames=0 console=tty0 console={console}"
+    )
 
 
 def qemu_argv(spec: QemuSpec) -> list[str]:
@@ -108,28 +94,16 @@ def qemu_argv(spec: QemuSpec) -> list[str]:
         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",
+        "-m", str(spec.memory_mb),
+        "-kernel", str(spec.kernel_path),
+        "-initrd", str(spec.initramfs_path),
+        "-append", kernel_cmdline(spec),
+        "-drive", f"file={spec.image_path},if=virtio,format=raw",
+        "-netdev", "user,id=net0",
+        "-device", f"{net_device},netdev=net0",
         "-nographic",
-        "-serial",
-        f"file:{spec.serial_log}",
+        "-serial", f"file:{spec.serial_log}",
         "-no-reboot",
     ]
     return argv
@@ -141,34 +115,18 @@ def ssh_proxy_command(socks_port: int) -> str:
 
 
 def ssh_argv(spec: QemuSpec) -> list[str]:
-    """Non-interactive SSH into the initramfs dropbear to feed the passphrase.
+    """Non-interactive SSH into the initramfs dropbear via the onion address.
 
-    ``-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.
+    ``-tt`` forces a pty so the cryptsetup askpass inside the encryptssh
+    session reads the passphrase we pipe on stdin. Host-key checking is off
+    because a throwaway onion has no known_hosts entry.
     """
-    argv = [
-        "ssh",
-        "-tt",
-        "-o",
-        "StrictHostKeyChecking=no",
-        "-o",
-        "UserKnownHostsFile=/dev/null",
-        "-o",
-        "BatchMode=yes",
-        "-i",
-        str(spec.ssh_key_path),
+    return [
+        "ssh", "-tt",
+        "-o", "StrictHostKeyChecking=no",
+        "-o", "UserKnownHostsFile=/dev/null",
+        "-o", "BatchMode=yes",
+        "-o", f"ProxyCommand={ssh_proxy_command(spec.socks_port)}",
+        "-i", str(spec.ssh_key_path),
+        f"root@{spec.onion_address}",
     ]
-    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

@@ -7,7 +7,6 @@ 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
@@ -32,34 +31,6 @@ class HarnessConfig:
     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:
@@ -114,7 +85,7 @@ def _acquire_sudo() -> _SudoKeepalive | None:
 
 
 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]
+    script = cfg.repo_root / "tests/e2e/qemu/build_image.sh"
     env = {
         **os.environ,
         "WORK_DIR": str(cfg.work_dir),
@@ -148,8 +119,6 @@ def _reclaim_work_dir(cfg: HarnessConfig, *, required: bool = False) -> None:
 
 
 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)
@@ -157,7 +126,6 @@ def _make_tor_net(cfg: HarnessConfig):
 
 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,
@@ -171,9 +139,6 @@ def _load_spec(cfg: HarnessConfig, ssh_key: Path, socks_port: int) -> QemuSpec:
         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,
     )
 
 
@@ -190,14 +155,11 @@ def run(cfg: HarnessConfig) -> QemuSpec:
     sudo = _acquire_sudo()
     net = _make_tor_net(cfg)
     try:
-        if cfg.unlock_transport != "direct":
-            print("\n[harness] bootstrapping Tor client (~30-60s)...", flush=True)
+        print("\n[harness] bootstrapping Tor client (~30-60s)...", flush=True)
         net.start()
-        print(
-            f"[harness] transport={cfg.unlock_transport}, building encrypted "
-            "image (several minutes)...",
-            flush=True,
-        )
+        print(f"[harness] Tor client ready on socks port {net.socks_port}.", flush=True)
+        print("[harness] building encrypted image "
+              "(pacstrap + mkinitcpio, several minutes)...", flush=True)
         _build_image(cfg, ssh_key, net.test_network_conf)
         spec = _load_spec(cfg, ssh_key, net.socks_port)
         boot_unlock.boot_and_unlock(spec)

tests/e2e/qemu/tor_net.py

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

tests/e2e/test_qemu_harness_unit.py

@@ -50,15 +50,6 @@ class TestKernelCmdline:
     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):
@@ -112,31 +103,6 @@ class TestSshInvocation:
         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."""
@@ -145,7 +111,9 @@ class TestBuildScriptStaysAligned:
         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="))
+        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
 
@@ -161,7 +129,9 @@ class TestBuildScriptStaysAligned:
         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="))
+        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()
@@ -170,35 +140,6 @@ class TestBuildScriptStaysAligned:
         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(

tests/e2e/test_qemu_unlock_e2e.py

@@ -35,20 +35,10 @@ 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 = [
+_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"]
+    "cryptsetup", "pacstrap", "tor", "ssh", "ncat",
+)
 
 
 def _missing_tools() -> list[str]:
@@ -58,24 +48,25 @@ def _missing_tools() -> list[str]:
 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)."
+        "needs LIM_E2E_QEMU=1 and "
+        + ", ".join(_REQUIRED)
+        + " (build stage needs root; slow)."
     ),
 )
 
 
-def test_full_build_boot_and_unlock(tmp_path):
+def test_full_build_boot_and_tor_unlock(tmp_path):
     chutney_path = os.environ.get("CHUTNEY_PATH")
     cfg = HarnessConfig(
         repo_root=REPO_ROOT,
         work_dir=tmp_path / "qemu-e2e",
         arch=_ARCH,
-        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.
+    # run() only returns after the marker was observed; re-assert for clarity.
+    assert spec.onion_address.endswith(".onion")
     serial = spec.serial_log.read_text(errors="replace")
     assert qemu_config.BOOT_OK_MARKER in serial

tests/e2e/test_tor_unlock_e2e.py

@@ -3,7 +3,7 @@
 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),
+  1. generate the v3 onion keys offline (as lim.image.tor does in the chroot),
   2. start a real Tor onion service from a torrc mirroring the baked-in one,
      forwarding the virtual port 22 to a local dropbear stand-in,
   3. connect to the .onion address through Tor and deliver the passphrase,
@@ -24,7 +24,7 @@ import time
 
 import pytest
 
-from lim.image.initramfs import keygen
+from lim.image import tor as tor_module
 from tests.e2e import tor_harness
 
 # The offline checks only need the tor binary (no network) and are
@@ -32,7 +32,9 @@ from tests.e2e import tor_harness
 # 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 = 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)",
@@ -50,7 +52,7 @@ def workdir(tmp_path):
 
 def test_offline_keygen_matches_production_flags():
     """Guard: the harness keygen mirrors the flags production actually runs."""
-    script = keygen._KEYGEN_SCRIPT
+    script = tor_module._KEYGEN_SCRIPT
     assert "--DisableNetwork 1" in script
     assert "HiddenServicePort" in script
     assert "22 127.0.0.1:22" in script
@@ -60,7 +62,9 @@ def test_offline_keygen_matches_production_flags():
 @_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")
+    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"):

tests/e2e/tor_harness.py

@@ -13,7 +13,7 @@ import threading
 import time
 from pathlib import Path
 
-# Matches lim.image.initramfs.keygen._KEYGEN_SCRIPT: an empty -f torrc keeps the host's
+# Matches lim.image.tor._KEYGEN_SCRIPT: an empty -f torrc keeps the host's
 # /etc/tor/torrc out of the run, --DisableNetwork 1 makes the keys appear
 # without ever touching the network.
 KEYGEN_TIMEOUT = 30
@@ -30,7 +30,7 @@ def free_port() -> int:
 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
+    Mirrors lim.image.tor._KEYGEN_SCRIPT, run directly on the host instead
     of inside the image chroot (the tor invocation is identical).
     """
     onion_dir.mkdir(parents=True, exist_ok=True)
@@ -39,23 +39,11 @@ def generate_onion_keys(onion_dir: Path, data_dir: Path) -> str:
     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",
+            "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,
     )
@@ -74,7 +62,9 @@ def generate_onion_keys(onion_dir: Path, data_dir: Path) -> str:
     return hostname.read_text().strip()
 
 
-def start_onion_service(torrc_path: Path, socks_port: int, log_path: Path) -> subprocess.Popen:
+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
@@ -83,13 +73,9 @@ def start_onion_service(torrc_path: Path, socks_port: int, log_path: Path) -> su
     """
     return subprocess.Popen(
         [
-            "tor",
-            "-f",
-            str(torrc_path),
-            "--SocksPort",
-            str(socks_port),
-            "--Log",
-            f"notice file {log_path}",
+            "tor", "-f", str(torrc_path),
+            "--SocksPort", str(socks_port),
+            "--Log", f"notice file {log_path}",
         ],
         stdout=subprocess.DEVNULL,
         stderr=subprocess.DEVNULL,
@@ -112,7 +98,11 @@ def _socks5_handshake(sock: socket.socket, onion_host: str, onion_port: int) ->
     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)
+    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)

tests/unit/test_cli.py

@@ -24,7 +24,9 @@ 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(cli.COMMANDS, "backup", Command(lambda: None, "d", needs_root=True))
+    monkeypatch.setitem(
+        cli.COMMANDS, "backup", Command(lambda: None, "d", needs_root=True)
+    )
     cli.main(["--type", "backup", "--auto-confirm"])
     assert escalated == [True]
 
@@ -57,10 +59,6 @@ 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

@@ -1,50 +0,0 @@
-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,25 +33,3 @@ 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,17 +60,3 @@ 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,17 +63,6 @@ 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):
@@ -112,8 +101,7 @@ class TestInstallPackages:
             for kind, cmd, input_text in fake_runner.calls
             if kind == "run" and cmd[0] == "chroot"
         ]
-        assert len(chroot_calls) == 1
-        assert "pacman --noconfirm -Sy --needed btrfs-progs" in chroot_calls[0]
+        assert chroot_calls == ["pacman --noconfirm -S --needed btrfs-progs"]
 
     def test_apt_for_retropie(self, tmp_path, fake_runner):
         install_packages("retropie", tmp_path, "btrfs-progs")
@@ -122,9 +110,7 @@ class TestInstallPackages:
             for kind, cmd, input_text in fake_runner.calls
             if kind == "run" and cmd[0] == "chroot"
         ]
-        assert len(chroot_calls) == 1
-        assert "apt-get update" in chroot_calls[0]
-        assert "apt-get install -y btrfs-progs" in chroot_calls[0]
+        assert chroot_calls == ["yes | apt install btrfs-progs"]
 
     def test_unsupported_distribution_raises(self, tmp_path, fake_runner):
         with pytest.raises(LimError):

tests/unit/test_storage.py

@@ -20,7 +20,9 @@ def test_storage_target_derives_all_paths():
     assert target.partition_path == "/dev/sdb1"
 
 
-def test_single_drive_setup_command_sequence(fake_runner, answers, block_devices, monkeypatch):
+def test_single_drive_setup_command_sequence(
+    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,19 +58,10 @@ 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"
     ]
 
 
@@ -104,12 +95,18 @@ def test_export_chowns_real_home_without_sudo(tmp_path, fake_runner, monkeypatch
 
     sync.export_to_system()
 
-    chown_calls = [(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)]
+    chown_calls = [
+        (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 (real_home / ".ssh/id_rsa").stat().st_mode & 0o777 == 0o600
 
 
-def test_execute_sync_plan_creates_folders_and_runs_rsync(tmp_path, fake_runner):
+def test_execute_sync_plan_creates_folders_and_runs_rsync(
+    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

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

@@ -1,42 +0,0 @@
-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

@@ -1,235 +0,0 @@
-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