Kang
446e1a768c
- Updated `release_metadata` function to support manual testing from the main branch without publishing. - Added tests to verify behavior for manual main runs, tag releases, and restrictions on other branches. - Modified workflow to conditionally publish images based on the event type and added logging for manual test results. - Enhanced documentation to clarify manual testing procedures and workflow behavior.
239 lines
8.8 KiB
Python
239 lines
8.8 KiB
Python
"""Version checks and registry operations for the Web release workflow."""
|
|
|
|
import argparse
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
SEMVER = re.compile(
|
|
r"(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)"
|
|
r"(?:-([0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?"
|
|
r"(?:\+([0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?"
|
|
)
|
|
DIGEST = re.compile(r"sha256:[0-9a-f]{64}")
|
|
|
|
|
|
def version_parts(version):
|
|
match = SEMVER.fullmatch(version)
|
|
if not match:
|
|
raise ValueError(f"Invalid SemVer: {version!r}")
|
|
prerelease = match[4]
|
|
if prerelease and any(
|
|
part.isdigit() and len(part) > 1 and part.startswith("0")
|
|
for part in prerelease.split(".")
|
|
):
|
|
raise ValueError(f"Invalid numeric prerelease identifier: {version!r}")
|
|
return tuple(int(match[i]) for i in (1, 2, 3)), prerelease
|
|
|
|
|
|
def release_metadata(ref, package_version, repository, event_name="push"):
|
|
test_only = ref == "refs/heads/main" and event_name == "workflow_dispatch"
|
|
if not test_only and not ref.startswith("refs/tags/v"):
|
|
raise ValueError(
|
|
"Select main for a manual test, or a version tag for a release."
|
|
)
|
|
version = package_version if test_only else ref.removeprefix("refs/tags/v")
|
|
_, prerelease = version_parts(version)
|
|
if version != package_version:
|
|
raise ValueError(
|
|
f"Tag version {version!r} differs from package.json {package_version!r}."
|
|
)
|
|
# Docker tags cannot contain '+'. Keep build metadata in the OCI version label.
|
|
image_tag = version.replace("+", "_")
|
|
if len(image_tag) > 128:
|
|
raise ValueError("Version exceeds Docker's 128-character tag limit.")
|
|
owner = repository.split("/")[0].lower()
|
|
return {
|
|
"version": version,
|
|
"image_tag": image_tag,
|
|
"stable": str(prerelease is None).lower(),
|
|
"publish": str(not test_only).lower(),
|
|
"image": f"ghcr.io/{owner}/nyaterm-web",
|
|
}
|
|
|
|
|
|
def should_promote(candidate, current):
|
|
candidate_version, candidate_prerelease = version_parts(candidate)
|
|
if candidate_prerelease:
|
|
return False
|
|
if current is None:
|
|
return True
|
|
current_version, current_prerelease = version_parts(current)
|
|
if current_prerelease:
|
|
raise ValueError(
|
|
"The existing latest image is a prerelease; refusing to replace it automatically."
|
|
)
|
|
return candidate_version > current_version
|
|
|
|
|
|
def verify_manifest(raw, expected_digests):
|
|
manifest = json.loads(raw)
|
|
actual = {
|
|
(item["platform"]["os"], item["platform"]["architecture"]): item["digest"]
|
|
for item in manifest["manifests"]
|
|
}
|
|
if len(manifest["manifests"]) != 2 or actual != expected_digests:
|
|
raise ValueError(f"Unexpected release platforms or digests: {actual}")
|
|
return "sha256:" + hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
class RegistryRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, request, fp, code, msg, headers, newurl):
|
|
redirected = super().redirect_request(request, fp, code, msg, headers, newurl)
|
|
if (
|
|
redirected
|
|
and urllib.parse.urlsplit(request.full_url).netloc
|
|
!= urllib.parse.urlsplit(newurl).netloc
|
|
):
|
|
redirected.remove_header("Authorization")
|
|
return redirected
|
|
|
|
|
|
class Registry:
|
|
def __init__(self, image, username, password):
|
|
registry, self.repository = image.split("/", 1)
|
|
if registry != "ghcr.io":
|
|
raise ValueError("Only GHCR images are supported.")
|
|
self.opener = urllib.request.build_opener(RegistryRedirect())
|
|
credentials = base64.b64encode(f"{username}:{password}".encode()).decode()
|
|
query = urllib.parse.urlencode(
|
|
{"service": "ghcr.io", "scope": f"repository:{self.repository}:pull"}
|
|
)
|
|
response = self.get_json(
|
|
f"https://ghcr.io/token?{query}", {"Authorization": f"Basic {credentials}"}
|
|
)
|
|
self.headers = {"Authorization": f"Bearer {response['token']}"}
|
|
|
|
def get_json(self, url, headers, missing_manifest=False):
|
|
try:
|
|
with self.opener.open(
|
|
urllib.request.Request(url, headers=headers), timeout=30
|
|
) as response:
|
|
return json.load(response)
|
|
except urllib.error.HTTPError as error:
|
|
if missing_manifest and error.code == 404:
|
|
payload = json.loads(error.read())
|
|
codes = [item.get("code") for item in payload.get("errors", [])]
|
|
if codes and all(code == "MANIFEST_UNKNOWN" for code in codes):
|
|
return None
|
|
raise
|
|
|
|
def manifest(self, reference, missing=False):
|
|
if reference != "latest" and not DIGEST.fullmatch(reference):
|
|
raise ValueError("Invalid manifest digest.")
|
|
headers = self.headers | {
|
|
"Accept": ", ".join(
|
|
[
|
|
"application/vnd.oci.image.index.v1+json",
|
|
"application/vnd.docker.distribution.manifest.list.v2+json",
|
|
"application/vnd.oci.image.manifest.v1+json",
|
|
"application/vnd.docker.distribution.manifest.v2+json",
|
|
]
|
|
)
|
|
}
|
|
return self.get_json(
|
|
f"https://ghcr.io/v2/{self.repository}/manifests/{reference}",
|
|
headers,
|
|
missing,
|
|
)
|
|
|
|
def image_version(self, reference, missing=False):
|
|
manifest = self.manifest(reference, missing)
|
|
if manifest is None:
|
|
return None
|
|
if "manifests" in manifest:
|
|
platform = next(
|
|
item
|
|
for item in manifest["manifests"]
|
|
if item.get("platform", {}).get("os") == "linux"
|
|
and item["platform"].get("architecture") == "amd64"
|
|
)
|
|
manifest = self.manifest(platform["digest"])
|
|
digest = manifest["config"]["digest"]
|
|
if not DIGEST.fullmatch(digest):
|
|
raise ValueError("Invalid image config digest.")
|
|
config = self.get_json(
|
|
f"https://ghcr.io/v2/{self.repository}/blobs/{digest}", self.headers
|
|
)
|
|
version = config["config"]["Labels"]["org.opencontainers.image.version"]
|
|
version_parts(version)
|
|
return version
|
|
|
|
|
|
def promote_latest(registry, image, version, digest):
|
|
if registry.image_version(digest) != version:
|
|
raise ValueError(
|
|
"Published image version does not match the candidate release."
|
|
)
|
|
current = registry.image_version("latest", missing=True)
|
|
promote = should_promote(version, current)
|
|
if promote:
|
|
subprocess.run(
|
|
[
|
|
"docker",
|
|
"buildx",
|
|
"imagetools",
|
|
"create",
|
|
"--tag",
|
|
f"{image}:latest",
|
|
f"{image}@{digest}",
|
|
],
|
|
check=True,
|
|
)
|
|
if registry.image_version("latest") != version:
|
|
raise ValueError("latest verification failed.")
|
|
return f"{'Updated' if promote else 'Kept'} latest: candidate={version}, previous={current or '(absent)'}"
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument(
|
|
"command", choices=["metadata", "check-manifest", "promote-latest"]
|
|
)
|
|
args = parser.parse_args()
|
|
if args.command == "metadata":
|
|
metadata = release_metadata(
|
|
os.environ["GITHUB_REF"],
|
|
json.loads(Path("package.json").read_text())["version"],
|
|
os.environ["GITHUB_REPOSITORY"],
|
|
os.environ["GITHUB_EVENT_NAME"],
|
|
)
|
|
with open(os.environ["GITHUB_OUTPUT"], "a") as output:
|
|
for key, value in metadata.items():
|
|
output.write(f"{key}={value}\n")
|
|
mode = "release" if metadata["publish"] == "true" else "test (no publishing)"
|
|
message = f"Validated Web {mode}: {metadata['version']}"
|
|
print(message)
|
|
with open(os.environ["GITHUB_STEP_SUMMARY"], "a") as summary:
|
|
summary.write(message + "\n")
|
|
elif args.command == "check-manifest":
|
|
digest = verify_manifest(
|
|
Path("manifest.json").read_bytes(),
|
|
{
|
|
("linux", arch): os.environ[f"DIGEST_{arch.upper()}"]
|
|
for arch in ("amd64", "arm64")
|
|
},
|
|
)
|
|
with open(os.environ["GITHUB_OUTPUT"], "a") as output:
|
|
output.write(f"digest={digest}\n")
|
|
else:
|
|
image, version, digest = (
|
|
os.environ[key] for key in ("IMAGE", "VERSION", "DIGEST")
|
|
)
|
|
registry = Registry(image, os.environ["GITHUB_ACTOR"], os.environ["GH_TOKEN"])
|
|
message = promote_latest(registry, image, version, digest)
|
|
print(message)
|
|
with open(os.environ["GITHUB_STEP_SUMMARY"], "a") as summary:
|
|
summary.write(message + "\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|