Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions projects/butter-backup/examples/config.json5
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
// JSON5 allows comments and trailing commas
{
SudoPassCmd: "gpg --decrypt ~/.local/share/passwords/sudo-password.gpg",
// OpenDirectory: "/media/ButterBackup", // optional: where to open devices
DeviceConfigurations: [
{
Name: "BtrFS Backup Example",
Expand Down
1 change: 1 addition & 0 deletions projects/butter-backup/examples/config.toml
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

[butter-backup]
SudoPassCmd = "gpg --decrypt ~/.local/share/passwords/sudo-password.gpg"
# OpenDirectory = "/media/ButterBackup" # optional: where to open devices

[[butter-backup.device-configurations]]
Name = "BtrFS Backup Example"
Expand Down
1 change: 1 addition & 0 deletions projects/butter-backup/examples/config.yaml
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
---
SudoPassCmd: "gpg --decrypt ~/.local/share/passwords/sudo-password.gpg"
# OpenDirectory: "/media/ButterBackup" # optional: where to open devices
DeviceConfigurations:
- Name: "BtrFS Backup Example"
UUID: "12345678-1234-5678-1234-567812345678"
Expand Down
25 changes: 16 additions & 9 deletions projects/butter-backup/src/butter_backup/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -141,9 +141,10 @@ def _open_device(
cfg: cp.DeviceConfiguration, base_dir: Path, sudo_pass_cmd: str | None
) -> None:
mount_dir = base_dir / cfg.Name
mount_dir.mkdir(exist_ok=True)
created_mount_dir = False
try:
_refresh_sudo(sudo_pass_cmd)
created_mount_dir = sdm.ensure_directory(mount_dir)
decrypted = sdm.open_encrypted_device(cfg.device(), cfg.DevicePassCmd)
sdm.mount_device(decrypted, mount_dir=mount_dir, compression=cfg.compression())
except Exception:
Expand All @@ -153,15 +154,16 @@ def _open_device(
typer.echo(
f"Speichermedium {cfg.Name} konnte nicht geöffnet werden. Es wird übersprungen."
)
with contextlib.suppress(OSError):
mount_dir.rmdir()
if created_mount_dir:
with contextlib.suppress(sh.ShellInterfaceError):
cmd: sh.StrPathList = ["sudo", "rmdir", mount_dir]
sh.run_cmd(cmd=cmd)
else:
typer.echo(f"Speichermedium {cfg.Name} wurde in {mount_dir} geöffnet.")


@app.command()
def open( # noqa: A001
dest: Path | None = typer.Argument(None), # noqa: B008
config: Path | None = CONFIG_OPTION,
verbose: int = VERBOSITY_OPTION,
) -> None:
Expand All @@ -179,13 +181,14 @@ def open( # noqa: A001
oder durch Verwendung von `restic`. Nach erfolgreicher Wiederherstellung
kann das Speichermedium mit `butter-backup close` wieder entfernt werden.

Optional kann ein Zielverzeichnis angegeben werden. Wenn angegeben, werden
die Speichermedien in Unterverzeichnissen dieses Verzeichnisses gemountet.
Andernfalls wird ein temporäres Verzeichnis erstellt.
Das Zielverzeichnis für die Speichermedien kann in der Konfiguration über
das Feld `OpenDirectory` festgelegt werden. Wenn nicht angegeben, wird ein
temporäres Verzeichnis erstellt.
"""
setup_logging(verbose)
parsed_config = _read_configuration(config)
base_dir = dest if dest is not None else Path(mkdtemp())
open_dir = parsed_config.OpenDirectory
base_dir = open_dir if open_dir is not None else Path(mkdtemp())
for cfg in parsed_config.DeviceConfigurations:
if _skip_device(
cfg,
Expand Down Expand Up @@ -265,9 +268,13 @@ def backup(
continue
backend = bb.BackupBackend.from_config(cfg)
_refresh_sudo(parsed_config.SudoPassCmd)
open_dir = parsed_config.OpenDirectory
dest = open_dir / cfg.Name if open_dir is not None else None
with (
sdm.decrypted_device(cfg.device(), cfg.DevicePassCmd) as decrypted,
sdm.mounted_device(decrypted, compression=cfg.compression()) as mount_dir,
sdm.mounted_device(
decrypted, dest, compression=cfg.compression()
) as mount_dir,
):
backend.do_backup(mount_dir, parsed_config.SudoPassCmd)
# A backup could take so long that the sudo session expires. In this
Expand Down
17 changes: 16 additions & 1 deletion projects/butter-backup/src/butter_backup/config_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -183,8 +183,16 @@ def compression(self) -> None:
class Configuration(BaseModel):
model_config = ConfigDict(frozen=True)
DeviceConfigurations: list[DeviceConfiguration]
OpenDirectory: Path | None = None
SudoPassCmd: str | None = None

@field_validator("OpenDirectory", mode="before")
@classmethod
def expand_tilde_in_open_directory(cls, v: str | None) -> str | None:
if v is None:
return None
return str(Path(v).expanduser())

@model_validator(mode="after")
def check_unique_names(self) -> t.Self:
name_counts = Counter(cfg.Name for cfg in self.DeviceConfigurations)
Expand All @@ -196,6 +204,9 @@ def check_unique_names(self) -> t.Self:
return self


_TOML_DEVICE_CONFIGURATIONS_KEY = "device-configurations"


def _parse_as_json(content: str) -> Any:
return json.loads(content)

Expand All @@ -207,7 +218,11 @@ def _parse_as_json5(content: str) -> Any:
def _parse_as_toml(content: str) -> Any:
data = tomllib.loads(content)
bb = data["butter-backup"]
return {"DeviceConfigurations": bb["device-configurations"]}
result: dict[str, Any] = {
k: v for k, v in bb.items() if k != _TOML_DEVICE_CONFIGURATIONS_KEY
}
result["DeviceConfigurations"] = bb[_TOML_DEVICE_CONFIGURATIONS_KEY]
return result


def _parse_as_yaml(content: str) -> Any:
Expand Down
20 changes: 10 additions & 10 deletions projects/butter-backup/tests/cli/test_cli_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -260,17 +260,17 @@ def test_open_with_explicit_dest(
) -> None:
config = encrypted_device
expected_cryptsetup_map = Path(f"/dev/mapper/{config.UUID}")
config_file = tmp_path / "config.json"
wrapped_config = cp.Configuration(DeviceConfigurations=[config])
config_file.write_text(wrapped_config.model_dump_json())
dest_dir = tmp_path / "mounts"
dest_dir.mkdir()
expected_mount_dir = dest_dir / config.Name
if create_dest_subdir:
expected_mount_dir.mkdir()
open_result = runner.invoke(
app, ["open", str(dest_dir), "--config", str(config_file)]
config_file = tmp_path / "config.json"
wrapped_config = cp.Configuration(
DeviceConfigurations=[config], OpenDirectory=dest_dir
)
config_file.write_text(wrapped_config.model_dump_json())
open_result = runner.invoke(app, ["open", "--config", str(config_file)])
assert open_result.exit_code == 0
assert str(expected_mount_dir) in open_result.stdout
assert expected_cryptsetup_map.exists()
Expand All @@ -290,14 +290,14 @@ def test_open_shows_error_on_failure(runner, encrypted_device, tmp_path: Path) -
config = encrypted_device.model_copy(
update={"DevicePassCmd": "echo wrong_password"}
)
config_file = tmp_path / "config.json"
wrapped_config = cp.Configuration(DeviceConfigurations=[config])
config_file.write_text(wrapped_config.model_dump_json())
dest_dir = tmp_path / "mounts"
dest_dir.mkdir()
open_result = runner.invoke(
app, ["open", str(dest_dir), "--config", str(config_file)]
config_file = tmp_path / "config.json"
wrapped_config = cp.Configuration(
DeviceConfigurations=[config], OpenDirectory=dest_dir
)
config_file.write_text(wrapped_config.model_dump_json())
open_result = runner.invoke(app, ["open", "--config", str(config_file)])
expected_msg = f"Speichermedium {config.Name} konnte nicht geöffnet werden. Es wird übersprungen."
assert open_result.exit_code == 0
assert expected_msg in open_result.stdout
Expand Down
10 changes: 6 additions & 4 deletions projects/butter-backup/tests/cli/test_sudo_pass_cmd.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,13 +35,15 @@ def test_sudo_pass_cmd_is_used_in_open(
tmp_path: Path,
) -> None:
sudo_pass_cmd = "echo test_password"
dest_dir = tmp_path / "mounts"
dest_dir.mkdir()
wrapped_config = cp.Configuration(
DeviceConfigurations=[encrypted_device], SudoPassCmd=sudo_pass_cmd
DeviceConfigurations=[encrypted_device],
OpenDirectory=dest_dir,
SudoPassCmd=sudo_pass_cmd,
)
config_file = tmp_path / "config.json"
config_file.write_text(wrapped_config.model_dump_json())
dest_dir = tmp_path / "mounts"
dest_dir.mkdir()

mock_pipe = mocker.patch("shell_interface.pipe_pass_cmd_to_real_cmd")
mocker.patch(
Expand All @@ -50,7 +52,7 @@ def test_sudo_pass_cmd_is_used_in_open(
)
mocker.patch("storage_device_managers.mount_device")

runner.invoke(app, ["open", str(dest_dir), "--config", str(config_file)])
runner.invoke(app, ["open", "--config", str(config_file)])

mock_pipe.assert_called_once_with(
sudo_pass_cmd, ["sudo", "-Sv"], capture_output=True
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -128,3 +128,19 @@ def test_parse_configuration_with_sudo_pass_cmd(
raw = cfg.model_dump_json()
result = cp.parse_configuration(raw)
assert result == cfg


def test_toml_parsing_preserves_sudo_pass_cmd() -> None:
toml_content = """\
[butter-backup]
SudoPassCmd = "echo sudo-password"

[[butter-backup.device-configurations]]
UUID = "12345678-1234-5678-1234-567812345678"
DevicePassCmd = "echo device-password"
BackupRepositoryFolder = "repo"
RepositoryPassCmd = "echo repo-password"
FilesAndFolders = []
"""
result = cp.parse_configuration(toml_content)
assert result.SudoPassCmd == "echo sudo-password"
11 changes: 11 additions & 0 deletions projects/storage-device-managers/tests/test_mount_unmount.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,17 @@ class MyCustomTestException(Exception):
pass


def test_ensure_directory_creates_missing_directory(tmp_path: Path) -> None:
destination = tmp_path / "nested" / "mount"
assert not destination.exists()
assert sdm.ensure_directory(destination)
assert destination.exists()


def test_ensure_directory_skips_existing_directory(tmp_path: Path) -> None:
assert not sdm.ensure_directory(tmp_path)


def test_mount_device(
device_with_fs, tmp_path: Path, compression_kwargs: CompressionKwargsT
) -> None:
Expand Down
Loading