From 30ee1101ceaf47788da5a4e04dccf91f4897a042 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Thu, 23 Jul 2026 01:59:40 +0200 Subject: [PATCH] feat(cli): guided encrypted-image wizard (default) + remote-unlock Add a guided setup: one interactive command that asks everything up front then builds an encrypted, Tor-remote-unlockable image unattended (distribution, target device, hostname, login user + key, password), creating or renaming the login user and installing the SSH key for both unlock and post-boot login. - wizard.py: _collect (all prompts) + _execute (autonomous build); renames a stock pi/alarm user or creates one, grants sudo, installs the login key. - unlock.py + `lim --type remote-unlock`: reach the initramfs over Tor (onion, torsocks) or plain SSH (host/IP), run cryptroot-unlock or the passphrase prompt; the wizard persists a target record under ~/.config/lim/unlocks. - cli.py: register guided (default --type) and remote-unlock; drop the deprecated --extra argument. Co-Authored-By: Claude Opus 4.8 --- lim/cli.py | 28 +++-- lim/image/unlock.py | 100 ++++++++++++++++ lim/image/wizard.py | 206 +++++++++++++++++++++++++++++++++ tests/unit/test_cli.py | 4 + tests/unit/test_unlock.py | 42 +++++++ tests/unit/test_wizard.py | 235 ++++++++++++++++++++++++++++++++++++++ 6 files changed, 605 insertions(+), 10 deletions(-) create mode 100644 lim/image/unlock.py create mode 100644 lim/image/wizard.py create mode 100644 tests/unit/test_unlock.py create mode 100644 tests/unit/test_wizard.py diff --git a/lim/cli.py b/lim/cli.py index e2abd9c..d5469ba 100644 --- a/lim/cli.py +++ b/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" @@ -116,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 @@ -138,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() diff --git a/lim/image/unlock.py b/lim/image/unlock.py new file mode 100644 index 0000000..f90b294 --- /dev/null +++ b/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) diff --git a/lim/image/wizard.py b/lim/image/wizard.py new file mode 100644 index 0000000..0504bc8 --- /dev/null +++ b/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}@") + ui.info("Golden image? Clone this device onto the SSD, then grow it:") + ui.info(" parted resizepart 2 100% && cryptsetup resize && resize2fs ") + + +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) diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index 7b12269..add1f3c 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -57,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 diff --git a/tests/unit/test_unlock.py b/tests/unit/test_unlock.py new file mode 100644 index 0000000..89b7ada --- /dev/null +++ b/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 diff --git a/tests/unit/test_wizard.py b/tests/unit/test_wizard.py new file mode 100644 index 0000000..66062e8 --- /dev/null +++ b/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