diff --git a/lim/cli.py b/lim/cli.py index 5841dec..e2abd9c 100644 --- a/lim/cli.py +++ b/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, ), } diff --git a/lim/data/sync.py b/lim/data/sync.py index 52cbda1..725dd1d 100644 --- a/lim/data/sync.py +++ b/lim/data/sync.py @@ -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) diff --git a/lim/device.py b/lim/device.py index bd30469..fe04b24 100644 --- a/lim/device.py +++ b/lim/device.py @@ -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: diff --git a/lim/image/choosers.py b/lim/image/choosers.py index f67f575..372e109 100644 --- a/lim/image/choosers.py +++ b/lim/image/choosers.py @@ -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" diff --git a/lim/image/setup.py b/lim/image/setup.py index 6defd89..32be9ed 100644 --- a/lim/image/setup.py +++ b/lim/image/setup.py @@ -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) diff --git a/lim/image/transfer.py b/lim/image/transfer.py index ed98fc8..a9f96ca 100644 --- a/lim/image/transfer.py +++ b/lim/image/transfer.py @@ -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( diff --git a/lim/image/verify.py b/lim/image/verify.py index 6a77e73..447a31b 100644 --- a/lim/image/verify.py +++ b/lim/image/verify.py @@ -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}") diff --git a/lim/luks.py b/lim/luks.py index b7c5202..133db8b 100644 --- a/lim/luks.py +++ b/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) diff --git a/lim/runner.py b/lim/runner.py index d984bae..2912cea 100644 --- a/lim/runner.py +++ b/lim/runner.py @@ -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}") diff --git a/lim/storage/raid1.py b/lim/storage/raid1.py index 0387261..80fb8a1 100644 --- a/lim/storage/raid1.py +++ b/lim/storage/raid1.py @@ -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) diff --git a/lim/storage/single_drive.py b/lim/storage/single_drive.py index 273ae0f..8122fb9 100644 --- a/lim/storage/single_drive.py +++ b/lim/storage/single_drive.py @@ -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) diff --git a/tests/e2e/tor_harness.py b/tests/e2e/tor_harness.py index 8597158..5055559 100644 --- a/tests/e2e/tor_harness.py +++ b/tests/e2e/tor_harness.py @@ -13,7 +13,7 @@ import threading import time from pathlib import Path -# Matches lim.image.tor._KEYGEN_SCRIPT: an empty -f torrc keeps the host's +# Matches lim.image.initramfs.keygen._KEYGEN_SCRIPT: an empty -f torrc keeps the host's # /etc/tor/torrc out of the run, --DisableNetwork 1 makes the keys appear # without ever touching the network. KEYGEN_TIMEOUT = 30 @@ -30,7 +30,7 @@ def free_port() -> int: def generate_onion_keys(onion_dir: Path, data_dir: Path) -> str: """Create v3 onion keys offline; return the .onion hostname. - Mirrors lim.image.tor._KEYGEN_SCRIPT, run directly on the host instead + Mirrors lim.image.initramfs.keygen._KEYGEN_SCRIPT, run directly on the host instead of inside the image chroot (the tor invocation is identical). """ onion_dir.mkdir(parents=True, exist_ok=True) @@ -39,11 +39,23 @@ def generate_onion_keys(onion_dir: Path, data_dir: Path) -> str: empty_torrc.write_text("") process = subprocess.Popen( [ - "tor", "-f", str(empty_torrc), "--DisableNetwork", "1", - "--DataDirectory", str(data_dir), - "--HiddenServiceDir", str(onion_dir), - "--HiddenServicePort", "22 127.0.0.1:22", - "--SocksPort", "0", "--RunAsDaemon", "0", "--Log", "notice stderr", + "tor", + "-f", + str(empty_torrc), + "--DisableNetwork", + "1", + "--DataDirectory", + str(data_dir), + "--HiddenServiceDir", + str(onion_dir), + "--HiddenServicePort", + "22 127.0.0.1:22", + "--SocksPort", + "0", + "--RunAsDaemon", + "0", + "--Log", + "notice stderr", ], stderr=subprocess.DEVNULL, ) @@ -62,9 +74,7 @@ def generate_onion_keys(onion_dir: Path, data_dir: Path) -> str: return hostname.read_text().strip() -def start_onion_service( - torrc_path: Path, socks_port: int, log_path: Path -) -> subprocess.Popen: +def start_onion_service(torrc_path: Path, socks_port: int, log_path: Path) -> subprocess.Popen: """Start Tor with the given torrc (service side) plus a SOCKS port. The torrc carries the HiddenServiceDir/HiddenServicePort just like the @@ -73,9 +83,13 @@ def start_onion_service( """ return subprocess.Popen( [ - "tor", "-f", str(torrc_path), - "--SocksPort", str(socks_port), - "--Log", f"notice file {log_path}", + "tor", + "-f", + str(torrc_path), + "--SocksPort", + str(socks_port), + "--Log", + f"notice file {log_path}", ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, @@ -98,11 +112,7 @@ def _socks5_handshake(sock: socket.socket, onion_host: str, onion_port: int) -> if sock.recv(2) != b"\x05\x00": raise RuntimeError("SOCKS5 handshake rejected.") host = onion_host.encode() - request = ( - b"\x05\x01\x00\x03" - + bytes([len(host)]) + host - + struct.pack(">H", onion_port) - ) + request = b"\x05\x01\x00\x03" + bytes([len(host)]) + host + struct.pack(">H", onion_port) sock.settimeout(ONION_CONNECT_TIMEOUT) sock.sendall(request) header = sock.recv(4) diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index cad7663..7b12269 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -24,9 +24,7 @@ def test_every_command_dispatches_to_its_function(monkeypatch): def test_needs_root_command_requests_root(monkeypatch): escalated = [] monkeypatch.setattr(system, "ensure_root", lambda: escalated.append(True)) - monkeypatch.setitem( - cli.COMMANDS, "backup", Command(lambda: None, "d", needs_root=True) - ) + monkeypatch.setitem(cli.COMMANDS, "backup", Command(lambda: None, "d", needs_root=True)) cli.main(["--type", "backup", "--auto-confirm"]) assert escalated == [True] diff --git a/tests/unit/test_storage.py b/tests/unit/test_storage.py index 8f7cab6..2d33068 100644 --- a/tests/unit/test_storage.py +++ b/tests/unit/test_storage.py @@ -20,9 +20,7 @@ def test_storage_target_derives_all_paths(): assert target.partition_path == "/dev/sdb1" -def test_single_drive_setup_command_sequence( - fake_runner, answers, block_devices, monkeypatch -): +def test_single_drive_setup_command_sequence(fake_runner, answers, block_devices, monkeypatch): monkeypatch.setattr("lim.system.real_user", lambda: "kevin") answers("sdb", "N") # device name, skip overwrite diff --git a/tests/unit/test_sync.py b/tests/unit/test_sync.py index 921aeea..2f886fd 100644 --- a/tests/unit/test_sync.py +++ b/tests/unit/test_sync.py @@ -58,10 +58,19 @@ def test_rsync_command_uses_delete_only_for_directories(): directory_op = sync.SyncOperation("src/", "dst/", "backup", is_directory=True) file_op = sync.SyncOperation("src", "dst", "backup", is_directory=False) assert sync.rsync_command(directory_op) == [ - "rsync", "-abcEPuvW", "--delete", "--backup-dir=backup", "src/", "dst/" + "rsync", + "-abcEPuvW", + "--delete", + "--backup-dir=backup", + "src/", + "dst/", ] assert sync.rsync_command(file_op) == [ - "rsync", "-abcEPuvW", "--backup-dir=backup", "src", "dst" + "rsync", + "-abcEPuvW", + "--backup-dir=backup", + "src", + "dst", ] @@ -95,18 +104,12 @@ def test_export_chowns_real_home_without_sudo(tmp_path, fake_runner, monkeypatch sync.export_to_system() - chown_calls = [ - (cmd, sudo) for cmd, sudo in fake_runner.sudo_log if cmd[0] == "chown" - ] - assert chown_calls == [ - (["chown", "-R", "kevin:kevin", str(real_home)], False) - ] + chown_calls = [(cmd, sudo) for cmd, sudo in fake_runner.sudo_log if cmd[0] == "chown"] + assert chown_calls == [(["chown", "-R", "kevin:kevin", str(real_home)], False)] assert (real_home / ".ssh/id_rsa").stat().st_mode & 0o777 == 0o600 -def test_execute_sync_plan_creates_folders_and_runs_rsync( - tmp_path, fake_runner -): +def test_execute_sync_plan_creates_folders_and_runs_rsync(tmp_path, fake_runner): operation = sync.SyncOperation( source=str(tmp_path / "src.txt"), destination=str(tmp_path / "deep/nested/dst.txt"),