Add public Forgejo static site deploy action

static_site.py

@@ -0,0 +1,311 @@
+#!/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 deploy_bunny(args: argparse.Namespace) -> dict[str, Any]:
+    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 or not result["changed"]:
+        return result
+
+    for path in upload:
+        storage.upload(files[path])
+    for path in delete:
+        storage.delete(path)
+    storage.upload_manifest(manifest_for(files))
+    if 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("--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())