Promote verified tag artifacts to Gitea releases

tools/publish_gitea_release.py

@@ -0,0 +1,299 @@
+#!/usr/bin/env python3
+"""Promote one verified Linux workflow artifact to a matching Gitea release."""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import subprocess
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+from pathlib import Path
+from typing import Any, Callable, Protocol
+
+
+EXPECTED_ASSETS = frozenset({
+    "OpenMaidEngine-Himegari-linux-x64.tar.gz",
+    "OpenMaidEngine-Himegari-linux-x64.tar.gz.sha256",
+    "BUILD-INFO.json",
+    "SHA256SUMS",
+    "package-smoke.log",
+})
+TAG_PATTERN = re.compile(r"v[0-9][0-9A-Za-z.+-]*\Z")
+COMMIT_PATTERN = re.compile(r"[0-9a-f]{40}\Z")
+TOKEN_PATTERN = re.compile(r"[A-Za-z0-9._-]+\Z")
+
+
+class ReleaseApi(Protocol):
+    def get_release(self, tag: str) -> dict[str, Any] | None: ...
+
+    def create_release(self, payload: dict[str, Any]) -> dict[str, Any]: ...
+
+    def list_assets(self, release_id: int) -> list[dict[str, Any]]: ...
+
+
+class GiteaApi:
+    def __init__(self, server: str, repository: str, token: str) -> None:
+        parsed = urllib.parse.urlsplit(server)
+        if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+            raise ValueError(f"invalid Gitea server URL: {server}")
+        if parsed.username or parsed.password or parsed.query or parsed.fragment:
+            raise ValueError("Gitea server URL must not contain credentials, a query, or a fragment")
+        parts = repository.split("/")
+        if len(parts) != 2 or not all(parts):
+            raise ValueError(f"repository must be owner/name: {repository}")
+        if not TOKEN_PATTERN.fullmatch(token):
+            raise ValueError("GITEA_TOKEN is missing or malformed")
+        owner, name = (urllib.parse.quote(part, safe="") for part in parts)
+        self.base_url = f"{server.rstrip('/')}/api/v1/repos/{owner}/{name}"
+        self.token = token
+
+    def _request(
+        self,
+        method: str,
+        path: str,
+        *,
+        payload: dict[str, Any] | None = None,
+        allow_not_found: bool = False,
+    ) -> dict[str, Any] | list[dict[str, Any]] | None:
+        data = None if payload is None else json.dumps(payload).encode("utf-8")
+        request = urllib.request.Request(
+            self.base_url + path,
+            data=data,
+            method=method,
+            headers={
+                "Accept": "application/json",
+                "Authorization": f"token {self.token}",
+                "Content-Type": "application/json",
+                "User-Agent": "OpenMaidEngine-release-promotion",
+            },
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=30) as response:
+                return json.load(response)
+        except urllib.error.HTTPError as error:
+            if allow_not_found and error.code == 404:
+                return None
+            detail = error.read().decode("utf-8", errors="replace")
+            raise RuntimeError(f"Gitea API {method} {path} failed ({error.code}): {detail}") from error
+
+    def get_release(self, tag: str) -> dict[str, Any] | None:
+        result = self._request(
+            "GET",
+            "/releases/tags/" + urllib.parse.quote(tag, safe=""),
+            allow_not_found=True,
+        )
+        assert result is None or isinstance(result, dict)
+        return result
+
+    def create_release(self, payload: dict[str, Any]) -> dict[str, Any]:
+        result = self._request("POST", "/releases", payload=payload)
+        if not isinstance(result, dict):
+            raise RuntimeError("Gitea create-release response was not an object")
+        return result
+
+    def list_assets(self, release_id: int) -> list[dict[str, Any]]:
+        result = self._request("GET", f"/releases/{release_id}/assets")
+        if not isinstance(result, list):
+            raise RuntimeError("Gitea release-assets response was not an array")
+        return result
+
+    def upload_asset(self, release_id: int, asset: Path) -> dict[str, Any]:
+        url = (
+            f"{self.base_url}/releases/{release_id}/assets?"
+            + urllib.parse.urlencode({"name": asset.name})
+        )
+        curl_config = (
+            f'header = "Authorization: token {self.token}"\n'
+            'header = "Accept: application/json"\n'
+        )
+        process = subprocess.run(
+            [
+                "curl",
+                "--config", "-",
+                "--fail-with-body",
+                "--silent",
+                "--show-error",
+                "--request", "POST",
+                "--form", f"attachment=@{asset}",
+                url,
+            ],
+            input=curl_config,
+            text=True,
+            encoding="utf-8",
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            check=False,
+        )
+        if process.returncode != 0:
+            raise RuntimeError(
+                f"Gitea asset upload failed for {asset.name}: "
+                f"{process.stderr.strip()} {process.stdout.strip()}".strip()
+            )
+        try:
+            result = json.loads(process.stdout)
+        except json.JSONDecodeError as error:
+            raise RuntimeError(f"Gitea asset upload returned invalid JSON for {asset.name}") from error
+        if not isinstance(result, dict):
+            raise RuntimeError(f"Gitea asset upload response was not an object for {asset.name}")
+        return result
+
+
+def validate_inputs(tag: str, target: str, assets: list[Path]) -> None:
+    if not TAG_PATTERN.fullmatch(tag):
+        raise ValueError(f"release tag must be v-prefixed and version-like: {tag}")
+    if not COMMIT_PATTERN.fullmatch(target):
+        raise ValueError(f"release target must be a lowercase SHA-1 commit: {target}")
+    names = [asset.name for asset in assets]
+    if len(names) != len(set(names)):
+        raise ValueError("release asset names must be unique")
+    if set(names) != EXPECTED_ASSETS:
+        missing = sorted(EXPECTED_ASSETS - set(names))
+        extra = sorted(set(names) - EXPECTED_ASSETS)
+        raise ValueError(f"unexpected release asset set; missing={missing}, extra={extra}")
+    for asset in assets:
+        if not asset.is_file():
+            raise ValueError(f"release asset was not found: {asset}")
+        if asset.stat().st_size <= 0:
+            raise ValueError(f"release asset is empty: {asset}")
+
+    by_name = {asset.name: asset for asset in assets}
+    try:
+        build_info = json.loads(by_name["BUILD-INFO.json"].read_text(encoding="utf-8"))
+    except (json.JSONDecodeError, UnicodeDecodeError) as error:
+        raise ValueError("BUILD-INFO.json is not valid UTF-8 JSON") from error
+    expected_build_info = {
+        "schema_version": 1,
+        "source_commit": target,
+        "source_dirty": False,
+        "target": "linux-x64",
+    }
+    build_mismatches = {
+        key: (build_info.get(key), value)
+        for key, value in expected_build_info.items()
+        if build_info.get(key) != value
+    }
+    if build_mismatches:
+        raise ValueError(f"BUILD-INFO.json does not match this promotion: {build_mismatches}")
+
+    archive = by_name["OpenMaidEngine-Himegari-linux-x64.tar.gz"]
+    checksum_path = by_name["OpenMaidEngine-Himegari-linux-x64.tar.gz.sha256"]
+    checksum_line = checksum_path.read_text(encoding="ascii").strip()
+    checksum_match = re.fullmatch(
+        r"([0-9a-f]{64})  OpenMaidEngine-Himegari-linux-x64\.tar\.gz",
+        checksum_line,
+    )
+    if checksum_match is None:
+        raise ValueError(f"archive checksum file has an unexpected format: {checksum_path}")
+    digest = hashlib.sha256()
+    with archive.open("rb") as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            digest.update(chunk)
+    if digest.hexdigest() != checksum_match.group(1):
+        raise ValueError("Linux archive does not match its external SHA-256 checksum")
+
+    smoke_log = by_name["package-smoke.log"].read_text(encoding="utf-8")
+    if "PACKAGE SMOKE OK: opcodes=548 ffmpeg-abi=3" not in smoke_log:
+        raise ValueError("package-smoke.log does not contain the accepted package smoke result")
+
+
+def _validate_release(release: dict[str, Any], tag: str, target: str, title: str) -> int:
+    expected = {
+        "tag_name": tag,
+        "target_commitish": target,
+        "name": title,
+        "draft": False,
+        "prerelease": False,
+    }
+    mismatches = {
+        key: (release.get(key), value)
+        for key, value in expected.items()
+        if release.get(key) != value
+    }
+    if mismatches:
+        raise ValueError(f"existing Gitea release does not match this promotion: {mismatches}")
+    release_id = release.get("id")
+    if not isinstance(release_id, int) or release_id <= 0:
+        raise ValueError("Gitea release has no valid numeric id")
+    return release_id
+
+
+def promote_release(
+    api: ReleaseApi,
+    upload: Callable[[int, Path], dict[str, Any]],
+    tag: str,
+    target: str,
+    assets: list[Path],
+) -> dict[str, Any]:
+    validate_inputs(tag, target, assets)
+    title = f"OpenMaidEngine Himegari {tag}"
+    release = api.get_release(tag)
+    if release is None:
+        release = api.create_release({
+            "tag_name": tag,
+            "target_commitish": target,
+            "name": title,
+            "body": (
+                "Automated Linux x64 release built from `" + target + "`.\n\n"
+                "The attached archive passed the packaged opcode-metadata and FFmpeg ABI smoke gate. "
+                "BUILD-INFO.json, SHA256SUMS, and package-smoke.log provide the external build evidence."
+            ),
+            "draft": False,
+            "prerelease": False,
+        })
+    release_id = _validate_release(release, tag, target, title)
+
+    existing_assets: dict[str, dict[str, Any]] = {}
+    for existing in api.list_assets(release_id):
+        name = existing.get("name")
+        if isinstance(name, str):
+            if name in existing_assets:
+                raise ValueError(f"Gitea release has duplicate asset names: {name}")
+            existing_assets[name] = existing
+
+    for asset in assets:
+        existing = existing_assets.get(asset.name)
+        if existing is not None:
+            if existing.get("size") != asset.stat().st_size:
+                raise ValueError(
+                    f"existing release asset differs in size and will not be overwritten: {asset.name}"
+                )
+            continue
+        uploaded = upload(release_id, asset)
+        if uploaded.get("name") != asset.name or uploaded.get("size") != asset.stat().st_size:
+            raise RuntimeError(f"Gitea reported an unexpected uploaded asset: {asset.name}")
+
+    final_assets = {asset.get("name"): asset for asset in api.list_assets(release_id)}
+    for asset in assets:
+        published = final_assets.get(asset.name)
+        if published is None or published.get("size") != asset.stat().st_size:
+            raise RuntimeError(f"release asset verification failed: {asset.name}")
+    return release
+
+
+def main(arguments: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--server", required=True)
+    parser.add_argument("--repository", required=True)
+    parser.add_argument("--tag", required=True)
+    parser.add_argument("--target", required=True)
+    parser.add_argument("--asset", action="append", required=True, type=Path)
+    args = parser.parse_args(arguments)
+
+    token = os.environ.get("GITEA_TOKEN", "")
+    api = GiteaApi(args.server, args.repository, token)
+    release = promote_release(api, api.upload_asset, args.tag, args.target, args.asset)
+    print(f"Gitea release ready: {release.get('html_url', args.tag)}")
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except (OSError, RuntimeError, ValueError) as error:
+        print(f"release promotion failed: {error}", file=sys.stderr)
+        raise SystemExit(1)

tools/test_publish_gitea_release.py

@@ -0,0 +1,136 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import hashlib
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from typing import Any
+
+import publish_gitea_release
+
+
+TARGET = "0123456789abcdef0123456789abcdef01234567"
+TAG = "v0.1.0"
+
+
+class FakeApi:
+    def __init__(self, release: dict[str, Any] | None = None) -> None:
+        self.release = release
+        self.assets: list[dict[str, Any]] = []
+        self.created_payload: dict[str, Any] | None = None
+
+    def get_release(self, tag: str) -> dict[str, Any] | None:
+        self.requested_tag = tag
+        return self.release
+
+    def create_release(self, payload: dict[str, Any]) -> dict[str, Any]:
+        self.created_payload = payload
+        self.release = matching_release()
+        return self.release
+
+    def list_assets(self, release_id: int) -> list[dict[str, Any]]:
+        self.requested_release_id = release_id
+        return list(self.assets)
+
+    def upload(self, release_id: int, asset: Path) -> dict[str, Any]:
+        uploaded = {"name": asset.name, "size": asset.stat().st_size}
+        self.assets.append(uploaded)
+        return uploaded
+
+
+def matching_release() -> dict[str, Any]:
+    return {
+        "id": 17,
+        "tag_name": TAG,
+        "target_commitish": TARGET,
+        "name": f"OpenMaidEngine Himegari {TAG}",
+        "draft": False,
+        "prerelease": False,
+        "html_url": "https://gitea.invalid/releases/tag/v0.1.0",
+    }
+
+
+def create_assets(root: Path) -> list[Path]:
+    archive = root / "OpenMaidEngine-Himegari-linux-x64.tar.gz"
+    archive.write_bytes(b"synthetic archive")
+    checksum = root / "OpenMaidEngine-Himegari-linux-x64.tar.gz.sha256"
+    checksum.write_text(
+        f"{hashlib.sha256(archive.read_bytes()).hexdigest()}  {archive.name}\n",
+        encoding="ascii",
+    )
+    build_info = root / "BUILD-INFO.json"
+    build_info.write_text(json.dumps({
+        "schema_version": 1,
+        "source_commit": TARGET,
+        "source_dirty": False,
+        "target": "linux-x64",
+    }), encoding="utf-8")
+    ledger = root / "SHA256SUMS"
+    ledger.write_text(f"{'0' * 64}  Himegari.x86_64\n", encoding="ascii")
+    smoke = root / "package-smoke.log"
+    smoke.write_text("PACKAGE SMOKE OK: opcodes=548 ffmpeg-abi=3\n", encoding="utf-8")
+    return [archive, checksum, build_info, ledger, smoke]
+
+
+class PublishGiteaReleaseTests(unittest.TestCase):
+    def test_creates_release_and_uploads_verified_asset_set(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            assets = create_assets(Path(temporary))
+            api = FakeApi()
+            result = publish_gitea_release.promote_release(api, api.upload, TAG, TARGET, assets)
+            self.assertEqual(17, result["id"])
+            self.assertEqual(TAG, api.created_payload["tag_name"])
+            self.assertEqual(TARGET, api.created_payload["target_commitish"])
+            self.assertEqual({asset.name for asset in assets}, {asset["name"] for asset in api.assets})
+
+    def test_retry_keeps_matching_assets_and_uploads_only_missing(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            assets = create_assets(Path(temporary))
+            api = FakeApi(matching_release())
+            api.assets.append({"name": assets[0].name, "size": assets[0].stat().st_size})
+            uploaded: list[str] = []
+
+            def upload(release_id: int, asset: Path) -> dict[str, Any]:
+                uploaded.append(asset.name)
+                return api.upload(release_id, asset)
+
+            publish_gitea_release.promote_release(api, upload, TAG, TARGET, assets)
+            self.assertNotIn(assets[0].name, uploaded)
+            self.assertEqual(len(assets) - 1, len(uploaded))
+            self.assertIsNone(api.created_payload)
+
+    def test_refuses_mismatched_release_and_asset_collision(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            assets = create_assets(Path(temporary))
+            wrong_release = matching_release()
+            wrong_release["target_commitish"] = "f" * 40
+            api = FakeApi(wrong_release)
+            with self.assertRaisesRegex(ValueError, "does not match"):
+                publish_gitea_release.promote_release(api, api.upload, TAG, TARGET, assets)
+
+            api = FakeApi(matching_release())
+            api.assets.append({"name": assets[0].name, "size": 999})
+            with self.assertRaisesRegex(ValueError, "will not be overwritten"):
+                publish_gitea_release.promote_release(api, api.upload, TAG, TARGET, assets)
+
+    def test_refuses_mismatched_build_evidence(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            assets = create_assets(Path(temporary))
+            by_name = {asset.name: asset for asset in assets}
+            by_name["OpenMaidEngine-Himegari-linux-x64.tar.gz"].write_bytes(b"changed")
+            with self.assertRaisesRegex(ValueError, "external SHA-256"):
+                publish_gitea_release.validate_inputs(TAG, TARGET, assets)
+
+            assets = create_assets(Path(temporary))
+            by_name = {asset.name: asset for asset in assets}
+            build_info = json.loads(by_name["BUILD-INFO.json"].read_text(encoding="utf-8"))
+            build_info["source_commit"] = "f" * 40
+            by_name["BUILD-INFO.json"].write_text(json.dumps(build_info), encoding="utf-8")
+            with self.assertRaisesRegex(ValueError, "does not match"):
+                publish_gitea_release.validate_inputs(TAG, TARGET, assets)
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/validate.py

@@ -33,6 +33,7 @@ CORE_TESTS = (
     "test_install_godot_templates.py",
     "test_package_linux_x64.py",
     "test_dotnet_publish_proxy.py",
+    "test_publish_gitea_release.py",
     "test_diff_optrace.py",
     "test_engine_ctx.py",
     "test_ghidra_handler_map.py",