style: apply ruff format across the tree; refresh moved-module doc refs
Bring 14 files that predated the ruff-format run into line with the configured formatter (line-length 100); provably formatting-only (ruff format of HEAD == working tree, and ruff format is semantics-preserving). Also refresh two lim.image.tor._KEYGEN_SCRIPT doc references in tor_harness.py to lim.image.initramfs.keygen after the module moved. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
18
lim/cli.py
18
lim/cli.py
@@ -59,14 +59,12 @@ COMMANDS: dict[str, Command] = {
|
||||
),
|
||||
"mount": Command(
|
||||
single_drive.mount,
|
||||
"Mount Encrypted Storage:\n"
|
||||
" - Unlocks a LUKS partition and mounts it.",
|
||||
"Mount Encrypted Storage:\n - Unlocks a LUKS partition and mounts it.",
|
||||
needs_root=True,
|
||||
),
|
||||
"umount": Command(
|
||||
single_drive.umount,
|
||||
"Unmount Encrypted Storage:\n"
|
||||
" - Unmounts a LUKS partition and closes the mapper.",
|
||||
"Unmount Encrypted Storage:\n - Unmounts a LUKS partition and closes the mapper.",
|
||||
needs_root=True,
|
||||
),
|
||||
"single-boot": Command(
|
||||
@@ -83,26 +81,22 @@ COMMANDS: dict[str, Command] = {
|
||||
),
|
||||
"unlock": Command(
|
||||
crypt.unlock,
|
||||
"Unlock Data:\n"
|
||||
" - Decrypts the encfs data store into the decrypted folder.",
|
||||
"Unlock Data:\n - Decrypts the encfs data store into the decrypted folder.",
|
||||
needs_root=False,
|
||||
),
|
||||
"lock": Command(
|
||||
crypt.lock,
|
||||
"Lock Data:\n"
|
||||
" - Unmounts the decrypted encfs folder.",
|
||||
"Lock Data:\n - Unmounts the decrypted encfs folder.",
|
||||
needs_root=False,
|
||||
),
|
||||
"import": Command(
|
||||
sync.import_from_system,
|
||||
"Import Data:\n"
|
||||
" - Copies personal data from the system into the encrypted store.",
|
||||
"Import Data:\n - Copies personal data from the system into the encrypted store.",
|
||||
needs_root=False,
|
||||
),
|
||||
"export": Command(
|
||||
sync.export_to_system,
|
||||
"Export Data:\n"
|
||||
" - Copies personal data from the encrypted store back to the system.",
|
||||
"Export Data:\n - Copies personal data from the encrypted store back to the system.",
|
||||
needs_root=False,
|
||||
),
|
||||
}
|
||||
|
||||
@@ -90,31 +90,22 @@ def execute_sync_plan(operations: list[SyncOperation]) -> None:
|
||||
if not operation.is_directory and Path(operation.destination).is_file():
|
||||
ui.info("The destination file already exists!")
|
||||
ui.info("Difference:")
|
||||
runner.run(
|
||||
["diff", operation.destination, operation.source], check=False
|
||||
)
|
||||
runner.run(["diff", operation.destination, operation.source], check=False)
|
||||
runner.run(rsync_command(operation))
|
||||
|
||||
|
||||
def ensure_unlocked() -> None:
|
||||
mounts = runner.output(["mount"], check=False)
|
||||
if str(config.DECRYPTED_PATH) not in mounts:
|
||||
ui.info(
|
||||
f"The decrypted folder {config.DECRYPTED_PATH} is locked. "
|
||||
"You need to unlock it!"
|
||||
)
|
||||
ui.info(f"The decrypted folder {config.DECRYPTED_PATH} is locked. You need to unlock it!")
|
||||
crypt.unlock()
|
||||
|
||||
|
||||
def import_from_system(mode: str = "import") -> None:
|
||||
ensure_unlocked()
|
||||
backup_folder = (
|
||||
config.BACKUP_PATH / mode / time.strftime("%Y%m%d%H%M%S")
|
||||
)
|
||||
backup_folder = config.BACKUP_PATH / mode / time.strftime("%Y%m%d%H%M%S")
|
||||
backup_folder.mkdir(parents=True, exist_ok=True)
|
||||
operations = build_sync_plan(
|
||||
mode, system.real_home(), config.DATA_PATH, backup_folder
|
||||
)
|
||||
operations = build_sync_plan(mode, system.real_home(), config.DATA_PATH, backup_folder)
|
||||
execute_sync_plan(operations)
|
||||
|
||||
|
||||
|
||||
@@ -119,9 +119,7 @@ def overwrite_device(device: Device) -> None:
|
||||
|
||||
def blkid_value(path: str, tag: str) -> str:
|
||||
"""Return a blkid tag value ("TYPE", "UUID", ...) or "" when unavailable."""
|
||||
return runner.output(
|
||||
["blkid", path, "-s", tag, "-o", "value"], sudo=True, check=False
|
||||
)
|
||||
return runner.output(["blkid", path, "-s", tag, "-o", "value"], sudo=True, check=False)
|
||||
|
||||
|
||||
def is_mounted(path_fragment: str) -> bool:
|
||||
|
||||
@@ -44,9 +44,7 @@ def choose_manjaro(plan: ImagePlan) -> None:
|
||||
def choose_moode(plan: ImagePlan) -> None:
|
||||
plan.boot_size = "+200M"
|
||||
plan.image_checksum = "185cbc9a4994534bb7a4bc2744c78197"
|
||||
plan.base_download_url = (
|
||||
"https://github.com/moode-player/moode/releases/download/r651prod/"
|
||||
)
|
||||
plan.base_download_url = "https://github.com/moode-player/moode/releases/download/r651prod/"
|
||||
plan.image_name = "moode-r651-iso.zip"
|
||||
|
||||
|
||||
@@ -57,9 +55,7 @@ def choose_retropie(plan: ImagePlan) -> None:
|
||||
if entry is None:
|
||||
raise LimError(f"Version {version} isn't supported.")
|
||||
plan.raspberry_pi_version = version
|
||||
plan.base_download_url = (
|
||||
"https://github.com/RetroPie/RetroPie-Setup/releases/download/4.8/"
|
||||
)
|
||||
plan.base_download_url = "https://github.com/RetroPie/RetroPie-Setup/releases/download/4.8/"
|
||||
plan.image_checksum = entry["checksum"]
|
||||
plan.image_name = entry["image"]
|
||||
|
||||
@@ -67,9 +63,7 @@ def choose_retropie(plan: ImagePlan) -> None:
|
||||
def choose_torbox(plan: ImagePlan) -> None:
|
||||
plan.base_download_url = "https://www.torbox.ch/data/"
|
||||
plan.image_name = "torbox-20220102-v050.gz"
|
||||
plan.image_checksum = (
|
||||
"0E1BA7FFD14AAAE5F0462C8293D95B62C3BF1D9E726E26977BD04772C55680D3"
|
||||
)
|
||||
plan.image_checksum = "0E1BA7FFD14AAAE5F0462C8293D95B62C3BF1D9E726E26977BD04772C55680D3"
|
||||
plan.boot_size = "+200M"
|
||||
|
||||
|
||||
@@ -78,9 +72,7 @@ def choose_android_x86(plan: ImagePlan) -> None:
|
||||
"https://www.fosshub.com/Android-x86.html?dwl=android-x86_64-9.0-r2.iso"
|
||||
)
|
||||
plan.image_name = "android-x86_64-9.0-r2.iso"
|
||||
plan.image_checksum = (
|
||||
"f7eb8fc56f29ad5432335dc054183acf086c539f3990f0b6e9ff58bd6df4604e"
|
||||
)
|
||||
plan.image_checksum = "f7eb8fc56f29ad5432335dc054183acf086c539f3990f0b6e9ff58bd6df4604e"
|
||||
plan.boot_size = "+500M"
|
||||
|
||||
|
||||
|
||||
@@ -36,8 +36,7 @@ def _select_unmounted_device() -> device_module.Device:
|
||||
device = device_module.select_device()
|
||||
if device_module.is_mounted(device.path):
|
||||
raise LimError(
|
||||
f'Device {device.path} is allready mounted. '
|
||||
f'Umount with "umount {device.path}*".'
|
||||
f'Device {device.path} is allready mounted. Umount with "umount {device.path}*".'
|
||||
)
|
||||
return device
|
||||
|
||||
@@ -83,9 +82,7 @@ def run_setup() -> None:
|
||||
session.make_working_folder()
|
||||
session.make_mount_folders()
|
||||
|
||||
plan.root_filesystem = ui.ask(
|
||||
"Which filesystem should be used? E.g.:btrfs,ext4... (none):"
|
||||
)
|
||||
plan.root_filesystem = ui.ask("Which filesystem should be used? E.g.:btrfs,ext4... (none):")
|
||||
|
||||
if ui.confirm(f"Should the image be transfered to {device.path}?"):
|
||||
transfer.transfer_image(plan, session)
|
||||
|
||||
@@ -15,10 +15,7 @@ def download_image(plan: ImagePlan) -> None:
|
||||
ui.info(f"Removing image {plan.image_path}.")
|
||||
plan.image_path.unlink()
|
||||
else:
|
||||
ui.info(
|
||||
"Forcing download wasn't neccessary. "
|
||||
f"File {plan.image_path} doesn't exist."
|
||||
)
|
||||
ui.info(f"Forcing download wasn't neccessary. File {plan.image_path} doesn't exist.")
|
||||
|
||||
ui.info("Start Download procedure...")
|
||||
if plan.image_path.is_file():
|
||||
@@ -35,13 +32,13 @@ def download_image(plan: ImagePlan) -> None:
|
||||
def arch_partition_input(boot_size: str) -> str:
|
||||
"""Fdisk answers: FAT32 boot partition of boot_size plus root partition."""
|
||||
return (
|
||||
"o\n" # clear out any partitions on the drive
|
||||
"p\n" # list partitions (should be empty)
|
||||
"o\n" # clear out any partitions on the drive
|
||||
"p\n" # list partitions (should be empty)
|
||||
"n\np\n1\n\n" # new primary partition 1, default start sector
|
||||
f"{boot_size}\n"
|
||||
"t\nc\n" # set partition 1 to type W95 FAT32 (LBA)
|
||||
"t\nc\n" # set partition 1 to type W95 FAT32 (LBA)
|
||||
"n\np\n2\n\n\n" # new primary partition 2 over the remaining space
|
||||
"w\n" # write partition table
|
||||
"w\n" # write partition table
|
||||
)
|
||||
|
||||
|
||||
@@ -61,9 +58,18 @@ def decompress_command(image_path: Path) -> list[str]:
|
||||
|
||||
def _luks_format_root(plan: ImagePlan, session: ImageSession) -> None:
|
||||
luks_format = [
|
||||
"cryptsetup", "-v", "luksFormat",
|
||||
"-c", "aes-xts-plain64", "-s", "512", "-h", "sha512",
|
||||
"--use-random", "-i", "1000",
|
||||
"cryptsetup",
|
||||
"-v",
|
||||
"luksFormat",
|
||||
"-c",
|
||||
"aes-xts-plain64",
|
||||
"-s",
|
||||
"512",
|
||||
"-h",
|
||||
"sha512",
|
||||
"--use-random",
|
||||
"-i",
|
||||
"1000",
|
||||
]
|
||||
if plan.luks_memory_cost:
|
||||
ui.info(
|
||||
|
||||
@@ -24,8 +24,7 @@ def resolve_checksum(download_url: str) -> str | None:
|
||||
for extension in ("sha1", "sha512", "md5"):
|
||||
checksum_url = f"{download_url}.{extension}"
|
||||
ui.info(
|
||||
"Image Checksum is not defined. "
|
||||
f"Try to download image signature from {checksum_url}."
|
||||
f"Image Checksum is not defined. Try to download image signature from {checksum_url}."
|
||||
)
|
||||
if url_exists(checksum_url):
|
||||
content = runner.output(["wget", checksum_url, "-q", "-O", "-"])
|
||||
@@ -62,8 +61,7 @@ def verify_signature(download_url: str, image_path: str | Path, image_folder: Pa
|
||||
"""Best-effort GPG signature verification of the downloaded image."""
|
||||
ui.info("Note: Checksums verify integrity but do not confirm authenticity.")
|
||||
ui.info(
|
||||
"Proceeding to signature verification, "
|
||||
"which ensures the file comes from a trusted source."
|
||||
"Proceeding to signature verification, which ensures the file comes from a trusted source."
|
||||
)
|
||||
signature_url = f"{download_url}.sig"
|
||||
ui.info(f"Attempting to download the image signature from: {signature_url}")
|
||||
|
||||
12
lim/luks.py
12
lim/luks.py
@@ -40,14 +40,10 @@ def create_luks_key_and_update_crypttab(
|
||||
runner.sync_disks()
|
||||
|
||||
ui.info("Opening and closing device to verify that everything works fine...")
|
||||
closed = runner.run(
|
||||
["cryptsetup", "-v", "luksClose", mapper_name], sudo=True, check=False
|
||||
)
|
||||
closed = runner.run(["cryptsetup", "-v", "luksClose", mapper_name], sudo=True, check=False)
|
||||
if closed.returncode != 0:
|
||||
ui.info(f"No need to luksClose {mapper_name}. Device isn't open.")
|
||||
runner.run(
|
||||
["cryptsetup", "luksAddKey", partition_path, str(secret_key_path)], sudo=True
|
||||
)
|
||||
runner.run(["cryptsetup", "luksAddKey", partition_path, str(secret_key_path)], sudo=True)
|
||||
runner.run(
|
||||
[
|
||||
"cryptsetup",
|
||||
@@ -70,9 +66,7 @@ def create_luks_key_and_update_crypttab(
|
||||
print(crypttab_path.read_text())
|
||||
|
||||
|
||||
def update_fstab(
|
||||
mapper_path: str, mount_path: str, *, fstab_path: Path = FSTAB_PATH
|
||||
) -> None:
|
||||
def update_fstab(mapper_path: str, mount_path: str, *, fstab_path: Path = FSTAB_PATH) -> None:
|
||||
entry = f"{mapper_path} {mount_path} btrfs defaults 0 2"
|
||||
ui.info("Adding fstab entry...")
|
||||
fsutil.ensure_line_in_file(entry, fstab_path)
|
||||
|
||||
@@ -50,8 +50,7 @@ def run(
|
||||
return subprocess.CompletedProcess(cmd, COMMAND_NOT_FOUND)
|
||||
if check and result.returncode != 0:
|
||||
raise LimError(
|
||||
error_msg
|
||||
or f"Command failed with code {result.returncode}: {shlex.join(cmd)}"
|
||||
error_msg or f"Command failed with code {result.returncode}: {shlex.join(cmd)}"
|
||||
)
|
||||
return result
|
||||
|
||||
@@ -98,8 +97,7 @@ def pipeline(
|
||||
) -> None:
|
||||
"""Run commands as a shell-style pipeline (cmd1 | cmd2 | ...)."""
|
||||
prepared = [
|
||||
_with_sudo(cmd, sudo=sudo_last and index == len(cmds) - 1)
|
||||
for index, cmd in enumerate(cmds)
|
||||
_with_sudo(cmd, sudo=sudo_last and index == len(cmds) - 1) for index, cmd in enumerate(cmds)
|
||||
]
|
||||
pretty = " | ".join(shlex.join(cmd) for cmd in prepared)
|
||||
ui.info(f"Running: {pretty}")
|
||||
|
||||
@@ -25,9 +25,7 @@ def setup() -> None:
|
||||
runner.run(["cryptsetup", "luksFormat", second.device.path], sudo=True)
|
||||
|
||||
runner.run(["cryptsetup", "luksOpen", first.device.path, first.mapper_name], sudo=True)
|
||||
runner.run(
|
||||
["cryptsetup", "luksOpen", second.device.path, second.mapper_name], sudo=True
|
||||
)
|
||||
runner.run(["cryptsetup", "luksOpen", second.device.path, second.mapper_name], sudo=True)
|
||||
runner.run(["cryptsetup", "status", first.mapper_path], sudo=True)
|
||||
runner.run(["cryptsetup", "status", second.mapper_path], sudo=True)
|
||||
|
||||
|
||||
@@ -32,9 +32,7 @@ def setup() -> None:
|
||||
runner.run(["cryptsetup", "-v", "-y", "luksFormat", target.partition_path], sudo=True)
|
||||
|
||||
ui.info("Unlock partition...")
|
||||
runner.run(
|
||||
["cryptsetup", "luksOpen", target.partition_path, target.mapper_name], sudo=True
|
||||
)
|
||||
runner.run(["cryptsetup", "luksOpen", target.partition_path, target.mapper_name], sudo=True)
|
||||
|
||||
ui.info("Create btrfs file system...")
|
||||
runner.run(["mkfs.btrfs", target.mapper_path], sudo=True)
|
||||
@@ -57,9 +55,7 @@ def mount() -> None:
|
||||
target = select_storage_target()
|
||||
|
||||
ui.info("Unlock partition...")
|
||||
runner.run(
|
||||
["cryptsetup", "luksOpen", target.partition_path, target.mapper_name], sudo=True
|
||||
)
|
||||
runner.run(["cryptsetup", "luksOpen", target.partition_path, target.mapper_name], sudo=True)
|
||||
|
||||
ui.info("Mount partition...")
|
||||
runner.run(["mount", target.mapper_path, target.mount_path], sudo=True)
|
||||
|
||||
Reference in New Issue
Block a user