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


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",
        ]
        if config.ssh_auth_sock:
            volumes += [f"{config.ssh_auth_sock}:{config.ssh_auth_sock}:Z"]

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

        volume_args = [a for vol in volumes for a in ["-v", vol]]

        secrets_args = [
            a
            for s in config.secret_sources
            for a in ["--secret", f"{s.name},mode=0{s.mode:o}"]
        ]

        args = (
            [
                "podman",
                "run",
                "-h",
                self.hostname,
                "--detach",
                "--name",
                container_name,
                "-e",
                "SSH_AUTH_SOCK",
                "-e",
                "TZ=Europe/Paris",
                "-e",
                "SSH_KEY_NAME",
                "-e",
                f"HOST_LOGIN={self.login}",
                "--security-opt=label=disable",
            ]
            + volume_args
            + secrets_args
            + [self.image]
        )
        print(" ".join(args))
        subprocess.run(args)

    def rm(self):
        subprocess.run(["podman", "rm", "-f", self.name])

    def exec(self, cmd: list[str], env_vars: list[str] = []):
        args = ["podman", "exec", "-ti"]
        args += [a for var in env_vars for a in ["-e", var]]
        subprocess.run(args + [self.name] + cmd)

    @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}")