stackpilot/backend/services/stack_assets_service.py

"""Inventory of everything a stack's data actually lives in.

A stack is more than its compose file: bind-mounted directories (``./config``)
and named volumes hold the real state. StackPilot itself runs in a container, so
it can only *see* what is mounted into it — a bind source like
``/opt/stacks/arr-stack/gluetun`` may exist on the host and still be invisible
here (that happens whenever the stacks directory is mounted under a different
host path than ``STACKS_DIR``, because compose resolves ``./gluetun`` against the
path *inside* this container and the daemon then creates it at that same path on
the **host**).

Everything in this module therefore reads and writes host paths through a
throwaway helper container: the daemon does the mounting, so the data is
reachable regardless of what StackPilot has mounted. That is what makes backups
complete instead of "just the compose file".
"""
from __future__ import annotations

import logging
import os
import re
from typing import Optional

import yaml

from config import settings
from docker_client import (
    HELPER_LABEL,
    DockerError,
    get_client,
    get_long_client,
    run_container,
    safe_call,
)
from services import compose_service, stack_job_service

logger = logging.getLogger("stackpilot.assets")

COMPOSE_PROJECT_LABEL = "com.docker.compose.project"
COMPOSE_SERVICE_LABEL = "com.docker.compose.service"
COMPOSE_VOLUME_LABEL = "com.docker.compose.volume"

# Paths that are plumbing, never stack data.
SYSTEM_PATHS = {
    "/var/run/docker.sock",
    "/run/docker.sock",
    "/etc/localtime",
    "/etc/timezone",
    "/etc/hosts",
    "/etc/resolv.conf",
}
SYSTEM_PREFIXES = ("/dev", "/proc", "/sys", "/run", "/var/run", "/var/lib/docker")

# Volume driver_opts types that point at storage which lives somewhere else
# entirely (a NAS). Pulling a media library through a tar.gz is never what the
# user wants, and *restoring* one would overwrite the share.
REMOTE_VOLUME_TYPES = {"nfs", "nfs4", "cifs", "smb", "smb3", "smbfs", "sshfs", "glusterfs"}

# The same for bind mounts and bind-type volumes, which only say *where* they
# are, not what: a host directory can be an NFS mount (fstab/autofs) just as
# well. Recognised by the filesystem magic `stat -f` reports for the path.
REMOTE_FS_MAGIC = {
    0x6969: "nfs",
    0x517B: "smb",
    0xFF534D42: "cifs",
    0xFE534D42: "smb2",
    0x00C36400: "ceph",
}
# Mount types in /proc/self/mountinfo that are remote storage.
REMOTE_MOUNT_TYPES = {"nfs", "nfs4", "cifs", "smb3", "smbfs", "ceph", "glusterfs",
                      "fuse.sshfs", "fuse.rclone", "fuse.glusterfs"}

# Volumes Docker names itself (64 hex chars): recreated with every container, so
# a restore could never hand them back to the service.
_ANONYMOUS_RE = re.compile(r"^[0-9a-f]{64}$")

# Bind directories larger than this are listed but not selected by default.
DEFAULT_MAX_BIND_BYTES = 2 * 1024**3

_ENV_RE = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)(?::?-([^}]*))?\}|\$([A-Za-z_][A-Za-z0-9_]*)")


class AssetError(Exception):
    pass


# --------------------------------------------------------------------------- #
# Helper container primitives (host-path I/O)
# --------------------------------------------------------------------------- #


def ensure_helper_image(client) -> None:
    image = settings.BACKUP_HELPER_IMAGE
    try:
        safe_call(client.images.get, image)
    except DockerError:
        logger.info("Pulling helper image %s", image)
        safe_call(client.images.pull, image)


def _create_helper(client, volumes: dict):
    container = safe_call(
        client.containers.create,
        settings.BACKUP_HELPER_IMAGE,
        command="true",
        volumes=volumes,
        labels={HELPER_LABEL: "1"},
    )
    # Streaming a large volume through this container is the part of a backup
    # that runs for hours; cancelling one works by removing it (see
    # :mod:`services.stack_job_service`), so the job has to know it exists.
    stack_job_service.register_helper(container.id)
    return container


def _remove(container) -> None:
    stack_job_service.drop_helper(getattr(container, "id", ""))
    try:
        container.remove(force=True)
    except Exception:  # noqa: BLE001 - cleanup is best effort
        pass


def _split(path: str) -> tuple[str, str]:
    clean = path.rstrip("/") or "/"
    return os.path.dirname(clean) or "/", os.path.basename(clean)


def _is_chown_error(exc: Exception) -> bool:
    text = str(exc)
    return "chown" in text.lower() and "not permitted" in text.lower()


def _put_archive(container, dest: str, src_file: str, volumes: dict) -> None:
    """Unpack an archive into a container path, coping with squashed mounts.

    The daemon restores ownership while extracting, which an NFS/CIFS export
    with ``root_squash`` refuses. In that case the archive is unpacked into a
    throwaway container's own filesystem first and the files are then copied
    across — ownership cannot be preserved there, but the restore completes
    instead of failing outright.
    """
    with open(src_file, "rb") as fh:
        try:
            container.put_archive(dest, fh)
            return
        except Exception as exc:  # noqa: BLE001
            if not _is_chown_error(exc):
                raise
    logger.warning("%s rejects ownership changes; restoring without it", dest)
    client = get_long_client()
    staging = _create_helper_with(
        client, volumes, ["sh", "-c", f"cp -R /tmp/. '{dest}/'"]
    )
    try:
        with open(src_file, "rb") as fh:
            staging.put_archive("/tmp", fh)
        staging.start()
        status = staging.wait(timeout=3600).get("StatusCode", 1)
        if status != 0:
            err = (staging.logs(stdout=True, stderr=True) or b"").decode("utf-8", "replace")
            raise AssetError(err.strip() or f"copy failed (exit {status})")
    finally:
        _remove(staging)


def _create_helper_with(client, volumes: dict, command: list[str]):
    ensure_helper_image(client)
    container = safe_call(
        client.containers.create,
        settings.BACKUP_HELPER_IMAGE,
        command=command,
        volumes=volumes,
        labels={HELPER_LABEL: "1"},
    )
    stack_job_service.register_helper(container.id)
    return container


def _unknown() -> dict:
    return {"kind": "unknown", "size": None, "fs": None, "remote": None, "measured": False}


def inspect_paths(paths: list[str]) -> dict[str, dict]:
    """Classify host paths via one helper container.

    ``{path: {"kind", "size", "fs", "remote", "measured"}}``: ``kind`` is dir /
    file / special, ``size`` bytes. ``remote`` names the network filesystem the
    path sits on (nfs, cifs, ...), whose contents are not walked at all.
    Directory sizes are capped at 20s of ``du``; ``measured`` is False when that
    ran out, which on its own says the directory is big.
    """
    unique = [p for p in dict.fromkeys(paths) if p]
    if not unique:
        return {}
    client = get_long_client()
    ensure_helper_image(client)
    mounts = {p: {"bind": f"/m/{i}", "mode": "ro"} for i, p in enumerate(unique)}
    remote_hex = "|".join(f"{m:x}" for m in REMOTE_FS_MAGIC)
    script_parts = []
    for i in range(len(unique)):
        script_parts.append(
            f'd=/m/{i}; t=$(stat -f -c %t "$d" 2>/dev/null); t=${{t:--}}; '
            f'if [ -d "$d" ]; then '
            f'case "$t" in {remote_hex}) echo "{i} dir - $t";; '
            f'*) s=$(timeout 20 du -sk "$d" 2>/dev/null | cut -f1); echo "{i} dir ${{s:--}} $t";; esac; '
            f'elif [ -f "$d" ]; then echo "{i} file $(stat -c %s "$d" 2>/dev/null) $t"; '
            f'else echo "{i} special - $t"; fi'
        )
    script = "; ".join(script_parts)
    try:
        out = run_container(
            client, settings.BACKUP_HELPER_IMAGE, ["sh", "-c", script], volumes=mounts
        )
    except DockerError as exc:
        logger.warning("Path inspection failed: %s", exc)
        return {p: _unknown() for p in unique}

    result: dict[str, dict] = {p: _unknown() for p in unique}
    for line in (out or b"").decode("utf-8", "replace").splitlines():
        parts = line.strip().split()
        if len(parts) < 2 or not parts[0].isdigit():
            continue
        idx = int(parts[0])
        if idx >= len(unique):
            continue
        kind = parts[1]
        size: Optional[int] = None
        if len(parts) > 2 and parts[2].isdigit():
            size = int(parts[2]) * 1024 if kind == "dir" else int(parts[2])
        fs = parts[3] if len(parts) > 3 and parts[3] != "-" else None
        try:
            remote = REMOTE_FS_MAGIC.get(int(fs, 16)) if fs else None
        except ValueError:
            remote = None
        result[unique[idx]] = {
            "kind": kind,
            "size": size,
            "fs": fs,
            "remote": remote,
            "measured": kind != "dir" or remote is not None or size is not None,
        }
    return result


def export_path(source: str, kind: str, dest_file: str) -> int:
    """Tar a host path (dir contents, or a single file) into ``dest_file``."""
    client = get_long_client()
    ensure_helper_image(client)
    if kind == "file":
        parent, base = _split(source)
        if not base:
            raise AssetError(f"Cannot archive {source}")
        container = _create_helper(client, {parent: {"bind": "/src", "mode": "ro"}})
        member = f"/src/{base}"
    else:
        container = _create_helper(client, {source: {"bind": "/src", "mode": "ro"}})
        # "/src/." archives the *contents*, so restore can unpack straight back
        # into the directory without a stray prefix.
        member = "/src/."
    written = 0
    try:
        bits, _ = container.get_archive(member)
        with open(dest_file, "wb") as fh:
            for chunk in bits:
                fh.write(chunk)
                written += len(chunk)
    finally:
        _remove(container)
    return written


def import_path(source: str, kind: str, src_file: str) -> None:
    """Unpack an archive produced by :func:`export_path` back to its host path."""
    client = get_long_client()
    ensure_helper_image(client)
    if kind == "file":
        parent, _base = _split(source)
        mounts = {parent: {"bind": "/dst", "mode": "rw"}}
    else:
        mounts = {source: {"bind": "/dst", "mode": "rw"}}
    container = _create_helper(client, mounts)
    try:
        _put_archive(container, "/dst", src_file, mounts)
    finally:
        _remove(container)


def export_volume(full_name: str, dest_file: str) -> int:
    """Stream a named volume's contents into ``dest_file`` (never into RAM)."""
    client = get_long_client()
    ensure_helper_image(client)
    container = _create_helper(client, {full_name: {"bind": "/v", "mode": "ro"}})
    written = 0
    try:
        bits, _ = container.get_archive("/v/.")
        with open(dest_file, "wb") as fh:
            for chunk in bits:
                fh.write(chunk)
                written += len(chunk)
    finally:
        _remove(container)
    return written


def volume_attrs(full_name: str) -> Optional[dict]:
    """A volume's inspect data, or None if it does not exist."""
    try:
        return safe_call(get_client().volumes.get, full_name).attrs or {}
    except DockerError:
        return None


def running_users(full_name: str) -> list[str]:
    """Names of running containers that have the volume mounted."""
    try:
        containers = safe_call(
            get_client().containers.list, filters={"volume": full_name, "status": "running"}
        )
    except DockerError:
        return []
    return sorted(c.name for c in containers)


def import_volume(
    full_name: str,
    labels: dict,
    src_file: str,
    wipe: bool = True,
    driver: Optional[str] = None,
    driver_opts: Optional[dict] = None,
) -> None:
    """Restore a volume from an archive, optionally clearing it first.

    A missing volume is recreated with the driver and options it had, not as a
    plain local volume: compose refuses to use an existing volume whose
    configuration differs from the compose file.
    """
    client = get_long_client()
    ensure_helper_image(client)
    existed = True
    try:
        safe_call(client.volumes.get, full_name)
    except DockerError:
        existed = False
        safe_call(
            client.volumes.create,
            name=full_name,
            driver=driver or "local",
            driver_opts=driver_opts or None,
            labels=labels or {},
        )
    if existed and wipe:
        # Restore means "back to the snapshot": drop files created since.
        run_container(
            client,
            settings.BACKUP_HELPER_IMAGE,
            ["sh", "-c", "find /v -mindepth 1 -delete"],
            volumes={full_name: {"bind": "/v", "mode": "rw"}},
        )
    mounts = {full_name: {"bind": "/v", "mode": "rw"}}
    container = _create_helper(client, mounts)
    try:
        _put_archive(container, "/v", src_file, mounts)
    finally:
        _remove(container)


# --------------------------------------------------------------------------- #
# Where does STACKS_DIR really live on the host?
# --------------------------------------------------------------------------- #


def host_stacks_dir() -> Optional[str]:
    """Host path backing ``STACKS_DIR`` inside this container, if detectable.

    Read from /proc/self/mountinfo (field 4 is the source subtree on the host
    filesystem). Returns None when not running in a container / not bind-mounted.
    """
    target = settings.STACKS_DIR.rstrip("/") or "/"
    try:
        with open("/proc/self/mountinfo", "r", encoding="utf-8") as fh:
            for line in fh:
                parts = line.split()
                if len(parts) < 5:
                    continue
                if parts[4].rstrip("/") == target:
                    return parts[3]
    except OSError:
        return None
    return None


def stacks_path_mismatch() -> Optional[dict]:
    """Report a host/container path mismatch for the stacks directory.

    When they differ, compose resolves a stack's relative bind mounts against
    the *container* path, so the daemon creates the data directories at that
    path on the host — invisible to StackPilot. Backups then only find the
    compose file unless bind sources are captured through a helper container.
    """
    host = host_stacks_dir()
    container = settings.STACKS_DIR.rstrip("/")
    if not host or host.rstrip("/") == container:
        return None
    return {"host": host, "container": container}


# --------------------------------------------------------------------------- #
# Compose / container mount discovery
# --------------------------------------------------------------------------- #


def _env_for_stack(stack_id: str) -> dict:
    env: dict[str, str] = {}
    path = os.path.join(compose_service.stack_dir(stack_id), ".env")
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            for raw in fh:
                line = raw.strip()
                if not line or line.startswith("#") or "=" not in line:
                    continue
                key, _, value = line.partition("=")
                env[key.strip()] = value.strip().strip('"').strip("'")
    except OSError:
        pass
    return env


def _interpolate(text: str, env: dict) -> str:
    def repl(m: re.Match) -> str:
        name = m.group(1) or m.group(3)
        default = m.group(2) or ""
        return env.get(name, default)

    return _ENV_RE.sub(repl, text)


def _bind_specs_from_compose(stack_id: str) -> list[dict]:
    """Bind sources declared in the compose file (used when no containers exist)."""
    directory = compose_service.stack_dir(stack_id)
    compose_file = compose_service.find_compose_file(directory)
    if not compose_file:
        return []
    try:
        with open(compose_file, "r", encoding="utf-8", errors="replace") as fh:
            data = yaml.safe_load(fh) or {}
    except (OSError, yaml.YAMLError):
        return []
    env = _env_for_stack(stack_id)
    out: list[dict] = []
    for service, spec in (data.get("services") or {}).items():
        if not isinstance(spec, dict):
            continue
        for entry in spec.get("volumes") or []:
            source = target = None
            if isinstance(entry, str):
                parts = _interpolate(entry, env).split(":")
                if len(parts) >= 2:
                    source, target = parts[0], parts[1]
            elif isinstance(entry, dict):
                if entry.get("type") not in (None, "bind"):
                    continue
                source = _interpolate(str(entry.get("source") or ""), env)
                target = _interpolate(str(entry.get("target") or ""), env)
            if not source or not target:
                continue
            if not (source.startswith("/") or source.startswith(".") or source.startswith("~")):
                continue  # named volume
            if source.startswith("~"):
                continue  # home-relative: resolved by the daemon's user, skip
            resolved = source if source.startswith("/") else os.path.normpath(
                os.path.join(directory, source)
            )
            out.append({"source": resolved, "service": str(service), "target": target})
    return out


def _bind_specs_from_containers(stack_id: str) -> list[dict]:
    """Bind sources as the daemon actually mounted them (authoritative)."""
    try:
        client = get_client()
        containers = safe_call(
            client.containers.list,
            all=True,
            filters={"label": f"{COMPOSE_PROJECT_LABEL}={stack_id}"},
        )
    except DockerError:
        return []
    out: list[dict] = []
    for c in containers:
        service = (c.labels or {}).get(COMPOSE_SERVICE_LABEL, c.name)
        for mount in c.attrs.get("Mounts") or []:
            if mount.get("Type") != "bind" or not mount.get("Source"):
                continue
            out.append(
                {
                    "source": mount["Source"],
                    "service": service,
                    "target": mount.get("Destination") or "",
                }
            )
    return out


def is_system_path(path: str) -> bool:
    if path in SYSTEM_PATHS:
        return True
    return any(path == p or path.startswith(p + "/") for p in SYSTEM_PREFIXES)


def _inside(path: str, parent: str) -> bool:
    parent = parent.rstrip("/")
    return path == parent or path.startswith(parent + "/")


def _volume_entry(name: str, attrs: dict, external: bool) -> dict:
    labels = attrs.get("Labels") or {}
    options = attrs.get("Options") or {}
    driver = attrs.get("Driver", "local")
    vtype = str(options.get("type") or "").lower()
    device = str(options.get("device") or "")
    remote = (
        vtype in REMOTE_VOLUME_TYPES
        or driver != "local"
        or device.startswith("//")
        or device.startswith(":")
    )
    # A local volume that binds a host directory (type none, o=bind) is only as
    # local as that directory; inventory() checks what it sits on.
    bind_device = (
        device
        if not remote and device.startswith("/") and "bind" in str(options.get("o") or "")
        else None
    )
    return {
        "name": name,
        "short": labels.get(COMPOSE_VOLUME_LABEL, name),
        "labels": labels,
        "driver": driver,
        "options": options,
        "remote": remote,
        "remote_type": vtype or (driver if driver != "local" else None),
        "external": external,
        "bind_device": bind_device,
    }


def _external_volume_names(stack_id: str) -> set[str]:
    """Named volumes the stack uses but compose did not create for it:
    ``external: true`` in the compose file, or mounted by its containers."""
    names: set[str] = set()
    try:
        client = get_client()
        containers = safe_call(
            client.containers.list,
            all=True,
            filters={"label": f"{COMPOSE_PROJECT_LABEL}={stack_id}"},
        )
        for c in containers:
            for mount in c.attrs.get("Mounts") or []:
                if mount.get("Type") == "volume" and mount.get("Name"):
                    names.add(mount["Name"])
    except DockerError:
        pass
    compose_file = compose_service.find_compose_file(compose_service.stack_dir(stack_id))
    if compose_file:
        try:
            with open(compose_file, "r", encoding="utf-8", errors="replace") as fh:
                data = yaml.safe_load(fh) or {}
        except (OSError, yaml.YAMLError):
            data = {}
        env = _env_for_stack(stack_id)
        for key, spec in (data.get("volumes") or {}).items():
            if isinstance(spec, dict) and spec.get("external"):
                names.add(_interpolate(str(spec.get("name") or key), env))
    return names


def compose_volumes(stack_id: str) -> list[dict]:
    """The stack's named volumes, with remote-storage detection.

    Compose-created ones carry the project label. Volumes the stack only uses
    (``external: true``, or created by hand) do not, and were silently missing
    from backups before; they are listed with ``external`` set. Anonymous
    volumes are listed too, so it is visible that data lives in one.
    """
    try:
        client = get_client()
        vols = safe_call(
            client.volumes.list,
            filters={"label": f"{COMPOSE_PROJECT_LABEL}={stack_id}"},
        )
    except DockerError:
        return []
    out = [_volume_entry(v.name, v.attrs or {}, external=False) for v in vols]
    known = {v["name"] for v in out}
    for name in sorted(_external_volume_names(stack_id) - known):
        attrs = volume_attrs(name)
        if attrs is None:
            continue  # declared external but not created yet
        out.append(_volume_entry(name, attrs, external=True))
    return out


def _mounts() -> list[tuple[str, str]]:
    """(mount point, filesystem type) of every mount this process sees."""
    out = []
    try:
        with open("/proc/self/mountinfo", "r", encoding="utf-8") as fh:
            for line in fh:
                left, _, right = line.partition(" - ")
                fields, rfields = left.split(), right.split()
                if len(fields) >= 5 and rfields:
                    # Spaces etc. are octal-escaped in mountinfo.
                    point = fields[4].encode().decode("unicode_escape")
                    out.append((point, rfields[0]))
    except OSError:
        pass
    return out


def remote_mount_type(path: str) -> Optional[str]:
    """The remote filesystem type ``path`` lives on as this process sees it,
    e.g. an NFS share mounted inside the stacks directory. None if local."""
    best, best_type = "", None
    for point, fstype in _mounts():
        if _inside(path, point) and len(point) >= len(best):
            best, best_type = point, fstype
    return best_type if best_type in REMOTE_MOUNT_TYPES else None


def inventory(stack_id: str, max_bind_bytes: int = DEFAULT_MAX_BIND_BYTES) -> dict:
    """What a backup of this stack would (and would not) capture.

    Bind sources are merged from the running containers (authoritative) and the
    compose file (covers stacks that were never started), classified through a
    helper container so host-only paths are seen too.
    """
    directory = compose_service.stack_dir(stack_id)
    specs = _bind_specs_from_containers(stack_id) or []
    seen = {(s["source"], s["service"], s["target"]) for s in specs}
    for spec in _bind_specs_from_compose(stack_id):
        if (spec["source"], spec["service"], spec["target"]) not in seen:
            specs.append(spec)

    grouped: dict[str, dict] = {}
    for spec in specs:
        entry = grouped.setdefault(spec["source"], {"source": spec["source"], "mounts": []})
        mount = {"service": spec["service"], "target": spec["target"]}
        if mount not in entry["mounts"]:
            entry["mounts"].append(mount)

    volumes_raw = compose_volumes(stack_id)
    real_paths = [p for p in grouped if not is_system_path(p)]
    stats = inspect_paths(
        real_paths + [v["bind_device"] for v in volumes_raw if v["bind_device"]]
    )

    binds = []
    for path, entry in sorted(grouped.items()):
        system = is_system_path(path)
        info = stats.get(path) or _unknown()
        kind, size = info["kind"], info["size"]
        remote = info.get("remote") or (remote_mount_type(path) if not system else None)
        inside = _inside(path, directory)
        # A path inside the stack directory that this process can actually read
        # is already covered by the compose/ tree in the archive.
        visible = inside and os.path.exists(path)
        include = True
        reason = None
        if system:
            include, reason = False, "system path"
        elif remote:
            include, reason = False, f"remote storage ({remote})"
        elif kind == "special":
            include, reason = False, "not a regular file or directory"
        elif kind == "unknown":
            include, reason = False, "could not inspect path"
        elif size is not None and size > max_bind_bytes:
            include, reason = False, f"larger than {max_bind_bytes // 1024**3} GiB"
        elif not info.get("measured", True):
            # du ran out of time: too big to have been walked in 20s.
            include, reason = False, "too large to measure"
        binds.append(
            {
                "source": path,
                "mounts": entry["mounts"],
                "kind": kind,
                "size": size,
                "inside_stack_dir": inside,
                # Readable from here and inside the stack folder → the compose/
                # tree already carries it, no separate archive needed.
                "covered_by_compose": visible,
                "via": "compose" if visible else "archive",
                "system": system,
                "remote": remote,
                "include_default": include,
                "reason": reason,
            }
        )

    volumes = []
    for vol in volumes_raw:
        if vol["bind_device"]:
            device_remote = (stats.get(vol["bind_device"]) or {}).get("remote")
            if device_remote:
                vol = {**vol, "remote": True, "remote_type": device_remote}
        include, reason = True, None
        if vol["remote"]:
            include, reason = False, f"remote storage ({vol['remote_type']})"
        elif _ANONYMOUS_RE.match(vol["name"]):
            include, reason = False, "anonymous volume (Docker recreates it with the container)"
        volumes.append({**vol, "include_default": include, "reason": reason})

    return {
        "stack_id": stack_id,
        "stack_dir": directory,
        "stack_dir_visible": os.path.isdir(directory),
        "path_mismatch": stacks_path_mismatch(),
        "binds": binds,
        "volumes": volumes,
    }