static-site-deploy/static_site.py

#!/usr/bin/env python3
"""Deploy a built static website to Bunny Storage or an SSH endpoint."""

from __future__ import annotations

import argparse
import hashlib
import json
import mimetypes
import os
import subprocess
import sys
import tarfile
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Any


MANIFEST_NAME = ".static-site-manifest.json"
MANIFEST_VERSION = 1


class DeployError(RuntimeError):
    """A static-site deployment could not be completed safely."""


@dataclass(frozen=True)
class LocalFile:
    path: str
    source: Path
    size: int
    sha256: str


def collect_files(root: Path) -> dict[str, LocalFile]:
    root = root.resolve()
    if not root.is_dir():
        raise DeployError(f"build directory does not exist: {root}")

    files: dict[str, LocalFile] = {}
    for source in sorted(root.rglob("*")):
        if source.is_symlink():
            raise DeployError(f"symbolic links are not supported: {source.relative_to(root)}")
        if not source.is_file():
            continue
        relative = source.relative_to(root).as_posix()
        if relative == MANIFEST_NAME:
            raise DeployError(f"{MANIFEST_NAME} is reserved by the deployer")
        digest = hashlib.sha256(source.read_bytes()).hexdigest()
        files[relative] = LocalFile(relative, source, source.stat().st_size, digest)
    if "index.html" not in files:
        raise DeployError("build directory must contain index.html")
    return files


def manifest_for(files: dict[str, LocalFile]) -> dict[str, Any]:
    return {
        "version": MANIFEST_VERSION,
        "files": {
            path: {"sha256": item.sha256, "size": item.size}
            for path, item in sorted(files.items())
        },
    }


def parse_manifest(raw: bytes | None) -> dict[str, dict[str, Any]]:
    if raw is None:
        return {}
    try:
        value = json.loads(raw)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise DeployError(f"remote {MANIFEST_NAME} is not valid JSON") from exc
    if not isinstance(value, dict) or value.get("version") != MANIFEST_VERSION:
        raise DeployError(f"remote {MANIFEST_NAME} has an unsupported version")
    files = value.get("files")
    if not isinstance(files, dict) or not all(isinstance(key, str) for key in files):
        raise DeployError(f"remote {MANIFEST_NAME} has an invalid files map")
    for path, metadata in files.items():
        parts = Path(path).parts
        if (
            not path
            or path.startswith("/")
            or ".." in parts
            or path == MANIFEST_NAME
            or not isinstance(metadata, dict)
        ):
            raise DeployError(f"remote {MANIFEST_NAME} contains an unsafe path")
    return files


class BunnyStorage:
    def __init__(self, zone: str, password: str, endpoint: str) -> None:
        if not zone or not password:
            raise DeployError("BUNNY_STORAGE_ZONE and BUNNY_STORAGE_PASSWORD are required")
        endpoint = endpoint.removeprefix("https://").rstrip("/")
        if not endpoint or "/" in endpoint:
            raise DeployError("Bunny storage endpoint must be a hostname")
        self.zone = zone
        self.password = password
        self.base_url = f"https://{endpoint}/{urllib.parse.quote(zone, safe='')}"

    def _request(
        self,
        method: str,
        path: str,
        *,
        body: bytes | None = None,
        headers: dict[str, str] | None = None,
        missing_ok: bool = False,
    ) -> bytes | None:
        url_path = urllib.parse.quote(path, safe="/")
        request_headers = {"AccessKey": self.password, **(headers or {})}
        request = urllib.request.Request(
            f"{self.base_url}/{url_path}",
            data=body,
            headers=request_headers,
            method=method,
        )
        try:
            with urllib.request.urlopen(request, timeout=60) as response:
                return response.read()
        except urllib.error.HTTPError as exc:
            if missing_ok and exc.code == 404:
                return None
            detail = exc.read().decode("utf-8", errors="replace")
            raise DeployError(
                f"Bunny Storage {method} {path} failed: HTTP {exc.code}: {detail}"
            ) from exc
        except urllib.error.URLError as exc:
            raise DeployError(f"Bunny Storage {method} {path} failed: {exc.reason}") from exc

    def get_manifest(self) -> bytes | None:
        return self._request("GET", MANIFEST_NAME, missing_ok=True)

    def upload(self, item: LocalFile) -> None:
        content_type = mimetypes.guess_type(item.path)[0] or "application/octet-stream"
        self._request(
            "PUT",
            item.path,
            body=item.source.read_bytes(),
            headers={
                "Checksum": item.sha256.upper(),
                "Content-Type": content_type,
            },
        )

    def upload_manifest(self, manifest: dict[str, Any]) -> None:
        body = (json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n").encode()
        self._request(
            "PUT",
            MANIFEST_NAME,
            body=body,
            headers={
                "Checksum": hashlib.sha256(body).hexdigest().upper(),
                "Content-Type": "application/json",
            },
        )

    def delete(self, path: str) -> None:
        self._request("DELETE", path, missing_ok=True)


def bunny_plan(
    local: dict[str, LocalFile], remote: dict[str, dict[str, Any]]
) -> tuple[list[str], list[str]]:
    upload = [
        path
        for path, item in local.items()
        if not isinstance(remote.get(path), dict)
        or remote[path].get("sha256") != item.sha256
        or remote[path].get("size") != item.size
    ]
    delete = sorted(set(remote) - set(local))
    # Publish assets before documents that may start referring to them.
    upload.sort(key=lambda path: (path.lower().endswith(('.html', '.htm')), path))
    return upload, delete


def purge_bunny_pull_zone(pull_zone_id: str, api_key: str) -> None:
    if not api_key:
        raise DeployError("BUNNY_API_KEY is required when BUNNY_PULL_ZONE_ID is set")
    request = urllib.request.Request(
        f"https://api.bunny.net/pullzone/{urllib.parse.quote(pull_zone_id, safe='')}/purgeCache",
        headers={"AccessKey": api_key},
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=60):
            return
    except urllib.error.HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        raise DeployError(f"Bunny cache purge failed: HTTP {exc.code}: {detail}") from exc
    except urllib.error.URLError as exc:
        raise DeployError(f"Bunny cache purge failed: {exc.reason}") from exc


def purge_bunny_gateway(endpoint: str, zone: str, password: str) -> None:
    parsed = urllib.parse.urlsplit(endpoint)
    if (
        parsed.scheme != "https"
        or not parsed.hostname
        or parsed.username is not None
        or parsed.password is not None
        or parsed.port is not None
        or parsed.path not in ("", "/")
        or parsed.query
        or parsed.fragment
    ):
        raise DeployError("BUNNY_PURGE_ENDPOINT must be an HTTPS origin without a path")
    url = (
        f"https://{parsed.hostname}/v1/purge/"
        f"{urllib.parse.quote(zone, safe='')}"
    )
    request = urllib.request.Request(
        url,
        data=b"",
        headers={"Authorization": f"Bearer {password}", "Accept": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=60):
            return
    except urllib.error.HTTPError as exc:
        exc.read()
        raise DeployError(f"Bunny purge gateway failed: HTTP {exc.code}") from exc
    except urllib.error.URLError as exc:
        raise DeployError(f"Bunny purge gateway failed: {exc.reason}") from exc


def deploy_bunny(args: argparse.Namespace) -> dict[str, Any]:
    if args.purge_endpoint and args.pull_zone_id:
        raise DeployError(
            "BUNNY_PURGE_ENDPOINT and BUNNY_PULL_ZONE_ID are mutually exclusive"
        )
    files = collect_files(args.directory)
    storage = BunnyStorage(args.zone, args.password, args.endpoint)
    remote = parse_manifest(storage.get_manifest())
    upload, delete = bunny_plan(files, remote)
    result = {
        "backend": "bunny",
        "changed": bool(upload or delete),
        "dry_run": args.dry_run,
        "upload": upload,
        "delete": delete,
        "purged": False,
    }
    if args.dry_run:
        return result

    if result["changed"]:
        for path in upload:
            storage.upload(files[path])
        for path in delete:
            storage.delete(path)
        storage.upload_manifest(manifest_for(files))
    if args.purge_endpoint:
        purge_bunny_gateway(args.purge_endpoint, args.zone, args.password)
        result["purged"] = True
    elif args.pull_zone_id:
        purge_bunny_pull_zone(args.pull_zone_id, args.api_key)
        result["purged"] = True
    return result


def deploy_ssh(args: argparse.Namespace) -> dict[str, Any]:
    files = collect_files(args.directory)
    if args.dry_run:
        return {
            "backend": "ssh",
            "changed": True,
            "dry_run": True,
            "files": len(files),
            "target": args.target,
            "site": args.site,
        }

    command = ["ssh", "-p", str(args.port)]
    if args.identity:
        command.extend(["-i", str(args.identity)])
    command.extend([args.target, args.site])
    process = subprocess.Popen(command, stdin=subprocess.PIPE)
    assert process.stdin is not None
    try:
        with tarfile.open(fileobj=process.stdin, mode="w|gz") as archive:
            for path, item in sorted(files.items()):
                archive.add(item.source, arcname=path, recursive=False)
    except BrokenPipeError:
        pass
    return_code = process.wait()
    if return_code:
        raise DeployError(f"SSH deployment failed with exit code {return_code}")
    return {
        "backend": "ssh",
        "changed": True,
        "dry_run": False,
        "files": len(files),
        "target": args.target,
        "site": args.site,
    }


def parser() -> argparse.ArgumentParser:
    result = argparse.ArgumentParser(
        description="Deploy a completed static-site build to Bunny or an SSH endpoint."
    )
    subparsers = result.add_subparsers(dest="backend", required=True)

    bunny = subparsers.add_parser("bunny", help="Sync a build to Bunny Storage.")
    bunny.add_argument("directory", type=Path)
    bunny.add_argument("--zone", default=os.environ.get("BUNNY_STORAGE_ZONE", ""))
    bunny.add_argument(
        "--endpoint",
        default=os.environ.get("BUNNY_STORAGE_ENDPOINT", "storage.bunnycdn.com"),
        help="Regional storage hostname (default: BUNNY_STORAGE_ENDPOINT or Frankfurt).",
    )
    bunny.add_argument(
        "--pull-zone-id", default=os.environ.get("BUNNY_PULL_ZONE_ID", "")
    )
    bunny.add_argument(
        "--purge-endpoint", default=os.environ.get("BUNNY_PURGE_ENDPOINT", "")
    )
    bunny.add_argument("--dry-run", action="store_true")
    bunny.set_defaults(
        handler=deploy_bunny,
        password=os.environ.get("BUNNY_STORAGE_PASSWORD", ""),
        api_key=os.environ.get("BUNNY_API_KEY", ""),
    )

    ssh = subparsers.add_parser("ssh", help="Atomically deploy through static-deploy SSH.")
    ssh.add_argument("directory", type=Path)
    ssh.add_argument("--target", required=True, help="SSH target, e.g. static-deploy@citadel.")
    ssh.add_argument("--site", required=True, help="Configured caddy_static_sites name.")
    ssh.add_argument("--port", type=int, default=22)
    ssh.add_argument("--identity", type=Path)
    ssh.add_argument("--dry-run", action="store_true")
    ssh.set_defaults(handler=deploy_ssh)
    return result


def main() -> int:
    args = parser().parse_args()
    try:
        report = args.handler(args)
    except DeployError as exc:
        print(f"static-site: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(report, indent=2, sort_keys=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())