blog/delivery/scripts/generic_package.py

#!/usr/bin/env python3
"""Publish a static release as immutable files in Gitea's generic registry."""

from __future__ import annotations

import argparse
import base64
import json
import os
from pathlib import Path
import re
import sys
import urllib.error
import urllib.parse
import urllib.request

try:
    from .static_release import ReleaseError, inspect_archive, validate_revision
except ImportError:  # Direct script execution adds scripts/ to sys.path.
    from static_release import ReleaseError, inspect_archive, validate_revision


COMPONENT_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+-]*$")
ARCHIVE_NAME = "site.tar.gz"
CHECKSUM_NAME = "site.tar.gz.sha256"
COORDINATE_NAME = "release-coordinate.json"
MAX_RESPONSE_BYTES = 200 * 1024 * 1024


class PackageError(RuntimeError):
    pass


def validate_component(value: str, label: str) -> None:
    if not COMPONENT_RE.fullmatch(value):
        raise PackageError(f"{label} contains unsupported characters")


def validate_registry(registry: str, *, allow_http: bool = False) -> str:
    parsed = urllib.parse.urlsplit(registry)
    allowed_schemes = {"https"} | ({"http"} if allow_http else set())
    if (
        parsed.scheme not in allowed_schemes
        or not parsed.netloc
        or parsed.query
        or parsed.fragment
        or parsed.username
        or parsed.password
    ):
        raise PackageError("registry must be an HTTPS origin without credentials, query or fragment")
    return registry.rstrip("/")


def package_file_url(
    registry: str,
    owner: str,
    package: str,
    version: str,
    filename: str,
    *,
    allow_http: bool = False,
) -> str:
    registry = validate_registry(registry, allow_http=allow_http)
    for value, label in ((owner, "owner"), (package, "package"), (filename, "filename")):
        validate_component(value, label)
    validate_revision(version)
    components = [owner, package, version, filename]
    owner_part, package_part, version_part, filename_part = (
        urllib.parse.quote(value, safe="") for value in components
    )
    return (
        f"{registry}/api/packages/{owner_part}/generic/"
        f"{package_part}/{version_part}/{filename_part}"
    )


def authorization(username: str | None, token: str | None) -> str | None:
    if bool(username) != bool(token):
        raise PackageError("package username and token must be supplied together")
    if not username:
        return None
    encoded = base64.b64encode(f"{username}:{token}".encode()).decode("ascii")
    return f"Basic {encoded}"


def request(
    method: str,
    url: str,
    *,
    username: str | None,
    token: str | None,
    data: bytes | None = None,
    timeout: float = 300.0,
) -> tuple[int, bytes]:
    headers = {"User-Agent": "donatella-static-publisher/1"}
    auth = authorization(username, token)
    if auth:
        headers["Authorization"] = auth
    if data is not None:
        headers["Content-Type"] = "application/octet-stream"
    outgoing = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(outgoing, timeout=timeout) as response:
            content = response.read(MAX_RESPONSE_BYTES + 1)
            if len(content) > MAX_RESPONSE_BYTES:
                raise PackageError("package registry response exceeds the size limit")
            return response.status, content
    except urllib.error.HTTPError as exc:
        try:
            content = exc.read(MAX_RESPONSE_BYTES + 1)
            if len(content) > MAX_RESPONSE_BYTES:
                raise PackageError("package registry error response exceeds the size limit")
            return exc.code, content
        finally:
            exc.close()
    except (OSError, urllib.error.URLError) as exc:
        raise PackageError(f"package registry request failed: {exc}") from exc


def put_immutable(
    url: str,
    content: bytes,
    *,
    username: str,
    token: str,
    timeout: float = 300.0,
) -> None:
    status, existing = request("GET", url, username=username, token=token, timeout=timeout)
    if status == 200:
        if existing != content:
            raise PackageError(f"immutable package file already exists with different content: {url}")
        return
    if status != 404:
        raise PackageError(f"package preflight returned HTTP {status}: {url}")

    status, _ = request(
        "PUT",
        url,
        username=username,
        token=token,
        data=content,
        timeout=timeout,
    )
    if status not in {201, 409}:
        raise PackageError(f"package upload returned HTTP {status}: {url}")

    status, published = request("GET", url, username=username, token=token, timeout=timeout)
    if status != 200 or published != content:
        raise PackageError(f"published package file did not verify byte-for-byte: {url}")


def coordinate_bytes(revision: str, artifact_url: str, digest: str) -> bytes:
    return (
        json.dumps(
            {
                "revision": revision,
                "artifact_url": artifact_url,
                "artifact_digest": digest,
            },
            separators=(",", ":"),
        )
        + "\n"
    ).encode("utf-8")


def publish_release(
    *,
    registry: str,
    owner: str,
    package: str,
    revision: str,
    archive: Path,
    checksum: Path,
    coordinate: Path,
    username: str,
    token: str,
    allow_http: bool = False,
) -> dict[str, str]:
    if not username or not token:
        raise PackageError("package username and token are required")
    if archive.name != ARCHIVE_NAME or checksum.name != CHECKSUM_NAME:
        raise PackageError(f"release files must be named {ARCHIVE_NAME} and {CHECKSUM_NAME}")
    try:
        _, _, digest = inspect_archive(archive, checksum, revision)
    except ReleaseError as exc:
        raise PackageError(str(exc)) from exc

    artifact_url = package_file_url(
        registry,
        owner,
        package,
        revision,
        ARCHIVE_NAME,
        allow_http=allow_http,
    )
    checksum_url = package_file_url(
        registry,
        owner,
        package,
        revision,
        CHECKSUM_NAME,
        allow_http=allow_http,
    )
    coordinate_url = package_file_url(
        registry,
        owner,
        package,
        revision,
        COORDINATE_NAME,
        allow_http=allow_http,
    )
    payload = coordinate_bytes(revision, artifact_url, digest)
    if coordinate.resolve() in {archive.resolve(), checksum.resolve()}:
        raise PackageError("coordinate output must be separate from the archive and checksum")
    coordinate.parent.mkdir(parents=True, exist_ok=True)
    coordinate.write_bytes(payload)

    for url, content in (
        (artifact_url, archive.read_bytes()),
        (checksum_url, checksum.read_bytes()),
        (coordinate_url, payload),
    ):
        put_immutable(url, content, username=username, token=token)

    return {
        "revision": revision,
        "artifact_url": artifact_url,
        "artifact_digest": digest,
    }


def parser() -> argparse.ArgumentParser:
    command = argparse.ArgumentParser(description=__doc__)
    command.add_argument("--registry", required=True)
    command.add_argument("--owner", required=True)
    command.add_argument("--package", required=True)
    command.add_argument("--revision", required=True)
    command.add_argument("--archive", type=Path, required=True)
    command.add_argument("--checksum", type=Path, required=True)
    command.add_argument("--coordinate", type=Path, required=True)
    return command


def main() -> int:
    args = parser().parse_args()
    try:
        result = publish_release(
            registry=args.registry,
            owner=args.owner,
            package=args.package,
            revision=args.revision,
            archive=args.archive,
            checksum=args.checksum,
            coordinate=args.coordinate,
            username=os.environ.get("STATIC_PACKAGE_USER", ""),
            token=os.environ.get("STATIC_PACKAGE_TOKEN", ""),
        )
    except PackageError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(result, separators=(",", ":")))
    return 0


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