Loading poolpay/vault/Vault.py 0 → 100644 +112 −0 Original line number Diff line number Diff line from dataclasses import dataclass from pathlib import Path from subprocess import run from uuid import uuid4 @dataclass(frozen=True) class Vault: image: Path mountpoint: Path # The name is not part of the interface, but we need to remember it during command execution, because it will be # identified by the operating system using this name _name: str = f"poolpay-vault-{uuid4()}" @property def _mapper(self) -> Path: """ :return: the path to the device mapper under which the vault should be mapped """ return Path("/dev/mapper") / self._name def create(self, password: str, size_mb: int = 64) -> None: image_absolute = self.image.resolve() if image_absolute.exists(): raise FileExistsError( f"{image_absolute} already exists, not overwriting it." ) # Follow instructions from https://opensource.com/article/21/4/linux-encryption run( [ "dd", "if=/dev/urandom", f"of={image_absolute}", "bs=1M", f"count={size_mb}", ], check=True, ) run( ["cryptsetup", "luksFormat", f"{image_absolute}", "-"], input=password.encode("utf-8"), check=True, ) self._cryptsetup_open(password) run(["sudo", "mkfs.ext4", "-L", "poolpay-vault", f"{self._mapper}"], check=True) self._cryptsetup_close() def _cryptsetup_open(self, password: str) -> None: """ Use `cryptsetup open` to open the vault and create a device mapper for it. :raise FileNotFoundError: if after opening attempt there is no mapper file, i.e. opening failed """ run( [ "sudo", "cryptsetup", "open", "--type", "luks", f"{self.image.resolve()}", self._name, "--key-file", "-", ], input=password.encode("utf-8"), check=True, ) if not self._mapper.exists(): raise FileNotFoundError( f"_cryptsetup_open failed, {self._mapper} should now be there but isn't." ) def _cryptsetup_close(self) -> None: """ Use `crypsetup close` to close the vault. :raise FileNotFoundError: if after closing attempt there still is a mapper file, i.e. closing failed """ run(["sudo", "cryptsetup", "close", self._name], check=True) if self._mapper.exists(): raise FileExistsError( f"_cryptsetup_close failed, {self._mapper} shouldn't be there anymore but still was." ) def _mount(self) -> None: self.mountpoint.mkdir(exist_ok=True) run( ["sudo", "mount", f"{self._mapper}", f"{self.mountpoint.resolve()}"], check=True, ) def _umount(self) -> None: run(["sudo", "umount", f"{self.mountpoint.resolve()}"], check=True) def open(self, password: str) -> None: self._cryptsetup_open(password) self._mount() def close(self) -> None: self._umount() self._cryptsetup_close() @property def is_open(self) -> bool: # TODO Parse the output of "mount" to see if our mapper is mounted at our mountpoint raise NotImplementedError() poolpay/vault/__init__.py 0 → 100644 +0 −0 Empty file added. Loading
poolpay/vault/Vault.py 0 → 100644 +112 −0 Original line number Diff line number Diff line from dataclasses import dataclass from pathlib import Path from subprocess import run from uuid import uuid4 @dataclass(frozen=True) class Vault: image: Path mountpoint: Path # The name is not part of the interface, but we need to remember it during command execution, because it will be # identified by the operating system using this name _name: str = f"poolpay-vault-{uuid4()}" @property def _mapper(self) -> Path: """ :return: the path to the device mapper under which the vault should be mapped """ return Path("/dev/mapper") / self._name def create(self, password: str, size_mb: int = 64) -> None: image_absolute = self.image.resolve() if image_absolute.exists(): raise FileExistsError( f"{image_absolute} already exists, not overwriting it." ) # Follow instructions from https://opensource.com/article/21/4/linux-encryption run( [ "dd", "if=/dev/urandom", f"of={image_absolute}", "bs=1M", f"count={size_mb}", ], check=True, ) run( ["cryptsetup", "luksFormat", f"{image_absolute}", "-"], input=password.encode("utf-8"), check=True, ) self._cryptsetup_open(password) run(["sudo", "mkfs.ext4", "-L", "poolpay-vault", f"{self._mapper}"], check=True) self._cryptsetup_close() def _cryptsetup_open(self, password: str) -> None: """ Use `cryptsetup open` to open the vault and create a device mapper for it. :raise FileNotFoundError: if after opening attempt there is no mapper file, i.e. opening failed """ run( [ "sudo", "cryptsetup", "open", "--type", "luks", f"{self.image.resolve()}", self._name, "--key-file", "-", ], input=password.encode("utf-8"), check=True, ) if not self._mapper.exists(): raise FileNotFoundError( f"_cryptsetup_open failed, {self._mapper} should now be there but isn't." ) def _cryptsetup_close(self) -> None: """ Use `crypsetup close` to close the vault. :raise FileNotFoundError: if after closing attempt there still is a mapper file, i.e. closing failed """ run(["sudo", "cryptsetup", "close", self._name], check=True) if self._mapper.exists(): raise FileExistsError( f"_cryptsetup_close failed, {self._mapper} shouldn't be there anymore but still was." ) def _mount(self) -> None: self.mountpoint.mkdir(exist_ok=True) run( ["sudo", "mount", f"{self._mapper}", f"{self.mountpoint.resolve()}"], check=True, ) def _umount(self) -> None: run(["sudo", "umount", f"{self.mountpoint.resolve()}"], check=True) def open(self, password: str) -> None: self._cryptsetup_open(password) self._mount() def close(self) -> None: self._umount() self._cryptsetup_close() @property def is_open(self) -> bool: # TODO Parse the output of "mount" to see if our mapper is mounted at our mountpoint raise NotImplementedError()
