refactor

zfs_backup/commands/cryptsetup.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from ._core import CommandInterface
+from ..messages import MAPPER_ENTRY_ALREADY_EXISTS, CANNOT_FIND_DEVICE_PATH
+from ..misc import MAPPER_PATH, DISK_BY_UUID
+
+
+class Cryptsetup(CommandInterface):
+    class _Subcommands:
+        close = "close"
+        status = "status"
+        open = "open"
+
+    _BINARY = Path("/usr/bin/cryptsetup")
+
+    @classmethod
+    def encrypt(cls, mapper_entry: str) -> None:
+        """Close a previously open LUKS container."""
+        cls._run_command(cls._Subcommands.close, mapper_entry)
+
+    @classmethod
+    def _status(cls, name: str) -> str:
+        """Return `cryptsetup status` of a device under `/dev/mapper`"""
+        return cls._get_output(cls._Subcommands.status, name)
+
+    @classmethod
+    def decrypt(cls, path: Path, mapper_entry: str) -> bool:
+        """Decrypt a disk, return whether it was already decrypted."""
+        if cls._mapper_entry_in_use(mapper_entry):
+            if cls._already_decrypted(path, mapper_entry):
+                return True
+            print(MAPPER_ENTRY_ALREADY_EXISTS)
+            raise FileExistsError()
+        cls._run_command(cls._Subcommands.open, str(path), mapper_entry)
+        return False
+
+    @classmethod
+    def _mapper_entry_in_use(cls, mapper_entry: str) -> bool:
+        return (MAPPER_PATH / mapper_entry).exists()
+
+    @classmethod
+    def _get_device_by_uuid_from_status_output(cls, status: str) -> Path:
+        """Given the output of `cryptsetup status [device]`,
+        get the unique name of the disk as path in /dev/disk/by-uuid."""
+        device = cls._get_device_path(status)
+        result = cls._resolve_uuid_of_device(device)
+        return result
+
+    @classmethod
+    def _resolve_uuid_of_device(cls, device):
+        """Given a device like `/dev/sda`, resolve its entry under `/dev/disk/by-uuid`.
+
+        Exit if none can be found."""
+        for item in DISK_BY_UUID.iterdir():
+            if item.readlink() == device:
+                result = item
+                break
+        else:
+            raise FileNotFoundError(
+                f"Could not resolve UUID of cryptsetup device {device}"
+            )
+        return result
+
+    @classmethod
+    def _get_device_path(cls, status) -> Path:
+        """Given the output of `cryptsetup status`, return the path of the device."""
+        for line in status.split("\n"):
+            if line.startswith("  device:"):
+                device = Path(line.split(":")[-1].strip())
+                break
+        else:
+            raise FileNotFoundError(CANNOT_FIND_DEVICE_PATH)
+        # noinspection PyUnboundLocalVariable
+        return device
+
+    @classmethod
+    def _already_decrypted(cls, path: Path, mapper_entry: str) -> bool:
+        """Given a path in `/dev/disk/by-uuid` and the corresponding name under `/dev/mapper`,
+        check if the disk is already decrypted."""
+        decrypted_disk_path = cls._get_device_by_uuid_from_status_output(
+            cls._status(mapper_entry)
+        )
+        return decrypted_disk_path == path