"""Push/pull stack backups to remote destinations (SFTP or S3-compatible). All operations are synchronous (paramiko / boto3); async callers should wrap them with ``asyncio.to_thread``. Destination config is a plain dict parsed from the ``BackupDestination.config`` JSON column. """ from __future__ import annotations import io import json import logging import os import posixpath import stat from typing import Any from models.backup_destination import BackupDestination logger = logging.getLogger("stackpilot.backup_dest") class DestinationError(Exception): pass def parse_config(dest: BackupDestination) -> dict: try: return json.loads(dest.config or "{}") except json.JSONDecodeError: return {} # --------------------------------------------------------------------------- # # 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: resp = client.list_objects_v2(**kwargs) except Exception as exc: # noqa: BLE001 raise DestinationError(f"S3 list failed: {exc}") from exc out = [] for obj in resp.get("Contents", []): 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)) # --------------------------------------------------------------------------- # # 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) 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) 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) 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) else: raise DestinationError(f"Unknown destination type '{dest.type}'") def test(dest: BackupDestination) -> bool: """Connectivity check — lists the target (cheap, validates auth + path).""" list_backups(dest) return True