393 lines
16 KiB
Python
393 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Publish extension release assets through the Gitea REST API."""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
from pathlib import Path
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import quote, urlencode
|
|
from urllib.request import Request, urlopen
|
|
import xml.etree.ElementTree as ET
|
|
|
|
|
|
class ApiError(RuntimeError):
|
|
pass
|
|
|
|
|
|
VERSION_RE = re.compile(r"^(?:0|[1-9][0-9]*)(?:\.(?:0|[1-9][0-9]*)){0,3}$")
|
|
EXPECTED_CRX_CONTENT_TYPE = "application/x-chrome-extension"
|
|
|
|
|
|
def _safe_error_detail(error, token):
|
|
try:
|
|
detail = error.read().decode("utf-8", errors="replace").strip()
|
|
except (OSError, UnicodeError):
|
|
detail = ""
|
|
if token:
|
|
detail = detail.replace(token, "[REDACTED]")
|
|
return detail[:500]
|
|
|
|
|
|
def api_call(base_url, token, method, path, payload=None, expected_errors=()):
|
|
url = base_url.rstrip("/") + path
|
|
body = None if payload is None else json.dumps(payload).encode("utf-8")
|
|
headers = {"Authorization": f"token {token}", "Accept": "application/json"}
|
|
if body is not None:
|
|
headers["Content-Type"] = "application/json"
|
|
request = Request(url, data=body, headers=headers, method=method)
|
|
try:
|
|
with urlopen(request) as response:
|
|
response_body = response.read()
|
|
status = response.status
|
|
except HTTPError as error:
|
|
if error.code in expected_errors:
|
|
return error.code, None
|
|
detail = _safe_error_detail(error, token)
|
|
suffix = f": {detail}" if detail else ""
|
|
reason = str(error.reason).replace(token, "[REDACTED]") if token else str(error.reason)
|
|
raise ApiError(f"Gitea API {method} {path}: HTTP {error.code} {reason}{suffix}") from None
|
|
except URLError as error:
|
|
detail = str(error.reason).replace(token, "[REDACTED]") if token else str(error.reason)
|
|
raise ApiError(f"Gitea API {method} {path}: {detail}") from None
|
|
if not response_body:
|
|
return status, None
|
|
try:
|
|
return status, json.loads(response_body)
|
|
except json.JSONDecodeError as error:
|
|
raise ApiError(f"Gitea API {method} {path}: invalid JSON response") from error
|
|
|
|
|
|
def repository_path(repository):
|
|
parts = repository.split("/", 1)
|
|
if len(parts) != 2 or not all(parts):
|
|
raise ValueError("GITEA_REPOSITORY must be OWNER/REPOSITORY")
|
|
owner, name = (quote(part, safe="") for part in parts)
|
|
return f"/repos/{owner}/{name}"
|
|
|
|
|
|
def get_or_create_release(base_url, token, repo_path, tag, prerelease):
|
|
encoded_tag = quote(tag, safe="")
|
|
status, release = api_call(
|
|
base_url, token, "GET", f"{repo_path}/releases/tags/{encoded_tag}", expected_errors=(404,)
|
|
)
|
|
if status == 404:
|
|
_, release = api_call(
|
|
base_url,
|
|
token,
|
|
"POST",
|
|
f"{repo_path}/releases",
|
|
{
|
|
"tag_name": tag,
|
|
"name": tag,
|
|
"body": "Automated Chrome extension release.",
|
|
"prerelease": prerelease,
|
|
},
|
|
)
|
|
if not isinstance(release, dict) or not isinstance(release.get("id"), int):
|
|
raise ApiError(f"Gitea API returned an invalid release for tag {tag}")
|
|
return release
|
|
|
|
|
|
def release_assets(base_url, token, repo_path, release_id):
|
|
_, assets = api_call(base_url, token, "GET", f"{repo_path}/releases/{release_id}/assets")
|
|
if not isinstance(assets, list):
|
|
raise ApiError("Gitea API returned an invalid release asset list")
|
|
return assets
|
|
|
|
|
|
def _asset_path(repo_path, release_id, attachment_id):
|
|
return f"{repo_path}/releases/{release_id}/assets/{attachment_id}"
|
|
|
|
|
|
def _upload_asset(base_url, token, repo_path, release_id, path, asset_name):
|
|
data = path.read_bytes()
|
|
# A random boundary is also checked against the file so it cannot terminate the part.
|
|
while True:
|
|
boundary = f"----history-collector-{secrets.token_hex(16)}"
|
|
if boundary.encode("ascii") not in data:
|
|
break
|
|
filename = path.name.replace("\\", "_").replace('"', "_").replace("\r", "_").replace("\n", "_")
|
|
disposition_name = asset_name.replace("\\", "_").replace('"', "_").replace("\r", "_").replace("\n", "_")
|
|
boundary_bytes = boundary.encode("ascii")
|
|
body = b"--" + boundary_bytes + b"\r\n"
|
|
body += f'Content-Disposition: form-data; name="attachment"; filename="{filename}"\r\n'.encode("utf-8")
|
|
body += b"Content-Type: application/octet-stream\r\n\r\n" + data
|
|
body += b"\r\n--" + boundary_bytes + b"--\r\n"
|
|
query = urlencode({"name": disposition_name})
|
|
url_path = f"{repo_path}/releases/{release_id}/assets?{query}"
|
|
request = Request(
|
|
base_url.rstrip("/") + url_path,
|
|
data=body,
|
|
headers={
|
|
"Authorization": f"token {token}",
|
|
"Accept": "application/json",
|
|
"Content-Type": f"multipart/form-data; boundary={boundary}",
|
|
},
|
|
method="POST",
|
|
)
|
|
try:
|
|
with urlopen(request) as response:
|
|
response_body = response.read()
|
|
status = response.status
|
|
except HTTPError as error:
|
|
detail = _safe_error_detail(error, token)
|
|
suffix = f": {detail}" if detail else ""
|
|
reason = str(error.reason).replace(token, "[REDACTED]") if token else str(error.reason)
|
|
raise ApiError(f"Gitea API POST {url_path}: HTTP {error.code} {reason}{suffix}") from None
|
|
except URLError as error:
|
|
detail = str(error.reason).replace(token, "[REDACTED]") if token else str(error.reason)
|
|
raise ApiError(f"Gitea API POST {url_path}: {detail}") from None
|
|
if status < 200 or status >= 300:
|
|
raise ApiError(f"Gitea API POST {url_path}: HTTP {status}")
|
|
try:
|
|
asset = json.loads(response_body)
|
|
except (json.JSONDecodeError, UnicodeDecodeError) as error:
|
|
raise ApiError(f"Gitea API POST {url_path}: invalid JSON response") from error
|
|
if not isinstance(asset, dict) or not isinstance(asset.get("id"), int):
|
|
raise ApiError(f"Gitea API POST {url_path}: invalid attachment response")
|
|
return asset
|
|
|
|
|
|
def _delete_asset(base_url, token, repo_path, release_id, attachment_id):
|
|
api_call(base_url, token, "DELETE", _asset_path(repo_path, release_id, attachment_id))
|
|
|
|
|
|
def _edit_asset(base_url, token, repo_path, release_id, attachment_id, name):
|
|
api_call(
|
|
base_url,
|
|
token,
|
|
"PATCH",
|
|
_asset_path(repo_path, release_id, attachment_id),
|
|
{"name": name},
|
|
)
|
|
|
|
|
|
def _cleanup_asset(base_url, token, repo_path, release_id, attachment_id):
|
|
if attachment_id is None:
|
|
return None
|
|
try:
|
|
_delete_asset(base_url, token, repo_path, release_id, attachment_id)
|
|
except ApiError as error:
|
|
return str(error)
|
|
return None
|
|
|
|
|
|
def replace_asset(base_url, token, repo_path, release_id, path):
|
|
existing = next(
|
|
(
|
|
asset
|
|
for asset in release_assets(base_url, token, repo_path, release_id)
|
|
if asset.get("name") == path.name and isinstance(asset.get("id"), int)
|
|
),
|
|
None,
|
|
)
|
|
if existing is None:
|
|
_upload_asset(base_url, token, repo_path, release_id, path, path.name)
|
|
return
|
|
|
|
old_id = existing["id"]
|
|
temporary_name = f".{path.name}.upload-{secrets.token_hex(12)}"
|
|
backup_name = f".{path.name}.backup-{secrets.token_hex(12)}"
|
|
temporary = _upload_asset(base_url, token, repo_path, release_id, path, temporary_name)
|
|
temporary_id = temporary["id"]
|
|
old_renamed = False
|
|
try:
|
|
_edit_asset(base_url, token, repo_path, release_id, old_id, backup_name)
|
|
old_renamed = True
|
|
try:
|
|
_edit_asset(base_url, token, repo_path, release_id, temporary_id, path.name)
|
|
except ApiError as error:
|
|
restore_error = None
|
|
try:
|
|
_edit_asset(base_url, token, repo_path, release_id, old_id, path.name)
|
|
except ApiError as restore_failure:
|
|
restore_error = str(restore_failure)
|
|
cleanup_error = _cleanup_asset(
|
|
base_url, token, repo_path, release_id, temporary_id
|
|
)
|
|
details = [str(error)]
|
|
if restore_error:
|
|
details.append(f"could not restore previous asset: {restore_error}")
|
|
if cleanup_error:
|
|
details.append(f"could not remove temporary asset: {cleanup_error}")
|
|
raise ApiError("; ".join(details)) from None
|
|
|
|
cleanup_error = _cleanup_asset(base_url, token, repo_path, release_id, old_id)
|
|
if cleanup_error:
|
|
raise ApiError(f"asset replaced, but could not remove backup asset: {cleanup_error}")
|
|
except ApiError:
|
|
if not old_renamed:
|
|
cleanup_error = _cleanup_asset(base_url, token, repo_path, release_id, temporary_id)
|
|
if cleanup_error:
|
|
raise ApiError(f"asset replacement failed; could not remove temporary asset: {cleanup_error}") from None
|
|
raise
|
|
|
|
|
|
def _chrome_version(value, label):
|
|
if not isinstance(value, str) or not VERSION_RE.fullmatch(value):
|
|
raise ApiError(f"{label} must contain 1 to 4 numeric Chrome version components")
|
|
components = tuple(int(component) for component in value.split("."))
|
|
if any(component > 65535 for component in components):
|
|
raise ApiError(f"{label} contains a component greater than 65535")
|
|
return components + (0,) * (4 - len(components))
|
|
|
|
|
|
def should_publish_updates(existing_version, candidate_version):
|
|
if existing_version is None:
|
|
return True
|
|
return _chrome_version(candidate_version, "candidate version") > _chrome_version(
|
|
existing_version, "existing updates.xml version"
|
|
)
|
|
|
|
|
|
def _updates_version(xml_body):
|
|
try:
|
|
root = ET.fromstring(xml_body)
|
|
except ET.ParseError as error:
|
|
raise ApiError(f"existing updates.xml is not valid XML: {error}") from error
|
|
for element in root.iter():
|
|
if element.tag.rsplit("}", 1)[-1] == "updatecheck" and "version" in element.attrib:
|
|
version = element.attrib["version"]
|
|
_chrome_version(version, "existing updates.xml version")
|
|
return version
|
|
raise ApiError("existing updates.xml has no updatecheck version")
|
|
|
|
|
|
def _public_asset_url(base_url, repo_path, asset):
|
|
for key in ("browser_download_url", "download_url"):
|
|
if isinstance(asset.get(key), str) and asset[key]:
|
|
return asset[key]
|
|
api_suffix = "/api/v1"
|
|
if base_url.rstrip("/").endswith(api_suffix):
|
|
server_url = base_url.rstrip("/")[:-len(api_suffix)]
|
|
return f"{server_url}{repo_path}/releases/download/updates/{quote(asset['name'], safe='')}"
|
|
raise ApiError("Gitea did not provide a public download URL for existing updates.xml")
|
|
|
|
|
|
def _download_anonymously(url, purpose):
|
|
request = Request(url, headers={"Accept": "application/xml"}, method="GET")
|
|
try:
|
|
with urlopen(request) as response:
|
|
return response.read()
|
|
except HTTPError as error:
|
|
raise ApiError(f"anonymous {purpose} request: HTTP {error.code} {error.reason}") from None
|
|
except URLError as error:
|
|
raise ApiError(f"anonymous {purpose} request failed: {error.reason}") from None
|
|
|
|
|
|
def existing_updates_version(base_url, token, repo_path, release_id):
|
|
asset = next(
|
|
(
|
|
asset
|
|
for asset in release_assets(base_url, token, repo_path, release_id)
|
|
if asset.get("name") == "updates.xml"
|
|
),
|
|
None,
|
|
)
|
|
if asset is None:
|
|
return None
|
|
return _updates_version(
|
|
_download_anonymously(_public_asset_url(base_url, repo_path, asset), "existing updates.xml")
|
|
)
|
|
|
|
|
|
def _updates_codebase(xml_path):
|
|
try:
|
|
root = ET.parse(xml_path).getroot()
|
|
except (OSError, ET.ParseError) as error:
|
|
raise ApiError(f"cannot read generated updates.xml: {error}") from error
|
|
for element in root.iter():
|
|
if element.tag.rsplit("}", 1)[-1] == "updatecheck":
|
|
codebase = element.get("codebase")
|
|
if codebase:
|
|
return codebase
|
|
raise ApiError("generated updates.xml has no codebase URL")
|
|
|
|
|
|
def verify_public_crx(updates_path):
|
|
codebase = _updates_codebase(updates_path)
|
|
request = Request(codebase, headers={"Accept": "application/octet-stream"}, method="GET")
|
|
try:
|
|
with urlopen(request) as response:
|
|
status = response.status
|
|
content_type = response.headers.get("Content-Type", "")
|
|
magic = response.read(4)
|
|
except HTTPError as error:
|
|
raise ApiError(
|
|
f"CRX codebase URL is not anonymously reachable (HTTP {error.code}); "
|
|
"make the release public and configure its MIME type"
|
|
) from None
|
|
except URLError as error:
|
|
raise ApiError(
|
|
f"CRX codebase URL is not anonymously reachable ({error.reason}); "
|
|
"make the release public and configure its MIME type"
|
|
) from None
|
|
if status != 200:
|
|
raise ApiError(f"CRX codebase URL returned HTTP {status}; make the release public")
|
|
if magic != b"Cr24":
|
|
raise ApiError("CRX codebase URL did not return a CRX3 file (missing Cr24 magic)")
|
|
if content_type.split(";", 1)[0].strip().lower() != EXPECTED_CRX_CONTENT_TYPE:
|
|
raise ApiError(
|
|
"CRX codebase URL returned an invalid Content-Type; configure Gitea "
|
|
"[repository.mimetype_mapping] or a reverse proxy .crx header as "
|
|
"application/x-chrome-extension"
|
|
)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--tag", required=True)
|
|
parser.add_argument("--directory", required=True, type=Path)
|
|
parser.add_argument("--repository", default=os.environ.get("GITEA_REPOSITORY"))
|
|
parser.add_argument("--api-url", default=os.environ.get("GITEA_API_URL"))
|
|
args = parser.parse_args()
|
|
token = os.environ.get("GITEA_TOKEN")
|
|
server_url = os.environ.get("GITEA_SERVER_URL")
|
|
api_url = args.api_url or (f"{server_url.rstrip('/')}/api/v1" if server_url else None)
|
|
if not token:
|
|
raise ValueError("GITEA_TOKEN is required")
|
|
if not api_url:
|
|
raise ValueError("GITEA_API_URL or GITEA_SERVER_URL is required")
|
|
if not args.repository:
|
|
raise ValueError("GITEA_REPOSITORY is required")
|
|
if not args.tag.startswith("v"):
|
|
raise ValueError("tag must start with v")
|
|
|
|
version = args.tag[1:]
|
|
_chrome_version(version, "tag version")
|
|
repo_path = repository_path(args.repository)
|
|
directory = args.directory
|
|
assets = {
|
|
"crx": directory / f"history-collector-{version}.crx",
|
|
"zip": directory / f"history-collector-{version}.zip",
|
|
"checksums": directory / "checksums.txt",
|
|
"extension_id": directory / "extension-id.txt",
|
|
"updates": directory / "updates.xml",
|
|
}
|
|
if any(not path.is_file() for path in assets.values()):
|
|
raise ValueError("build directory is missing one or more release assets")
|
|
|
|
release = get_or_create_release(api_url, token, repo_path, args.tag, prerelease=False)
|
|
for key in ("crx", "zip", "checksums", "extension_id"):
|
|
replace_asset(api_url, token, repo_path, release["id"], assets[key])
|
|
# Verify the URL emitted in the new XML only after the versioned CRX is public.
|
|
verify_public_crx(assets["updates"])
|
|
|
|
updates_release = get_or_create_release(api_url, token, repo_path, "updates", prerelease=True)
|
|
old_version = existing_updates_version(api_url, token, repo_path, updates_release["id"])
|
|
if should_publish_updates(old_version, version):
|
|
replace_asset(api_url, token, repo_path, updates_release["id"], assets["updates"])
|
|
print(f"Published {args.tag} and updates.xml")
|
|
else:
|
|
print(f"Published {args.tag}; kept existing updates.xml version {old_version}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except (ApiError, OSError, ValueError) as error:
|
|
raise SystemExit(f"publish failed: {error}") from error
|