menzeljandClaude Opus 5
34458c702c
An NFSv3 mount with locking makes the kernel register its lock manager with rpcbind on the Docker host itself. On a host without nfs-common the mount fails with a bare "connection refused" that reads like the NAS is unreachable, even though the NAS and its ports are fine. Only StackPilot ever writes the backup files on an NFS destination, one writer per file, so nothing needs locking. mount_options() now adds nolock unless the options already choose (lock, nolock, local_lock=) or ask for NFSv4, which locks in-protocol and needs no rpcbind. Existing destinations get their volume recreated once with the new options. The refused-mount hint now points at rpcbind/nfs-common on the Docker host first, and "protocol not supported" explains the missing kernel module (modprobe nfsv4). The form labels the option. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
621 lines
21 KiB
Python
621 lines
21 KiB
Python
"""Push/pull stack backups to remote destinations (SFTP, S3-compatible or NFS).
|
|
|
|
All operations are synchronous (paramiko / boto3 / docker); async callers
|
|
should wrap them with ``asyncio.to_thread``. Destination config is a dict
|
|
stored in the ``BackupDestination.config`` column as JSON, encrypted at rest
|
|
(:mod:`services.crypto_service`) because it carries SFTP passwords, SSH keys
|
|
and S3 secret keys. Always go through :func:`parse_config` / :func:`dump_config`
|
|
— never touch the column directly.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import logging
|
|
import posixpath
|
|
import stat
|
|
import tarfile
|
|
import tempfile
|
|
from typing import Any, Optional
|
|
|
|
from models.backup_destination import BackupDestination
|
|
from services import crypto_service
|
|
|
|
logger = logging.getLogger("stackpilot.backup_dest")
|
|
|
|
|
|
class DestinationError(Exception):
|
|
pass
|
|
|
|
|
|
def parse_config(dest: BackupDestination) -> dict:
|
|
"""Decrypt and parse a destination's config.
|
|
|
|
Tolerates plaintext (pre-encryption rows) and returns ``{}`` rather than
|
|
raising if the value can't be decrypted — a destination whose key is gone
|
|
should show up as unconfigured in the UI, not take the whole list down with
|
|
a 500. The failure is logged with the destination name so it's findable.
|
|
"""
|
|
try:
|
|
raw = crypto_service.decrypt(dest.config or "{}")
|
|
except crypto_service.DecryptError as exc:
|
|
logger.error("Destination '%s': %s", dest.name, exc)
|
|
return {}
|
|
try:
|
|
return json.loads(raw or "{}")
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
|
|
|
|
def dump_config(config: dict) -> str:
|
|
"""Serialise and encrypt a config dict for storage."""
|
|
return crypto_service.encrypt(json.dumps(config or {}))
|
|
|
|
|
|
def migrate_plaintext_configs(session) -> int:
|
|
"""Encrypt destination configs written before encryption existed.
|
|
|
|
Runs once at startup. Returns how many rows were rewritten.
|
|
"""
|
|
from sqlmodel import select
|
|
|
|
migrated = 0
|
|
for dest in session.exec(select(BackupDestination)).all():
|
|
if crypto_service.is_encrypted(dest.config):
|
|
continue
|
|
dest.config = crypto_service.encrypt(dest.config or "{}")
|
|
session.add(dest)
|
|
migrated += 1
|
|
if migrated:
|
|
session.commit()
|
|
return migrated
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# SFTP (paramiko)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _sftp_connect(cfg: dict):
|
|
import paramiko
|
|
|
|
host = cfg.get("host")
|
|
if not host:
|
|
raise DestinationError("SFTP host is required")
|
|
port = int(cfg.get("port") or 22)
|
|
username = cfg.get("username")
|
|
transport = paramiko.Transport((host, port))
|
|
try:
|
|
pkey = None
|
|
if cfg.get("private_key"):
|
|
pkey = _load_key(cfg["private_key"])
|
|
transport.connect(username=username, password=cfg.get("password") or None, pkey=pkey)
|
|
except Exception as exc: # noqa: BLE001
|
|
transport.close()
|
|
raise DestinationError(f"SFTP connection failed: {exc}") from exc
|
|
return paramiko.SFTPClient.from_transport(transport), transport
|
|
|
|
|
|
def _load_key(key_str: str):
|
|
import paramiko
|
|
|
|
for cls in (paramiko.Ed25519Key, paramiko.ECDSAKey, paramiko.RSAKey):
|
|
try:
|
|
return cls.from_private_key(io.StringIO(key_str))
|
|
except Exception: # noqa: BLE001
|
|
continue
|
|
raise DestinationError("Could not parse SFTP private key")
|
|
|
|
|
|
def _sftp_makedirs(sftp, path: str) -> None:
|
|
if not path or path in (".", "/"):
|
|
return
|
|
parts = path.strip("/").split("/")
|
|
cur = "/" if path.startswith("/") else ""
|
|
for p in parts:
|
|
cur = posixpath.join(cur, p) if cur else p
|
|
try:
|
|
sftp.stat(cur)
|
|
except IOError:
|
|
sftp.mkdir(cur)
|
|
|
|
|
|
def _sftp_upload(cfg: dict, local_path: str, filename: str) -> str:
|
|
sftp, transport = _sftp_connect(cfg)
|
|
try:
|
|
base = cfg.get("path") or "."
|
|
if base not in (".", ""):
|
|
_sftp_makedirs(sftp, base)
|
|
remote = posixpath.join(base, filename) if base not in (".", "") else filename
|
|
sftp.put(local_path, remote)
|
|
return remote
|
|
finally:
|
|
sftp.close()
|
|
transport.close()
|
|
|
|
|
|
def _sftp_list(cfg: dict) -> list[dict]:
|
|
sftp, transport = _sftp_connect(cfg)
|
|
try:
|
|
base = cfg.get("path") or "."
|
|
out = []
|
|
try:
|
|
entries = sftp.listdir_attr(base)
|
|
except IOError:
|
|
return []
|
|
for e in entries:
|
|
if stat.S_ISDIR(e.st_mode):
|
|
continue
|
|
if not e.filename.endswith(".tar.gz"):
|
|
continue
|
|
out.append({"name": e.filename, "size": e.st_size, "modified": e.st_mtime})
|
|
return sorted(out, key=lambda x: x["modified"] or 0, reverse=True)
|
|
finally:
|
|
sftp.close()
|
|
transport.close()
|
|
|
|
|
|
def _sftp_download(cfg: dict, name: str, local_path: str) -> None:
|
|
sftp, transport = _sftp_connect(cfg)
|
|
try:
|
|
base = cfg.get("path") or "."
|
|
remote = posixpath.join(base, name) if base not in (".", "") else name
|
|
sftp.get(remote, local_path)
|
|
finally:
|
|
sftp.close()
|
|
transport.close()
|
|
|
|
|
|
def _sftp_delete(cfg: dict, name: str) -> None:
|
|
sftp, transport = _sftp_connect(cfg)
|
|
try:
|
|
base = cfg.get("path") or "."
|
|
remote = posixpath.join(base, name) if base not in (".", "") else name
|
|
sftp.remove(remote)
|
|
finally:
|
|
sftp.close()
|
|
transport.close()
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# S3-compatible (boto3)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _s3_client(cfg: dict):
|
|
import boto3
|
|
|
|
bucket = cfg.get("bucket")
|
|
if not bucket:
|
|
raise DestinationError("S3 bucket is required")
|
|
return boto3.client(
|
|
"s3",
|
|
endpoint_url=cfg.get("endpoint_url") or None,
|
|
region_name=cfg.get("region") or None,
|
|
aws_access_key_id=cfg.get("access_key") or None,
|
|
aws_secret_access_key=cfg.get("secret_key") or None,
|
|
)
|
|
|
|
|
|
def _s3_key(cfg: dict, filename: str) -> str:
|
|
prefix = (cfg.get("prefix") or "").strip("/")
|
|
return f"{prefix}/{filename}" if prefix else filename
|
|
|
|
|
|
def _s3_upload(cfg: dict, local_path: str, filename: str) -> str:
|
|
client = _s3_client(cfg)
|
|
key = _s3_key(cfg, filename)
|
|
try:
|
|
client.upload_file(local_path, cfg["bucket"], key)
|
|
except Exception as exc: # noqa: BLE001
|
|
raise DestinationError(f"S3 upload failed: {exc}") from exc
|
|
return key
|
|
|
|
|
|
def _s3_list(cfg: dict) -> list[dict]:
|
|
client = _s3_client(cfg)
|
|
prefix = (cfg.get("prefix") or "").strip("/")
|
|
kwargs: dict[str, Any] = {"Bucket": cfg["bucket"]}
|
|
if prefix:
|
|
kwargs["Prefix"] = prefix + "/"
|
|
try:
|
|
# Paginated: one call returns at most 1000 keys, and retention sorts
|
|
# by age, so the oldest backups were the ones it never saw.
|
|
pages = client.get_paginator("list_objects_v2").paginate(**kwargs)
|
|
objects = [obj for page in pages for obj in page.get("Contents", [])]
|
|
except Exception as exc: # noqa: BLE001
|
|
raise DestinationError(f"S3 list failed: {exc}") from exc
|
|
out = []
|
|
for obj in objects:
|
|
name = obj["Key"].split("/")[-1]
|
|
if not name.endswith(".tar.gz"):
|
|
continue
|
|
out.append(
|
|
{
|
|
"name": name,
|
|
"size": obj.get("Size", 0),
|
|
"modified": obj["LastModified"].timestamp() if obj.get("LastModified") else None,
|
|
}
|
|
)
|
|
return sorted(out, key=lambda x: x["modified"] or 0, reverse=True)
|
|
|
|
|
|
def _s3_download(cfg: dict, name: str, local_path: str) -> None:
|
|
client = _s3_client(cfg)
|
|
try:
|
|
client.download_file(cfg["bucket"], _s3_key(cfg, name), local_path)
|
|
except Exception as exc: # noqa: BLE001
|
|
raise DestinationError(f"S3 download failed: {exc}") from exc
|
|
|
|
|
|
def _s3_delete(cfg: dict, name: str) -> None:
|
|
client = _s3_client(cfg)
|
|
client.delete_object(Bucket=cfg["bucket"], Key=_s3_key(cfg, name))
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# NFS — the Docker daemon mounts the export as a named volume; file I/O runs
|
|
# through a throwaway helper container (same pattern as volume backups), so
|
|
# the backend itself needs no mount privileges.
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
_NFS_VOLUME_PREFIX = "stackpilot-nfs-dest-"
|
|
|
|
|
|
def _nfs_check_name(name: str) -> None:
|
|
import re
|
|
|
|
# Same safe charset as the subdir parts: the name is interpolated into the
|
|
# helper container's shell commands.
|
|
if not name or not re.fullmatch(r"[A-Za-z0-9._-]+", name) or name.startswith("."):
|
|
raise DestinationError(f"Invalid backup file name '{name}'")
|
|
|
|
|
|
def _nfs_subdir(cfg: dict) -> str:
|
|
"""Sanitized relative directory inside the export ('' = export root).
|
|
|
|
Parts are restricted to a safe charset because the path is interpolated
|
|
into helper-container shell commands.
|
|
"""
|
|
import re
|
|
|
|
raw = (cfg.get("subdir") or "").strip().strip("/")
|
|
if not raw:
|
|
return ""
|
|
parts = [p for p in raw.split("/") if p]
|
|
for p in parts:
|
|
if p == ".." or not re.fullmatch(r"[A-Za-z0-9._-]+", p):
|
|
raise DestinationError(
|
|
"Subdirectory may only contain letters, digits, '.', '_' and '-'"
|
|
)
|
|
return "/".join(parts)
|
|
|
|
|
|
def mount_options(cfg: dict) -> str:
|
|
"""The destination's mount options, with ``nolock`` added for NFSv2/3.
|
|
|
|
An NFSv3 mount with locking makes the kernel register its lock manager
|
|
with rpcbind on the *Docker host*, which is only there when nfs-common (or
|
|
the distribution's equivalent) is installed; without it the mount fails
|
|
with a bare "connection refused". Only StackPilot ever writes these backup
|
|
files, one writer per file, so there is nothing to lock. An explicit
|
|
lock/nolock/local_lock option is left as given, and NFSv4 has locking
|
|
built into the protocol and needs no rpcbind at all.
|
|
"""
|
|
raw = (cfg.get("options") or "rw").strip().strip(",")
|
|
parts = [p.strip() for p in raw.split(",") if p.strip()]
|
|
chosen = any(p in ("lock", "nolock") or p.startswith("local_lock=") for p in parts)
|
|
v4 = any(p.split("=", 1)[0] in ("nfsvers", "vers") and p.split("=", 1)[-1].startswith("4")
|
|
for p in parts)
|
|
if not chosen and not v4:
|
|
parts.append("nolock")
|
|
return ",".join(parts)
|
|
|
|
|
|
def _nfs_volume(dest: BackupDestination, cfg: dict) -> str:
|
|
"""Ensure the named volume describing this NFS mount exists; recreate it
|
|
when the destination's server/path/options changed (opts are immutable)."""
|
|
from docker_client import DockerError, get_client, safe_call
|
|
|
|
server = (cfg.get("server") or "").strip()
|
|
path = (cfg.get("path") or "").strip()
|
|
if not server:
|
|
raise DestinationError("NFS server is required")
|
|
if not path.startswith("/"):
|
|
raise DestinationError("NFS export path must be absolute (start with /)")
|
|
driver_opts = {"type": "nfs", "o": f"addr={server},{mount_options(cfg)}", "device": f":{path}"}
|
|
|
|
name = f"{_NFS_VOLUME_PREFIX}{dest.id}"
|
|
client = get_client()
|
|
try:
|
|
vol = safe_call(client.volumes.get, name)
|
|
except DockerError:
|
|
vol = None
|
|
if vol is not None and (vol.attrs.get("Options") or {}) != driver_opts:
|
|
# Stopped containers still hold the volume; before 0.66.1 every failed
|
|
# mount left one behind, and then new settings never took effect.
|
|
clear_nfs_leftovers(client, name)
|
|
try:
|
|
safe_call(vol.remove)
|
|
except DockerError as exc:
|
|
raise DestinationError(
|
|
f"The NFS settings changed, but the Docker volume '{name}' is still in use "
|
|
f"and cannot be recreated with them: {exc.detail or exc.error}"
|
|
) from exc
|
|
vol = None
|
|
if vol is None:
|
|
safe_call(
|
|
client.volumes.create,
|
|
name=name,
|
|
driver="local",
|
|
driver_opts=driver_opts,
|
|
labels={"stackpilot.nfs-destination": str(dest.id)},
|
|
)
|
|
return name
|
|
|
|
|
|
def clear_nfs_leftovers(client, volume: Optional[str] = None) -> int:
|
|
"""Remove stopped containers holding an NFS destination volume.
|
|
|
|
Only StackPilot's helpers ever mount these volumes, so any container on
|
|
one that is not running is a leftover (older versions left one behind per
|
|
failed mount). ``volume=None`` covers every NFS destination.
|
|
"""
|
|
from docker_client import DockerError, safe_call
|
|
|
|
if volume:
|
|
names = [volume]
|
|
else:
|
|
try:
|
|
vols = safe_call(
|
|
client.volumes.list, filters={"label": "stackpilot.nfs-destination"}
|
|
)
|
|
except DockerError:
|
|
return 0
|
|
names = [v.name for v in vols]
|
|
removed = 0
|
|
for name in names:
|
|
try:
|
|
containers = safe_call(client.containers.list, all=True, filters={"volume": name})
|
|
except DockerError:
|
|
continue
|
|
for container in containers:
|
|
if container.status == "running":
|
|
continue
|
|
try:
|
|
container.remove(force=True)
|
|
removed += 1
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
return removed
|
|
|
|
|
|
def _nfs_target(cfg: dict) -> str:
|
|
sub = _nfs_subdir(cfg)
|
|
return f"/nfs/{sub}" if sub else "/nfs"
|
|
|
|
|
|
def _mount_error(exc: Exception) -> DestinationError:
|
|
"""A mount failure, with what to do about the common one."""
|
|
text = str(exc)
|
|
message = f"NFS mount failed: {text}"
|
|
lowered = text.lower()
|
|
if "connection refused" in lowered:
|
|
message += (
|
|
" — a connection was refused. With locking enabled ('lock' in the mount "
|
|
"options) NFSv3 needs rpcbind running on the Docker host itself: install "
|
|
"nfs-common there, or drop 'lock'. Otherwise check that NFS is enabled on the "
|
|
"server and that ports 111, 2049 and mountd are reachable from the Docker host."
|
|
)
|
|
elif "protocol not supported" in lowered:
|
|
message += (
|
|
" — the Docker host's kernel cannot speak this NFS version: load the module "
|
|
"(e.g. 'modprobe nfsv4') or remove nfsvers from the mount options."
|
|
)
|
|
return DestinationError(message)
|
|
|
|
|
|
def _nfs_run(volume: str, command: list[str]) -> str:
|
|
"""Run a helper container with the NFS volume at /nfs; return stdout."""
|
|
from config import settings
|
|
from docker_client import DockerError, get_long_client, run_container
|
|
from services.stack_assets_service import ensure_helper_image
|
|
|
|
client = get_long_client()
|
|
ensure_helper_image(client)
|
|
try:
|
|
out = run_container(
|
|
client,
|
|
settings.BACKUP_HELPER_IMAGE,
|
|
command,
|
|
volumes={volume: {"bind": "/nfs", "mode": "rw"}},
|
|
)
|
|
return out.decode("utf-8", "replace")
|
|
except DockerError as exc:
|
|
if exc.error == "container_failed":
|
|
raise DestinationError(f"NFS operation failed: {exc.detail}") from exc
|
|
# Mount errors surface here (unreachable server, bad export, ...).
|
|
raise _mount_error(exc) from exc
|
|
|
|
|
|
def _nfs_helper(volume: str, command: list[str] | str = "true"):
|
|
"""A created (not started) helper container for archive I/O on /nfs."""
|
|
import docker.errors
|
|
|
|
from config import settings
|
|
from docker_client import HELPER_LABEL, DockerError, get_client, safe_call
|
|
from services.stack_assets_service import ensure_helper_image
|
|
|
|
client = get_client()
|
|
ensure_helper_image(client)
|
|
try:
|
|
return safe_call(
|
|
client.containers.create,
|
|
settings.BACKUP_HELPER_IMAGE,
|
|
command=command,
|
|
volumes={volume: {"bind": "/nfs", "mode": "rw"}},
|
|
labels={HELPER_LABEL: "1"},
|
|
)
|
|
except (docker.errors.APIError, DockerError) as exc:
|
|
raise _mount_error(exc) from exc
|
|
|
|
|
|
def _nfs_upload(dest: BackupDestination, cfg: dict, local_path: str, filename: str) -> str:
|
|
import docker.errors
|
|
|
|
_nfs_check_name(filename)
|
|
volume = _nfs_volume(dest, cfg)
|
|
target = _nfs_target(cfg)
|
|
# Creates the subdir if needed AND fails early with a clear mount error.
|
|
_nfs_run(volume, ["mkdir", "-p", target])
|
|
# Unpack into the container's own filesystem, then copy the file across:
|
|
# extracting straight into the NFS mount makes the daemon chown the file,
|
|
# which a root_squash export refuses ("failed to Lchown ... for UID 0").
|
|
container = _nfs_helper(
|
|
volume, ["sh", "-c", f"cat '/tmp/{filename}' > '{target}/{filename}'"]
|
|
)
|
|
try:
|
|
with tempfile.TemporaryFile() as tmp:
|
|
with tarfile.open(fileobj=tmp, mode="w") as tar:
|
|
tar.add(local_path, arcname=filename)
|
|
tmp.seek(0)
|
|
container.put_archive("/tmp", tmp)
|
|
container.start()
|
|
status = container.wait(timeout=3600).get("StatusCode", 1)
|
|
if status != 0:
|
|
err = (container.logs(stdout=True, stderr=True) or b"").decode("utf-8", "replace")
|
|
raise DestinationError(f"NFS upload failed: {err.strip() or f'exit {status}'}")
|
|
except docker.errors.APIError as exc:
|
|
raise DestinationError(f"NFS upload failed: {exc}") from exc
|
|
finally:
|
|
try:
|
|
container.remove(force=True)
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
sub = _nfs_subdir(cfg)
|
|
return posixpath.join(sub, filename) if sub else filename
|
|
|
|
|
|
def _nfs_list(dest: BackupDestination, cfg: dict) -> list[dict]:
|
|
volume = _nfs_volume(dest, cfg)
|
|
target = _nfs_target(cfg)
|
|
out = _nfs_run(
|
|
volume,
|
|
["sh", "-c", f"cd {target} 2>/dev/null && stat -c '%n|%s|%Y' *.tar.gz 2>/dev/null; true"],
|
|
)
|
|
entries = []
|
|
for line in out.splitlines():
|
|
parts = line.strip().split("|")
|
|
if len(parts) != 3 or parts[0] == "*.tar.gz":
|
|
continue
|
|
try:
|
|
entries.append({"name": parts[0], "size": int(parts[1]), "modified": int(parts[2])})
|
|
except ValueError:
|
|
continue
|
|
return sorted(entries, key=lambda x: x["modified"] or 0, reverse=True)
|
|
|
|
|
|
def _nfs_download(dest: BackupDestination, cfg: dict, name: str, local_path: str) -> None:
|
|
import docker.errors
|
|
|
|
_nfs_check_name(name)
|
|
volume = _nfs_volume(dest, cfg)
|
|
target = _nfs_target(cfg)
|
|
container = _nfs_helper(volume)
|
|
try:
|
|
bits, _ = container.get_archive(f"{target}/{name}")
|
|
with tempfile.TemporaryFile() as tmp:
|
|
for chunk in bits:
|
|
tmp.write(chunk)
|
|
tmp.seek(0)
|
|
with tarfile.open(fileobj=tmp) as tar:
|
|
member = next((m for m in tar.getmembers() if m.isreg()), None)
|
|
fh = tar.extractfile(member) if member else None
|
|
if fh is None:
|
|
raise DestinationError(f"'{name}' not found on NFS destination")
|
|
with open(local_path, "wb") as out:
|
|
while chunk := fh.read(1024 * 1024):
|
|
out.write(chunk)
|
|
except docker.errors.APIError as exc:
|
|
raise DestinationError(f"NFS download failed (does '{name}' exist?): {exc}") from exc
|
|
finally:
|
|
try:
|
|
container.remove(force=True)
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
|
|
def _nfs_delete(dest: BackupDestination, cfg: dict, name: str) -> None:
|
|
_nfs_check_name(name)
|
|
volume = _nfs_volume(dest, cfg)
|
|
_nfs_run(volume, ["rm", "-f", f"{_nfs_target(cfg)}/{name}"])
|
|
|
|
|
|
def _nfs_test(dest: BackupDestination, cfg: dict) -> bool:
|
|
volume = _nfs_volume(dest, cfg)
|
|
target = _nfs_target(cfg)
|
|
_nfs_run(
|
|
volume,
|
|
["sh", "-c", f"mkdir -p {target} && touch {target}/.stackpilot-test && rm -f {target}/.stackpilot-test"],
|
|
)
|
|
return True
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Dispatch
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def upload(dest: BackupDestination, local_path: str, filename: str) -> str:
|
|
cfg = parse_config(dest)
|
|
if dest.type == "sftp":
|
|
return _sftp_upload(cfg, local_path, filename)
|
|
if dest.type == "s3":
|
|
return _s3_upload(cfg, local_path, filename)
|
|
if dest.type == "nfs":
|
|
return _nfs_upload(dest, cfg, local_path, filename)
|
|
raise DestinationError(f"Unknown destination type '{dest.type}'")
|
|
|
|
|
|
def list_backups(dest: BackupDestination) -> list[dict]:
|
|
cfg = parse_config(dest)
|
|
if dest.type == "sftp":
|
|
return _sftp_list(cfg)
|
|
if dest.type == "s3":
|
|
return _s3_list(cfg)
|
|
if dest.type == "nfs":
|
|
return _nfs_list(dest, cfg)
|
|
raise DestinationError(f"Unknown destination type '{dest.type}'")
|
|
|
|
|
|
def download(dest: BackupDestination, name: str, local_path: str) -> None:
|
|
cfg = parse_config(dest)
|
|
if dest.type == "sftp":
|
|
_sftp_download(cfg, name, local_path)
|
|
elif dest.type == "s3":
|
|
_s3_download(cfg, name, local_path)
|
|
elif dest.type == "nfs":
|
|
_nfs_download(dest, cfg, name, local_path)
|
|
else:
|
|
raise DestinationError(f"Unknown destination type '{dest.type}'")
|
|
|
|
|
|
def delete(dest: BackupDestination, name: str) -> None:
|
|
cfg = parse_config(dest)
|
|
if dest.type == "sftp":
|
|
_sftp_delete(cfg, name)
|
|
elif dest.type == "s3":
|
|
_s3_delete(cfg, name)
|
|
elif dest.type == "nfs":
|
|
_nfs_delete(dest, cfg, name)
|
|
else:
|
|
raise DestinationError(f"Unknown destination type '{dest.type}'")
|
|
|
|
|
|
def test(dest: BackupDestination) -> bool:
|
|
"""Connectivity check — validates reachability, auth and write access."""
|
|
if dest.type == "nfs":
|
|
return _nfs_test(dest, parse_config(dest))
|
|
list_backups(dest)
|
|
return True
|