diff --git a/sync-alfred.py b/sync-alfred.py new file mode 100755 index 0000000..66ed2e7 --- /dev/null +++ b/sync-alfred.py @@ -0,0 +1,472 @@ +#!/usr/bin/env python3 +"""Create and restore a marked ZIP backup of Alfred preferences over WebDAV.""" + +from __future__ import annotations + +import argparse +import base64 +import json +import logging +import os +import shutil +import socket +import stat +import tempfile +import time +import urllib.error +import urllib.parse +import urllib.request +import zipfile +from datetime import datetime, timezone +from pathlib import Path, PurePosixPath + + +PACKAGE_NAME = "Alfred.alfredpreferences" +ARCHIVE_NAME = f"{PACKAGE_NAME}.sync-backup.zip" +MARKER_NAME = ".alfred-sync-backup" +FORMAT_VERSION = 1 +DEFAULT_WEBDAV_URL = ( + "https://u201686:T6672ICVoWAedECH@u201686.your-storagebox.de/" +) +DEFAULT_REMOTE_PATH = "backup/macs/alfred" +WEBDAV_TIMEOUT = 30 +LOGGER = logging.getLogger("sync-alfred") + + +class SyncError(RuntimeError): + pass + + +class ProgressIO: + """Wrap a readable binary file and report transfer progress once per second.""" + + def __init__( + self, + file_obj, + *, + total: int | None = None, + label: str = "Progress", + ) -> None: + self.file_obj = file_obj + self.total = total + self.label = label + self.transferred = 0 + self.started_at = time.monotonic() + self.last_report_at = self.started_at - 1 + self.finished = False + self._report(force=True) + + def __getattr__(self, name): + return getattr(self.file_obj, name) + + def __len__(self) -> int: + return self.total or 0 + + def read(self, size: int = -1): + chunk = self.file_obj.read(size) + if chunk: + self.transferred += len(chunk) + self._report() + else: + self._report(force=True, complete=True) + return chunk + + def finish(self) -> None: + self._report(force=True, complete=True) + + @staticmethod + def _format_bytes(value: float) -> str: + units = ("B", "KiB", "MiB", "GiB") + for unit in units: + if value < 1024 or unit == units[-1]: + return f"{value:.1f} {unit}" + value /= 1024 + return f"{value:.1f} GiB" + + def _report(self, *, force: bool = False, complete: bool = False) -> None: + if self.finished: + return + now = time.monotonic() + if not force and now - self.last_report_at < 1: + return + + self.last_report_at = now + elapsed = max(now - self.started_at, 0.001) + transferred = self._format_bytes(self.transferred) + rate = self._format_bytes(self.transferred / elapsed) + if self.total is None: + message = f"{self.label}: {transferred} at {rate}/s" + else: + percent = 100 * self.transferred / self.total if self.total else 100 + total = self._format_bytes(self.total) + message = f"{self.label}: {transferred} / {total} ({percent:.1f}%) at {rate}/s" + + LOGGER.info(message) + if complete: + self.finished = True + + +class WebDAVClient: + def __init__(self, base_url: str, remote_path: str) -> None: + parsed = urllib.parse.urlsplit(base_url) + if parsed.scheme not in ("http", "https") or not parsed.hostname: + raise SyncError("invalid WebDAV URL") + if parsed.query or parsed.fragment: + raise SyncError("WebDAV URL must not contain a query or fragment") + + username = urllib.parse.unquote(parsed.username or "") + password = urllib.parse.unquote(parsed.password or "") + if not username: + username = os.environ.get("ALFRED_WEBDAV_USER", "") + if not password: + password = os.environ.get("ALFRED_WEBDAV_PASSWORD", "") + + # Strip credentials from request URLs and diagnostic output. + netloc = parsed.netloc.rsplit("@", 1)[-1] + base_path = parsed.path.rstrip("/") + self.base_url = urllib.parse.urlunsplit( + (parsed.scheme, netloc, f"{base_path}/", "", "") + ) + self.remote_parts = self._parse_remote_path(remote_path) + self.auth_header = None + + password_manager = urllib.request.HTTPPasswordMgrWithDefaultRealm() + if username and password: + encoded_credentials = base64.b64encode( + f"{username}:{password}".encode("utf-8") + ).decode("ascii") + self.auth_header = f"Basic {encoded_credentials}" + origin = urllib.parse.urlunsplit( + (parsed.scheme, netloc, "/", "", "") + ) + password_manager.add_password(None, origin, username, password) + self.opener = urllib.request.build_opener( + urllib.request.HTTPBasicAuthHandler(password_manager) + ) + + @staticmethod + def _parse_remote_path(remote_path: str) -> tuple[str, ...]: + parts = tuple(remote_path.strip("/").split("/")) + if not parts or any(not part or part in (".", "..") for part in parts): + raise SyncError(f"invalid WebDAV remote path: {remote_path!r}") + return parts + + def _url_for(self, parts: tuple[str, ...]) -> str: + encoded = "/".join(urllib.parse.quote(part, safe="") for part in parts) + return f"{self.base_url.rstrip('/')}/{encoded}" + + @property + def archive_url(self) -> str: + return self._url_for(self.remote_parts + (ARCHIVE_NAME,)) + + @property + def display_location(self) -> str: + return self.archive_url + + def _request( + self, + method: str, + url: str, + *, + data: object | None = None, + headers: dict[str, str] | None = None, + ) -> urllib.response.addinfourl: + request_headers = dict(headers or {}) + if self.auth_header: + request_headers.setdefault("Authorization", self.auth_header) + request = urllib.request.Request( + url, data=data, headers=request_headers, method=method + ) + try: + return self.opener.open(request, timeout=WEBDAV_TIMEOUT) + except urllib.error.HTTPError as exc: + detail = f"HTTP {exc.code} {exc.reason}" + raise SyncError( + f"WebDAV {method} failed for remote archive: {detail}" + ) from exc + except urllib.error.URLError as exc: + raise SyncError(f"WebDAV {method} failed: {exc.reason}") from exc + + def ensure_remote_directory(self) -> None: + for index in range(1, len(self.remote_parts) + 1): + # WebDAV collections are directory-like resources and this server + # redirects collection URLs without their trailing slash. + url = f"{self._url_for(self.remote_parts[:index])}/" + headers = {"Authorization": self.auth_header} if self.auth_header else {} + request = urllib.request.Request(url, headers=headers, method="MKCOL") + try: + response = self.opener.open(request, timeout=WEBDAV_TIMEOUT) + except urllib.error.HTTPError as exc: + # WebDAV uses 405 when MKCOL is sent for an existing collection. + if exc.code != 405: + detail = f"HTTP {exc.code} {exc.reason}" + raise SyncError( + f"WebDAV MKCOL failed for remote directory: {detail}" + ) from exc + exc.close() + except urllib.error.URLError as exc: + raise SyncError(f"WebDAV MKCOL failed: {exc.reason}") from exc + else: + response.close() + + def upload(self, archive_path: Path) -> None: + self.ensure_remote_directory() + try: + total = archive_path.stat().st_size + source = archive_path.open("rb") + except OSError as exc: + raise SyncError(f"could not open archive {archive_path}: {exc}") from exc + + progress = ProgressIO(source, total=total, label="Uploading") + try: + with source: + with self._request( + "PUT", + self.archive_url, + data=progress, + headers={ + "Content-Type": "application/zip", + "Content-Length": str(total), + }, + ) as response: + response.read() + finally: + progress.finish() + + def download(self, archive_path: Path) -> None: + with self._request("GET", self.archive_url) as response: + try: + with archive_path.open("wb") as output: + shutil.copyfileobj(response, output) + output.flush() + os.fsync(output.fileno()) + except OSError as exc: + raise SyncError( + f"could not write archive {archive_path}: {exc}" + ) from exc + + +def configured_paths() -> tuple[Path, WebDAVClient]: + local_dir = Path( + os.environ.get("ALFRED_LOCAL_DIR", "~/Documents/alfred") + ).expanduser() + webdav = WebDAVClient( + os.environ.get("ALFRED_WEBDAV_URL", DEFAULT_WEBDAV_URL), + os.environ.get("ALFRED_REMOTE_PATH", DEFAULT_REMOTE_PATH), + ) + return local_dir, webdav + + +def marker(kind: str) -> bytes: + contents = { + "format": FORMAT_VERSION, + "package": PACKAGE_NAME, + "kind": kind, + "created_utc": datetime.now(timezone.utc).isoformat(), + "source_host": socket.gethostname(), + } + return (json.dumps(contents, sort_keys=True) + "\n").encode("utf-8") + + +def create_archive(source_dir: Path, archive_path: Path, kind: str) -> None: + if not source_dir.is_dir(): + raise SyncError(f"local preferences directory not found: {source_dir}") + + try: + with zipfile.ZipFile( + archive_path, "w", compression=zipfile.ZIP_DEFLATED + ) as archive: + archive.write(source_dir, PACKAGE_NAME) + for path in sorted(source_dir.rglob("*"), key=lambda item: item.as_posix()): + if path.is_symlink(): + raise SyncError(f"symlinks are not supported in preferences: {path}") + relative = path.relative_to(source_dir) + archive.write( + path, + PurePosixPath(PACKAGE_NAME, *relative.parts).as_posix(), + ) + archive.writestr(MARKER_NAME, marker(kind)) + except OSError as exc: + raise SyncError(f"could not create archive {archive_path}: {exc}") from exc + + +def validate_archive(archive_path: Path) -> None: + try: + with zipfile.ZipFile(archive_path) as archive: + broken_file = archive.testzip() + if broken_file is not None: + raise SyncError(f"backup archive is corrupt: {broken_file}") + + names = archive.namelist() + if names.count(MARKER_NAME) != 1: + raise SyncError(f"backup archive must contain one {MARKER_NAME} marker") + + try: + metadata = json.loads(archive.read(MARKER_NAME).decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError, KeyError) as exc: + raise SyncError("backup marker is invalid") from exc + + if metadata.get("format") != FORMAT_VERSION: + raise SyncError("unsupported Alfred backup format") + if metadata.get("package") != PACKAGE_NAME: + raise SyncError("backup marker is for a different package") + if metadata.get("kind") != "backup": + raise SyncError("archive is not a primary Alfred backup") + + seen: set[str] = set() + has_package = False + for name in names: + if name in seen: + raise SyncError(f"duplicate file in backup archive: {name}") + seen.add(name) + + if "\\" in name: + raise SyncError(f"unsafe path in backup archive: {name}") + path = PurePosixPath(name) + if path.is_absolute() or ".." in path.parts: + raise SyncError(f"unsafe path in backup archive: {name}") + + if name == MARKER_NAME: + continue + if name == PACKAGE_NAME or name.startswith(f"{PACKAGE_NAME}/"): + has_package = True + continue + raise SyncError(f"unexpected file in backup archive: {name}") + + if not has_package: + raise SyncError(f"backup archive does not contain {PACKAGE_NAME}") + except zipfile.BadZipFile as exc: + raise SyncError(f"backup archive is not a valid ZIP: {archive_path}") from exc + + +def extract_archive(archive_path: Path, destination: Path) -> Path: + package_destination = destination / PACKAGE_NAME + destination.mkdir(parents=True, exist_ok=True) + + with zipfile.ZipFile(archive_path) as archive: + for info in archive.infolist(): + if info.filename == MARKER_NAME: + continue + + relative = PurePosixPath(info.filename) + target = destination.joinpath(*relative.parts) + if info.is_dir() or info.filename.endswith("/"): + target.mkdir(parents=True, exist_ok=True) + continue + + mode = (info.external_attr >> 16) & 0xFFFF + if stat.S_ISLNK(mode): + raise SyncError(f"symlinks are not supported in backup archives: {info.filename}") + + target.parent.mkdir(parents=True, exist_ok=True) + with archive.open(info) as source, target.open("wb") as output: + shutil.copyfileobj(source, output) + permissions = mode & 0o777 + if permissions: + target.chmod(permissions) + + if not package_destination.is_dir(): + raise SyncError(f"backup archive does not contain {PACKAGE_NAME}") + return package_destination + + +def unique_backup_path(local_dir: Path) -> Path: + timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") + candidate = local_dir / f"{PACKAGE_NAME}.before-restore-{timestamp}.zip" + if not candidate.exists(): + return candidate + return local_dir / f"{PACKAGE_NAME}.before-restore-{timestamp}-{os.getpid()}.zip" + + +def backup_current_preferences(local_dir: Path, temporary_archive: Path) -> Path | None: + package = local_dir / PACKAGE_NAME + if not package.exists(): + return None + if not package.is_dir(): + raise SyncError(f"local preferences path is not a directory: {package}") + + backup_path = unique_backup_path(local_dir) + create_archive(package, temporary_archive, "pre-restore") + os.replace(temporary_archive, backup_path) + return backup_path + + +def backup(local_dir: Path, webdav: WebDAVClient) -> None: + local_dir.mkdir(parents=True, exist_ok=True) + + with tempfile.TemporaryDirectory(prefix=".alfred-backup-", dir=local_dir) as temp: + archive = Path(temp) / ARCHIVE_NAME + create_archive(local_dir / PACKAGE_NAME, archive, "backup") + webdav.upload(archive) + + LOGGER.info("Backed up %s to %s", PACKAGE_NAME, webdav.display_location) + + +def restore(local_dir: Path, webdav: WebDAVClient) -> None: + local_dir.mkdir(parents=True, exist_ok=True) + with tempfile.TemporaryDirectory(prefix=".alfred-restore-", dir=local_dir) as temp: + temp_dir = Path(temp) + archive = temp_dir / ARCHIVE_NAME + webdav.download(archive) + validate_archive(archive) + restored_package = extract_archive(archive, temp_dir) + previous_backup = backup_current_preferences( + local_dir, temp_dir / "current-pre-restore.zip" + ) + if previous_backup is not None: + LOGGER.info("Backed up current preferences to %s", previous_backup) + + old_container = Path( + tempfile.mkdtemp(prefix=".alfred-old-", dir=local_dir) + ) + old_package = old_container / PACKAGE_NAME + current_package = local_dir / PACKAGE_NAME + try: + if current_package.exists(): + os.replace(current_package, old_package) + os.replace(restored_package, current_package) + except OSError as exc: + if old_package.exists() and not current_package.exists(): + os.replace(old_package, current_package) + raise SyncError(f"could not install restored preferences: {exc}") from exc + else: + shutil.rmtree(old_container) + + LOGGER.info("Restored %s from %s", PACKAGE_NAME, webdav.display_location) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Back up or restore Alfred preferences using a marked ZIP archive." + ) + parser.add_argument( + "mode", + nargs="?", + default="backup", + choices=("backup", "push", "restore", "pull", "apply"), + help="backup (default) or restore; push/pull/apply are aliases", + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + logging.basicConfig(level=logging.INFO, format="%(message)s") + + try: + local_dir, webdav = configured_paths() + if args.mode in ("backup", "push"): + backup(local_dir, webdav) + else: + restore(local_dir, webdav) + except SyncError as exc: + LOGGER.error("sync-alfred: %s", exc) + return 1 + except OSError as exc: + LOGGER.error("sync-alfred: filesystem error: %s", exc) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())