cloud-sync/cloud_sync/restic.py

"""restic binary discovery + auto-download + invocation.

Discovery order:
  1. ``--restic-binary <path>`` (explicit override)
  2. ``<pack-folder>/.cloud-sync/restic-<RESTIC_VERSION>`` (pinned cached copy)
  3. ``$PATH`` (only if version matches RESTIC_VERSION)
  4. download from GitHub releases (unless ``--no-download``)

The version is pinned because repos written by one restic version can have
features another version can't read. Cache the pinned binary per-instance
so deleting the instance dir wipes everything cloud-sync owns.
"""

from __future__ import annotations

import bz2
import hashlib
import os
import platform
import re
import shutil
import stat
import subprocess
import sys
import tempfile
import urllib.request
import zipfile
from dataclasses import dataclass
from pathlib import Path

from .cli import Args


RESTIC_VERSION = "0.18.0"
RELEASE_TAG = f"v{RESTIC_VERSION}"

_HTTP_TIMEOUT = 30


@dataclass(frozen=True)
class Platform:
    os_tag: str       # linux, darwin, windows
    arch_tag: str     # amd64, arm64
    is_windows: bool


def _detect_platform() -> Platform:
    name = platform.system().lower()
    arch_raw = platform.machine().lower()
    if name.startswith("linux"):
        os_tag = "linux"
    elif name.startswith("darwin"):
        os_tag = "darwin"
    elif name.startswith("windows"):
        os_tag = "windows"
    else:
        raise RuntimeError(f"unsupported platform: {name}")

    if arch_raw in ("amd64", "x86_64"):
        arch_tag = "amd64"
    elif arch_raw in ("aarch64", "arm64"):
        arch_tag = "arm64"
    else:
        raise RuntimeError(f"unsupported architecture: {arch_raw}")
    return Platform(os_tag, arch_tag, os_tag == "windows")


def _binary_filename(plat: Platform) -> str:
    suffix = ".exe" if plat.is_windows else ""
    return f"restic-{RESTIC_VERSION}{suffix}"


def resolve_binary(args: Args) -> Path:
    """Return a usable restic binary path, downloading if needed."""
    if args.restic_binary is not None:
        if not args.restic_binary.exists():
            raise FileNotFoundError(
                f"--restic-binary path does not exist: {args.restic_binary}"
            )
        return args.restic_binary

    plat = _detect_platform()
    cache_dir = args.pack_folder / ".cloud-sync"
    cached = cache_dir / _binary_filename(plat)
    if cached.exists() and os.access(cached, os.X_OK):
        return cached

    # Try $PATH only when version matches exactly
    system = _find_system_binary_matching_version()
    if system is not None:
        return system

    if not args.allow_download:
        raise RuntimeError(
            f"no usable restic binary at {cached} or on $PATH, "
            f"and --no-download disabled auto-fetch"
        )

    cache_dir.mkdir(parents=True, exist_ok=True)
    _download_restic_to(cached, plat)
    return cached


def run(
    binary: Path,
    args: list[str],
    env: dict[str, str] | None = None,
    cwd: Path | None = None,
    timeout: int = 900,
) -> tuple[int, str]:
    """Run restic. Inherits stderr to caller's terminal for live progress.
    Returns (returncode, captured_stdout)."""
    merged_env = dict(os.environ)
    if env:
        merged_env.update(env)
    p = subprocess.run(  # noqa: S603 — controlled invocation
        [str(binary), *args],
        cwd=str(cwd) if cwd else None,
        env=merged_env,
        stdout=subprocess.PIPE,
        stderr=sys.stderr,
        text=True,
        timeout=timeout,
        check=False,
    )
    return p.returncode, p.stdout


# ---------------------------------------------------------------------------
# discovery + download helpers
# ---------------------------------------------------------------------------


def _find_system_binary_matching_version() -> Path | None:
    name = "restic.exe" if _detect_platform().is_windows else "restic"
    found = shutil.which(name)
    if not found:
        return None
    path = Path(found)
    return path if _binary_version(path) == RESTIC_VERSION else None


def _binary_version(path: Path) -> str | None:
    try:
        out = subprocess.run(  # noqa: S603
            [str(path), "version"],
            capture_output=True, text=True, timeout=5, check=False,
        )
    except (subprocess.SubprocessError, OSError):
        return None
    text = out.stdout + out.stderr
    m = re.search(r"restic\s+(\d+\.\d+\.\d+)", text)
    return m.group(1) if m else None


def _download_restic_to(target: Path, plat: Platform) -> None:
    ext = "zip" if plat.is_windows else "bz2"
    asset = f"restic_{RESTIC_VERSION}_{plat.os_tag}_{plat.arch_tag}.{ext}"
    asset_url = (
        f"https://github.com/restic/restic/releases/download/"
        f"{RELEASE_TAG}/{asset}"
    )
    sums_url = (
        f"https://github.com/restic/restic/releases/download/"
        f"{RELEASE_TAG}/SHA256SUMS"
    )
    print(
        f"cloud-sync: downloading restic {RESTIC_VERSION} from {asset_url}",
        file=sys.stderr,
    )
    with tempfile.NamedTemporaryFile(suffix=f".{ext}", delete=False) as tmp:
        tmp_path = Path(tmp.name)
    try:
        with urllib.request.urlopen(asset_url, timeout=_HTTP_TIMEOUT) as r:
            tmp_path.write_bytes(r.read())
        expected = _expected_sha(sums_url, asset)
        actual = _sha256_file(tmp_path)
        if expected.lower() != actual.lower():
            raise RuntimeError(
                f"restic download sha mismatch: expected {expected}, got {actual}"
            )
        _decompress_into(tmp_path, target, ext)
        if not plat.is_windows:
            target.chmod(target.stat().st_mode | stat.S_IXUSR | stat.S_IRUSR | stat.S_IWUSR)
    finally:
        tmp_path.unlink(missing_ok=True)


def _expected_sha(sums_url: str, asset: str) -> str:
    with urllib.request.urlopen(sums_url, timeout=_HTTP_TIMEOUT) as r:
        body = r.read().decode("utf-8")
    for line in body.splitlines():
        line = line.strip()
        if line.endswith(asset):
            return line.split()[0]
    raise RuntimeError(f"restic SHA256SUMS missing entry for {asset}")


def _sha256_file(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(65536), b""):
            h.update(chunk)
    return h.hexdigest()


def _decompress_into(archive: Path, target: Path, ext: str) -> None:
    target.parent.mkdir(parents=True, exist_ok=True)
    if ext == "bz2":
        with bz2.open(archive, "rb") as src, target.open("wb") as dst:
            shutil.copyfileobj(src, dst)
    elif ext == "zip":
        with zipfile.ZipFile(archive) as zf:
            inner = next(
                (n for n in zf.namelist() if n.endswith("restic.exe")),
                None,
            )
            if inner is None:
                raise RuntimeError("restic.exe not found in downloaded zip")
            with zf.open(inner) as src, target.open("wb") as dst:
                shutil.copyfileobj(src, dst)
    else:
        raise RuntimeError(f"unsupported archive ext: {ext}")