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"

    @classmethod
    def read(cls, hostname: str, login: str, config_dir: Path):
        secret_sources_file = config_dir / f"secret_sources_{hostname}_{login}"
        data_sources_file = config_dir / f"data_sources_{hostname}_{login}"
        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=config_dir / "borgmatic.d",
            borgmatic_path=config_dir / "borgmatic",
            history_file=config_dir / f".bash_history_{login}",
            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(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}")