203 lines
7.6 KiB
Python
203 lines
7.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Publish immutable GovOPlaN runtime evidence as Gitea release assets."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import mimetypes
|
|
import os
|
|
from pathlib import Path
|
|
import secrets
|
|
import sys
|
|
from typing import Any
|
|
from urllib.error import HTTPError
|
|
from urllib.parse import quote, urlencode
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
MAX_ASSET_BYTES = 256 * 1024 * 1024
|
|
|
|
|
|
class PublishError(RuntimeError):
|
|
"""A release asset cannot be published immutably."""
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--base-url", default="https://git.add-ideas.de")
|
|
parser.add_argument("--owner", default="GovOPlaN")
|
|
parser.add_argument("--repo", default="govoplan")
|
|
parser.add_argument("--tag", required=True)
|
|
parser.add_argument("--title", required=True)
|
|
parser.add_argument("--body", default="Signed GovOPlaN runtime distribution.")
|
|
parser.add_argument("--asset", type=Path, action="append", default=[], required=True)
|
|
parser.add_argument("--token-env", default="GITEA_RELEASE_TOKEN")
|
|
return parser
|
|
|
|
|
|
class GiteaReleasePublisher:
|
|
def __init__(self, *, base_url: str, owner: str, repo: str, token: str) -> None:
|
|
if not base_url.startswith("https://"):
|
|
raise PublishError("Gitea release publication requires HTTPS")
|
|
if not token:
|
|
raise PublishError("Gitea release token is empty")
|
|
self.base_url = base_url.rstrip("/")
|
|
self.owner = owner
|
|
self.repo = repo
|
|
self.token = token
|
|
|
|
def release(self, *, tag: str, title: str, body: str) -> dict[str, Any]:
|
|
path = self._repo_path(f"/releases/tags/{quote(tag, safe='')}")
|
|
try:
|
|
return self._json("GET", path)
|
|
except HTTPError as exc:
|
|
if exc.code != 404:
|
|
raise
|
|
return self._json(
|
|
"POST",
|
|
self._repo_path("/releases"),
|
|
payload={
|
|
"tag_name": tag,
|
|
"name": title,
|
|
"body": body,
|
|
"draft": False,
|
|
"prerelease": False,
|
|
},
|
|
expected=201,
|
|
)
|
|
|
|
def upload_assets(self, release: dict[str, Any], assets: tuple[Path, ...]) -> None:
|
|
release_id = release.get("id")
|
|
if isinstance(release_id, bool) or not isinstance(release_id, int):
|
|
raise PublishError("Gitea release response has no numeric id")
|
|
existing = self._json(
|
|
"GET",
|
|
self._repo_path(f"/releases/{release_id}/assets"),
|
|
)
|
|
if not isinstance(existing, list):
|
|
raise PublishError("Gitea release assets response is invalid")
|
|
existing_by_name = {
|
|
str(item.get("name")): item for item in existing if isinstance(item, dict)
|
|
}
|
|
for asset in assets:
|
|
path = asset.expanduser().resolve()
|
|
if path.is_symlink() or not path.is_file():
|
|
raise PublishError(f"release asset is not a regular file: {path}")
|
|
size = path.stat().st_size
|
|
if size > MAX_ASSET_BYTES:
|
|
raise PublishError(f"release asset exceeds size limit: {path.name}")
|
|
prior = existing_by_name.get(path.name)
|
|
if prior is not None:
|
|
self._require_same_existing_asset(prior, path)
|
|
continue
|
|
self._upload(release_id, path)
|
|
|
|
def _require_same_existing_asset(self, prior: dict[str, Any], path: Path) -> None:
|
|
url = prior.get("browser_download_url")
|
|
size = prior.get("size")
|
|
if not isinstance(url, str) or not url.startswith("https://") or size != path.stat().st_size:
|
|
raise PublishError(f"release asset already exists with another identity: {path.name}")
|
|
request = Request(url, headers=self._headers())
|
|
digest = hashlib.sha256()
|
|
total = 0
|
|
with urlopen(request, timeout=30) as response: # noqa: S310
|
|
while True:
|
|
chunk = response.read(1024 * 1024)
|
|
if not chunk:
|
|
break
|
|
total += len(chunk)
|
|
if total > MAX_ASSET_BYTES:
|
|
raise PublishError("existing release asset exceeds size limit")
|
|
digest.update(chunk)
|
|
if digest.hexdigest() != _sha256_file(path):
|
|
raise PublishError(f"release asset already exists with another digest: {path.name}")
|
|
|
|
def _upload(self, release_id: int, path: Path) -> None:
|
|
boundary = "govoplan-" + secrets.token_hex(16)
|
|
content_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
|
|
prefix = (
|
|
f"--{boundary}\r\n"
|
|
f'Content-Disposition: form-data; name="attachment"; filename="{path.name}"\r\n'
|
|
f"Content-Type: {content_type}\r\n\r\n"
|
|
).encode("utf-8")
|
|
suffix = f"\r\n--{boundary}--\r\n".encode("ascii")
|
|
data = prefix + path.read_bytes() + suffix
|
|
query = urlencode({"name": path.name})
|
|
request = Request(
|
|
self._repo_path(f"/releases/{release_id}/assets") + "?" + query,
|
|
data=data,
|
|
method="POST",
|
|
headers={
|
|
**self._headers(),
|
|
"Content-Type": f"multipart/form-data; boundary={boundary}",
|
|
},
|
|
)
|
|
try:
|
|
with urlopen(request, timeout=120) as response: # noqa: S310
|
|
if response.status != 201:
|
|
raise PublishError(
|
|
f"Gitea asset upload returned HTTP {response.status}"
|
|
)
|
|
except HTTPError as exc:
|
|
raise PublishError(f"Gitea asset upload failed with HTTP {exc.code}") from exc
|
|
|
|
def _json(
|
|
self,
|
|
method: str,
|
|
url: str,
|
|
*,
|
|
payload: dict[str, Any] | None = None,
|
|
expected: int = 200,
|
|
) -> Any:
|
|
data = None
|
|
headers = self._headers()
|
|
if payload is not None:
|
|
data = json.dumps(payload).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
request = Request(url, data=data, method=method, headers=headers)
|
|
with urlopen(request, timeout=30) as response: # noqa: S310
|
|
if response.status != expected:
|
|
raise PublishError(f"Gitea API returned HTTP {response.status}")
|
|
return json.load(response)
|
|
|
|
def _headers(self) -> dict[str, str]:
|
|
return {"Authorization": f"token {self.token}", "Accept": "application/json"}
|
|
|
|
def _repo_path(self, suffix: str) -> str:
|
|
return (
|
|
f"{self.base_url}/api/v1/repos/{quote(self.owner, safe='')}/"
|
|
f"{quote(self.repo, safe='')}{suffix}"
|
|
)
|
|
|
|
|
|
def _sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def main() -> int:
|
|
args = build_parser().parse_args()
|
|
try:
|
|
publisher = GiteaReleasePublisher(
|
|
base_url=args.base_url,
|
|
owner=args.owner,
|
|
repo=args.repo,
|
|
token=os.environ.get(args.token_env, ""),
|
|
)
|
|
release = publisher.release(tag=args.tag, title=args.title, body=args.body)
|
|
publisher.upload_assets(release, tuple(args.asset))
|
|
except (HTTPError, OSError, PublishError, ValueError) as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
return 1
|
|
print(f"Published {len(args.asset)} immutable asset(s) to {args.owner}/{args.repo} {args.tag}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|