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

refactor!: port shell scripts to Python package

Browse Source 
Bash scripts were untestable and duplicated device/LUKS/mount logic;
the lim/ package centralizes it behind one subprocess wrapper and a
YAML image catalog (single point of truth).

BREAKING CHANGE: scripts/*.sh removed. Use `lim --type <cmd>`; new
types mount/umount/single-boot/raid1-boot/lock/unlock/import/export
replace direct script calls. --extra is deprecated and ignored.

- distributions.yml + lim/catalog.py hold the image catalog (PyYAML)
- pytest suite: 102 tests with mocked subprocess (tests/unit) and a
  250-line max file-length guard (tests/lint)
- ruff strict (select ALL), GitHub Actions CI, Dependabot; Travis gone
- Makefile: install (symlink ~/.local/bin/lim) and test targets
- fixes over bash: SUDO_USER-aware chown, mmcblk/nvme partition paths,
  sha512 checksum support, whole-pipeline failure detection, blkid
  UUID fallback for pre-mounted images, conditional fstab seeding for
  PARTUUID/LABEL images, clean errors for missing binaries
This commit is contained in:
Kevin Veen-Birkenbach
2026-07-14 11:27:49 +02:00
parent c420dd164d
commit ccdef065df
77 changed files with 3402 additions and 1614 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
lim/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""Linux Image Manager — administration tool for Linux images and encrypted storage."""
__version__ = "1.0.0"

31
lim/catalog.py Normal file
View File

@@ -0,0 +1,31 @@
"""Read-only access to the image catalog (distributions.yml in the repo root)."""
from functools import cache
import yaml
from lim import config
CATALOG_PATH = config.REPOSITORY_PATH / "distributions.yml"
@cache
def _load() -> dict:
with CATALOG_PATH.open() as handle:
return yaml.safe_load(handle)
def arch_rpi_images() -> dict[str, dict[str, str]]:
return _load()["arch_rpi_images"]
def manjaro_gnome_releases() -> dict[str, dict[str, str]]:
return _load()["manjaro_gnome_releases"]
def retropie_images() -> dict[str, dict[str, str]]:
return _load()["retropie_images"]
def mkinitcpio_modules_by_rpi() -> dict[str, str]:
return _load()["mkinitcpio_modules_by_rpi"]

167
lim/cli.py Normal file
View File

@@ -0,0 +1,167 @@
"""Command line interface: `lim --type <command>`."""
import argparse
import sys
from collections.abc import Callable
from dataclasses import dataclass
from lim import system, ui
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.storage import raid1, single_drive
@dataclass(frozen=True)
class Command:
func: Callable[[], None]
description: str
needs_root: bool
COMMANDS: dict[str, Command] = {
"image": Command(
image_setup.run_setup,
"Linux Image Setup:\n"
" - Creates partitions and formats them.\n"
" - Transfers the Linux image file to the device.\n"
" - Configures boot and root partitions.",
needs_root=True,
),
"single": Command(
single_drive.setup,
"Single Drive Encryption Setup:\n"
" - Sets up disk encryption using LUKS on one drive.\n"
" - Configures a Btrfs file system for secure storage.",
needs_root=True,
),
"raid1": Command(
raid1.setup,
"RAID1 Encryption Setup:\n"
" - Configures a virtual RAID1 with two drives.\n"
" - Uses LUKS encryption and a Btrfs RAID1 file system for redundancy.",
needs_root=True,
),
"backup": Command(
backup.run_backup,
"Backup Image Setup:\n"
" - Creates an image backup from a memory device to a file.\n"
" - Uses dd to transfer the image from the specified device to an image file.",
needs_root=True,
),
"chroot": Command(
chroot.run_chroot,
"Chroot Environment Setup:\n"
" - Mounts partitions and configures the chroot environment for a Linux image.\n"
" - Provides a shell within the Linux image for system maintenance.",
needs_root=True,
),
"mount": Command(
single_drive.mount,
"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.",
needs_root=True,
),
"single-boot": Command(
single_drive.mount_on_boot,
"Single Drive Automount Setup:\n"
" - Creates a LUKS keyfile and registers it in crypttab and fstab.",
needs_root=True,
),
"raid1-boot": Command(
raid1.mount_on_boot,
"RAID1 Automount Setup:\n"
" - Creates LUKS keyfiles for both drives and registers them in crypttab/fstab.",
needs_root=True,
),
"unlock": Command(
crypt.unlock,
"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.",
needs_root=False,
),
"import": Command(
sync.import_from_system,
"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.",
needs_root=False,
),
}
def build_parser() -> argparse.ArgumentParser:
epilog_lines = ["Available script types:"]
epilog_lines += [
f" {name:<12} - {command.description.splitlines()[0].rstrip(':')}"
for name, command in COMMANDS.items()
]
parser = argparse.ArgumentParser(
prog="lim",
description="Linux Image Manager — manages Linux images and encrypted storage.",
epilog="\n".join(epilog_lines),
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"--type",
required=True,
choices=list(COMMANDS.keys()),
help="Select the command to execute.",
)
parser.add_argument(
"--auto-confirm",
action="store_true",
help="Automatically confirm execution without prompting the user.",
)
parser.add_argument(
"--extra",
nargs=argparse.REMAINDER,
default=[],
help="Deprecated, ignored. Former extra parameters for the shell scripts.",
)
return parser
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()
ui.header()
ui.info(f"Selected command: {args.type}")
ui.info("Description:")
print(command.description)
try:
if not args.auto_confirm:
input("Press Enter to execute or Ctrl+C to cancel...")
command.func()
except LimError as exc:
ui.error(f"{exc} -> Leaving program.")
sys.exit(1)
except KeyboardInterrupt:
print()
ui.error("Execution aborted by user.")
sys.exit(1)

11
lim/config.py Normal file
View File

@@ -0,0 +1,11 @@
"""Repository paths shared by all modules."""
from pathlib import Path
REPOSITORY_PATH = Path(__file__).resolve().parent.parent
CONFIGURATION_PATH = REPOSITORY_PATH / "configuration"
PACKAGE_PATH = CONFIGURATION_PATH / "packages"
ENCRYPTED_PATH = REPOSITORY_PATH / ".encrypted"
DECRYPTED_PATH = REPOSITORY_PATH / "decrypted"
DATA_PATH = DECRYPTED_PATH / "data"
BACKUP_PATH = DECRYPTED_PATH / "backup"

0
lim/data/__init__.py Normal file
View File

24
lim/data/crypt.py Normal file
View File

@@ -0,0 +1,24 @@
"""Lock/unlock the encfs-encrypted data directory."""
from lim import config, runner, ui
def unlock() -> None:
ui.info(f"Unlocking directory {config.DECRYPTED_PATH}...")
if not config.DECRYPTED_PATH.is_dir():
ui.info(f"Creating directory {config.DECRYPTED_PATH}...")
config.DECRYPTED_PATH.mkdir()
ui.info(f"Decrypting directory {config.ENCRYPTED_PATH} to {config.DECRYPTED_PATH}...")
runner.run(["encfs", str(config.ENCRYPTED_PATH), str(config.DECRYPTED_PATH)])
print("ATTENTION: DATA IS NOW DECRYPTED!")
def lock() -> None:
ui.info(f"Locking directory {config.DECRYPTED_PATH}...")
runner.run(
["fusermount", "-u", str(config.DECRYPTED_PATH)],
error_msg="Unmounting failed.",
)
ui.info("Data is now encrypted.")
ui.info(f"Removing directory {config.DECRYPTED_PATH}...")
config.DECRYPTED_PATH.rmdir()

135
lim/data/sync.py Normal file
View File

@@ -0,0 +1,135 @@
"""Import personal data from the system into the encrypted store, or export it back."""
import time
from dataclasses import dataclass
from pathlib import Path
from lim import config, runner, system, ui
from lim.data import crypt
from lim.errors import LimError
# Paths relative to $HOME; trailing "/" marks a directory tree.
BACKUP_ITEMS = [
".ssh/",
".gitconfig",
".atom/config.cson",
".projectlibre/projectlibre.conf",
".local/share/rhythmbox/rhythmdb.xml",
".config/keepassxc/keepassxc.ini",
"Documents/certificates/",
"Documents/security/",
"Documents/identity/",
"Documents/health/",
"Documents/licenses/",
]
@dataclass(frozen=True)
class SyncOperation:
source: str
destination: str
backup_dir: str
is_directory: bool
def build_sync_plan(
mode: str,
home: Path,
data_path: Path,
backup_folder: Path,
) -> list[SyncOperation]:
"""One rsync operation per existing backup item; missing sources are skipped."""
operations = []
for item in BACKUP_ITEMS:
is_directory = item.endswith("/")
system_path = home / item
# The store mirrors the absolute system path below DATA_PATH.
data_path_item = Path(f"{data_path}{system_path}")
if mode == "export":
source, destination = data_path_item, system_path
elif mode == "import":
source, destination = system_path, data_path_item
else:
raise LimError(f"Unknown sync mode: {mode}")
ui.info(f"{mode.capitalize()} data from {source} to {destination}...")
if not (source.is_file() or source.is_dir()):
ui.warning(f"{source} doesn't exist. Copying data is not possible.")
continue
if is_directory:
backup_dir = Path(f"{backup_folder}{system_path}")
else:
backup_dir = Path(f"{backup_folder}{system_path}").parent
operations.append(
SyncOperation(
source=f"{source}/" if is_directory else str(source),
destination=f"{destination}/" if is_directory else str(destination),
backup_dir=str(backup_dir),
is_directory=is_directory,
)
)
return operations
def rsync_command(operation: SyncOperation) -> list[str]:
command = ["rsync", "-abcEPuvW"]
if operation.is_directory:
command.append("--delete")
command += [f"--backup-dir={operation.backup_dir}", operation.source, operation.destination]
return command
def execute_sync_plan(operations: list[SyncOperation]) -> None:
for operation in operations:
Path(operation.backup_dir).mkdir(parents=True, exist_ok=True)
destination_dir = (
Path(operation.destination)
if operation.is_directory
else Path(operation.destination).parent
)
destination_dir.mkdir(parents=True, exist_ok=True)
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(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!"
)
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.mkdir(parents=True, exist_ok=True)
operations = build_sync_plan(
mode, system.real_home(), config.DATA_PATH, backup_folder
)
execute_sync_plan(operations)
def export_to_system() -> None:
import_from_system(mode="export")
user = system.real_user()
home = system.real_home()
ui.info("Setting right permissions for imported files...")
try:
(home / ".ssh").chmod(0o700)
for child in (home / ".ssh").rglob("*"):
child.chmod(0o700 if child.is_dir() else 0o600)
except OSError as exc:
raise LimError(f"Failed to set correct ssh permissions: {exc}") from exc
# Best effort without privilege escalation, like the original script.
chowned = runner.run(["chown", "-R", f"{user}:{user}", str(home)], check=False)
if chowned.returncode != 0:
ui.warning(f'Not all files could be owned by user "{user}"...')

130
lim/device.py Normal file
View File

@@ -0,0 +1,130 @@
"""Block-device selection and low-level device helpers."""
import stat as stat_module
from dataclasses import dataclass
from pathlib import Path
from lim import runner, ui
from lim.errors import LimError
SYS_BLOCK_PATH = Path("/sys/block")
PARTITION_TABLE_INFO = """\
##########################################################################################
Note on Partition Table Deletion:
---------------------------------------------
• MBR (Master Boot Record):
- Typically occupies the first sector (512 bytes), i.e., 1 block.
• GPT (GUID Partition Table):
- Uses a protective MBR (1 block), a GPT header (1 block),
and usually a partition entry array that takes up about 32 blocks.
- Total: approximately 34 blocks (assuming a 512-byte block size).
Recommendation: For deleting a GPT partition table, use a block size of 512 bytes
and overwrite at least 34 blocks to ensure the entire table is cleared.
##########################################################################################"""
def optimal_blocksize(name: str, sys_block_path: Path = SYS_BLOCK_PATH) -> str:
"""64 * physical block size, or "4K" when the size cannot be read.
See https://www.heise.de/ct/hotline/Optimale-Blockgroesse-fuer-dd-2056768.html
"""
size_path = sys_block_path / name / "queue" / "physical_block_size"
try:
return str(64 * int(size_path.read_text().strip()))
except (OSError, ValueError):
return "4K"
@dataclass(frozen=True)
class Device:
name: str # e.g. "sda" or "mmcblk0"
@property
def path(self) -> str:
return f"/dev/{self.name}"
@property
def optimal_blocksize(self) -> str:
return optimal_blocksize(self.name)
def partition(self, number: int) -> str:
"""/dev/sda -> /dev/sda1, /dev/mmcblk0 -> /dev/mmcblk0p1."""
if self.name[-1].isdigit():
return f"{self.path}p{number}"
return f"{self.path}{number}"
def is_block_device(path: str) -> bool:
try:
return stat_module.S_ISBLK(Path(path).stat().st_mode)
except OSError:
return False
def select_device() -> Device:
ui.info("Available devices:")
runner.run(["lsblk", "-o", "NAME,SIZE,TYPE,MODEL"], check=False)
name = ui.ask("Please type in the name of the device: /dev/")
device = Device(name)
if not name or not is_block_device(device.path):
raise LimError(f"{device.path} is not a valid device.")
ui.info(f"Device path set to: {device.path}")
ui.info(f"Optimal blocksize set to: {device.optimal_blocksize}")
return device
def overwrite_device(device: Device) -> None:
"""Optionally overwrite the device (or its first blocks) with zeros."""
print(PARTITION_TABLE_INFO)
answer = ui.ask(
f"Should {device.path} be overwritten with zeros before copying? (y/N/block count)"
)
if answer == "y":
ui.info("Overwriting entire device...")
runner.run(
[
"dd",
"if=/dev/zero",
f"of={device.path}",
f"bs={device.optimal_blocksize}",
"status=progress",
],
sudo=True,
error_msg=f"Overwriting {device.path} failed.",
)
runner.sync_disks()
elif answer in ("", "N"):
ui.info("Skipping Overwriting...")
elif answer.isdigit():
ui.info(f"Overwriting {answer} blocks...")
runner.run(
[
"dd",
"if=/dev/zero",
f"of={device.path}",
f"bs={device.optimal_blocksize}",
f"count={answer}",
"status=progress",
],
sudo=True,
error_msg=f"Overwriting {device.path} failed.",
)
runner.sync_disks()
else:
raise LimError("Invalid input. Block count must be a number.")
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
)
def is_mounted(path_fragment: str) -> bool:
"""Whether any current mount line contains the given fragment."""
mounts = runner.output(["mount"], check=False)
return any(path_fragment in line for line in mounts.splitlines())

2
lim/errors.py Normal file
View File

@@ -0,0 +1,2 @@
class LimError(Exception):
"""Fatal error that aborts the current lim command."""

32
lim/fsutil.py Normal file
View File

@@ -0,0 +1,32 @@
"""Small file manipulation helpers."""
from pathlib import Path
from lim import ui
from lim.errors import LimError
def replace_in_file(search: str, replace: str, path: str | Path) -> None:
"""Replace every literal occurrence of ``search``; fail when absent."""
path = Path(path)
text = path.read_text()
new_text = text.replace(search, replace)
if new_text == text:
raise LimError(f"Search string '{search}' not found in {path}.")
path.write_text(new_text)
def ensure_line_in_file(line: str, path: str | Path) -> bool:
"""Append ``line`` unless already present. Returns True when appended."""
path = Path(path)
content = path.read_text() if path.exists() else ""
if line in content.splitlines():
ui.warning(f"File {path} already contains the following entry:")
print(line)
ui.info("Skipped.")
return False
with path.open("a") as handle:
if content and not content.endswith("\n"):
handle.write("\n")
handle.write(line + "\n")
return True

0
lim/image/__init__.py Normal file
View File

32
lim/image/backup.py Normal file
View File

@@ -0,0 +1,32 @@
"""Create an image backup from a memory device into a file."""
from pathlib import Path
from lim import device as device_module
from lim import runner, ui
def run_backup() -> None:
ui.info("Backupscript for memory devices started...")
print()
device = device_module.select_device()
working_dir = Path.cwd()
path = ""
while not path:
path = ui.ask(f"Please type in backup image path+name relative to {working_dir}:")
output_file = f"{path}.img" if path.startswith("/") else f"{working_dir}/{path}.img"
ui.info(f"Input file: {device.path}")
ui.info(f"Output file: {output_file}")
ui.ask('Please confirm by pushing "Enter". To cancel use "Ctrl + C"')
ui.info("Imagetransfer starts. This can take a while...")
runner.run(
["dd", f"if={device.path}", f"of={output_file}", "bs=1M", "status=progress"],
sudo=True,
error_msg='"dd" failed.',
)
runner.sync_disks()
ui.success("Imagetransfer successfull.")

111
lim/image/choosers.py Normal file
View File

@@ -0,0 +1,111 @@
"""Interactive selection of the distribution image to install."""
from lim import catalog, runner, ui
from lim.errors import LimError
from lim.image.plan import ImagePlan
def choose_arch(plan: ImagePlan) -> None:
version = ui.ask("Which Raspberry Pi will be used (e.g.: 1, 2, 3b, 3b+, 4...):")
entry = catalog.arch_rpi_images().get(version)
if entry is None:
raise LimError(f"Version {version} isn't supported.")
plan.raspberry_pi_version = version
plan.boot_size = "+500M"
plan.base_download_url = "http://os.archlinuxarm.org/os/"
plan.image_name = entry["image"]
plan.luks_memory_cost = entry["luks_memory_cost"]
def choose_manjaro(plan: ImagePlan) -> None:
plan.boot_size = "+500M"
flavour = ui.ask("Which version(e.g.:architect,gnome) should be used:")
if flavour == "architect":
plan.image_checksum = "6b1c2fce12f244c1e32212767a9d3af2cf8263b2"
plan.base_download_url = (
"https://osdn.net/frs/redir.php?m=dotsrc&f=%2Fstorage%2Fg%2Fm%2Fma"
"%2Fmanjaro%2Farchitect%2F20.0%2F"
)
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:")
entry = catalog.manjaro_gnome_releases().get(release)
if entry is None:
raise LimError(f"Gnome release {release} isn't supported.")
plan.image_checksum = entry.get("checksum")
plan.base_download_url = entry["url"]
plan.image_name = entry["image"]
plan.luks_memory_cost = entry.get("luks_memory_cost")
plan.raspberry_pi_version = entry.get("raspberry_pi_version")
else:
raise LimError(f"Manjaro version {flavour} isn't supported.")
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.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:")
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.image_checksum = entry["checksum"]
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.boot_size = "+200M"
def choose_android_x86(plan: ImagePlan) -> None:
plan.base_download_url = (
"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.boot_size = "+500M"
DISTRIBUTION_CHOOSERS = {
"android-x86": choose_android_x86,
"torbox": choose_torbox,
"arch": choose_arch,
"manjaro": choose_manjaro,
"moode": choose_moode,
"retropie": choose_retropie,
}
def choose_linux_image(plan: ImagePlan) -> None:
distribution = ui.ask(
"Which distribution should be used [arch,moode,retropie,manjaro,torbox...]?"
)
chooser = DISTRIBUTION_CHOOSERS.get(distribution)
if chooser is None:
raise LimError(f"Distribution {distribution} isn't supported.")
plan.distribution = distribution
chooser(plan)
def choose_local_image(plan: ImagePlan) -> None:
ui.info("Available images:")
runner.run(["ls", "-l", str(plan.image_folder)], check=False)
plan.image_name = ui.ask("Which image would you like to use?")

23
lim/image/chroot.py Normal file
View File

@@ -0,0 +1,23 @@
"""Mount an image and open an interactive shell inside it."""
from lim import device as device_module
from lim import runner, ui
from lim.image.session import ImageSession
def run_chroot() -> None:
ui.info("Starting chroot...")
device = device_module.select_device()
session = ImageSession(device)
try:
session.make_working_folder()
session.make_mount_folders()
session.decrypt_root()
session.mount_partitions()
session.mount_chroot_binds()
session.copy_qemu()
session.copy_resolv_conf()
ui.info("Bash shell starts...")
runner.run(["chroot", str(session.root_mount_path), "/bin/bash"], sudo=True)
finally:
session.destructor()

117
lim/image/encryption.py Normal file
View File

@@ -0,0 +1,117 @@
"""Remote-unlockable LUKS boot configuration (dropbear + mkinitcpio).
See https://gist.github.com/gea0/4fc2be0cb7a74d0e7cc4322aed710d38 and
https://gist.github.com/EnigmaCurry/2f9bed46073da8e38057fe78a61e7994
"""
from pathlib import Path
from lim import catalog, fsutil, packages, runner, ui
from lim.image.plan import ImagePlan
from lim.image.session import ImageSession, chroot_bash, install_packages
MKINITCPIO_HOOKS_PREFIX = (
"base udev autodetect microcode modconf kms keyboard keymap consolefont block"
)
MKINITCPIO_HOOKS_SUFFIX = "filesystems fsck"
def _configure_mkinitcpio(plan: ImagePlan, root: Path) -> None:
# Concerning mkinitcpio warnings, see
# https://gist.github.com/imrvelj/c65cd5ca7f5505a65e59204f5a3f7a6d
mkinitcpio_path = root / "etc/mkinitcpio.conf"
ui.info(f"Configuring {mkinitcpio_path}...")
additional_modules = catalog.mkinitcpio_modules_by_rpi().get(
plan.raspberry_pi_version
)
if additional_modules is None:
ui.warning(f"Version {plan.raspberry_pi_version} isn't supported.")
additional_modules = ""
modules = f"g_cdc usb_f_acm usb_f_ecm {additional_modules} g_ether".replace(" ", " ")
fsutil.replace_in_file("MODULES=()", f"MODULES=({modules})", mkinitcpio_path)
fsutil.replace_in_file(
"BINARIES=()", "BINARIES=(/usr/lib/libgcc_s.so.1)", mkinitcpio_path
)
fsutil.replace_in_file(
f"HOOKS=({MKINITCPIO_HOOKS_PREFIX} {MKINITCPIO_HOOKS_SUFFIX})",
f"HOOKS=({MKINITCPIO_HOOKS_PREFIX} sleep netconf dropbear encryptssh "
f"{MKINITCPIO_HOOKS_SUFFIX})",
mkinitcpio_path,
)
ui.info(f"Content of {mkinitcpio_path}:{mkinitcpio_path.read_text()}")
ui.info("Generating mkinitcpio...")
chroot_bash(root, "mkinitcpio -vP")
def _register_encrypted_root(plan: ImagePlan, session: ImageSession, root: Path) -> None:
fstab_path = root / "etc/fstab"
fstab_line = (
f"UUID={session.root_partition_uuid} / {plan.root_filesystem}"
" defaults,noatime 0 1"
)
ui.info(f"Configuring {fstab_path}...")
fsutil.ensure_line_in_file(fstab_line, fstab_path)
ui.info(f"Content of {fstab_path}:{fstab_path.read_text()}")
crypttab_path = root / "etc/crypttab"
crypttab_line = (
f"{session.root_mapper_name} UUID={session.root_partition_uuid} none luks"
)
ui.info(f"Configuring {crypttab_path}...")
fsutil.ensure_line_in_file(crypttab_line, crypttab_path)
ui.info(f"Content of {crypttab_path}:{crypttab_path.read_text()}")
def _configure_bootloader(plan: ImagePlan, session: ImageSession, root: Path) -> None:
cryptdevice = (
f"cryptdevice=UUID={session.root_partition_uuid}:{session.root_mapper_name} "
f"root={session.root_mapper_path}"
)
boot_txt_path = session.boot_mount_path / "boot.txt"
if boot_txt_path.is_file():
ui.info(f"Configuring {boot_txt_path}...")
hostname = (root / "etc/hostname").read_text().strip()
fsutil.replace_in_file(
"part uuid ${devtype} ${devnum}:2 uuid", "", boot_txt_path
)
fsutil.replace_in_file(
"setenv bootargs console=ttyS1,115200 console=tty0 root=PARTUUID=${uuid} "
'rw rootwait smsc95xx.macaddr="${usbethaddr}"',
f"setenv bootargs console=ttyS1,115200 console=tty0 "
f"ip=::::{hostname}:eth0:dhcp {cryptdevice} rw rootwait "
# Concerning issues with network adapter names, see
# https://forum.iobroker.net/topic/40542/raspberry-pi4-kein-eth0-mehr/16
'smsc95xx.macaddr="${usbethaddr}" net.ifnames=0 biosdevname=0',
boot_txt_path,
)
ui.info(f"Content of {boot_txt_path}:{boot_txt_path.read_text()}")
ui.info("Generating...")
chroot_bash(root, "cd /boot/ && ./mkscr || exit 1")
else:
cmdline_txt_path = session.boot_mount_path / "cmdline.txt"
ui.info(f"Configuring {cmdline_txt_path}...")
fsutil.replace_in_file(
"root=/dev/mmcblk0p2",
f"{cryptdevice} rootfstype={plan.root_filesystem}",
cmdline_txt_path,
)
ui.info(f"Content of {cmdline_txt_path}:{cmdline_txt_path.read_text()}")
def configure_encryption(
plan: ImagePlan, session: ImageSession, authorized_keys: Path
) -> None:
root = session.root_mount_path
ui.info("Setup encryption...")
ui.info("Installing neccessary software...")
install_packages(
plan.distribution, root, " ".join(packages.get_packages("server/luks"))
)
dropbear_root_key_path = root / "etc/dropbear/root_key"
ui.info(f"Adding {authorized_keys} to dropbear...")
runner.run(["cp", "-v", str(authorized_keys), str(dropbear_root_key_path)], sudo=True)
_configure_mkinitcpio(plan, root)
_register_encrypted_root(plan, session, root)
_configure_bootloader(plan, session, root)

31
lim/image/plan.py Normal file
View File

@@ -0,0 +1,31 @@
"""The image setup plan collected from the user's answers."""
from dataclasses import dataclass, field
from pathlib import Path
DEFAULT_BOOT_SIZE = "+500M"
@dataclass
class ImagePlan:
operation_system: str = "linux"
distribution: str | None = None
base_download_url: str | None = None
image_name: str | None = None
image_checksum: str | None = None
boot_size: str = ""
luks_memory_cost: str | None = None
raspberry_pi_version: str | None = None
encrypt_system: bool = False
root_filesystem: str = ""
image_folder: Path = field(default_factory=Path)
@property
def download_url(self) -> str | None:
if self.base_download_url is None or self.image_name is None:
return None
return f"{self.base_download_url}{self.image_name}"
@property
def image_path(self) -> Path:
return self.image_folder / self.image_name

236
lim/image/raspberry.py Normal file
View File

@@ -0,0 +1,236 @@
"""Raspberry-Pi-specific configuration of a freshly transferred image."""
from pathlib import Path
from lim import device as device_module
from lim import fsutil, runner, ui
from lim.errors import LimError
from lim.image.encryption import configure_encryption
from lim.image.plan import ImagePlan
from lim.image.session import ImageSession, chroot_bash, install_packages
ADMINISTRATOR_USERNAME = "administrator"
def configure_sudoers(root_mount_path: Path, username: str) -> None:
sudo_config_dir = root_mount_path / "etc/sudoers.d"
sudo_config_file = sudo_config_dir / username
sudo_config_dir.mkdir(parents=True, exist_ok=True)
try:
sudo_config_file.write_text(f"{username} ALL=(ALL:ALL) ALL\n")
sudo_config_file.chmod(0o440)
except OSError as exc:
raise LimError(f"Failed to create sudoers file for {username}: {exc}") from exc
def configure_ssh_key(
public_key_path: str, target_ssh_folder: Path, authorized_keys: Path
) -> None:
source = Path(public_key_path)
if not source.is_file():
raise LimError(
f'The ssh key "{public_key_path}" can\'t be copied to '
f'"{authorized_keys}" because it doesn\'t exist.'
)
ui.info("Copy ssh key to target...")
target_ssh_folder.mkdir(parents=True, exist_ok=True)
authorized_keys.write_text(source.read_text())
ui.info(f"{authorized_keys} contains the following: {authorized_keys.read_text()}")
ui.info("Set permissions with chmod...")
target_ssh_folder.chmod(0o700)
authorized_keys.chmod(0o600)
def _ensure_image_mounted(session: ImageSession) -> None:
ui.info("Start regular mounting procedure...")
if device_module.is_mounted(session.boot_partition_path):
ui.info(f"{session.boot_partition_path} is allready mounted...")
elif device_module.is_mounted(session.root_mapper_path):
ui.info(f"{session.root_mapper_path} is allready mounted...")
else:
session.decrypt_root()
session.mount_partitions()
if session.boot_partition_uuid is None:
# Partitions were mounted by an earlier run; fetch what mount_partitions
# would have provided so fstab/crypttab never see a None UUID.
session.read_partition_uuids()
def _seed_boot_uuid(session: ImageSession) -> None:
fstab_path = session.root_mount_path / "etc/fstab"
if "/dev/mmcblk0p1" not in fstab_path.read_text():
# PARTUUID=/LABEL= based images (e.g. RetroPie, Manjaro ARM) have
# nothing to seed — that is a valid state, not an error.
ui.info(f"{fstab_path} does not reference /dev/mmcblk0p1. Skipping UUID seeding.")
return
ui.info(f"Seeding UUID to {fstab_path} to avoid path conflicts...")
fsutil.replace_in_file(
"/dev/mmcblk0p1", f"UUID={session.boot_partition_uuid}", fstab_path
)
ui.info(f"Content of {fstab_path}:{fstab_path.read_text()}")
def _select_target_user(root: Path) -> tuple[str, str]:
"""Return (default username, target username), renaming the home on request."""
target_home_path = root / "home"
home_entries = sorted(path.name for path in target_home_path.iterdir())
if not home_entries:
raise LimError(f"No home directory found under {target_home_path}.")
default_username = home_entries[0]
rename = default_username != ADMINISTRATOR_USERNAME and ui.confirm(
f"Should the {default_username} be renamed to {ADMINISTRATOR_USERNAME}? "
)
if not rename:
return default_username, default_username
ui.info(
f"Rename home directory from {target_home_path / default_username} "
f"to {target_home_path / ADMINISTRATOR_USERNAME}..."
)
runner.run(
[
"mv",
"-v",
str(target_home_path / default_username),
str(target_home_path / ADMINISTRATOR_USERNAME),
],
sudo=True,
error_msg="Failed to rename home directory",
)
return default_username, ADMINISTRATOR_USERNAME
def _change_passwords(root: Path, target_username: str) -> None:
password = ui.ask(
f"Type in new password for user root and {target_username} (leave empty to skip): "
)
if not password:
ui.info("No password change requested, skipped password change...")
return
repeated = ui.ask(f'Repeat new password for "{target_username}": ')
if password != repeated:
raise LimError("Passwords didn't match.")
ui.info("Changing passwords on target system...")
chroot_bash(
root,
f"( echo '{password}'; echo '{password}' ) | passwd {target_username}\n"
f"( echo '{password}'; echo '{password}' ) | passwd\n",
error_msg="Failed to change password.",
)
def _configure_hostname(root: Path) -> None:
hostname_path = root / "etc/hostname"
hostname = ui.ask("Type in the hostname (leave empty to skip): ")
if hostname:
hostname_path.write_text(hostname + "\n")
else:
hostname = hostname_path.read_text().strip()
ui.info("No hostname change requested, skipped hostname change...")
ui.info(f"Used hostname is: {hostname}")
def _update_system(plan: ImagePlan, root: Path) -> None:
if not ui.confirm("Should the system be updated?"):
return
ui.info("Updating system...")
if plan.distribution in ("arch", "manjaro"):
chroot_bash(root, "pacman --noconfirm -Syyu")
elif plan.distribution in ("moode", "retropie"):
chroot_bash(root, "yes | apt update\nyes | apt upgrade\n")
else:
ui.warning(
f'System update for operation system "{plan.distribution}" '
"is not supported yet. Skipped."
)
def _run_retropie_procedures(session: ImageSession, public_key_path: str) -> None:
if public_key_path:
(session.boot_mount_path / "ssh").write_text("\n")
if not ui.confirm("Should the RetroFlag specific procedures be executed?"):
return
ui.info("Executing RetroFlag specific procedures...")
chroot_bash(
session.root_mount_path,
'wget -O - "https://raw.githubusercontent.com/RetroFlag/'
'retroflag-picase/master/install_gpi.sh" | bash\n',
)
def _configure_users_and_keys(session: ImageSession) -> tuple[str, Path]:
"""Handle user rename, sudo rights and SSH key; return (public key path, authorized_keys)."""
root = session.root_mount_path
ui.info("Define target paths...")
default_username, target_username = _select_target_user(root)
renamed = default_username != target_username
target_user_ssh_folder = root / "home" / target_username / ".ssh"
target_authorized_keys = target_user_ssh_folder / "authorized_keys"
if ui.confirm(f"Should the {target_username} have sudo rights? "):
configure_sudoers(root, target_username)
public_key_path = ui.ask(
"Enter the path to the SSH key to be added to the image (default: none):"
)
if public_key_path:
configure_ssh_key(public_key_path, target_user_ssh_folder, target_authorized_keys)
else:
ui.info("Skipped SSH-key copying..")
ui.info("Start chroot procedures...")
session.mount_chroot_binds()
session.copy_qemu()
session.copy_resolv_conf()
chroot_user_home = f"/home/{target_username}/"
if renamed:
ui.info("Delete old user and create new user")
chroot_bash(
root,
f"userdel -r {default_username}\n"
f"useradd -m -d {chroot_user_home} -s /bin/bash {target_username}\n"
f"chown -R {target_username}:{target_username} {chroot_user_home}\n",
error_msg="Failed to delete old user and create new user",
)
if public_key_path:
ui.info("Chroot to set ownership...")
chroot_bash(
root,
f"chown -vR {target_username}:{target_username} {chroot_user_home}.ssh\n",
)
_change_passwords(root, target_username)
return public_key_path, target_authorized_keys
def configure_raspberry_image(plan: ImagePlan, session: ImageSession) -> None:
_ensure_image_mounted(session)
root = session.root_mount_path
_seed_boot_uuid(session)
public_key_path, target_authorized_keys = _configure_users_and_keys(session)
_configure_hostname(root)
if plan.distribution in ("arch", "manjaro"):
ui.info("Populating keys...")
chroot_bash(
root,
"yes | pacman-key --init\nyes | pacman-key --populate archlinuxarm\n",
)
_update_system(plan, root)
ui.info(f"Installing software for filesystem {plan.root_filesystem}...")
if plan.root_filesystem == "btrfs":
install_packages(plan.distribution, root, "btrfs-progs")
else:
ui.info("Skipped.")
if plan.encrypt_system:
configure_encryption(plan, session, target_authorized_keys)
ui.info("Running system specific procedures...")
if plan.distribution == "retropie":
_run_retropie_procedures(session, public_key_path)

190
lim/image/session.py Normal file
View File

@@ -0,0 +1,190 @@
"""Mount/unmount lifecycle for working on an image on a block device."""
import time
from pathlib import Path
from lim import device as device_module
from lim import runner, ui
from lim.device import Device
from lim.errors import LimError
class ImageSession:
"""Working folder, partition paths and (chroot) mounts for one device.
Call ``destructor()`` in a finally block: it unmounts and removes
everything best-effort, mirroring partially completed setups.
"""
def __init__(self, device: Device) -> None:
self.device = device
self.working_folder: Path | None = None
self.boot_mount_path: Path | None = None
self.root_mount_path: Path | None = None
self.boot_partition_path = device.partition(1)
self.root_partition_path = device.partition(2)
# Overridden by decrypt_root() when the root partition is LUKS.
self.root_mapper_path = self.root_partition_path
self.root_mapper_name: str | None = None
self.boot_partition_uuid: str | None = None
self.root_partition_uuid: str | None = None
def make_working_folder(self) -> None:
# Mount points must live on the root fs; mkdir() fails on collisions.
self.working_folder = Path(f"/tmp/linux-image-manager-{int(time.time())}") # noqa: S108
ui.info(f"Create temporary working folder in {self.working_folder}")
self.working_folder.mkdir(parents=True)
def make_mount_folders(self) -> None:
if self.working_folder is None:
self.make_working_folder()
ui.info("Preparing mount paths...")
self.boot_mount_path = self.working_folder / "boot"
self.root_mount_path = self.working_folder / "root"
self.boot_mount_path.mkdir()
self.root_mount_path.mkdir()
def root_is_luks(self) -> bool:
return device_module.blkid_value(self.root_partition_path, "TYPE") == "crypto_LUKS"
def read_partition_uuids(self) -> None:
"""Fetch partition UUIDs via blkid; derive the LUKS mapper name if unset."""
self.root_partition_uuid = device_module.blkid_value(
self.root_partition_path, "UUID"
)
self.boot_partition_uuid = device_module.blkid_value(
self.boot_partition_path, "UUID"
)
if self.root_mapper_name is None and self.root_is_luks():
# Same deterministic name decrypt_root() would have used.
self.root_mapper_name = f"linux-image-manager-{self.root_partition_uuid}"
self.root_mapper_path = f"/dev/mapper/{self.root_mapper_name}"
def decrypt_root(self) -> None:
if not self.root_is_luks():
return
self.root_partition_uuid = device_module.blkid_value(
self.root_partition_path, "UUID"
)
self.root_mapper_name = f"linux-image-manager-{self.root_partition_uuid}"
self.root_mapper_path = f"/dev/mapper/{self.root_mapper_name}"
ui.info(f"Decrypting of {self.root_partition_path} is neccessary...")
runner.run(
[
"cryptsetup",
"-v",
"luksOpen",
self.root_partition_path,
self.root_mapper_name,
],
sudo=True,
)
def mount_partitions(self) -> None:
if self.boot_mount_path is None:
raise LimError("Mount folders are not prepared yet.")
ui.info("Mount boot and root partition...")
runner.run(
["mount", "-v", self.boot_partition_path, str(self.boot_mount_path)],
sudo=True,
)
runner.run(
["mount", "-v", self.root_mapper_path, str(self.root_mount_path)], sudo=True
)
ui.info("Setting uuid variables...")
self.read_partition_uuids()
ui.info("The following mounts refering this setup exist:")
runner.run(["findmnt", "-R", str(self.working_folder)], check=False)
def 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", "/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)
runner.run(["mount", "--bind", "/dev/pts", f"{root}/dev/pts"], sudo=True)
def copy_qemu(self) -> None:
ui.info("Copy qemu binary...")
runner.run(
["cp", "-v", "/usr/bin/qemu-arm-static", f"{self.root_mount_path}/usr/bin/"],
sudo=True,
)
def copy_resolv_conf(self) -> None:
ui.info("Copy resolv.conf...")
copied = runner.run(
[
"cp",
"--remove-destination",
"-v",
"/etc/resolv.conf",
f"{self.root_mount_path}/etc/",
],
sudo=True,
check=False,
)
if copied.returncode != 0:
ui.warning("Failed. Probably there is no internet connection available.")
def _umount(self, path: str, *, lazy: bool = False) -> None:
flags = ["-lv"] if lazy else ["-v"]
result = runner.run(["umount", *flags, path], sudo=True, check=False)
if result.returncode != 0:
ui.warning(f"Umounting {path} failed!")
def _rmdir(self, path: Path | None) -> None:
if path is None:
return
result = runner.run(["rmdir", "-v", str(path)], sudo=True, check=False)
if result.returncode != 0:
ui.warning(f"Removing {path} failed!")
def destructor(self) -> None:
ui.info("Cleaning up...")
ui.info("Unmounting everything...")
root = self.root_mount_path
if root is not None:
self._umount(f"{root}/dev/pts", lazy=True)
self._umount(f"{root}/dev", lazy=True)
self._umount(f"{root}/proc")
self._umount(f"{root}/sys")
self._umount(f"{root}/boot")
self._umount(str(root))
if self.boot_mount_path is not None:
self._umount(str(self.boot_mount_path))
ui.info("Deleting mount folders...")
self._rmdir(self.root_mount_path)
self._rmdir(self.boot_mount_path)
self._rmdir(self.working_folder)
if self.root_mapper_name and self.root_is_luks():
ui.info(f"Trying to close decrypted {self.root_mapper_name}...")
closed = runner.run(
["cryptsetup", "-v", "luksClose", self.root_mapper_name],
sudo=True,
check=False,
)
if closed.returncode != 0:
ui.warning("Failed.")
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."""
runner.run(
["chroot", str(root_mount_path), "/bin/bash"],
input_text=script,
sudo=True,
error_msg=error_msg,
)
def install_packages(distribution: str, root_mount_path: Path, package_names: str) -> None:
"""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}")
elif distribution in ("moode", "retropie"):
chroot_bash(root_mount_path, f"yes | apt install {package_names}")
else:
raise LimError("Package manager not supported.")

96
lim/image/setup.py Normal file
View File

@@ -0,0 +1,96 @@
"""Interactive Linux image setup: download, verify, transfer, configure.
Reference for encrypted Raspberry Pi images:
https://wiki.polaire.nl/doku.php?id=archlinux-raspberry-encrypted
"""
import pwd
from pathlib import Path
from lim import device as device_module
from lim import system, ui
from lim.errors import LimError
from lim.image import choosers, transfer, verify
from lim.image.plan import ImagePlan
from lim.image.raspberry import configure_raspberry_image
from lim.image.session import ImageSession
def _prepare_image_folder(plan: ImagePlan) -> None:
ui.info("Configure user...")
origin_username = ui.ask("Please type in a valid working username:")
try:
pwd.getpwnam(origin_username)
except KeyError:
raise LimError(f"User {origin_username} doesn't exist.") from None
ui.info("Image routine starts...")
plan.image_folder = Path(f"/home/{origin_username}/Software/Images")
ui.info(f'The images will be stored in "{plan.image_folder}".')
if not plan.image_folder.is_dir():
ui.info(f'Folder "{plan.image_folder}" doesn\'t exist. It will be created now.')
plan.image_folder.mkdir(parents=True)
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}*".'
)
return device
def _choose_and_verify_image(plan: ImagePlan) -> None:
plan.operation_system = ui.ask(
"Which operation system would you like to use [linux,windows,...]?"
)
if plan.operation_system == "linux":
choosers.choose_linux_image(plan)
plan.encrypt_system = ui.confirm("Should the system be encrypted?")
ui.info("Generating os-image...")
transfer.download_image(plan)
else:
choosers.choose_local_image(plan)
ui.info("Verifying image...")
ui.info("Verifying checksum...")
if plan.image_checksum is None and plan.download_url is not None:
plan.image_checksum = verify.resolve_checksum(plan.download_url)
verify.verify_checksum(plan.image_path, plan.image_checksum)
if plan.download_url is not None:
verify.verify_signature(plan.download_url, plan.image_path, plan.image_folder)
def run_setup() -> None:
ui.info("Setupscript for images started...")
ui.info("Checking if root...")
if not system.is_root():
raise LimError("This script must be executed as root!")
plan = ImagePlan()
_prepare_image_folder(plan)
device = _select_unmounted_device()
_choose_and_verify_image(plan)
session = ImageSession(device)
try:
session.make_working_folder()
session.make_mount_folders()
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)
else:
ui.info("Skipping image transfer...")
if plan.raspberry_pi_version:
configure_raspberry_image(plan, session)
finally:
session.destructor()
ui.success("Setup successfull :)")

140
lim/image/transfer.py Normal file
View File

@@ -0,0 +1,140 @@
"""Download the selected image and transfer it onto the target device."""
from pathlib import Path
from lim import device as device_module
from lim import runner, ui
from lim.errors import LimError
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?"):
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("Start Download procedure...")
if plan.image_path.is_file():
ui.info("Image exist local. Download skipped.")
return
ui.info(f'Image "{plan.image_name}" doesn\'t exist under local path "{plan.image_path}".')
ui.info(f'Image "{plan.image_name}" gets downloaded from "{plan.download_url}"...')
runner.run(
["wget", plan.download_url, "-O", str(plan.image_path)],
error_msg=f'Download from "{plan.download_url}" failed.',
)
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)
"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)
"n\np\n2\n\n\n" # new primary partition 2 over the remaining space
"w\n" # write partition table
)
def decompress_command(image_path: Path) -> list[str]:
"""Build the command that streams the raw image to stdout for dd."""
suffix = image_path.suffix
if suffix == ".zip":
return ["unzip", "-p", str(image_path)]
if suffix == ".gz":
return ["gunzip", "-c", str(image_path)]
if suffix == ".iso":
return ["pv", str(image_path)]
if suffix == ".xz":
return ["unxz", "-c", str(image_path)]
raise LimError(f'Image transfer for "{image_path.name}" is not supported yet!')
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",
]
if plan.luks_memory_cost:
ui.info(
f"Formating {session.root_partition_path} with LUKS "
f"with --pbkdf-memory set to {plan.luks_memory_cost}"
)
luks_format += ["--pbkdf-memory", plan.luks_memory_cost]
else:
ui.info(f"Formating {session.root_partition_path} with LUKS")
runner.run([*luks_format, session.root_partition_path], sudo=True)
def transfer_arch_image(plan: ImagePlan, session: ImageSession) -> None:
boot_size = plan.boot_size or DEFAULT_BOOT_SIZE
ui.info(f"The boot partition will be set to {boot_size}.")
ui.info("Creating partitions...")
runner.run(
["fdisk", session.device.path],
input_text=arch_partition_input(boot_size),
sudo=True,
)
ui.info("Format boot partition...")
runner.run(["mkfs.vfat", session.boot_partition_path], sudo=True)
if plan.encrypt_system:
_luks_format_root(plan, session)
session.decrypt_root()
ui.info("Format root partition...")
runner.run([f"mkfs.{plan.root_filesystem}", "-f", session.root_mapper_path], sudo=True)
session.mount_partitions()
ui.info("Root files will be transfered to device...")
runner.run(
["bsdtar", "-xpf", str(plan.image_path), "-C", str(session.root_mount_path)],
sudo=True,
)
runner.sync_disks()
ui.info("Boot files will be transfered to device...")
boot_source = session.root_mount_path / "boot"
for entry in sorted(boot_source.iterdir()):
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...")
device_module.overwrite_device(session.device)
ui.info("Starting image transfer...")
if plan.distribution == "arch":
transfer_arch_image(plan, session)
return
blocksize = session.device.optimal_blocksize
ui.info(f"Transfering {plan.image_path.suffix} file...")
runner.pipeline(
decompress_command(plan.image_path),
[
"dd",
f"of={session.device.path}",
f"bs={blocksize}",
"conv=fsync",
"status=progress",
],
sudo_last=True,
error_msg=f"DD {plan.image_path} to {session.device.path} failed.",
)
runner.sync_disks()

100
lim/image/verify.py Normal file
View File

@@ -0,0 +1,100 @@
"""Image integrity (checksums) and authenticity (GPG signatures) checks."""
import hashlib
import re
from pathlib import Path
from lim import runner, ui
from lim.errors import LimError
ALGORITHM_BY_DIGEST_LENGTH = {
32: "md5",
40: "sha1",
64: "sha256",
128: "sha512",
}
def url_exists(url: str) -> bool:
return runner.succeeds(["wget", "-q", "--method=HEAD", url])
def resolve_checksum(download_url: str) -> str | None:
"""Try to fetch a published checksum next to the image download."""
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}."
)
if url_exists(checksum_url):
content = runner.output(["wget", checksum_url, "-q", "-O", "-"])
checksum = content.split()[0] if content.split() else ""
if checksum:
ui.info(f"Defined image_checksum as {checksum}")
return checksum
ui.warning(f"No checksum found under {checksum_url}.")
return None
def verify_checksum(image_path: str | Path, checksum: str | None) -> None:
"""Verify the image against an md5/sha1/sha256/sha512 hex checksum."""
if not checksum:
ui.warning("No checksum is defined. Skipping checksum verification.")
return
algorithm = ALGORITHM_BY_DIGEST_LENGTH.get(len(checksum))
if algorithm is None:
raise LimError(
f"Checksum '{checksum}' has no recognized digest length "
"(expected md5, sha1, sha256 or sha512)."
)
ui.info(f"Checking {algorithm} checksum...")
digest = hashlib.new(algorithm)
with Path(image_path).open("rb") as handle:
while chunk := handle.read(1024 * 1024):
digest.update(chunk)
if digest.hexdigest() != checksum.lower():
raise LimError("Verification failed. HINT: Force the download of the image.")
ui.info(f"{algorithm} checksum verified.")
def verify_signature(download_url: str, image_path: str | Path, image_folder: Path) -> None:
"""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."
)
signature_url = f"{download_url}.sig"
ui.info(f"Attempting to download the image signature from: {signature_url}")
if not url_exists(signature_url):
ui.warning(f"No signature found under {signature_url}.")
return
signature_path = image_folder / (Path(str(image_path)).name + ".sig")
if not runner.succeeds(["wget", "-q", "-O", str(signature_path), signature_url]):
ui.warning("Failed to download the signature file.")
return
ui.info("Extract the key ID from the signature file")
verification = runner.output(
["gpg", "--status-fd", "1", "--verify", str(signature_path), str(image_path)],
check=False,
)
missing_keys = re.findall(r"NO_PUBKEY (\S+)", verification)
if missing_keys:
key_id = missing_keys[-1]
if runner.succeeds(["gpg", "--list-keys", key_id]):
ui.info(f"Key {key_id} already in keyring.")
else:
ui.info("Import the public key")
runner.run(
["gpg", "--keyserver", "keyserver.ubuntu.com", "--recv-keys", key_id],
check=False,
)
ui.info("Verify the signature")
if runner.succeeds(["gpg", "--verify", str(signature_path), str(image_path)]):
ui.info("Signature verification succeeded.")
else:
ui.warning("Signature verification failed.")

80
lim/luks.py Normal file
View File

@@ -0,0 +1,80 @@
"""LUKS key management plus crypttab/fstab bookkeeping."""
import re
from pathlib import Path
from lim import fsutil, runner, ui
from lim.errors import LimError
LUKS_KEY_DIRECTORY = Path("/etc/luks-keys")
CRYPTTAB_PATH = Path("/etc/crypttab")
FSTAB_PATH = Path("/etc/fstab")
def luks_uuid(partition_path: str) -> str:
dump = runner.output(["cryptsetup", "luksDump", partition_path], sudo=True)
match = re.search(r"UUID:\s*(\S+)", dump)
if not match:
raise LimError(f"Could not read LUKS UUID of {partition_path}.")
return match.group(1)
def create_luks_key_and_update_crypttab(
mapper_name: str,
partition_path: str,
*,
key_directory: Path = LUKS_KEY_DIRECTORY,
crypttab_path: Path = CRYPTTAB_PATH,
) -> None:
"""Generate a random keyfile, register it with LUKS and /etc/crypttab."""
ui.info("Creating luks-key-directory...")
key_directory.mkdir(parents=True, exist_ok=True)
secret_key_path = key_directory / f"{mapper_name}.keyfile"
ui.info(f"Generate secret key under: {secret_key_path}")
if secret_key_path.exists():
ui.warning("File already exists. Overwriting!")
runner.run(
["dd", "if=/dev/urandom", f"of={secret_key_path}", "bs=512", "count=8"],
sudo=True,
)
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
)
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",
"-v",
"luksOpen",
partition_path,
mapper_name,
f"--key-file={secret_key_path}",
],
sudo=True,
)
runner.run(["cryptsetup", "-v", "luksClose", mapper_name], sudo=True)
ui.info("Reading UUID...")
uuid = luks_uuid(partition_path)
entry = f"{mapper_name} UUID={uuid} {secret_key_path} luks"
ui.info("Adding crypttab entry...")
fsutil.ensure_line_in_file(entry, crypttab_path)
ui.info(f"The file {crypttab_path} contains now the following:")
print(crypttab_path.read_text())
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)
ui.info(f"The file {fstab_path} contains now the following:")
print(fstab_path.read_text())

18
lim/packages.py Normal file
View File

@@ -0,0 +1,18 @@
"""Reads package collections from configuration/packages/."""
from lim import config
from lim.errors import LimError
def get_packages(*collections: str) -> list[str]:
"""Package names from the given collections, comments stripped."""
names: list[str] = []
for collection in collections:
path = config.PACKAGE_PATH / f"{collection}.txt"
if not path.is_file():
raise LimError(f"Package collection {path} does not exist.")
for line in path.read_text().splitlines():
name = line.split("#", 1)[0].strip()
if name:
names.append(name)
return names

134
lim/runner.py Normal file
View File

@@ -0,0 +1,134 @@
"""Thin wrapper around subprocess for all external commands.
Every module calls external tools through this module so tests can
replace ``run``/``output``/``succeeds``/``pipeline`` with fakes.
Missing binaries never surface as tracebacks: actions (``check=True``,
``pipeline``) abort with a clear LimError, tolerated calls
(``check=False``, ``succeeds``) degrade to a warning plus shell-style
exit code 127.
"""
import os
import shlex
import subprocess
from lim import ui
from lim.errors import LimError
COMMAND_NOT_FOUND = 127
def _with_sudo(cmd: list[str], *, sudo: bool) -> list[str]:
parts = [str(part) for part in cmd]
if sudo and os.geteuid() != 0:
return ["sudo", *parts]
return parts
def _not_found_message(cmd: list[str]) -> str:
return f"Command not found: {cmd[0]} — please install it."
def run(
cmd: list[str],
*,
sudo: bool = False,
input_text: str | None = None,
check: bool = True,
error_msg: str | None = None,
) -> subprocess.CompletedProcess:
"""Run a command inheriting stdio so progress output stays visible."""
cmd = _with_sudo(cmd, sudo=sudo)
ui.info(f"Running: {shlex.join(cmd)}")
try:
result = subprocess.run(cmd, input=input_text, text=True, check=False)
except FileNotFoundError:
if check:
raise LimError(error_msg or _not_found_message(cmd)) from None
ui.warning(_not_found_message(cmd))
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)}"
)
return result
def output(
cmd: list[str],
*,
sudo: bool = False,
check: bool = True,
error_msg: str | None = None,
) -> str:
"""Run a command and return its stripped stdout."""
cmd = _with_sudo(cmd, sudo=sudo)
try:
result = subprocess.run(cmd, text=True, capture_output=True, check=False)
except FileNotFoundError:
if check:
raise LimError(error_msg or _not_found_message(cmd)) from None
ui.warning(_not_found_message(cmd))
return ""
if check and result.returncode != 0:
raise LimError(
error_msg
or f"Command failed with code {result.returncode}: {shlex.join(cmd)}"
+ (f"\n{result.stderr.strip()}" if result.stderr else "")
)
return result.stdout.strip()
def succeeds(cmd: list[str], *, sudo: bool = False) -> bool:
"""Run a command silently, report whether it exited with 0."""
cmd = _with_sudo(cmd, sudo=sudo)
try:
return subprocess.run(cmd, capture_output=True, check=False).returncode == 0
except FileNotFoundError:
ui.warning(_not_found_message(cmd))
return False
def pipeline(
*cmds: list[str],
sudo_last: bool = False,
error_msg: str | None = None,
) -> 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)
]
pretty = " | ".join(shlex.join(cmd) for cmd in prepared)
ui.info(f"Running: {pretty}")
processes: list[subprocess.Popen] = []
previous_stdout = None
try:
for index, cmd in enumerate(prepared):
last = index == len(prepared) - 1
process = subprocess.Popen(
cmd,
stdin=previous_stdout,
stdout=None if last else subprocess.PIPE,
)
if previous_stdout is not None:
previous_stdout.close()
previous_stdout = process.stdout
processes.append(process)
except FileNotFoundError as exc:
if previous_stdout is not None:
previous_stdout.close()
for process in processes:
process.kill()
process.wait()
raise LimError(f"Command not found: {exc.filename} — please install it.") from None
for process in processes:
process.wait()
if any(process.returncode != 0 for process in processes):
raise LimError(error_msg or f"Pipeline failed: {pretty}")
def sync_disks() -> None:
run(["sync"])

0
lim/storage/__init__.py Normal file
View File

38
lim/storage/common.py Normal file
View File

@@ -0,0 +1,38 @@
"""Shared path derivation for encrypted storage setups."""
from dataclasses import dataclass
from lim import device as device_module
from lim import ui
from lim.device import Device
@dataclass(frozen=True)
class StorageTarget:
"""An encrypted drive plus all derived mapper/mount/partition paths."""
device: Device
@property
def mapper_name(self) -> str:
return f"encrypteddrive-{self.device.name}"
@property
def mapper_path(self) -> str:
return f"/dev/mapper/{self.mapper_name}"
@property
def mount_path(self) -> str:
return f"/media/{self.mapper_name}"
@property
def partition_path(self) -> str:
return self.device.partition(1)
def select_storage_target() -> StorageTarget:
target = StorageTarget(device_module.select_device())
ui.info(f"mapper name set to : {target.mapper_name}")
ui.info(f"mapper path set to : {target.mapper_path}")
ui.info(f"mount path set to : {target.mount_path}")
return target

68
lim/storage/raid1.py Normal file
View File

@@ -0,0 +1,68 @@
"""Encrypted Btrfs RAID1 across two drives.
See https://balaskas.gr/btrfs/raid1.html and
https://mutschler.eu/linux/install-guides/ubuntu-btrfs-raid1/
"""
from lim import luks, runner, ui
from lim.storage.common import StorageTarget, select_storage_target
def _select_pair() -> tuple[StorageTarget, StorageTarget]:
ui.info("RAID1 partition 1...")
first = select_storage_target()
ui.info("RAID1 partition 2...")
second = select_storage_target()
return first, second
def setup() -> None:
first, second = _select_pair()
ui.info(f"Encrypting {first.device.path}...")
runner.run(["cryptsetup", "luksFormat", first.device.path], sudo=True)
ui.info(f"Encrypting {second.device.path}...")
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", "status", first.mapper_path], sudo=True)
runner.run(["cryptsetup", "status", second.mapper_path], sudo=True)
runner.run(
[
"mkfs.btrfs",
"-m",
"raid1",
"-d",
"raid1",
first.mapper_path,
second.mapper_path,
],
sudo=True,
)
ui.success("Encryption successfull :)")
def _show_luks_devices() -> None:
for name in runner.output(["lsblk", "-dno", "NAME"], check=False).split():
if runner.succeeds(["cryptsetup", "isLuks", f"/dev/{name}"], sudo=True):
ui.info(f"/dev/{name} is a LUKS encrypted storage device.")
def mount_on_boot() -> None:
ui.info("Activate Automount raid1 encrypted storages...")
_show_luks_devices()
first, second = _select_pair()
luks.create_luks_key_and_update_crypttab(first.mapper_name, first.device.path)
ui.info(f'Creating mount folder under "{first.mount_path}"...')
runner.run(["mkdir", "-vp", first.mount_path], sudo=True)
luks.create_luks_key_and_update_crypttab(second.mapper_name, second.device.path)
luks.update_fstab(first.mapper_path, first.mount_path)
ui.success("Installation finished. Please restart :)")

88
lim/storage/single_drive.py Normal file
View File

@@ -0,0 +1,88 @@
"""Single-drive LUKS + Btrfs storage: setup, mount, umount, mount-on-boot."""
from lim import device as device_module
from lim import luks, runner, system, ui
from lim.storage.common import select_storage_target
# fdisk answer sequences; empty lines accept the defaults.
CREATE_GPT_TABLE_INPUT = "g\nw\n"
CREATE_PARTITION_INPUT = "n\n\n\n\np\nw\n"
def setup() -> None:
print("Setups disk encryption")
target = select_storage_target()
device_module.overwrite_device(target.device)
ui.info("Creating new GPT partition table...")
runner.run(
["fdisk", "--wipe", "always", target.device.path],
input_text=CREATE_GPT_TABLE_INPUT,
sudo=True,
)
ui.info("Creating partition table...")
runner.run(
["fdisk", "--wipe", "always", target.device.path],
input_text=CREATE_PARTITION_INPUT,
sudo=True,
)
ui.info(f"Encrypt {target.device.path}...")
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
)
ui.info("Create btrfs file system...")
runner.run(["mkfs.btrfs", target.mapper_path], sudo=True)
ui.info(f'Creating mount folder under "{target.mount_path}"...')
runner.run(["mkdir", "-p", target.mount_path], sudo=True)
ui.info("Mount partition...")
runner.run(["mount", target.mapper_path, target.mount_path], sudo=True)
user = system.real_user()
ui.info("Own partition by user...")
runner.run(["chown", "-R", f"{user}:{user}", target.mount_path], sudo=True)
ui.success("Encryption successfull :)")
def mount() -> None:
print("Mounts encrypted storages")
target = select_storage_target()
ui.info("Unlock partition...")
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)
ui.success("Mounting successfull :)")
def umount() -> None:
print("Unmount encrypted storages")
target = select_storage_target()
ui.info(f"Unmount {target.mapper_path}...")
runner.run(["umount", target.mapper_path], sudo=True)
runner.run(["cryptsetup", "luksClose", target.mapper_name], sudo=True)
ui.success("Successfull :)")
def mount_on_boot() -> None:
print("Automount encrypted storages")
target = select_storage_target()
luks.create_luks_key_and_update_crypttab(target.mapper_name, target.partition_path)
luks.update_fstab(target.mapper_path, target.mount_path)
ui.success("Installation finished. Please restart :)")

35
lim/system.py Normal file
View File

@@ -0,0 +1,35 @@
"""Process-level helpers: privilege handling and user resolution."""
import getpass
import os
import sys
from pathlib import Path
from lim import ui
def is_root() -> bool:
return os.geteuid() == 0
def ensure_root() -> None:
"""Re-execute the current command with sudo when not running as root."""
if is_root():
return
ui.info("Root privileges required. Re-executing with sudo...")
script = str(Path(sys.argv[0]).resolve())
# Deliberate privilege escalation: replace this process with sudo.
os.execvp("sudo", ["sudo", sys.executable, script, *sys.argv[1:]]) # noqa: S606
def real_user() -> str:
"""Return the invoking user, even when running under sudo."""
return os.environ.get("SUDO_USER") or getpass.getuser()
def real_home() -> Path:
"""Home directory of the invoking user, even when running under sudo."""
sudo_user = os.environ.get("SUDO_USER")
if sudo_user:
return Path("/home") / sudo_user
return Path.home()

62
lim/ui.py Normal file
View File

@@ -0,0 +1,62 @@
"""Colored console messages and interactive prompts."""
import sys
_USE_COLOR = sys.stdout.isatty()
def _color(code: str) -> str:
return f"\033[{code}m" if _USE_COLOR else ""
COLOR_RED = _color("31")
COLOR_GREEN = _color("32")
COLOR_YELLOW = _color("33")
COLOR_BLUE = _color("34")
COLOR_MAGENTA = _color("35")
COLOR_CYAN = _color("36")
COLOR_WHITE = _color("37")
COLOR_RESET = _color("0")
def message(color: str, tag: str, text: str) -> None:
print(f"{color}[{tag}]:{COLOR_RESET} {text}")
def question(text: str) -> None:
message(COLOR_MAGENTA, "QUESTION", text)
def info(text: str) -> None:
message(COLOR_BLUE, "INFO", text)
def warning(text: str) -> None:
message(COLOR_YELLOW, "WARNING", text)
def success(text: str) -> None:
message(COLOR_GREEN, "SUCCESS", text)
def error(text: str) -> None:
message(COLOR_RED, "ERROR", text)
def ask(prompt: str) -> str:
question(prompt)
return input().strip()
def confirm(prompt: str) -> bool:
return ask(f"{prompt}(y/N)") == "y"
def header() -> None:
print(
f"\n{COLOR_YELLOW}The\n"
"LINUX IMAGE MANAGER\n"
"is an administration tool designed from and for Kevin Veen-Birkenbach.\n\n"
"Licensed under GNU GENERAL PUBLIC LICENSE Version 3"
f"{COLOR_RESET}\n"
)