#!/usr/bin/env python3 """Atomically transfer encrypted GuestOps backups to restricted storage.""" from __future__ import annotations import argparse import hashlib import json import os from pathlib import Path import re import shlex import shutil import stat import subprocess import tempfile import time import uuid BACKUP_NAME = re.compile(r"guestops-[0-9]{8}T[0-9]{6}Z\.tar\.gpg") SAFE_HOST = re.compile(r"[A-Za-z0-9.-]{1,253}") SAFE_USER = re.compile(r"[A-Za-z_][A-Za-z0-9_-]{0,31}") SAFE_REMOTE_PATH = re.compile(r"/[A-Za-z0-9._/-]{1,500}") SHA256 = re.compile(r"[0-9a-f]{64}") def require(condition: bool, message: str) -> None: if not condition: raise RuntimeError(message) def digest(path: Path) -> str: with path.open("rb") as stream: return hashlib.file_digest(stream, "sha256").hexdigest() def private_directory(value: str) -> Path: requested = Path(value) require(requested.is_absolute() and not requested.is_symlink(), "BACKUP_DIRECTORY must be an absolute, non-symlink path.") directory = requested.resolve() require(directory.is_dir(), "BACKUP_DIRECTORY must exist.") require(stat.S_IMODE(directory.stat().st_mode) & 0o077 == 0, "BACKUP_DIRECTORY must not be accessible to group or other users.") return directory def regular_file(value: str, field: str, *, private: bool) -> Path: requested = Path(value) require(requested.is_absolute() and not requested.is_symlink(), f"{field} must be an absolute, non-symlink path.") path = requested.resolve() require(path.is_file(), f"{field} must be an existing regular file.") if private: require(stat.S_IMODE(path.stat().st_mode) & 0o077 == 0, f"{field} must not be accessible to group or other users.") return path def configuration(environment: dict[str, str]) -> dict[str, object]: directory = private_directory(environment.get("BACKUP_DIRECTORY", "")) host = environment.get("BACKUP_REMOTE_HOST", "") user = environment.get("BACKUP_REMOTE_USER", "") remote = environment.get("BACKUP_REMOTE_DIRECTORY", "") require(SAFE_HOST.fullmatch(host) is not None, "BACKUP_REMOTE_HOST is invalid.") require(SAFE_USER.fullmatch(user) is not None, "BACKUP_REMOTE_USER is invalid.") require(SAFE_REMOTE_PATH.fullmatch(remote) is not None and "//" not in remote and "/../" not in remote + "/" and not remote.endswith("/.."), "BACKUP_REMOTE_DIRECTORY must be a safe absolute path.") identity = regular_file(environment.get("BACKUP_SSH_IDENTITY", ""), "BACKUP_SSH_IDENTITY", private=True) known_hosts = regular_file(environment.get("BACKUP_SSH_KNOWN_HOSTS", ""), "BACKUP_SSH_KNOWN_HOSTS", private=False) require(shutil.which("ssh") is not None and shutil.which("rsync") is not None, "ssh and rsync are required.") return { "directory": directory, "host": host, "user": user, "remote": remote.rstrip("/"), "identity": identity, "known_hosts": known_hosts, } def ssh_base(config: dict[str, object]) -> list[str]: return [ "ssh", "-o", "BatchMode=yes", "-o", "IdentitiesOnly=yes", "-o", "StrictHostKeyChecking=yes", "-o", "ConnectTimeout=15", "-o", f"UserKnownHostsFile={config['known_hosts']}", "-i", str(config["identity"]), f"{config['user']}@{config['host']}", ] def run(args: list[str], *, environment: dict[str, str] | None = None) -> bytes: result = subprocess.run(args, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=900, env=environment) require(result.returncode == 0, f"{Path(args[0]).name} step failed; review the restricted operator logs.") return result.stdout def remote_digest(config: dict[str, object], remote_path: str) -> str | None: command = f"if test -f {remote_path} && test ! -L {remote_path}; then sha256sum -- {remote_path}; fi" output = run([*ssh_base(config), command]).decode("utf-8", "strict").strip() if not output: return None value = output.split()[0] require(SHA256.fullmatch(value) is not None, "Remote checksum response was invalid.") return value def marker_path(backup: Path) -> Path: return backup.with_name(backup.name + ".transferred.json") def write_marker(backup: Path, checksum: str) -> None: marker = marker_path(backup) payload = json.dumps({ "schemaVersion": 1, "backup": backup.name, "sha256": checksum, "verifiedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), }, sort_keys=True) + "\n" fd, temporary = tempfile.mkstemp(prefix=marker.name + ".", dir=marker.parent) try: os.fchmod(fd, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as stream: stream.write(payload) stream.flush() os.fsync(stream.fileno()) os.replace(temporary, marker) finally: try: os.unlink(temporary) except FileNotFoundError: pass def transfer_one(config: dict[str, object], backup: Path) -> None: require(backup.is_file() and not backup.is_symlink() and BACKUP_NAME.fullmatch(backup.name) is not None, "Refusing to transfer an unexpected backup path.") checksum = digest(backup) remote_final = f"{config['remote']}/{backup.name}" existing = remote_digest(config, remote_final) if existing is not None: require(existing == checksum, "A remote backup with this name has a different checksum.") write_marker(backup, checksum) return remote_partial = f"{config['remote']}/.{backup.name}.partial-{uuid.uuid4().hex}" rsh = shlex.join([ "ssh", "-o", "BatchMode=yes", "-o", "IdentitiesOnly=yes", "-o", "StrictHostKeyChecking=yes", "-o", "ConnectTimeout=15", "-o", f"UserKnownHostsFile={config['known_hosts']}", "-i", str(config["identity"]), ]) rsync_environment = os.environ.copy() rsync_environment["RSYNC_RSH"] = rsh try: run(["rsync", "--archive", "--chmod=F600", "--protect-args", "--", str(backup), f"{config['user']}@{config['host']}:{remote_partial}"], environment=rsync_environment) require(remote_digest(config, remote_partial) == checksum, "Transferred backup checksum does not match the local file.") command = (f"test ! -e {remote_final} && mv -T -- {remote_partial} {remote_final} " f"&& chmod 600 -- {remote_final}") run([*ssh_base(config), command]) require(remote_digest(config, remote_final) == checksum, "Final remote backup checksum does not match the local file.") write_marker(backup, checksum) except Exception: # The name contains a fresh random suffix and is the only remote object this run may remove. subprocess.run([*ssh_base(config), f"rm -f -- {remote_partial}"], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=30) raise def transfer_pending(config: dict[str, object]) -> int: directory = config["directory"] backups = sorted(path for path in directory.iterdir() if path.is_file() and not path.is_symlink() and BACKUP_NAME.fullmatch(path.name) is not None) for backup in backups: marker = marker_path(backup) if marker.is_file() and not marker.is_symlink(): try: recorded = json.loads(marker.read_text(encoding="utf-8")) if recorded.get("sha256") == digest(backup): continue except (OSError, ValueError, json.JSONDecodeError): pass transfer_one(config, backup) return len(backups) def prune_verified(config: dict[str, object], retention_days: int, now: float | None = None) -> int: require(1 <= retention_days <= 365, "Local retention must be between 1 and 365 days.") threshold = (time.time() if now is None else now) - retention_days * 86400 removed = 0 for backup in config["directory"].iterdir(): if not backup.is_file() or backup.is_symlink() or BACKUP_NAME.fullmatch(backup.name) is None: continue marker = marker_path(backup) if backup.stat().st_mtime >= threshold or not marker.is_file() or marker.is_symlink(): continue try: recorded = json.loads(marker.read_text(encoding="utf-8")) except (OSError, ValueError, json.JSONDecodeError): continue if recorded.get("backup") != backup.name or recorded.get("sha256") != digest(backup): continue backup.unlink() marker.unlink() removed += 1 return removed def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--prune-verified", action="store_true") parser.add_argument("--retention-days", type=int, default=7) args = parser.parse_args() config = configuration(dict(os.environ)) observed = transfer_pending(config) removed = prune_verified(config, args.retention_days) if args.prune_verified else 0 print(f"Backup transfer completed: {observed} encrypted backup(s) inspected; " f"{removed} verified local backup(s) expired.") if __name__ == "__main__": try: main() except (OSError, RuntimeError, subprocess.SubprocessError, json.JSONDecodeError) as error: print(f"Backup transfer failed: {error}", file=__import__("sys").stderr) raise SystemExit(1)