pc-backup/pc_backup/container.py

import os
import subprocess
from pathlib import Path, PurePosixPath
from dataclasses import dataclass

from pc_backup.secret import Secret
from pc_backup.env import is_windows
from pc_backup.podman import Podman


def read_data_sources(file: Path) -> list[Path]:
    with open(file) as f:
        paths = f.readlines()
        return [Path(p_str.strip()).expanduser() for p_str in paths]


@dataclass
class Configuration:
    secret_sources: list[Secret]
    data_sources: list[Path]
    borgmatic_d_path: Path
    borgmatic_path: Path
    history_file: Path
    ssh_auth_sock: Path | None

    @staticmethod
    def get_config_dir() -> Path:
        if is_windows():
            program_data = Path(os.getenv("ProgramData"))
            return program_data / "pc_backup"
        else:
            return Path.home() / ".config" / "pc_backup"

    def __str__(self) -> str:
        ret = []
        ret += ["secret_sources:"]
        for s in self.secret_sources:
            ret += [f"  {s.name}: {s}"]

        ret += ["data_sources:"]
        for d in self.data_sources:
            ret += [f"  {d}"]

        ret += ["borgmatic setups:"]
        for f in self.borgmatic_d_path.iterdir():
            ret += [f"  {f.name}"]

        ret += ["paths:"]
        for p in ["borgmatic_d_path", "borgmatic_path", "history_file"]:
            ret += [f"  {p}: {getattr(self, p)}"]

        return "\n".join(ret)

    @classmethod
    def read(cls, hostname: str, login: str, config_dir: Path):
        # The configuration directory has to be organized like this :
        #
        # .
        # ├── borgmatic
        # │   └── common.yaml
        # ├── <host1>
        # │   ├── <user1>
        # │   │   ├── borgmatic.d
        # │   │   │   ├── <config1>.yaml
        # │   │   │   ├── <config2>.yaml
        # │   │   │   └── windows.yaml
        # │   │   ├── data_sources
        # │   │   └── secret_sources
        # │   └── <user2>
        # │       ├── borgmatic.d
        # │       │   ├── ...
        # │       ├── data_sources
        # │       └── secret_sources
        # └── <host2>
        #     ├── <user1>
        #     │   ├── borgmatic.d
        #     │   │   ├── ...
        #     │   ├── data_sources
        #     │   └── secret_sources
        #     └── <user2>
        #         ├── borgmatic.d
        #         │   ├── ...
        #         ├── data_sources
        #         └── secret_sources
        specific_config_dir = config_dir / hostname / login

        secret_sources_file = specific_config_dir / "secret_sources"
        data_sources_file = specific_config_dir / "data_sources"
        ssh_auth_sock = os.getenv("SSH_AUTH_SOCK")

        return cls(
            secret_sources=Secret.read_sources(secret_sources_file),
            data_sources=read_data_sources(data_sources_file),
            borgmatic_d_path=specific_config_dir / "borgmatic.d",
            borgmatic_path=config_dir / "borgmatic",
            history_file=specific_config_dir / ".bash_history",
            ssh_auth_sock=Path(ssh_auth_sock) if ssh_auth_sock else None,
        )


@dataclass
class BorgmaticContainer:
    hostname: str
    login: str
    name: str
    image: str = "ghcr.io/borgmatic-collective/borgmatic"

    def run(self, config: Configuration):
        container_name = f"borgmatic_{self.login}"

        config.history_file.touch()
        volumes = [
            f"{config.borgmatic_d_path}:/etc/borgmatic.d/",
            f"{config.borgmatic_path}:/etc/borgmatic/",
            f"{config.history_file}:/root/.bash_history",
            "borg_ssh_dir:/root/.ssh",
            "borg_config:/root/.config/borg",
            "borg_cache:/root/.cache/borg",
            "borgmatic_state:/root/.local/state/borgmatic",
            "borgmatic_log:/root/.local/share/borgmatic",
        ]

        volumes += [
            f"{vol}:{self.to_source_path(vol)}:ro" for vol in config.data_sources
        ]

        Podman.run(
            self.image,
            container_name,
            hostname=self.hostname,
            env=["TZ=Europe/Paris", f"HOST_LOGIN={self.login}"],
            ssh_auth_sock=config.ssh_auth_sock,
            volumes=volumes,
            secrets=config.secret_sources,
        )

    def rm(self):
        Podman.rm(self.name)

    def exec(self, cmd: list[str], *, interactive=False):
        Podman.exec(self.name, *cmd, interactive=interactive)

    @staticmethod
    def to_source_path(path: Path):
        mount_base = PurePosixPath("/mnt") / "source"
        inner_path = PurePosixPath(path)
        with_drive = PurePosixPath(inner_path.parts[0].replace(":", "")).joinpath(
            *inner_path.parts[1:]
        )
        return mount_base / with_drive.relative_to(with_drive.anchor)

    @classmethod
    def new(cls, hostname: str, login: str):
        return cls(hostname, login, f"borgmatic_{login}")