ekklesiaandClaude Opus 5.5
0f96019714
Gitea lets anyone with write access swap a release asset for another of the same name. tools/gitea-release, which the workflows use for every release write, uploads only under a name the release lacks, leaves identical bytes alone, refuses different ones, never moves a tag, and only appends to the record. Measured on a throwaway Gitea 1.27.2 with the automatic Actions token; tested here against a stand-in API. Co-Authored-By: Claude Opus 5.5 <[email protected]>
116 lines
5.2 KiB
Python
Executable File
116 lines
5.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
"""A stand-in for the part of Gitea 1.27's API that tools/gitea-release uses.
|
|
|
|
Tags, releases and assets in memory. Writes need the token
|
|
"not-a-real-token", as Gitea needs a real one. Like Gitea, it takes a
|
|
second asset under a name a release already has: refusing that is
|
|
tools/gitea-release's job, and the tests check that it does. GET
|
|
/test/tamper/TAG/NAME changes an asset's bytes, the way somebody could in
|
|
the web UI. Prints the port it listens on, then serves until killed.
|
|
"""
|
|
|
|
import json, re
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from urllib.parse import urlparse, parse_qs, unquote
|
|
|
|
TOKEN = "token not-a-real-token"
|
|
PREFIX = "/api/v1/repos/church/classroom"
|
|
state = {"tags": {"v0.1.5": "1" * 40}, "releases": [], "next": 1}
|
|
|
|
|
|
def find(tag):
|
|
return next((r for r in state["releases"] if r["tag_name"] == tag), None)
|
|
|
|
|
|
class Gitea(BaseHTTPRequestHandler):
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
def reply(self, status, body=None, raw=None):
|
|
data = raw if raw is not None else json.dumps(body).encode() if body is not None else b""
|
|
self.send_response(status)
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def authorised(self):
|
|
if self.headers.get("Authorization") != TOKEN:
|
|
self.reply(401, {"message": "token is required"})
|
|
return False
|
|
return True
|
|
|
|
def release_json(self, r):
|
|
port = self.server.server_port
|
|
return {**{k: v for k, v in r.items() if k != "assets"},
|
|
"html_url": f"http://127.0.0.1:{port}/church/classroom/releases/tag/{r['tag_name']}"}
|
|
|
|
def do_GET(self):
|
|
path = urlparse(self.path).path
|
|
if m := re.fullmatch(PREFIX + r"/releases/tags/(.+)", path):
|
|
r = find(unquote(m.group(1)))
|
|
return self.reply(200, self.release_json(r)) if r else self.reply(404, {})
|
|
if m := re.fullmatch(PREFIX + r"/tags/(.+)", path):
|
|
sha = state["tags"].get(unquote(m.group(1)))
|
|
return self.reply(200, {"name": m.group(1), "commit": {"sha": sha}}) if sha else self.reply(404, {})
|
|
if m := re.fullmatch(PREFIX + r"/releases/(\d+)/assets", path):
|
|
r = next((r for r in state["releases"] if r["id"] == int(m.group(1))), None)
|
|
if not r:
|
|
return self.reply(404, {})
|
|
port = self.server.server_port
|
|
return self.reply(200, [{"id": i, "name": n, "size": len(d),
|
|
"browser_download_url": f"http://127.0.0.1:{port}/download/{r['id']}/{n}"}
|
|
for i, (n, d) in enumerate(r["assets"].items())])
|
|
if m := re.fullmatch(r"/download/(\d+)/(.+)", path):
|
|
r = next((r for r in state["releases"] if r["id"] == int(m.group(1))), None)
|
|
data = r and r["assets"].get(unquote(m.group(2)))
|
|
return self.reply(200, raw=data) if data is not None else self.reply(404, {})
|
|
# For the test: somebody replaces an asset by hand in the web UI.
|
|
if m := re.fullmatch(r"/test/tamper/(.+)/(.+)", path):
|
|
find(unquote(m.group(1)))["assets"][unquote(m.group(2))] = b"tampered"
|
|
return self.reply(200, {})
|
|
return self.reply(404, {})
|
|
|
|
def do_POST(self):
|
|
if not self.authorised():
|
|
return
|
|
url = urlparse(self.path)
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
body = self.rfile.read(length)
|
|
if url.path == PREFIX + "/releases":
|
|
data = json.loads(body)
|
|
if find(data["tag_name"]):
|
|
return self.reply(409, {"message": "release exists"})
|
|
state["tags"].setdefault(data["tag_name"], data["target_commitish"])
|
|
r = {"id": state["next"], "tag_name": data["tag_name"], "name": data["name"],
|
|
"body": data["body"], "prerelease": data["prerelease"], "draft": False, "assets": {}}
|
|
state["next"] += 1
|
|
state["releases"].append(r)
|
|
return self.reply(201, self.release_json(r))
|
|
if m := re.fullmatch(PREFIX + r"/releases/(\d+)/assets", url.path):
|
|
r = next(r for r in state["releases"] if r["id"] == int(m.group(1)))
|
|
name = parse_qs(url.query)["name"][0]
|
|
boundary = self.headers["Content-Type"].split("boundary=")[1].encode()
|
|
part = body.split(b"--" + boundary)[1]
|
|
content = part.split(b"\r\n\r\n", 1)[1].rsplit(b"\r\n", 1)[0]
|
|
# Gitea itself happily takes a second asset of the same name.
|
|
r["assets"][name] = content
|
|
return self.reply(201, {"name": name})
|
|
return self.reply(404, {})
|
|
|
|
def do_PATCH(self):
|
|
if not self.authorised():
|
|
return
|
|
path = urlparse(self.path).path
|
|
if m := re.fullmatch(PREFIX + r"/releases/(\d+)", path):
|
|
r = next(r for r in state["releases"] if r["id"] == int(m.group(1)))
|
|
data = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
|
|
r.update({k: v for k, v in data.items() if k in ("body", "prerelease")})
|
|
return self.reply(200, self.release_json(r))
|
|
return self.reply(404, {})
|
|
|
|
|
|
server = HTTPServer(("127.0.0.1", 0), Gitea)
|
|
print(server.server_port, flush=True)
|
|
server.serve_forever()
|