Release version 2.3.0

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

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

.github/workflows/test.yml

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

CHANGELOG.md

@@ -1,5 +1,54 @@
 # Changelog
 
+## [2.3.0] - 2026-07-24
+
+**✨ New**
+
+- Guided setup wizard, now the default command: *lim* with no *--type* (or
+  *lim --type guided*) asks everything up front — distribution, target device,
+  hostname, login user + SSH key, password — then builds an encrypted,
+  Tor-remote-unlockable image unattended.
+- Debian / Raspberry Pi OS remote LUKS unlock. The initramfs unlock stack is
+  now distro-agnostic behind an *InitramfsBackend* abstraction: *mkinitcpio*
+  for Arch/Manjaro, *initramfs-tools* (cryptsetup-initramfs + dropbear-initramfs
+  + a Tor onion) for Raspberry Pi OS / moode / RetroPie.
+- Flash full disk images into LUKS. Distributions that ship a *.img* rather
+  than a rootfs tarball (Raspberry Pi OS, moode, RetroPie, Manjaro ARM) are
+  loop-mounted and their boot + root partitions copied into a fresh LUKS
+  container. New Raspberry Pi OS catalog (*lite64* / *desktop64* / *lite32*).
+- New *remote-unlock* command: reach a waiting initramfs over Tor (onion, via
+  *torsocks*) or plain SSH (host/IP), running *cryptroot-unlock* or presenting
+  the passphrase prompt. The wizard saves a target record under
+  *~/.config/lim/unlocks* so *lim --type remote-unlock* offers it directly.
+- Cross-architecture builds. Building a foreign-arch image (e.g. arm64
+  Raspberry Pi OS on an x86 host) auto-installs *qemu-user-static* and registers
+  binfmt, or aborts before the target is erased with per-distro instructions.
+- RTC-less clock hardening. A clock floor (build epoch baked into the initramfs)
+  plus a numeric *tor_ntp=* cmdline let the clock sync without DNS, so the Tor
+  onion still publishes on boards without a real-time clock (which boot at 1970).
+- Numbered selection menus for distribution and version prompts; typing the
+  name still works.
+- CI: a full Debian *build → boot → LUKS-unlock* QEMU end-to-end job runs on
+  push (*make test-qemu-debian*).
+
+**🐛 Fixes**
+
+- Raspberry Pi OS Bookworm boots the encrypted image: the initramfs is placed on
+  the *​/boot/firmware* FAT partition instead of the ext4 root, so cryptsetup is
+  present at boot (no more *(initramfs)* emergency shell).
+- Chroot package installs are robust cross-host: a Debian-safe *PATH* (so
+  *update-initramfs* / *useradd* resolve) and *apt-get update* before install
+  (stock images ship stale lists whose superseded *.deb* URLs 404).
+- Stock-image *fstab* is normalized (the existing *​/* line is replaced, not
+  duplicated) when copying a distro image into LUKS.
+- Large copies show progress (*rsync --info=progress2*, *cp* fallback).
+
+**💥 Breaking / behavior**
+
+- *lim* with no *--type* now runs the guided wizard (previously *--type* was
+  required).
+- The long-deprecated *--extra* argument is removed.
+
 ## [2.2.0] - 2026-07-21
 
 - Optional remote LUKS unlock via a Tor onion service in the initramfs.

Makefile

@@ -1,4 +1,4 @@
-.PHONY: install test test-tor test-qemu test-all
+.PHONY: install test test-tor test-qemu test-qemu-debian test-all
 
 PREFIX ?= $(HOME)/.local
 # Interpreter used for the tests. Override if your active venv lacks pytest,
@@ -20,11 +20,18 @@ test-tor:
 	LIM_E2E_TOR=1 $(PYTHON) -m pytest tests/e2e/test_tor_unlock_e2e.py -v
 
 # Full virtualized build+boot+unlock e2e. Needs qemu, arch-install-scripts,
-# cryptsetup, tor and ncat; the build stage runs under sudo. Set CHUTNEY_PATH
-# to use a private, offline Tor network instead of the public one.
+# cryptsetup and ssh; the build stage runs under sudo. The default "direct"
+# transport delivers the passphrase deterministically via a QEMU port-forward;
+# set LIM_E2E_TRANSPORT=tor for the full onion path (also needs tor + ncat, and
+# is flaky over public Tor inside QEMU).
 test-qemu:
 	LIM_E2E_QEMU=1 $(PYTHON) -m pytest tests/e2e/test_qemu_unlock_e2e.py -v -s
 
+# Same, but builds a Debian image with debootstrap (initramfs-tools backend)
+# instead of Arch. Needs debootstrap + cryptsetup + ssh; runs under sudo.
+test-qemu-debian:
+	LIM_E2E_QEMU=1 LIM_E2E_OS=debian $(PYTHON) -m pytest tests/e2e/test_qemu_unlock_e2e.py -v -s
+
 # Everything: mocked suite plus both opt-in e2e stages, in one run.
 test-all:
 	LIM_E2E_TOR=1 LIM_E2E_QEMU=1 $(PYTHON) -m pytest -v -s

lim/catalog.py

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

lim/cli.py

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

lim/configuration/initramfs-tools/tor_hook

@@ -25,6 +25,8 @@ 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

lim/configuration/initramfs-tools/tor_premount

@@ -34,8 +34,12 @@ if [ ! -s /etc/resolv.conf ]; then
     done
 fi
 
-# RTC-less boards boot at 1970; Tor rejects the consensus otherwise. Bounded so
-# a failed sync never blocks boot.
+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}" \

lim/data/sync.py

@@ -90,31 +90,22 @@ def execute_sync_plan(operations: list[SyncOperation]) -> None:
         if not operation.is_directory and Path(operation.destination).is_file():
             ui.info("The destination file already exists!")
             ui.info("Difference:")
-            runner.run(
-                ["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,9 +119,7 @@ def overwrite_device(device: Device) -> None:
 
 def blkid_value(path: str, tag: str) -> str:
     """Return a blkid tag value ("TYPE", "UUID", ...) or "" when unavailable."""
-    return runner.output(
-        ["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,6 +57,19 @@ retropie_images:
     checksum: b5daa6e7660a99c246966f3f09b4014b
     image: retropie-buster-4.8-rpi4_400.img.gz
 
+# Raspberry Pi OS. The "_latest" endpoints 302-redirect to the current release;
+# image is only the local filename (must end .img.xz for decompress_command).
+raspios_images:
+  lite64:
+    source_url: https://downloads.raspberrypi.com/raspios_lite_arm64_latest
+    image: raspios_lite_arm64_latest.img.xz
+  desktop64:
+    source_url: https://downloads.raspberrypi.com/raspios_arm64_latest
+    image: raspios_arm64_latest.img.xz
+  lite32:
+    source_url: https://downloads.raspberrypi.com/raspios_lite_armhf_latest
+    image: raspios_lite_armhf_latest.img.xz
+
 # Extra kernel modules the initramfs needs for early network access,
 # see https://raspberrypi.stackexchange.com/questions/67051
 mkinitcpio_modules_by_rpi:

lim/fsutil.py

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

lim/image/choosers.py

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

lim/image/crossarch.py

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

lim/image/encryption.py

@@ -13,7 +13,10 @@ from lim.image.plan import ImagePlan
 from lim.image.session import ImageSession, install_packages
 
 
-def configure_encryption(plan: ImagePlan, session: ImageSession, authorized_keys: Path) -> None:
+def configure_encryption(
+    plan: ImagePlan, session: ImageSession, authorized_keys: Path
+) -> str | None:
+    """Configure the remote-unlock LUKS stack; return the onion address, if any."""
     root = session.root_mount_path
     backend = get_backend(plan.distribution)
     ui.info("Setup encryption...")
@@ -26,12 +29,14 @@ def configure_encryption(plan: ImagePlan, session: ImageSession, authorized_keys
 
     backend.install_authorized_key(root, authorized_keys)
 
+    onion_address = None
     if plan.tor_unlock:
         # Hook files and onion keys must exist before the initramfs is baked.
-        backend.install_tor_unlock(plan, root)
+        onion_address = backend.install_tor_unlock(plan, root)
 
     # crypttab must be written before the initramfs is (re)generated so the
     # Debian cryptsetup hook can bake the mapping in.
     backend.register_encrypted_root(plan, session, root)
     backend.configure_initramfs(plan, root)
     backend.configure_bootloader(plan, session, root)
+    return onion_address

lim/image/initramfs/initramfs_tools.py

@@ -20,6 +20,9 @@ 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:
@@ -68,6 +71,8 @@ class InitramfsToolsBackend(InitramfsBackend):
             "  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"
@@ -89,12 +94,7 @@ class InitramfsToolsBackend(InitramfsBackend):
             "update-initramfs -c -k all 2>/dev/null || update-initramfs -u -k all",
         )
 
-    def configure_bootloader(
-        self,
-        plan: ImagePlan,  # noqa: ARG002
-        session: ImageSession,
-        root: Path,
-    ) -> None:
+    def configure_bootloader(self, plan: ImagePlan, session: ImageSession, root: Path) -> None:
         hostname = (root / "etc/hostname").read_text().strip()
         boot = session.boot_mount_path
 
@@ -111,5 +111,7 @@ class InitramfsToolsBackend(InitramfsBackend):
             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/loopimg.py

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

lim/image/plan.py

@@ -12,6 +12,10 @@ 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
@@ -23,6 +27,8 @@ 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/session.py

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

lim/image/setup.py

@@ -36,8 +36,7 @@ def _select_unmounted_device() -> device_module.Device:
     device = device_module.select_device()
     if device_module.is_mounted(device.path):
         raise LimError(
-            f'Device {device.path} is allready mounted. '
-            f'Umount with "umount {device.path}*".'
+            f'Device {device.path} is allready mounted. Umount with "umount {device.path}*".'
         )
     return device
 
@@ -83,9 +82,7 @@ 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/transfer.py

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

lim/image/unlock.py

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

lim/image/verify.py

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

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

lim/luks.py

@@ -40,14 +40,10 @@ def create_luks_key_and_update_crypttab(
     runner.sync_disks()
 
     ui.info("Opening and closing device to verify that everything works fine...")
-    closed = runner.run(
-        ["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",
@@ -70,9 +66,7 @@ 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,8 +50,7 @@ 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
 
@@ -98,8 +97,7 @@ def pipeline(
 ) -> None:
     """Run commands as a shell-style pipeline (cmd1 | cmd2 | ...)."""
     prepared = [
-        _with_sudo(cmd, sudo=sudo_last and index == len(cmds) - 1)
-        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,9 +25,7 @@ def setup() -> None:
     runner.run(["cryptsetup", "luksFormat", second.device.path], sudo=True)
 
     runner.run(["cryptsetup", "luksOpen", first.device.path, first.mapper_name], sudo=True)
-    runner.run(
-        ["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,9 +32,7 @@ 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)
@@ -57,9 +55,7 @@ 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

@@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "linux-image-manager"
-version = "2.2.0"
+version = "2.3.0"
 description = "CLI (`lim`) for downloading, verifying and flashing Linux images, LUKS/Btrfs encrypted storage (single drive and RAID1), image backups and chroot maintenance"
 readme = "README.md"
 requires-python = ">=3.10"

tests/e2e/qemu/README.md

@@ -27,7 +27,8 @@ CPU architecture.
 
 | Stage | File | Privilege |
 |---|---|---|
-| Build encrypted image | `build_image.sh` | **root** (loop, cryptsetup, chroot) |
+| Build encrypted image (Arch) | `build_image.sh` | **root** (loop, cryptsetup, chroot) |
+| Build encrypted image (Debian) | `build_image_debian.sh` | **root** (debootstrap, loop, cryptsetup, chroot) |
 | Tor network | `tor_net.py` | rootless |
 | Boot + unlock | `boot_unlock.py` | rootless (QEMU user-mode net) |
 | Orchestration | `harness.py` | mixed (uses `sudo` for the build only) |
@@ -53,18 +54,41 @@ 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
-# Public Tor network (simplest; non-deterministic, needs internet):
+# Arch (mkinitcpio), direct transport (deterministic, default):
 LIM_E2E_QEMU=1 pytest tests/e2e/test_qemu_unlock_e2e.py -v -s
 
-# Private, offline, deterministic Tor network via chutney:
-git clone https://gitlab.torproject.org/tpo/core/chutney
-CHUTNEY_PATH=$PWD/chutney LIM_E2E_QEMU=1 \
-  pytest tests/e2e/test_qemu_unlock_e2e.py -v -s
+# Debian (initramfs-tools) via debootstrap, direct transport:
+LIM_E2E_QEMU=1 LIM_E2E_OS=debian pytest tests/e2e/test_qemu_unlock_e2e.py -v -s
+
+# Full onion transport (opt-in, flaky over public Tor):
+LIM_E2E_QEMU=1 LIM_E2E_TRANSPORT=tor pytest tests/e2e/test_qemu_unlock_e2e.py -v -s
 ```
 
+`make test-qemu` / `make test-qemu-debian` wrap the direct-transport runs.
+
 Environment knobs: `LIM_E2E_ARCH` (`x86_64` default, or `aarch64`),
 `CHUTNEY_PATH` (enables the private network), `CHUTNEY_PATH` unset → public.
 

tests/e2e/qemu/boot_unlock.py

@@ -17,8 +17,13 @@ from tests.e2e.qemu.config import QemuSpec
 
 # One generous budget covers guest boot + tor bootstrap + onion publish +
 # unlock. TCG (no KVM) is slow, so keep it roomy.
-UNLOCK_TIMEOUT = 420
+UNLOCK_TIMEOUT = 480
 DELIVERY_INTERVAL = 8
+# Hold off the first delivery until the guest onion has had time to publish.
+# A connect to a not-yet-published onion makes the client Tor cache the failed
+# descriptor fetch and back off, so early hammering can lock us out for
+# minutes; waiting until the onion is up means the first fetch succeeds.
+INITIAL_DELIVERY_DELAY = 100
 _SSH_ATTEMPT_TIMEOUT = 45
 _POLL_INTERVAL = 2
 _PROMPT_WAIT = 4  # let the encryptssh prompt appear before sending
@@ -63,18 +68,49 @@ 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. We wait for the remote to
-    close it once the boot proceeds and dropbear is torn down.
+    its argon2 KDF, killing the unlock mid-flight. Session output is appended to
+    ssh-delivery.log so a failed unlock (onion unreachable, cryptroot-unlock
+    error) is diagnosable.
     """
+    log_file = _ssh_log_path(spec).open("a")
+    log_file.write("--- delivery attempt ---\n")
+    log_file.flush()
     proc = subprocess.Popen(
         config.ssh_argv(spec),
-        stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+        stdin=subprocess.PIPE,
+        stdout=log_file,
+        stderr=log_file,
         text=True,
     )
     try:
@@ -92,6 +128,7 @@ 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:
@@ -103,20 +140,33 @@ 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"Target onion: {spec.onion_address}:22  (unlock budget {UNLOCK_TIMEOUT}s)")
+    print(f"Unlock target: {target}  (budget {UNLOCK_TIMEOUT}s)")
     print("---- guest serial console ----", flush=True)
     # 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
-        next_delivery = 0.0
+        # Hold off the first delivery until the unlock endpoint is up (see above).
+        next_delivery = time.monotonic() + initial_delay
+        print(
+            f"[harness] waiting {initial_delay}s for the unlock endpoint before delivering...",
+            flush=True,
+        )
         delivery: threading.Thread | None = None
         while time.monotonic() < deadline:
             offset = _stream_new_serial(spec, offset)
@@ -132,17 +182,15 @@ def boot_and_unlock(spec: QemuSpec) -> None:
             # background while we keep streaming serial and watching the marker.
             idle = delivery is None or not delivery.is_alive()
             if idle and time.monotonic() >= next_delivery:
-                print(f"\n[harness] delivering passphrase over Tor to "
-                      f"{spec.onion_address}:22 (holding session open) ...", flush=True)
-                delivery = threading.Thread(
-                    target=_deliver_worker, args=(spec,), daemon=True
-                )
+                print("\n[harness] delivering passphrase (holding session open) ...", flush=True)
+                delivery = threading.Thread(target=_deliver_worker, args=(spec,), daemon=True)
                 delivery.start()
                 next_delivery = time.monotonic() + DELIVERY_INTERVAL
             time.sleep(_POLL_INTERVAL)
         raise RuntimeError(
             f"Boot-ok marker never appeared within {UNLOCK_TIMEOUT}s.\n"
-            f"{_serial_tail(spec)}{_qemu_stderr_tail(spec)}"
+            f"{_serial_tail(spec, lines=45)}{_qemu_stderr_tail(spec)}"
+            f"{_ssh_log_tail(spec)}{_client_tor_tail(spec)}"
         )
     finally:
         qemu.terminate()

tests/e2e/qemu/build_image_debian.sh

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

tests/e2e/qemu/config.py

@@ -42,6 +42,15 @@ 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:
@@ -63,18 +72,23 @@ def kernel_cmdline(spec: QemuSpec) -> str:
     QEMU cannot emulate — hence virtio here (we model the stack, not the board).
     """
     console = _SERIAL_CONSOLE[spec.arch]
-    # The netconf hook needs the explicit ip= form
+    # Arch's mkinitcpio encrypt/encryptssh finds the LUKS device from the
+    # cryptdevice= param; Debian's cryptsetup-initramfs reads /etc/crypttab
+    # baked into the initramfs, so it needs only root=/dev/mapper/<name>.
+    if spec.os_family == "debian":
+        crypt = f"root=/dev/mapper/{spec.mapper_name} rw"
+    else:
+        crypt = (
+            f"cryptdevice=UUID={spec.luks_uuid}:{spec.mapper_name} "
+            f"root=/dev/mapper/{spec.mapper_name} rw"
+        )
+    # The netconf/dropbear-initramfs hooks need the explicit ip= form
     # <client>:<server>:<gw>:<netmask>:<host>:<device>:<proto> — the device
-    # name is mandatory (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.
+    # name is mandatory; a bare "ip=dhcp" leaves netconf looping on
+    # "SIOCGIFFLAGS: No such device". net.ifnames=0 keeps the NIC named eth0.
     # The LAST console= owns /dev/console, which the boot-ok marker echoes to
     # and QEMU captures via -serial file, so the serial console must come last.
-    return (
-        f"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}"
-    )
+    return f"{crypt} ip=::::lim-e2e:eth0:dhcp net.ifnames=0 console=tty0 console={console}"
 
 
 def qemu_argv(spec: QemuSpec) -> list[str]:
@@ -94,16 +108,28 @@ 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", "user,id=net0",
-        "-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",
+        netdev,
+        "-device",
+        f"{net_device},netdev=net0",
         "-nographic",
-        "-serial", f"file:{spec.serial_log}",
+        "-serial",
+        f"file:{spec.serial_log}",
         "-no-reboot",
     ]
     return argv
@@ -115,18 +141,34 @@ def ssh_proxy_command(socks_port: int) -> str:
 
 
 def ssh_argv(spec: QemuSpec) -> list[str]:
-    """Non-interactive SSH into the initramfs dropbear via the onion address.
+    """Non-interactive SSH into the initramfs dropbear to feed the passphrase.
 
-    ``-tt`` forces a pty so the cryptsetup askpass inside the encryptssh
-    session reads the passphrase we pipe on stdin. Host-key checking is off
-    because a throwaway onion has no known_hosts entry.
+    ``-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.
     """
-    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}",
+    argv = [
+        "ssh",
+        "-tt",
+        "-o",
+        "StrictHostKeyChecking=no",
+        "-o",
+        "UserKnownHostsFile=/dev/null",
+        "-o",
+        "BatchMode=yes",
+        "-i",
+        str(spec.ssh_key_path),
     ]
+    if spec.direct_ssh_port:
+        argv += ["-p", str(spec.direct_ssh_port), "root@127.0.0.1"]
+    else:
+        argv += [
+            "-o",
+            f"ProxyCommand={ssh_proxy_command(spec.socks_port)}",
+            f"root@{spec.onion_address}",
+        ]
+    if spec.unlock_command:
+        argv.append(spec.unlock_command)
+    return argv

tests/e2e/qemu/harness.py

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

tests/e2e/qemu/tor_net.py

@@ -17,7 +17,7 @@ import subprocess
 import time
 from pathlib import Path
 
-BOOTSTRAP_TIMEOUT = 180
+BOOTSTRAP_TIMEOUT = 300  # public Tor bootstrap can stall on slow guards
 _GUEST_HOST_ALIAS = "10.0.2.2"  # QEMU user-net alias for the host's 127.0.0.1
 
 
@@ -41,12 +41,21 @@ 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}",
+                "tor",
+                "-f",
+                str(empty_torrc),
+                "--SocksPort",
+                str(self.socks_port),
+                "--DataDirectory",
+                str(self._data),
+                "--Log",
+                f"notice file {self._log}",
+                # HS-level log so onion descriptor fetch / rendezvous is visible.
+                "--Log",
+                f"[rend]info file {self._work_dir / 'tor-client-hs.log'}",
             ],
-            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
         )
         _wait_for_bootstrap(self._log, self._process)
 
@@ -75,7 +84,8 @@ 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,6 +50,15 @@ 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):
@@ -103,6 +112,31 @@ 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."""
@@ -111,9 +145,7 @@ 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
 
@@ -129,9 +161,7 @@ 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()
@@ -140,6 +170,35 @@ 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,10 +35,20 @@ REPO_ROOT = Path(__file__).resolve().parents[2]
 # The QEMU binary is arch-specific, so derive it from LIM_E2E_ARCH instead of
 # hardcoding x86_64 — otherwise the documented aarch64 run is never reachable.
 _ARCH = os.environ.get("LIM_E2E_ARCH", "x86_64")
-_REQUIRED = (
+# The OS family selects the build tool: pacstrap (Arch) or debootstrap (Debian).
+_OS = os.environ.get("LIM_E2E_OS", "arch")
+# "direct" (deterministic port-forward) is the default; "tor" runs the full,
+# opt-in onion transport (flaky over public Tor inside QEMU).
+_TRANSPORT = os.environ.get("LIM_E2E_TRANSPORT", "direct")
+_BUILD_TOOL = {"arch": "pacstrap", "debian": "debootstrap"}
+_REQUIRED = [
     qemu_config.qemu_binary(_ARCH),
-    "cryptsetup", "pacstrap", "tor", "ssh", "ncat",
-)
+    _BUILD_TOOL.get(_OS, "pacstrap"),
+    "cryptsetup",
+    "ssh",
+]
+if _TRANSPORT == "tor":  # the onion transport also needs a Tor client + ncat
+    _REQUIRED += ["tor", "ncat"]
 
 
 def _missing_tools() -> list[str]:
@@ -48,25 +58,24 @@ 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_tor_unlock(tmp_path):
+def test_full_build_boot_and_unlock(tmp_path):
     chutney_path = os.environ.get("CHUTNEY_PATH")
     cfg = HarnessConfig(
         repo_root=REPO_ROOT,
         work_dir=tmp_path / "qemu-e2e",
         arch=_ARCH,
+        os_family=_OS,
+        unlock_transport=_TRANSPORT,
         chutney_path=Path(chutney_path) if chutney_path else None,
     )
 
     spec = run(cfg)
 
-    # run() only returns after the marker was observed; re-assert for clarity.
-    assert spec.onion_address.endswith(".onion")
+    # run() only returns after the boot-ok marker was observed on the console.
     serial = spec.serial_log.read_text(errors="replace")
     assert qemu_config.BOOT_OK_MARKER in serial

tests/e2e/tor_harness.py

@@ -13,7 +13,7 @@ import threading
 import time
 from pathlib import Path
 
-# Matches lim.image.tor._KEYGEN_SCRIPT: an empty -f torrc keeps the host's
+# Matches lim.image.initramfs.keygen._KEYGEN_SCRIPT: an empty -f torrc keeps the host's
 # /etc/tor/torrc out of the run, --DisableNetwork 1 makes the keys appear
 # without ever touching the network.
 KEYGEN_TIMEOUT = 30
@@ -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.tor._KEYGEN_SCRIPT, run directly on the host instead
+    Mirrors lim.image.initramfs.keygen._KEYGEN_SCRIPT, run directly on the host instead
     of inside the image chroot (the tor invocation is identical).
     """
     onion_dir.mkdir(parents=True, exist_ok=True)
@@ -39,11 +39,23 @@ 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,
     )
@@ -62,9 +74,7 @@ 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
@@ -73,9 +83,13 @@ def start_onion_service(
     """
     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,
@@ -98,11 +112,7 @@ 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,9 +24,7 @@ def test_every_command_dispatches_to_its_function(monkeypatch):
 def test_needs_root_command_requests_root(monkeypatch):
     escalated = []
     monkeypatch.setattr(system, "ensure_root", lambda: escalated.append(True))
-    monkeypatch.setitem(
-        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]
 
@@ -59,6 +57,10 @@ def test_unknown_type_is_rejected_by_argparse():
     assert excinfo.value.code == 2
 
 
+def test_type_defaults_to_guided():
+    assert cli.build_parser().parse_args([]).type == "guided"
+
+
 def test_all_registered_commands_have_descriptions():
     for name, command in cli.COMMANDS.items():
         assert command.description, name

tests/unit/test_cross_arch.py

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

tests/unit/test_fsutil.py

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

tests/unit/test_image_session.py

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

tests/unit/test_image_setup.py

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

tests/unit/test_storage.py

@@ -20,9 +20,7 @@ def test_storage_target_derives_all_paths():
     assert target.partition_path == "/dev/sdb1"
 
 
-def test_single_drive_setup_command_sequence(
-    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,10 +58,19 @@ def test_rsync_command_uses_delete_only_for_directories():
     directory_op = sync.SyncOperation("src/", "dst/", "backup", is_directory=True)
     file_op = sync.SyncOperation("src", "dst", "backup", is_directory=False)
     assert sync.rsync_command(directory_op) == [
-        "rsync", "-abcEPuvW", "--delete", "--backup-dir=backup", "src/", "dst/"
+        "rsync",
+        "-abcEPuvW",
+        "--delete",
+        "--backup-dir=backup",
+        "src/",
+        "dst/",
     ]
     assert sync.rsync_command(file_op) == [
-        "rsync", "-abcEPuvW", "--backup-dir=backup", "src", "dst"
+        "rsync",
+        "-abcEPuvW",
+        "--backup-dir=backup",
+        "src",
+        "dst",
     ]
 
 
@@ -95,18 +104,12 @@ def test_export_chowns_real_home_without_sudo(tmp_path, fake_runner, monkeypatch
 
     sync.export_to_system()
 
-    chown_calls = [
-        (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

@@ -174,6 +174,7 @@ class TestInitramfsToolsBackend:
         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()
 
@@ -189,7 +190,7 @@ class TestInitramfsToolsBackend:
         apt = [
             text
             for _, _, text in fake_runner.calls
-            if text and "apt install" in text and "tor busybox" in text
+            if text and "apt-get install" in text and "tor busybox" in text
         ]
         assert len(apt) == 1
 
@@ -209,3 +210,11 @@ class TestInitramfsToolsHookHardening:
         premount = self._read("tor_premount")
         assert '. "$conf"' not in premount
         assert "sed -n 's/^IPV4DNS" in premount
+
+    def test_hook_stamps_a_clock_floor(self):
+        assert "/etc/tor-clock-floor" in self._read("tor_hook")
+
+    def test_premount_raises_clock_to_floor_before_ntp(self):
+        premount = self._read("tor_premount")
+        assert "/etc/tor-clock-floor" in premount
+        assert premount.index("tor-clock-floor") < premount.index("ntpd")

tests/unit/test_unlock.py

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

tests/unit/test_wizard.py

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