Commit aced0ea7 authored by Jakob Moser's avatar Jakob Moser
Browse files

Merge branch 'backend/improve-vault' into 'master'

Improve Vault

See merge request moser/poolpay!19
parents 4062e235 a5092861
Loading
Loading
Loading
Loading
+25 −3
Original line number Diff line number Diff line
from contextlib import AbstractContextManager
from dataclasses import dataclass
from os import getgid, getuid
from pathlib import Path
from subprocess import CalledProcessError, run
from typing import Any, Self
from uuid import uuid4


@dataclass(frozen=True)
class Vault:
class Vault(AbstractContextManager):
    image: Path
    mountpoint: Path

@@ -21,10 +23,14 @@ class Vault:
        """
        return Path("/dev/mapper") / self._name

    @property
    def exists(self) -> bool:
        return self.image.exists()

    def create(self, password: str, size_mb: int = 64) -> None:
        image_absolute = self.image.resolve()

        if image_absolute.exists():
        if self.exists:
            raise FileExistsError(
                f"{image_absolute} already exists, not overwriting it."
            )
@@ -123,14 +129,30 @@ class Vault:
        except CalledProcessError:
            return False

    def open(self, password: str) -> None:
    def open(self, password: str, create_if_not_exists: bool = True) -> Self:
        """
        Open the vault (creating it if desired and necessary) and return self, i.e., the instance.

        This method returns the Vault instance so that it can be used neatly in a `with` statement.
        """
        if create_if_not_exists and not self.exists:
            self.create(password)

        self._cryptsetup_open(password)
        self._mount()

        return self

    def close(self) -> None:
        self._umount()
        self._cryptsetup_close()

    def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
        """
        Ignore any potentially passed exceptions and close the vault.
        """
        self.close()

    @property
    def is_open(self) -> bool:
        return self._is_mounted