feat(web-release): implement web release workflow and testing scripts
- Added a GitHub Actions workflow for building and releasing the web application on version tag pushes. - Introduced Python and JavaScript scripts for release metadata generation, validation, and smoke testing of the web release image. - Updated .gitignore to include new artifacts and Python cache directories. - Created Docker Compose configuration for deploying the web application with environment variable management. - Enhanced documentation for deployment instructions and usage of the official web release image.
This commit is contained in:
Kang committed
1 parent
b35e71906a
commit
29be430b6a
8 files changed
+927
-1
No files matched your search
@@ -0,0 +1,276 @@
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
import unittest
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"web_release", Path(__file__).with_name("web-release.py")
|
||||
)
|
||||
release = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(release)
|
||||
|
||||
|
||||
class VersionTests(unittest.TestCase):
|
||||
def test_metadata_for_stable_and_prerelease(self):
|
||||
for version, stable in [("1.2.12", "true"), ("1.3.0-beta.1", "false")]:
|
||||
with self.subTest(version=version):
|
||||
metadata = release.release_metadata(
|
||||
f"refs/tags/v{version}", version, "NyaKang/nyaterm"
|
||||
)
|
||||
self.assertEqual(metadata["version"], version)
|
||||
self.assertEqual(metadata["image_tag"], version)
|
||||
self.assertEqual(metadata["stable"], stable)
|
||||
self.assertEqual(metadata["image"], "ghcr.io/nyakang/nyaterm-web")
|
||||
|
||||
def test_build_metadata_is_preserved_in_version_and_encoded_in_tag(self):
|
||||
metadata = release.release_metadata(
|
||||
"refs/tags/v1.2.3+build.4", "1.2.3+build.4", "nyakang/nyaterm"
|
||||
)
|
||||
self.assertEqual(metadata["version"], "1.2.3+build.4")
|
||||
self.assertEqual(metadata["image_tag"], "1.2.3_build.4")
|
||||
|
||||
def test_branch_dispatch_and_mismatched_version_fail(self):
|
||||
for ref in ["refs/heads/main", "refs/tags/1.2.12", "refs/tags/v1.2.11"]:
|
||||
with self.subTest(ref=ref), self.assertRaises(ValueError):
|
||||
release.release_metadata(ref, "1.2.12", "nyakang/nyaterm")
|
||||
|
||||
def test_invalid_semver_fails(self):
|
||||
for version in [
|
||||
"1.2",
|
||||
"01.2.3",
|
||||
"1.2.3-01",
|
||||
"1.2.3-alpha.01",
|
||||
"1.2.3-",
|
||||
"1.2.3+",
|
||||
"1.2.3\n",
|
||||
]:
|
||||
with self.subTest(version=version), self.assertRaises(ValueError):
|
||||
release.version_parts(version)
|
||||
|
||||
def test_docker_tag_length_limit(self):
|
||||
version = "1.2.3-" + "a" * 123
|
||||
with self.assertRaises(ValueError):
|
||||
release.release_metadata(
|
||||
f"refs/tags/v{version}", version, "nyakang/nyaterm"
|
||||
)
|
||||
|
||||
def test_latest_creation_and_numeric_upgrade(self):
|
||||
for candidate, current in [
|
||||
("1.2.12", None),
|
||||
("1.2.12", "1.2.9"),
|
||||
("1.10.0", "1.9.99"),
|
||||
("2.0.0", "1.99.99"),
|
||||
]:
|
||||
with self.subTest(candidate=candidate, current=current):
|
||||
self.assertTrue(release.should_promote(candidate, current))
|
||||
|
||||
def test_latest_never_regresses_or_accepts_prerelease(self):
|
||||
for candidate, current in [
|
||||
("1.2.11", "1.2.12"),
|
||||
("1.2.12", "1.2.12"),
|
||||
("1.2.12+rebuild", "1.2.12"),
|
||||
("2.0.0-beta.1", "1.2.12"),
|
||||
("2.0.0-beta.1", None),
|
||||
]:
|
||||
with self.subTest(candidate=candidate, current=current):
|
||||
self.assertFalse(release.should_promote(candidate, current))
|
||||
|
||||
def test_invalid_existing_latest_fails_closed(self):
|
||||
for current in ["unknown", "1.2.12-beta.1"]:
|
||||
with self.subTest(current=current), self.assertRaises(ValueError):
|
||||
release.should_promote("1.3.0", current)
|
||||
|
||||
|
||||
class ManifestTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.digests = {
|
||||
("linux", "amd64"): "sha256:" + "a" * 64,
|
||||
("linux", "arm64"): "sha256:" + "b" * 64,
|
||||
}
|
||||
self.manifests = [
|
||||
{"platform": {"os": os, "architecture": arch}, "digest": digest}
|
||||
for (os, arch), digest in self.digests.items()
|
||||
]
|
||||
|
||||
def test_both_tested_platforms_required(self):
|
||||
raw = json.dumps({"manifests": self.manifests}).encode()
|
||||
self.assertRegex(
|
||||
release.verify_manifest(raw, self.digests), r"^sha256:[0-9a-f]{64}$"
|
||||
)
|
||||
for manifests in [self.manifests[:1], self.manifests + self.manifests[:1]]:
|
||||
with self.subTest(manifests=manifests), self.assertRaises(ValueError):
|
||||
release.verify_manifest(
|
||||
json.dumps({"manifests": manifests}).encode(), self.digests
|
||||
)
|
||||
|
||||
def test_untested_digest_rejected(self):
|
||||
self.manifests[0]["digest"] = "sha256:" + "c" * 64
|
||||
with self.assertRaises(ValueError):
|
||||
release.verify_manifest(
|
||||
json.dumps({"manifests": self.manifests}).encode(), self.digests
|
||||
)
|
||||
|
||||
|
||||
class PromotionTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.image = "ghcr.io/nyakang/nyaterm-web"
|
||||
self.digest = "sha256:" + "a" * 64
|
||||
self.registry = Mock()
|
||||
|
||||
def test_first_release_and_upgrade_publish_verified_digest(self):
|
||||
for current in [None, "1.2.9"]:
|
||||
with self.subTest(current=current), patch.object(
|
||||
release.subprocess, "run"
|
||||
) as run:
|
||||
self.registry.image_version.side_effect = ["1.2.12", current, "1.2.12"]
|
||||
release.promote_latest(self.registry, self.image, "1.2.12", self.digest)
|
||||
run.assert_called_once_with(
|
||||
[
|
||||
"docker",
|
||||
"buildx",
|
||||
"imagetools",
|
||||
"create",
|
||||
"--tag",
|
||||
f"{self.image}:latest",
|
||||
f"{self.image}@{self.digest}",
|
||||
],
|
||||
check=True,
|
||||
)
|
||||
|
||||
def test_old_equal_and_prerelease_do_not_write_registry(self):
|
||||
for candidate, current in [
|
||||
("1.2.11", "1.2.12"),
|
||||
("1.2.12", "1.2.12"),
|
||||
("1.3.0-beta.1", "1.2.12"),
|
||||
]:
|
||||
with self.subTest(candidate=candidate), patch.object(
|
||||
release.subprocess, "run"
|
||||
) as run:
|
||||
self.registry.image_version.side_effect = [candidate, current]
|
||||
release.promote_latest(
|
||||
self.registry, self.image, candidate, self.digest
|
||||
)
|
||||
run.assert_not_called()
|
||||
|
||||
def test_latest_read_failure_cannot_trigger_publication(self):
|
||||
self.registry.image_version.side_effect = [
|
||||
"1.2.12",
|
||||
urllib.error.URLError("fixture"),
|
||||
]
|
||||
with patch.object(release.subprocess, "run") as run, self.assertRaises(
|
||||
urllib.error.URLError
|
||||
):
|
||||
release.promote_latest(self.registry, self.image, "1.2.12", self.digest)
|
||||
run.assert_not_called()
|
||||
|
||||
def test_wrong_candidate_version_cannot_trigger_publication(self):
|
||||
self.registry.image_version.return_value = "1.2.11"
|
||||
with patch.object(release.subprocess, "run") as run, self.assertRaises(
|
||||
ValueError
|
||||
):
|
||||
release.promote_latest(self.registry, self.image, "1.2.12", self.digest)
|
||||
run.assert_not_called()
|
||||
|
||||
def test_publication_error_and_failed_verification_fail_job(self):
|
||||
self.registry.image_version.side_effect = ["1.2.12", None]
|
||||
with patch.object(
|
||||
release.subprocess, "run", side_effect=RuntimeError("fixture")
|
||||
), self.assertRaises(RuntimeError):
|
||||
release.promote_latest(self.registry, self.image, "1.2.12", self.digest)
|
||||
self.registry.image_version.side_effect = ["1.2.12", "1.2.11", "1.2.11"]
|
||||
with patch.object(release.subprocess, "run"), self.assertRaises(ValueError):
|
||||
release.promote_latest(self.registry, self.image, "1.2.12", self.digest)
|
||||
|
||||
|
||||
class RegistryTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.registry = release.Registry.__new__(release.Registry)
|
||||
self.registry.opener = Mock()
|
||||
self.registry.repository = "nyakang/nyaterm-web"
|
||||
self.registry.headers = {"Authorization": "Bearer disposable-test-token"}
|
||||
|
||||
def http_error(self, status, code):
|
||||
self.registry.opener.open.side_effect = urllib.error.HTTPError(
|
||||
"https://ghcr.io/v2/nyakang/nyaterm-web/manifests/latest",
|
||||
status,
|
||||
"fixture",
|
||||
{},
|
||||
io.BytesIO(json.dumps({"errors": [{"code": code}]}).encode()),
|
||||
)
|
||||
|
||||
def test_only_missing_latest_can_be_treated_as_first_release(self):
|
||||
self.http_error(404, "MANIFEST_UNKNOWN")
|
||||
self.assertIsNone(self.registry.manifest("latest", missing=True))
|
||||
self.http_error(404, "MANIFEST_UNKNOWN")
|
||||
with self.assertRaises(urllib.error.HTTPError):
|
||||
self.registry.manifest("sha256:" + "a" * 64)
|
||||
|
||||
def test_auth_server_and_repository_errors_fail_closed(self):
|
||||
for status, code in [
|
||||
(401, "UNAUTHORIZED"),
|
||||
(403, "DENIED"),
|
||||
(429, "TOOMANYREQUESTS"),
|
||||
(500, "UNKNOWN"),
|
||||
(404, "NAME_UNKNOWN"),
|
||||
]:
|
||||
with self.subTest(status=status, code=code):
|
||||
self.http_error(status, code)
|
||||
with self.assertRaises(urllib.error.HTTPError):
|
||||
self.registry.manifest("latest", missing=True)
|
||||
|
||||
def test_network_failure_propagates(self):
|
||||
self.registry.opener.open.side_effect = urllib.error.URLError("fixture timeout")
|
||||
with self.assertRaises(urllib.error.URLError):
|
||||
self.registry.manifest("latest", missing=True)
|
||||
|
||||
def test_version_is_read_from_amd64_config_in_index(self):
|
||||
digest = "sha256:" + "a" * 64
|
||||
self.registry.manifest = Mock(
|
||||
side_effect=[
|
||||
{
|
||||
"manifests": [
|
||||
{
|
||||
"platform": {"os": "linux", "architecture": "arm64"},
|
||||
"digest": "sha256:" + "b" * 64,
|
||||
},
|
||||
{
|
||||
"platform": {"os": "linux", "architecture": "amd64"},
|
||||
"digest": digest,
|
||||
},
|
||||
]
|
||||
},
|
||||
{"config": {"digest": digest}},
|
||||
]
|
||||
)
|
||||
self.registry.get_json = Mock(
|
||||
return_value={
|
||||
"config": {"Labels": {"org.opencontainers.image.version": "1.2.12"}}
|
||||
}
|
||||
)
|
||||
self.assertEqual(self.registry.image_version("latest"), "1.2.12")
|
||||
self.registry.manifest.assert_called_with(digest)
|
||||
|
||||
def test_missing_version_label_is_an_error(self):
|
||||
self.registry.manifest = Mock(
|
||||
return_value={"config": {"digest": "sha256:" + "a" * 64}}
|
||||
)
|
||||
self.registry.get_json = Mock(return_value={"config": {"Labels": {}}})
|
||||
with self.assertRaises(KeyError):
|
||||
self.registry.image_version("latest")
|
||||
|
||||
def test_registry_redirect_does_not_forward_token_to_blob_storage(self):
|
||||
request = urllib.request.Request(
|
||||
"https://ghcr.io/v2/example/blobs/fixture", headers=self.registry.headers
|
||||
)
|
||||
redirect = release.RegistryRedirect().redirect_request(
|
||||
request, None, 307, "redirect", {}, "https://storage.example/blob"
|
||||
)
|
||||
self.assertFalse(redirect.has_header("Authorization"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,113 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { spawnSync } from "node:child_process";
|
||||
import { randomBytes } from "node:crypto";
|
||||
import { mkdirSync, writeFileSync } from "node:fs";
|
||||
import { resolve } from "node:path";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
|
||||
const image = process.env.NYATERM_SMOKE_IMAGE;
|
||||
assert(image, "NYATERM_SMOKE_IMAGE is required");
|
||||
const name = "nyaterm-web-release-smoke";
|
||||
const artifacts = resolve("artifacts/web-release-smoke");
|
||||
mkdirSync(artifacts, { recursive: true });
|
||||
const password = randomBytes(32).toString("base64");
|
||||
const encryptionKey = randomBytes(32).toString("base64");
|
||||
const docker = (args, optional = false) => {
|
||||
const result = spawnSync("docker", args, {
|
||||
encoding: "utf8",
|
||||
env: {
|
||||
...process.env,
|
||||
NYATERM_WEB_PASSWORD: password,
|
||||
NYATERM_WEB_ENCRYPTION_KEY: encryptionKey,
|
||||
},
|
||||
});
|
||||
if (!optional && (result.error || result.status !== 0))
|
||||
throw new Error(`Docker ${args[0]} failed: ${result.stderr}`);
|
||||
return result;
|
||||
};
|
||||
const fetchWithTimeout = (url, options = {}) =>
|
||||
fetch(url, { ...options, signal: AbortSignal.timeout(5000) });
|
||||
|
||||
try {
|
||||
docker([
|
||||
"run",
|
||||
"-d",
|
||||
"--name",
|
||||
name,
|
||||
"--init",
|
||||
"-p",
|
||||
"127.0.0.1::8080",
|
||||
"-e",
|
||||
"NYATERM_WEB_PASSWORD",
|
||||
"-e",
|
||||
"NYATERM_WEB_ENCRYPTION_KEY",
|
||||
"--read-only",
|
||||
"--tmpfs",
|
||||
"/tmp",
|
||||
"--cap-drop",
|
||||
"ALL",
|
||||
"--security-opt",
|
||||
"no-new-privileges:true",
|
||||
image,
|
||||
]);
|
||||
const binding = docker(["port", name, "8080/tcp"]).stdout.trim();
|
||||
const base = `http://${binding}/`;
|
||||
const deadline = Date.now() + 60_000;
|
||||
let response;
|
||||
while (Date.now() < deadline) {
|
||||
try {
|
||||
response = await fetchWithTimeout(base);
|
||||
if (response.ok) break;
|
||||
} catch {}
|
||||
await delay(500);
|
||||
}
|
||||
assert(response?.ok, "Container readiness timed out");
|
||||
const html = await response.text();
|
||||
assert(html.includes('<div id="root">'), "Missing application HTML");
|
||||
const assets = [...html.matchAll(/(?:src|href)="(\/assets\/[^"\s]+)"/g)];
|
||||
assert(assets.length > 0, "No built assets found");
|
||||
for (const [, path] of assets) {
|
||||
const asset = await fetchWithTimeout(new URL(path, base));
|
||||
assert.equal(asset.status, 200, `Asset unavailable: ${path}`);
|
||||
assert(
|
||||
!asset.headers.get("content-type")?.includes("text/html"),
|
||||
"Asset returned SPA fallback",
|
||||
);
|
||||
assert((await asset.arrayBuffer()).byteLength > 0, "Empty asset");
|
||||
}
|
||||
const sessionUrl = new URL("api/auth/session", base);
|
||||
assert.equal((await fetchWithTimeout(sessionUrl)).status, 401);
|
||||
const headers = {
|
||||
Origin: new URL(base).origin,
|
||||
"X-Nyaterm-Request": "1",
|
||||
"Content-Type": "application/json",
|
||||
};
|
||||
const loginUrl = new URL("api/auth/login", base);
|
||||
const rejected = await fetchWithTimeout(loginUrl, {
|
||||
method: "POST",
|
||||
headers,
|
||||
body: JSON.stringify({ password: "invalid" }),
|
||||
});
|
||||
assert.equal(rejected.status, 401);
|
||||
const login = await fetchWithTimeout(loginUrl, {
|
||||
method: "POST",
|
||||
headers,
|
||||
body: JSON.stringify({ password }),
|
||||
});
|
||||
assert.equal(login.status, 200, "Login failed");
|
||||
const { csrf } = await login.json();
|
||||
assert(csrf, "Missing CSRF token");
|
||||
const cookie = login.headers.get("set-cookie")?.split(";")[0];
|
||||
assert(cookie, "Missing session cookie");
|
||||
const session = await fetchWithTimeout(sessionUrl, {
|
||||
headers: { Cookie: cookie },
|
||||
});
|
||||
assert.equal(session.status, 200);
|
||||
assert.equal((await session.json()).csrf, csrf);
|
||||
console.log("Web release image smoke checks passed.");
|
||||
} finally {
|
||||
const logs = docker(["logs", name], true);
|
||||
writeFileSync(resolve(artifacts, "web.log"), logs.stdout + logs.stderr);
|
||||
// -v removes the anonymous /data volume created by the Dockerfile.
|
||||
docker(["rm", "-f", "-v", name], true);
|
||||
}
|
||||
@@ -0,0 +1,229 @@
|
||||
"""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):
|
||||
if not ref.startswith("refs/tags/v"):
|
||||
raise ValueError("Dispatch against a version tag (refs/tags/v<SemVer>).")
|
||||
version = 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(),
|
||||
"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"],
|
||||
)
|
||||
with open(os.environ["GITHUB_OUTPUT"], "a") as output:
|
||||
for key, value in metadata.items():
|
||||
output.write(f"{key}={value}\n")
|
||||
print(f"Validated Web release {metadata['version']}")
|
||||
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()
|
||||
Reference in new issue
Block a user