kevinveenbirkenbach/linux-image-manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(image): flash full disk images into LUKS; add Raspberry Pi OS

Browse Source 
Support distributions that ship a full .img (Raspberry Pi OS, moode, RetroPie,
Manjaro ARM) rather than a rootfs tarball, by loop-mounting the image and
copying its boot + root partitions into a fresh LUKS container.

- distributions.yml/catalog.py/choosers.py: raspios catalog (lite64/desktop64/
  lite32 via the stable _latest redirects) + choose_raspios.
- plan.py: source_url override so a _latest redirect downloads under an .img.xz
  name that decompress_command recognises.
- transfer.py: transfer_disk_image (loop-mount -> repartition -> LUKS -> rsync
  copy with progress, cp fallback -> fix boot fstab); transfer_image gains
  interactive= and routes encrypted non-arch images here; download_image gains
  force_prompt.
- loopimg.py: losetup attach/detach/partition helper.
- fsutil.drop_fstab_mount + register_encrypted_root: replace a stock image's
  existing / fstab line instead of colliding with it.
- encryption.configure_encryption returns the onion address.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Kevin Veen-Birkenbach
2026-07-23 01:57:28 +02:00
parent 68ea39b784
commit 5bef6177ac
10 changed files with 211 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
lim/catalog.py
View File

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

13
lim/distributions.yml
View File

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

17
lim/fsutil.py
View File

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

14
lim/image/choosers.py
View File

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

9
lim/image/encryption.py
View File

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

2
lim/image/initramfs/initramfs_tools.py
View File

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

28
lim/image/loopimg.py Normal file
View File

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

6
lim/image/plan.py
View File

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

108
lim/image/transfer.py
View File

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

22
tests/unit/test_fsutil.py
View File

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