#!/usr/bin/env python3 """The Gitea release that is the permanent record of one version. A Gitea release is where a version's package, its checksum, how it was built and what happened to it afterwards are kept for good: which commit, which gate, when it went to Preview (canary), when it was promoted to Stable and on whose approval. The update server is where classrooms fetch it; this is where a person months from now finds out why. PUBLISHED BYTES ARE NEVER REPLACED. Gitea lets anyone with write access delete an asset and upload another under the same name, and nothing in Gitea stops a workflow doing it either, so this does: an asset is only ever uploaded under a name the release does not have yet. An asset that is already there with exactly the same bytes is left alone and reported; one with different bytes fails, loudly, and nothing is changed. The same rule the update server has had since 0.2.6 was republished over itself (tools/release-lib.sh). A tag, likewise, is never moved: a release is only created for a tag that does not exist, or that already names the commit being released. The record is append-only. `record` adds a dated line under "## Record"; nothing here rewrites what an earlier line says. gitea-release show TAG the release, or exit 3 if none gitea-release notes TAG its notes and record, as text gitea-release create TAG COMMIT TITLE NOTES_FILE [--prerelease] gitea-release upload TAG FILE... each only if not there yet gitea-release download TAG NAME DEST an asset, by name gitea-release record TAG LINE append to the record gitea-release stable TAG no longer a pre-release Environment: GITEA_URL https://git.riverofglorymn.org (GITHUB_SERVER_URL in CI) GITEA_REPO River_of_Glory_Church/Ekklesia_Classroom (GITHUB_REPOSITORY) GITEA_TOKEN a token that may write releases; never printed """ import hashlib import json import os import sys import urllib.error import urllib.parse import urllib.request import uuid from datetime import datetime, timezone RECORD = "## Record" class Refused(Exception): """Something that must not happen, said in words.""" def say(message): print(f"gitea-release: {message}", flush=True) def config(): url = os.environ.get("GITEA_URL") or os.environ.get("GITHUB_SERVER_URL") repo = os.environ.get("GITEA_REPO") or os.environ.get("GITHUB_REPOSITORY") token = os.environ.get("GITEA_TOKEN", "") if not url or not repo: raise Refused("GITEA_URL and GITEA_REPO must be set") return url.rstrip("/"), repo, token def request(method, path, body=None, headers=None, raw=False): """(status, parsed body). 404 is an answer, not an error.""" url, repo, token = config() target = path if path.startswith("http") else f"{url}/api/v1/repos/{repo}{path}" headers = dict(headers or {}) if token: headers["Authorization"] = f"token {token}" if isinstance(body, (dict, list)): body = json.dumps(body).encode() headers["Content-Type"] = "application/json" call = urllib.request.Request(target, data=body, method=method, headers=headers) try: with urllib.request.urlopen(call, timeout=120) as response: data = response.read() status = response.status except urllib.error.HTTPError as error: data = error.read() status = error.code if status == 404: return 404, None if status >= 400: detail = data[:300].decode("utf-8", "replace") raise Refused(f"{method} {path} answered {status}: {detail}") if raw: return status, data return status, json.loads(data) if data else None def release(tag): status, found = request("GET", f"/releases/tags/{urllib.parse.quote(tag)}") return None if status == 404 else found def need_release(tag): found = release(tag) if found is None: raise Refused(f"there is no release {tag}") return found def tag_commit(tag): status, found = request("GET", f"/tags/{urllib.parse.quote(tag)}") if status == 404: return None return found["commit"]["sha"] def sha256(data): return hashlib.sha256(data).hexdigest() def now(): return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") def assets(found): status, listed = request("GET", f"/releases/{found['id']}/assets") return {} if status == 404 else {asset["name"]: asset for asset in listed or []} def fetch(asset): _status, data = request("GET", asset["browser_download_url"], raw=True) return data # --------------------------------------------------------------------------- def show(tag): found = release(tag) if found is None: say(f"no release {tag}") return 3 print(json.dumps({ "tag": found["tag_name"], "commit": tag_commit(tag), "prerelease": found.get("prerelease"), "assets": sorted(assets(found)), "url": found.get("html_url"), }, indent=2)) return 0 def notes(tag): print(need_release(tag).get("body") or "") return 0 def create(tag, commit, title, notes_file, prerelease): notes = open(notes_file, encoding="utf-8").read() at = tag_commit(tag) if at is not None and at != commit: raise Refused(f"tag {tag} already names {at}, not {commit}. " "A tag is never moved; release a new version.") found = release(tag) if found is not None: say(f"release {tag} already exists at {commit}; nothing created") return 0 request("POST", "/releases", { "tag_name": tag, "target_commitish": commit, "name": title, "body": notes.rstrip() + f"\n\n{RECORD}\n\n- {now()}: release created at `{commit}`\n", "draft": False, "prerelease": prerelease, }) if tag_commit(tag) != commit: raise Refused(f"release {tag} was created, but its tag does not name {commit}") say(f"created release {tag} at {commit}") return 0 def upload(tag, files): found = need_release(tag) present = assets(found) for path in files: name = os.path.basename(path) data = open(path, "rb").read() digest = sha256(data) if name in present: there = sha256(fetch(present[name])) if there == digest: say(f"{name} is already attached, byte for byte; left alone") continue raise Refused(f"{name} is already attached to {tag} with sha256 {there}; " f"this file is {digest}. A published asset is never replaced.") boundary = uuid.uuid4().hex body = (f"--{boundary}\r\n" f'Content-Disposition: form-data; name="attachment"; filename="{name}"\r\n' f"Content-Type: application/octet-stream\r\n\r\n").encode() \ + data + f"\r\n--{boundary}--\r\n".encode() request("POST", f"/releases/{found['id']}/assets?name={urllib.parse.quote(name)}", body, {"Content-Type": f"multipart/form-data; boundary={boundary}"}) # Read back: what Gitea serves is what was sent. there = assets(need_release(tag)).get(name) if there is None or sha256(fetch(there)) != digest: raise Refused(f"{name} was uploaded but does not read back as sha256 {digest}") say(f"attached {name}, sha256 {digest}") return 0 def download(tag, name, dest): present = assets(need_release(tag)) if name not in present: raise Refused(f"{tag} has no asset {name}") data = fetch(present[name]) with open(dest, "wb") as handle: handle.write(data) say(f"{name} -> {dest}, sha256 {sha256(data)}") return 0 def record(tag, line): found = need_release(tag) body = found.get("body") or "" if RECORD not in body: body = body.rstrip() + f"\n\n{RECORD}\n" body = body.rstrip() + f"\n- {now()}: {line}\n" request("PATCH", f"/releases/{found['id']}", {"body": body}) say(f"recorded on {tag}: {line}") return 0 def stable(tag): found = need_release(tag) request("PATCH", f"/releases/{found['id']}", {"prerelease": False}) say(f"{tag} is no longer marked as a pre-release") return 0 def main(argv): try: match argv: case ["show", tag]: return show(tag) case ["notes", tag]: return notes(tag) case ["create", tag, commit, title, notes_file, *rest] if rest in ([], ["--prerelease"]): return create(tag, commit, title, notes_file, rest == ["--prerelease"]) case ["upload", tag, *files] if files: return upload(tag, files) case ["download", tag, name, dest]: return download(tag, name, dest) case ["record", tag, line]: return record(tag, line) case ["stable", tag]: return stable(tag) case _: print(__doc__, file=sys.stderr) return 2 except Refused as refusal: say(f"REFUSED: {refusal}") return 1 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))