diff --git a/lim/catalog.py b/lim/catalog.py index cb14226..080fda5 100644 --- a/lim/catalog.py +++ b/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"] diff --git a/lim/distributions.yml b/lim/distributions.yml index 99ef8d1..1fe3136 100644 --- a/lim/distributions.yml +++ b/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: diff --git a/lim/fsutil.py b/lim/fsutil.py index 1ebf17e..2e8435c 100644 --- a/lim/fsutil.py +++ b/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) diff --git a/lim/image/choosers.py b/lim/image/choosers.py index 372e109..6d74a8c 100644 --- a/lim/image/choosers.py +++ b/lim/image/choosers.py @@ -60,6 +60,17 @@ def choose_retropie(plan: ImagePlan) -> None: plan.image_name = entry["image"] +def choose_raspios(plan: ImagePlan) -> None: + plan.boot_size = "+512M" + plan.root_filesystem = "ext4" + variant = ui.ask("Which Raspberry Pi OS image [lite64,desktop64,lite32]:") + 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" @@ -83,12 +94,13 @@ 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...]?" + "Which distribution should be used [arch,raspios,moode,retropie,manjaro,torbox...]?" ) chooser = DISTRIBUTION_CHOOSERS.get(distribution) if chooser is None: diff --git a/lim/image/encryption.py b/lim/image/encryption.py index 431a476..b9b38e0 100644 --- a/lim/image/encryption.py +++ b/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 diff --git a/lim/image/initramfs/initramfs_tools.py b/lim/image/initramfs/initramfs_tools.py index 9b503d0..e07265f 100644 --- a/lim/image/initramfs/initramfs_tools.py +++ b/lim/image/initramfs/initramfs_tools.py @@ -68,6 +68,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" diff --git a/lim/image/loopimg.py b/lim/image/loopimg.py new file mode 100644 index 0000000..bee33f5 --- /dev/null +++ b/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}" diff --git a/lim/image/plan.py b/lim/image/plan.py index 5653d71..14bb07f 100644 --- a/lim/image/plan.py +++ b/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}" diff --git a/lim/image/transfer.py b/lim/image/transfer.py index a9f96ca..8ab929a 100644 --- a/lim/image/transfer.py +++ b/lim/image/transfer.py @@ -1,16 +1,18 @@ """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() @@ -116,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( diff --git a/tests/unit/test_fsutil.py b/tests/unit/test_fsutil.py index 73fd1eb..1ced44d 100644 --- a/tests/unit/test_fsutil.py +++ b/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