ci(server): validate and publish versioned GHCR images

scripts/server_image_release.py

@@ -0,0 +1,452 @@
+#!/usr/bin/env python3
+"""Fail-closed, read-only source/GHCR checks. This tool never tags or pushes.
+
+Credentials stay in memory; outputs contain only validated provenance/digests.
+The workflow alone owns Docker login and the two explicit push commands.
+"""
+import argparse
+import base64
+import gzip
+import hashlib
+import io
+import json
+import os
+import pathlib
+import re
+import subprocess
+import sys
+import tarfile
+import tempfile
+import urllib.error
+import urllib.parse
+import urllib.request
+
+REPO = "ajisaacs/OpenNest"
+SOURCE = "https://github.com/" + REPO
+PACKAGE = "opennest-server"
+REGISTRY_NAME = "ajisaacs/" + PACKAGE
+IMAGE = "ghcr.io/" + REGISTRY_NAME
+REPO_API = "https://api.github.com/repos/" + REPO
+PACKAGE_API = "https://api.github.com/users/ajisaacs/packages/container/" + PACKAGE
+MANIFEST_TYPES = ("application/vnd.oci.image.manifest.v1+json",
+                  "application/vnd.docker.distribution.manifest.v2+json")
+# Any stored image representation is a collision; published readback stays strict.
+REGISTRY_MANIFEST_TYPES = MANIFEST_TYPES + ("application/vnd.oci.image.index.v1+json",
+                                          "application/vnd.docker.distribution.manifest.list.v2+json")
+
+
+class ReleaseError(Exception):
+    pass
+
+
+def require(condition, message):
+    if not condition:
+        raise ReleaseError(message)
+
+
+def version(tag):
+    require(isinstance(tag, str) and re.fullmatch(
+        r"v(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)", tag), "invalid release tag")
+    return tag[1:]
+
+
+def commit(value):
+    require(isinstance(value, str) and re.fullmatch(r"[0-9a-f]{40}", value), "invalid full commit SHA")
+    return value
+
+
+def sha256(value):
+    require(isinstance(value, str) and re.fullmatch(r"sha256:[0-9a-f]{64}", value), "invalid SHA-256 digest")
+    return value
+
+
+def json_body(body):
+    def unique(pairs):
+        result = {}
+        for key, value in pairs:
+            require(key not in result, "duplicate JSON key")
+            result[key] = value
+        return result
+    try:
+        return json.loads(body, object_pairs_hook=unique)
+    except (ValueError, UnicodeError, TypeError):
+        raise ReleaseError("malformed JSON response") from None
+
+
+def command(*args):
+    try:
+        result = subprocess.run(args, check=False, capture_output=True, text=True, timeout=300)
+    except (OSError, subprocess.TimeoutExpired):
+        raise ReleaseError("local command unavailable or timed out") from None
+    require(result.returncode == 0, "local command failed: " + args[0])
+    return result.stdout.strip()
+
+
+def git(*args):
+    return command("git", *args)
+
+
+def resolve_source(repo, event, ref, tag, event_sha=None):
+    version(tag)  # Validate before constructing a ref or stripping v.
+    require(repo == REPO, "publication requires the primary repository")
+    require((event == "workflow_dispatch" and ref == "refs/heads/master") or
+            (event == "release" and ref == "refs/tags/" + tag), "unapproved publication event/ref")
+    expected = None
+    if event == "release":
+        expected = commit(event_sha if event_sha is not None else os.environ.get("GITHUB_SHA", ""))
+    sha = commit(git("rev-parse", "--verify", "refs/tags/" + tag + "^{commit}"))
+    require(expected is None or sha == expected, "release tag differs from event commit")
+    git("merge-base", "--is-ancestor", sha, "refs/remotes/origin/master")
+    return sha
+
+
+def registry_path(suffix):
+    return "https://ghcr.io/v2/" + REGISTRY_NAME + "/" + suffix
+
+
+def absent(status, headers, body):
+    if status == 200:
+        return False
+    require(status == 404, "registry lookup failed; not proven absent")
+    data = json_body(body)
+    require(isinstance(data, dict) and isinstance(data.get("errors"), list) and data["errors"],
+            "malformed registry absence response")
+    require(all(isinstance(error, dict) and error.get("code") in ("MANIFEST_UNKNOWN", "NAME_UNKNOWN")
+                for error in data["errors"]), "registry denial/error is not absence")
+    return True
+
+
+def package_policy(status, headers, body):
+    data = json_body(body)
+    require(isinstance(data, dict), "malformed package metadata")
+    require(status == 200, "package metadata inaccessible; privacy unknown")
+    require(data.get("name") == PACKAGE and data.get("package_type") == "container" and
+            data.get("visibility") == "private" and isinstance(data.get("repository"), dict) and
+            data["repository"].get("full_name") == REPO, "package privacy/association mismatch")
+    return False
+
+
+def registry_token(status, body):
+    # Opaque token issuance does not prove the granted scope: GHCR may reduce it.
+    # Never use token success or registry absence to override package metadata.
+    require(status == 200, "registry authentication/scope request failed")
+    data = json_body(body)
+    require(isinstance(data, dict) and isinstance(data.get("token"), str) and
+            bool(data["token"]) and not any(c.isspace() for c in data["token"]),
+            "malformed token response")
+    if "scope" in data:
+        require(data["scope"] == "repository:" + REGISTRY_NAME + ":pull,push", "reduced registry scope")
+    return data["token"]
+
+
+class NoRedirect(urllib.request.HTTPRedirectHandler):
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        return None
+
+
+class Client:
+    def __init__(self):
+        self.github_token = os.environ.get("GITHUB_TOKEN", "")
+        require(bool(self.github_token), "GITHUB_TOKEN required")
+        require(os.environ.get("GITHUB_REPOSITORY") == REPO, "unexpected authenticated repository")
+        self.opener = urllib.request.build_opener(NoRedirect())
+        credentials = base64.b64encode((os.environ.get("GITHUB_ACTOR", "") + ":" +
+                                        self.github_token).encode()).decode()
+        url = "https://ghcr.io/token?" + urllib.parse.urlencode({
+            "service": "ghcr.io", "scope": "repository:" + REGISTRY_NAME + ":pull,push"})
+        status, _, body = self.request(url, {"Authorization": "Basic " + credentials})
+        self.registry_token = registry_token(status, body)
+
+    def request(self, url, headers):
+        try:
+            with self.opener.open(urllib.request.Request(url, headers=headers), timeout=60) as result:
+                return result.status, {k.lower(): v for k, v in result.headers.items()}, result.read()
+        except urllib.error.HTTPError as error:
+            return error.code, {k.lower(): v for k, v in error.headers.items()}, error.read()
+        except (OSError, ValueError):
+            raise ReleaseError("network/transport failure (not absence)") from None
+
+    def get(self, path):
+        if path.startswith("https://api.github.com/"):
+            return self.request(path, {"Authorization": "Bearer " + self.github_token,
+                                     "Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28"})
+        require(path.startswith(registry_path("")), "unexpected registry endpoint")
+        result = self.request(path, {"Authorization": "Bearer " + self.registry_token,
+                                    "Accept": ", ".join(REGISTRY_MANIFEST_TYPES)})
+        if "/blobs/" in path and result[0] in (302, 307):
+            location = result[1].get("location", "")
+            parsed = urllib.parse.urlparse(location)
+            require(parsed.scheme == "https" and parsed.hostname == "pkg-containers.githubusercontent.com",
+                    "unexpected blob redirect")
+            # Never forward credentials to redirected storage or record signed URLs.
+            return self.request(location, {})
+        return result
+
+
+def preflight(client, release_version, sha):
+    status, _, body = client.get(REPO_API)
+    repo = json_body(body)
+    require(status == 200 and isinstance(repo, dict) and repo.get("full_name") == REPO and
+            repo.get("default_branch") == "master", "repository/default branch authorization unproven")
+    package_policy(*client.get(PACKAGE_API))
+    tags_response = client.get(registry_path("tags/list"))
+    require(tags_response[0] == 200, "existing package registry read access unproven")
+    data = json_body(tags_response[2])
+    require(isinstance(data, dict) and data.get("name") == REGISTRY_NAME and
+            (data.get("tags") is None or (isinstance(data.get("tags"), list) and
+             all(isinstance(tag, str) for tag in data["tags"]))), "malformed registry tag list")
+    require("tags" in data and "link" not in tags_response[1], "incomplete registry tag list")
+    reserved_tags = (release_version, "sha-" + sha)
+    require(not any(tag in (data["tags"] or []) for tag in reserved_tags), "refusing existing immutable tag")
+    for tag in reserved_tags:
+        require(absent(*client.get(registry_path("manifests/" + tag))), "refusing existing immutable tag")
+    return {"package": IMAGE, "visibility": "private"}
+
+
+def content_digest(body):
+    return "sha256:" + hashlib.sha256(body).hexdigest()
+
+
+def stream_digest(stream):
+    digest = hashlib.sha256()
+    for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+        digest.update(chunk)
+    return "sha256:" + digest.hexdigest()
+
+
+def layer_diff_id(stream, media_type):
+    try:
+        if media_type in ("application/vnd.oci.image.layer.v1.tar+gzip",
+                          "application/vnd.docker.image.rootfs.diff.tar.gzip"):
+            with gzip.GzipFile(fileobj=stream) as decoded:
+                return stream_digest(decoded)
+        require(media_type == "application/vnd.oci.image.layer.v1.tar", "unsupported layer encoding")
+        return stream_digest(stream)
+    except (OSError, EOFError):
+        raise ReleaseError("invalid layer compression") from None
+
+
+def image_manifest(body):
+    data = json_body(body)
+    require(isinstance(data, dict) and data.get("schemaVersion") == 2 and
+            data.get("mediaType") in MANIFEST_TYPES and isinstance(data.get("config"), dict) and
+            isinstance(data.get("layers"), list), "unexpected manifest (single-platform required)")
+    return data
+
+
+def rootfs(config):
+    data = config.get("rootfs")
+    require(isinstance(data, dict) and data.get("type") == "layers" and
+            isinstance(data.get("diff_ids"), list) and data["diff_ids"], "missing config rootfs chain")
+    return [sha256(value) for value in data["diff_ids"]]
+
+
+def archive_identity(path, local):
+    """Read Docker's saved config/layers, never infer config identity from Id.
+
+    containerd save has an OCI layout envelope; its *one referenced image* must
+    be a manifest, not an index. Classic Docker save has only manifest.json.
+    No archive extraction, and all temporary image bytes are removed by caller.
+    """
+    try:
+        with tarfile.open(path) as archive:
+            def member(name):
+                entry = archive.getmember(name)
+                require(entry.isfile(), "archive member is not a regular file")
+                stream = archive.extractfile(entry)
+                if stream is None:
+                    raise ReleaseError("missing archive file")
+                return stream
+
+            def blob(descriptor):
+                require(isinstance(descriptor, dict), "invalid blob descriptor")
+                digest = sha256(descriptor.get("digest"))
+                stream = member("blobs/sha256/" + digest[7:])
+                require(type(descriptor.get("size")) is int and descriptor["size"] == archive.getmember("blobs/sha256/" + digest[7:]).size,
+                        "archive blob size mismatch")
+                require(stream_digest(stream) == digest, "archive blob digest mismatch")
+                stream.close()
+                return member("blobs/sha256/" + digest[7:])
+
+            names = archive.getnames()
+            require(len(names) == len(set(names)), "duplicate archive members")
+            if "index.json" in names:
+                index = json_body(member("index.json").read())
+                require(isinstance(index, dict) and index.get("schemaVersion") == 2 and
+                        isinstance(index.get("manifests"), list) and len(index["manifests"]) == 1,
+                        "ambiguous image archive")
+                descriptor = index["manifests"][0]
+                require(isinstance(descriptor, dict) and descriptor.get("mediaType") in MANIFEST_TYPES,
+                        "unexpected archive image index/attestation")
+                manifest = image_manifest(blob(descriptor).read())
+                config_body = blob(manifest["config"]).read()
+                layers = manifest["layers"]
+                diffs = [layer_diff_id(blob(layer), layer.get("mediaType")) for layer in layers]
+            else:
+                entries = json_body(member("manifest.json").read())
+                require(isinstance(entries, list) and len(entries) == 1 and isinstance(entries[0], dict),
+                        "ambiguous classic image archive")
+                config_body = member(entries[0]["Config"]).read()
+                diffs = [stream_digest(member(name)) for name in entries[0]["Layers"]]
+            config = json_body(config_body)
+            require(isinstance(config, dict) and config.get("os") == local.get("Os") and
+                    config.get("architecture") == local.get("Architecture") and config.get("config") == local.get("Config"),
+                    "saved config differs from inspected image")
+            require(diffs == rootfs(config) == local.get("RootFS", {}).get("Layers"), "saved layer/rootfs chain mismatch")
+            return {"config_digest": content_digest(config_body), "rootfs_diff_ids": diffs}
+    except (OSError, tarfile.TarError, KeyError, TypeError, AttributeError):
+        raise ReleaseError("invalid image archive") from None
+
+
+def local_identity(image):
+    local = inspect(image)
+    # Use the inspected immutable store ID, not a potentially retargeted tag.
+    with tempfile.TemporaryDirectory(prefix="opennest-image-identity-") as directory:
+        path = pathlib.Path(directory) / "image.tar"
+        command("docker", "image", "save", "-o", str(path), sha256(local.get("Id")))
+        local.update(archive_identity(path, local))
+    require(inspect(image)["Id"] == local["Id"], "local image changed while inspecting")
+    return local
+
+
+def check_local(local, release_version, sha):
+    require(isinstance(local, dict), "invalid local image inspect")
+    image_id = sha256(local.get("Id"))
+    require(local.get("Os") == "linux" and local.get("Architecture") == "amd64", "wrong image platform")
+    config = local.get("Config")
+    require(isinstance(config, dict) and isinstance(config.get("Labels"), dict), "missing image config/labels")
+    labels = config["Labels"]
+    for key, expected in (("source", SOURCE), ("version", release_version), ("revision", sha)):
+        require(labels.get("org.opencontainers.image." + key) == expected, "OCI label mismatch: " + key)
+    base = labels.get("org.opencontainers.image.base.name", "")
+    require(isinstance(base, str) and re.fullmatch(
+        r"mcr\.microsoft\.com/dotnet/aspnet@sha256:[0-9a-f]{64}", base), "runtime base not digest pinned")
+    config_digest = sha256(local.get("config_digest"))
+    diffs = local.get("rootfs_diff_ids")
+    require(isinstance(diffs, list) and diffs and
+            diffs == local.get("RootFS", {}).get("Layers"), "local rootfs identity mismatch")
+    for value in diffs:
+        sha256(value)
+    return {"image_id": image_id, "config_digest": config_digest, "rootfs_diff_ids": diffs,
+            "platform": "linux/amd64", "version": release_version,
+            "source_sha": sha, "runtime_image": base, "source": SOURCE}
+
+
+def readback(client, release_version, sha, local):
+    result = check_local(local, release_version, sha)
+    manifests = []
+    digests = []
+    for tag in (release_version, "sha-" + sha):
+        status, headers, body = client.get(registry_path("manifests/" + tag))
+        require(status == 200, "published tag readback failed")
+        remote_digest = sha256(headers.get("docker-content-digest"))
+        require(remote_digest == "sha256:" + hashlib.sha256(body).hexdigest(), "manifest content digest mismatch")
+        manifest = image_manifest(body)
+        manifests.append(manifest)
+        digests.append(remote_digest)
+    require(digests[0] == digests[1], "published tags have different digests")
+    descriptor = manifests[0]["config"]
+    config_digest = sha256(descriptor.get("digest"))
+    require(config_digest == result["config_digest"], "registry config differs from smoked local config")
+    status, _, body = client.get(registry_path("blobs/" + config_digest))
+    require(status == 200 and type(descriptor.get("size")) is int and descriptor["size"] == len(body) and
+            config_digest == "sha256:" + hashlib.sha256(body).hexdigest(), "config blob readback mismatch")
+    remote = json_body(body)
+    require(isinstance(remote, dict) and remote.get("os") == "linux" and remote.get("architecture") == "amd64" and
+            remote.get("config") == local["Config"], "remote platform/image config mismatch")
+    diffs = rootfs(remote)
+    require(diffs == result["rootfs_diff_ids"] and len(manifests[0]["layers"]) == len(diffs),
+            "remote rootfs chain differs from smoked image")
+    for layer, expected in zip(manifests[0]["layers"], diffs):
+        require(isinstance(layer, dict), "invalid remote layer descriptor")
+        digest = sha256(layer.get("digest"))
+        status, _, body = client.get(registry_path("blobs/" + digest))
+        require(status == 200 and type(layer.get("size")) is int and layer["size"] == len(body) and
+                digest == content_digest(body), "layer blob readback mismatch")
+        require(layer_diff_id(io.BytesIO(body), layer.get("mediaType")) == expected,
+                "remote layer does not match smoked rootfs")
+    result.update(manifest_digest=digests[0], config_digest=config_digest, package=IMAGE)
+    return result
+
+
+def inspect(image):
+    result = json_body(command("docker", "image", "inspect", image))
+    require(isinstance(result, list) and len(result) == 1, "ambiguous local image")
+    return result[0]
+
+
+def outputs(values):
+    path = os.environ.get("GITHUB_OUTPUT")
+    if path:
+        with open(path, "a", encoding="utf-8") as file:
+            for key, value in values.items():
+                require(isinstance(value, str) and "\n" not in value and "\r" not in value, "unsafe workflow output")
+                file.write(key + "=" + value + "\n")
+
+
+def save(name, data):
+    directory = pathlib.Path(os.environ.get("METADATA_DIR", ".hermes/server-image-metadata"))
+    directory.mkdir(parents=True, exist_ok=True)
+    (directory / name).write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("operation", choices=("source", "bases", "local", "preflight", "readback", "pulled"))
+    args = parser.parse_args()
+    if args.operation == "source":
+        tag = os.environ.get("TAG", "")
+        sha = resolve_source(os.environ.get("GITHUB_REPOSITORY"), os.environ.get("GITHUB_EVENT_NAME"),
+                             os.environ.get("GITHUB_REF"), tag, os.environ.get("GITHUB_SHA"))
+        outputs({"source_sha": sha, "version": version(tag)})
+        save("source.json", {"source_sha": sha, "version": version(tag), "tag": tag})
+        return
+    if args.operation == "bases":
+        bases = {}
+        for key, kind in (("sdk_image", "sdk"), ("runtime_image", "aspnet")):
+            name = "mcr.microsoft.com/dotnet/" + kind
+            command("docker", "pull", "--platform", "linux/amd64", name + ":8.0")
+            data = inspect(name + ":8.0")
+            candidates = [value for value in data.get("RepoDigests", []) if value.startswith(name + "@")]
+            require(len(candidates) == 1, "base digest not uniquely resolved")
+            sha256(candidates[0].split("@", 1)[1])
+            bases[key] = candidates[0]
+        outputs(bases)
+        save("bases.json", bases)
+        return
+    release_version = version("v" + os.environ.get("VERSION", ""))
+    sha = commit(os.environ.get("SOURCE_SHA", ""))
+    require(git("rev-parse", "HEAD") == sha, "checkout is not the exact tested source")
+    if args.operation == "preflight":
+        save("preflight.json", preflight(Client(), release_version, sha))
+        return
+    local = local_identity(os.environ.get("LOCAL_IMAGE", ""))
+    if args.operation == "readback":
+        client = Client()
+        package_policy(*client.get(PACKAGE_API))
+        directory = pathlib.Path(os.environ.get("METADATA_DIR", ".hermes/server-image-metadata"))
+        try:
+            previous = json_body((directory / "local.json").read_bytes())
+        except OSError:
+            raise ReleaseError("pre-smoke identity unavailable") from None
+        require(previous == check_local(local, release_version, sha), "image differs from pre-smoke identity")
+        result = readback(client, release_version, sha, local)
+        outputs({"manifest_digest": result["manifest_digest"], "config_digest": result["config_digest"]})
+        save("registry.json", result)
+    else:
+        result = check_local(local, release_version, sha)
+        if args.operation == "pulled":
+            require(result["config_digest"] == sha256(os.environ.get("CONFIG_DIGEST", "")), "pulled config digest mismatch")
+            expected = sha256(os.environ.get("MANIFEST_DIGEST", ""))
+            require(IMAGE + "@" + expected in local.get("RepoDigests", []), "pulled registry digest mismatch")
+        if args.operation == "local":
+            outputs({"image_id": result["image_id"]})
+        save("pulled.json" if args.operation == "pulled" else "local.json", result)
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except ReleaseError as error:
+        print("STOP: " + str(error), file=sys.stderr)
+        sys.exit(1)

scripts/test_server_image_release.py

@@ -0,0 +1,654 @@
+"""Behavioral gates for the read-only server image release helper."""
+import copy
+import gzip
+import hashlib
+import io
+import os
+import tarfile
+import importlib.util
+import json
+import pathlib
+import subprocess
+import tempfile
+import unittest
+from unittest import mock
+
+SPEC = importlib.util.spec_from_file_location(
+    "release", pathlib.Path(__file__).with_name("server_image_release.py"))
+assert SPEC is not None and SPEC.loader is not None
+release = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(release)
+SHA = "a" * 40
+BASE = "mcr.microsoft.com/dotnet/aspnet@sha256:" + "b" * 64
+LAYER = b"deterministic synthetic rootfs layer"
+LAYER_GZIP = gzip.compress(LAYER, mtime=0)
+
+
+def encoded(value):
+    return json.dumps(value, separators=(",", ":")).encode()
+
+
+def digest(data):
+    return "sha256:" + hashlib.sha256(data).hexdigest()
+
+
+def config():
+    return {"os": "linux", "architecture": "amd64",
+            "rootfs": {"type": "layers", "diff_ids": [digest(LAYER)]}, "config": {
+        "User": "1654", "Entrypoint": ["dotnet", "OpenNest.Server.dll"],
+        "Labels": {"org.opencontainers.image.source": release.SOURCE,
+                   "org.opencontainers.image.revision": SHA,
+                   "org.opencontainers.image.version": "1.2.3",
+                   "org.opencontainers.image.base.name": BASE}}}
+
+
+class FakeClient:
+    def __init__(self, responses):
+        self.responses = responses
+        self.calls = []
+
+    def get(self, path, **kwargs):
+        self.calls.append(path)
+        response = self.responses[path]
+        if isinstance(response, Exception):
+            raise response
+        return response
+
+
+def response(status, value, headers=None):
+    return status, headers or {}, encoded(value)
+
+
+class StrictInputs(unittest.TestCase):
+    def test_tag_positive(self):
+        for value in ("v0.0.0", "v1.2.3", "v10.20.30"):
+            self.assertEqual(release.version(value), value[1:])
+
+    def test_tag_rejects_shell_and_noncanonical_versions(self):
+        for value in ("1.2.3", "v01.2.3", "v1.02.3", "v1.2.03", "v1.2.3\n",
+                      "v1.2.3-rc1", "v1.2.3+meta", "v1.2", "v1.2.3;id", ""):
+            with self.subTest(value=value), self.assertRaises(release.ReleaseError):
+                release.version(value)
+
+    def test_sha_and_digest_are_full_lowercase(self):
+        self.assertEqual(release.commit(SHA), SHA)
+        self.assertEqual(release.sha256("sha256:" + "a" * 64), "sha256:" + "a" * 64)
+        for value in ("a" * 39, "A" * 40, SHA + "\n", "--help"):
+            with self.assertRaises(release.ReleaseError):
+                release.commit(value)
+        for value in ("sha256:abc", "sha512:" + "a" * 64, "sha256:" + "A" * 64):
+            with self.assertRaises(release.ReleaseError):
+                release.sha256(value)
+
+    def test_source_requires_primary_repo_and_approved_event(self):
+        with mock.patch.object(release, "git", return_value=SHA):
+            self.assertEqual(release.resolve_source(release.REPO, "workflow_dispatch",
+                                                   "refs/heads/master", "v1.2.3"), SHA)
+            self.assertEqual(release.resolve_source(release.REPO, "release",
+                                                   "refs/tags/v1.2.3", "v1.2.3", SHA), SHA)
+            for repo, event, ref in (("fork/OpenNest", "release", "refs/tags/v1.2.3"),
+                                     (release.REPO, "push", "refs/tags/v1.2.3"),
+                                     (release.REPO, "workflow_dispatch", "refs/heads/other")):
+                with self.assertRaises(release.ReleaseError):
+                    release.resolve_source(repo, event, ref, "v1.2.3")
+
+    def test_source_rejects_missing_tag_or_non_master_ancestry(self):
+        for calls in ([release.ReleaseError("missing")], [SHA, release.ReleaseError("ancestry")]):
+            with mock.patch.object(release, "git", side_effect=calls):
+                with self.assertRaises(release.ReleaseError):
+                    release.resolve_source(release.REPO, "release", "refs/tags/v1.2.3", "v1.2.3", SHA)
+
+    def test_source_validates_tag_before_git(self):
+        with mock.patch.object(release, "git") as git:
+            with self.assertRaises(release.ReleaseError):
+                release.resolve_source(release.REPO, "release", "refs/tags/x", "x;id")
+            git.assert_not_called()
+
+    def test_json_rejects_malformed_and_duplicate_keys(self):
+        for value in (b"<html>", b'{"a":1,"a":2}'):
+            with self.assertRaises(release.ReleaseError):
+                release.json_body(value)
+
+
+class RegistrySafety(unittest.TestCase):
+    def test_only_authenticated_structured_404_is_absence(self):
+        for code in ("NAME_UNKNOWN", "MANIFEST_UNKNOWN"):
+            self.assertTrue(release.absent(*response(404, {"errors": [{"code": code}]})))
+        for status, body in ((401, {"errors": [{"code": "UNAUTHORIZED"}]}),
+                             (403, {}), (429, {}), (500, {}), (404, {}),
+                             (404, {"errors": []}),
+                             (404, {"errors": [{"code": "DENIED"}]}),
+                             (404, {"errors": [{"code": "NAME_UNKNOWN"}, {"code": "DENIED"}]})):
+            with self.subTest(status=status, body=body), self.assertRaises(release.ReleaseError):
+                release.absent(*response(status, body))
+        self.assertFalse(release.absent(*response(200, {"schemaVersion": 2})))
+
+    def test_private_associated_package(self):
+        package = {"name": release.PACKAGE, "package_type": "container", "visibility": "private",
+                   "repository": {"full_name": release.REPO}}
+        self.assertFalse(release.package_policy(*response(200, package)))
+        for key, value in (("visibility", "public"), ("visibility", "internal"),
+                           ("repository", None), ("repository", {"full_name": "other/OpenNest"}),
+                           ("name", "other"), ("package_type", "npm")):
+            wrong = dict(package, **{key: value})
+            with self.assertRaises(release.ReleaseError):
+                release.package_policy(*response(200, wrong))
+
+    def test_package_metadata_404_never_authorizes_bootstrap(self):
+        with self.assertRaises(release.ReleaseError):
+            release.package_policy(*response(404, {
+                "message": "Not Found", "documentation_url": "https://docs.github.com/rest/packages"}))
+        for status, body in ((403, {"message": "Resource not accessible"}), (404, {}),
+                             (404, {"message": "Not Found", "documentation_url": "https://evil.test"})):
+            with self.assertRaises(release.ReleaseError):
+                release.package_policy(*response(status, body))
+
+    def test_authenticated_scope_response_must_succeed_and_be_well_formed(self):
+        self.assertEqual(release.registry_token(200, encoded({"token": "opaque-v1-token"})), "opaque-v1-token")
+        for status, body in ((401, {}), (403, {"errors": [{"code": "DENIED"}]}),
+                             (500, {}), (200, {}), (200, {"token": ""}), (200, {"token": "a\nb"})):
+            with self.assertRaises(release.ReleaseError):
+                release.registry_token(status, encoded(body))
+
+    def preflight_client(self, package_status=200, tags=None):
+        package = {"name": release.PACKAGE, "package_type": "container", "visibility": "private",
+                   "repository": {"full_name": release.REPO}}
+        missing = {"message": "Not Found", "documentation_url": "https://docs.github.com/rest/packages"}
+        return FakeClient({
+            release.REPO_API: response(200, {"full_name": release.REPO, "default_branch": "master"}),
+            release.PACKAGE_API: response(package_status, package if package_status == 200 else missing),
+            release.registry_path("tags/list"): response(200, {"name": release.REGISTRY_NAME, "tags": tags or []}),
+            release.registry_path("manifests/1.2.3"): response(404, {"errors": [{"code": "MANIFEST_UNKNOWN"}]}),
+            release.registry_path("manifests/sha-" + SHA): response(404, {"errors": [{"code": "MANIFEST_UNKNOWN"}]})})
+
+    def test_preflight_checks_both_collision_tags(self):
+        client = self.preflight_client()
+        release.preflight(client, "1.2.3", SHA)
+        self.assertIn(release.registry_path("manifests/sha-" + SHA), client.calls)
+        for tag in ("1.2.3", "sha-" + SHA):
+            client = self.preflight_client()
+            client.responses[release.registry_path("manifests/" + tag)] = response(200, {})
+            with self.assertRaises(release.ReleaseError):
+                release.preflight(client, "1.2.3", SHA)
+
+    def test_missing_metadata_denies_regardless_of_registry_tags(self):
+        with self.assertRaises(release.ReleaseError):
+            release.preflight(self.preflight_client(404), "1.2.3", SHA)
+        with self.assertRaises(release.ReleaseError):
+            release.preflight(self.preflight_client(404, ["old"]), "1.2.3", SHA)
+
+    def test_preflight_transport_metadata_and_tag_list_errors_fail_closed(self):
+        for path in (release.REPO_API, release.PACKAGE_API, release.registry_path("tags/list")):
+            for bad in (release.ReleaseError("network"), response(401, {}), response(200, {})):
+                client = self.preflight_client()
+                client.responses[path] = bad
+                with self.subTest(path=path), self.assertRaises(release.ReleaseError):
+                    release.preflight(client, "1.2.3", SHA)
+
+
+class ReadbackSafety(unittest.TestCase):
+    def fixture(self):
+        blob = encoded(config())
+        manifest = {"schemaVersion": 2, "mediaType": release.MANIFEST_TYPES[0],
+                    "config": {"digest": digest(blob), "size": len(blob)},
+                    "layers": [{"mediaType": "application/vnd.oci.image.layer.v1.tar+gzip",
+                                "digest": digest(LAYER_GZIP), "size": len(LAYER_GZIP)}]}
+        body = encoded(manifest)
+        responses = {release.registry_path("manifests/" + tag):
+                     (200, {"docker-content-digest": digest(body)}, body)
+                     for tag in ("1.2.3", "sha-" + SHA)}
+        responses[release.registry_path("blobs/" + digest(blob))] = (200, {}, blob)
+        responses[release.registry_path("blobs/" + digest(LAYER_GZIP))] = (200, {}, LAYER_GZIP)
+        local = {"Id": digest(blob), "config_digest": digest(blob), "Os": "linux", "Architecture": "amd64",
+                 "Config": config()["config"], "RootFS": {"Type": "layers", "Layers": [digest(LAYER)]},
+                 "rootfs_diff_ids": [digest(LAYER)]}
+        return FakeClient(responses), local
+
+    def test_readback_matches_both_tags_and_full_config(self):
+        client, local = self.fixture()
+        result = release.readback(client, "1.2.3", SHA, local)
+        self.assertEqual(result["config_digest"], local["Id"])
+        self.assertTrue(result["manifest_digest"].startswith("sha256:"))
+        self.assertNotEqual(result["manifest_digest"], result["config_digest"])
+
+    def test_tag_digest_header_body_and_config_mismatch(self):
+        for mode in ("tag", "header", "blob", "local"):
+            client, local = self.fixture()
+            path = release.registry_path("manifests/sha-" + SHA)
+            if mode in ("tag", "header"):
+                status, headers, body = client.responses[path]
+                client.responses[path] = (status, {"docker-content-digest": "sha256:" + "f" * 64}, body)
+            elif mode == "blob":
+                client.responses[release.registry_path("blobs/" + local["Id"])] = (200, {}, b"{}")
+            else:
+                local["Config"]["User"] = "0"
+            with self.subTest(mode=mode), self.assertRaises(release.ReleaseError):
+                release.readback(client, "1.2.3", SHA, local)
+
+    def test_both_valid_but_different_tag_manifests_are_rejected(self):
+        client, local = self.fixture()
+        path = release.registry_path("manifests/sha-" + SHA)
+        _, _, body = client.responses[path]
+        changed = json.loads(body)
+        changed["annotations"] = {"test": "different manifest, same config"}
+        body = encoded(changed)
+        client.responses[path] = (200, {"docker-content-digest": digest(body)}, body)
+        with self.assertRaisesRegex(release.ReleaseError, "different digests"):
+            release.readback(client, "1.2.3", SHA, local)
+
+    def test_remote_platform_and_config_mismatches_even_with_valid_hashes(self):
+        for field, value in (("os", "windows"), ("architecture", "arm64"), ("config", {})):
+            client, local = self.fixture()
+            remote = config()
+            remote[field] = value
+            blob = encoded(remote)
+            local["config_digest"] = digest(blob)
+            manifest = {"schemaVersion": 2, "mediaType": release.MANIFEST_TYPES[0],
+                        "config": {"digest": digest(blob), "size": len(blob)},
+                        "layers": [{"mediaType": "application/vnd.oci.image.layer.v1.tar+gzip",
+                                    "digest": digest(LAYER_GZIP), "size": len(LAYER_GZIP)}]}
+            body = encoded(manifest)
+            for tag in ("1.2.3", "sha-" + SHA):
+                client.responses[release.registry_path("manifests/" + tag)] = (
+                    200, {"docker-content-digest": digest(body)}, body)
+            client.responses[release.registry_path("blobs/" + digest(blob))] = (200, {}, blob)
+            with self.subTest(field=field), self.assertRaises(release.ReleaseError):
+                release.readback(client, "1.2.3", SHA, local)
+
+    def test_platform_labels_and_image_config_are_exact(self):
+        for field, value in (("Os", "windows"), ("Architecture", "arm64"), ("Id", "sha256:short")):
+            _, local = self.fixture()
+            local[field] = value
+            with self.assertRaises(release.ReleaseError):
+                release.check_local(local, "1.2.3", SHA)
+        for label in ("source", "version", "revision", "base.name"):
+            _, local = self.fixture()
+            local["Config"]["Labels"]["org.opencontainers.image." + label] = "wrong"
+            with self.assertRaises(release.ReleaseError):
+                release.check_local(local, "1.2.3", SHA)
+
+    def test_registry_rejects_index_and_missing_config(self):
+        for bad in ({"schemaVersion": 2, "mediaType": "application/vnd.oci.image.index.v1+json"},
+                    {"schemaVersion": 2, "mediaType": release.MANIFEST_TYPES[0]}):
+            client, local = self.fixture()
+            body = encoded(bad)
+            for tag in ("1.2.3", "sha-" + SHA):
+                client.responses[release.registry_path("manifests/" + tag)] = (
+                    200, {"docker-content-digest": digest(body)}, body)
+            with self.assertRaises(release.ReleaseError):
+                release.readback(client, "1.2.3", SHA, local)
+
+    def test_cli_failure_is_nonzero_without_secret_logging(self):
+        with tempfile.TemporaryDirectory() as directory:
+            result = subprocess.run(["python3", str(pathlib.Path(release.__file__)), "source"],
+                                    env={"PATH": "/usr/bin", "TAG": "bad;id", "GITHUB_TOKEN": "secret-sentinel"},
+                                    capture_output=True, text=True, cwd=directory)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertNotIn("secret-sentinel", result.stdout + result.stderr)
+
+
+class ArchiveIdentity(unittest.TestCase):
+    def archive(self, path, mode="oci", mutation=None):
+        client, local = ReadbackSafety().fixture()
+        config_body = encoded(config())
+        manifest_body = client.responses[release.registry_path("manifests/1.2.3")][2]
+        descriptor = {"mediaType": release.MANIFEST_TYPES[0], "digest": digest(manifest_body), "size": len(manifest_body)}
+        index = {"schemaVersion": 2, "manifests": [descriptor]}
+        members = {"blobs/sha256/" + digest(config_body)[7:]: config_body,
+                   "blobs/sha256/" + digest(manifest_body)[7:]: manifest_body,
+                   "blobs/sha256/" + digest(LAYER_GZIP)[7:]: LAYER_GZIP}
+        if mode == "classic":
+            members = {"config.json": config_body, "layer.tar": LAYER,
+                       "manifest.json": encoded([{"Config": "config.json", "Layers": ["layer.tar"]}])}
+        else:
+            if mutation == "index":
+                descriptor["mediaType"] = "application/vnd.oci.image.index.v1+json"
+            if mutation == "multiple":
+                index["manifests"].append(copy.deepcopy(descriptor))
+            members["index.json"] = encoded(index)
+        if mutation == "blob":
+            members["blobs/sha256/" + digest(LAYER_GZIP)[7:]] = b"bad layer"
+        if mutation == "rootfs":
+            local["RootFS"]["Layers"] = ["sha256:" + "c" * 64]
+        with tarfile.open(path, "w") as archive:
+            for name, body in members.items():
+                entry = tarfile.TarInfo(name)
+                entry.size = len(body)
+                archive.addfile(entry, io.BytesIO(body))
+        local["Id"] = digest(manifest_body) if mode == "oci" else digest(config_body)
+        return local
+
+    def test_archive_config_identity_on_classic_and_containerd_stores(self):
+        for mode in ("classic", "oci"):
+            with self.subTest(mode=mode), tempfile.TemporaryDirectory() as directory:
+                path = pathlib.Path(directory) / "image.tar"
+                local = self.archive(path, mode)
+                identity = release.archive_identity(path, local)
+                self.assertEqual(identity["config_digest"], digest(encoded(config())))
+                self.assertEqual(identity["rootfs_diff_ids"], [digest(LAYER)])
+                self.assertEqual(local["Id"] == identity["config_digest"], mode == "classic")
+
+    def test_archive_rejects_indexes_attestations_corruption_and_wrong_rootfs(self):
+        for mutation in ("index", "multiple", "blob", "rootfs"):
+            with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as directory:
+                path = pathlib.Path(directory) / "image.tar"
+                local = self.archive(path, mutation=mutation)
+                with self.assertRaises(release.ReleaseError):
+                    release.archive_identity(path, local)
+
+    def test_remote_layers_verify_compressed_digest_and_ordered_diff_ids(self):
+        for mode in ("hash", "size", "diff", "encoding", "count"):
+            client, local = ReadbackSafety().fixture()
+            path = release.registry_path("manifests/1.2.3")
+            manifest = json.loads(client.responses[path][2])
+            if mode in ("hash", "size"):
+                client.responses[release.registry_path("blobs/" + digest(LAYER_GZIP))] = (200, {}, b"bad")
+                if mode == "hash":
+                    manifest["layers"][0]["size"] = 3
+            elif mode == "diff":
+                body = gzip.compress(b"different rootfs", mtime=0)
+                manifest["layers"][0].update(digest=digest(body), size=len(body))
+                client.responses[release.registry_path("blobs/" + digest(body))] = (200, {}, body)
+            elif mode == "encoding":
+                manifest["layers"][0]["mediaType"] = "unsupported"
+            else:
+                manifest["layers"] = []
+            body = encoded(manifest)
+            for tag in ("1.2.3", "sha-" + SHA):
+                client.responses[release.registry_path("manifests/" + tag)] = (200, {"docker-content-digest": digest(body)}, body)
+            with self.subTest(mode=mode), self.assertRaises(release.ReleaseError):
+                release.readback(client, "1.2.3", SHA, local)
+
+    def test_pulled_cli_compares_archive_config_not_store_id(self):
+        _, local = ReadbackSafety().fixture()
+        manifest_digest = "sha256:" + "e" * 64
+        local["Id"] = manifest_digest
+        local["RepoDigests"] = [release.IMAGE + "@" + manifest_digest]
+        env = {"VERSION": "1.2.3", "SOURCE_SHA": SHA, "CONFIG_DIGEST": local["config_digest"],
+               "MANIFEST_DIGEST": manifest_digest, "LOCAL_IMAGE": local["Id"]}
+        with mock.patch.dict(os.environ, env), mock.patch.object(release, "git", return_value=SHA), \
+                mock.patch.object(release, "local_identity", return_value=local), \
+                mock.patch.object(release, "save") as save, mock.patch("sys.argv", ["helper", "pulled"]):
+            try:
+                release.main()
+            except release.ReleaseError as error:
+                self.fail("pulled config identity must not depend on Docker Id: " + str(error))
+            self.assertEqual(save.call_args.args[1]["config_digest"], env["CONFIG_DIGEST"])
+            with mock.patch.dict(os.environ, {"CONFIG_DIGEST": manifest_digest}), self.assertRaises(release.ReleaseError):
+                release.main()
+
+    def test_release_event_sha_is_validated_before_git(self):
+        for value in ("", SHA + "\n", SHA.upper(), SHA[:-1], " " + SHA):
+            with mock.patch.object(release, "git") as git, self.subTest(value=value), self.assertRaises(release.ReleaseError):
+                release.resolve_source(release.REPO, "release", "refs/tags/v1.2.3", "v1.2.3", value)
+            git.assert_not_called()
+
+    def test_remote_rootfs_config_mismatch_with_valid_blob_hashes(self):
+        client, local = ReadbackSafety().fixture()
+        remote = config()
+        remote["rootfs"]["diff_ids"] = ["sha256:" + "c" * 64]
+        blob = encoded(remote)
+        # Isolate the rootfs guard after the config digest comparison.
+        local["config_digest"] = digest(blob)
+        manifest = json.loads(client.responses[release.registry_path("manifests/1.2.3")][2])
+        manifest["config"].update(digest=digest(blob), size=len(blob))
+        body = encoded(manifest)
+        for tag in ("1.2.3", "sha-" + SHA):
+            client.responses[release.registry_path("manifests/" + tag)] = (200, {"docker-content-digest": digest(body)}, body)
+        client.responses[release.registry_path("blobs/" + digest(blob))] = (200, {}, blob)
+        with self.assertRaisesRegex(release.ReleaseError, "rootfs chain"):
+            release.readback(client, "1.2.3", SHA, local)
+
+    def test_readback_requires_the_pre_smoke_identity_stamp(self):
+        client, local = ReadbackSafety().fixture()
+        client.responses[release.PACKAGE_API] = response(200, {"name": release.PACKAGE,
+            "package_type": "container", "visibility": "private", "repository": {"full_name": release.REPO}})
+        with tempfile.TemporaryDirectory() as directory:
+            env = {"VERSION": "1.2.3", "SOURCE_SHA": SHA, "METADATA_DIR": directory}
+            stamp = pathlib.Path(directory) / "local.json"
+            with mock.patch.dict(os.environ, env), mock.patch.object(release, "git", return_value=SHA), \
+                    mock.patch.object(release, "Client", return_value=client), \
+                    mock.patch.object(release, "local_identity", return_value=local), \
+                    mock.patch("sys.argv", ["helper", "readback"]):
+                with self.assertRaisesRegex(release.ReleaseError, "identity unavailable"):
+                    release.main()
+                previous = release.check_local(local, "1.2.3", SHA)
+                previous["image_id"] = "sha256:" + "e" * 64
+                stamp.write_text(json.dumps(previous))
+                with self.assertRaisesRegex(release.ReleaseError, "pre-smoke identity"):
+                    release.main()
+                self.assertNotIn(release.registry_path("manifests/1.2.3"), client.calls)
+                stamp.write_text(json.dumps(release.check_local(local, "1.2.3", SHA)))
+                with mock.patch.object(release, "outputs"):
+                    release.main()
+                self.assertEqual(json.loads((pathlib.Path(directory) / "registry.json").read_text())["config_digest"], local["config_digest"])
+
+    def test_reduced_registry_read_access_fails_before_collision_checks(self):
+        client = RegistrySafety().preflight_client()
+        client.responses[release.registry_path("tags/list")] = response(404, {"errors": [{"code": "NAME_UNKNOWN"}]})
+        with self.assertRaisesRegex(release.ReleaseError, "read access"):
+            release.preflight(client, "1.2.3", SHA)
+        self.assertNotIn(release.registry_path("manifests/1.2.3"), client.calls)
+
+
+class SpecRegressions(unittest.TestCase):
+    def test_masked_package_404_denies_even_empty_or_missing_registry(self):
+        safety = RegistrySafety()
+        for registry in (response(200, {"name": release.REGISTRY_NAME, "tags": []}),
+                         response(404, {"errors": [{"code": "NAME_UNKNOWN"}]})):
+            client = safety.preflight_client(404)
+            client.responses[release.registry_path("tags/list")] = registry
+            with self.subTest(registry=registry), self.assertRaises(release.ReleaseError):
+                release.preflight(client, "1.2.3", SHA)
+            self.assertNotIn(release.registry_path("tags/list"), client.calls)
+
+    def test_opaque_reduced_scope_token_cannot_authorize_masked_metadata(self):
+        with mock.patch.dict("os.environ", {"GITHUB_TOKEN": "test-only", "GITHUB_REPOSITORY": release.REPO}):
+            for scope in (None, "", "repository:" + release.REGISTRY_NAME + ":pull"):
+                token = {"token": "opaque-test-only"}
+                if scope is not None:
+                    token["scope"] = scope
+                responses = [response(200, token),
+                             response(200, {"full_name": release.REPO, "default_branch": "master"}),
+                             response(404, {"message": "Not Found", "documentation_url": "https://docs.github.com/rest/packages"}),
+                             response(200, {"name": release.REGISTRY_NAME, "tags": []}),
+                             response(404, {"errors": [{"code": "MANIFEST_UNKNOWN"}]}),
+                             response(404, {"errors": [{"code": "MANIFEST_UNKNOWN"}]})]
+                with self.subTest(scope=scope), mock.patch.object(release.Client, "request", side_effect=responses):
+                    with self.assertRaises(release.ReleaseError):
+                        release.preflight(release.Client(), "1.2.3", SHA)
+
+    def test_store_image_id_is_not_config_digest(self):
+        client, local = ReadbackSafety().fixture()
+        local["config_digest"] = local["Id"]
+        local["Id"] = "sha256:" + "e" * 64  # containerd identifies the manifest, not config
+        try:
+            result = release.readback(client, "1.2.3", SHA, local)
+        except release.ReleaseError as error:
+            self.fail("same config must verify independently of Docker Id: " + str(error))
+        self.assertEqual(result["config_digest"], local["config_digest"])
+        self.assertEqual(result["image_id"], local["Id"])
+
+    def test_remote_rootfs_must_match_smoked_image(self):
+        client, local = ReadbackSafety().fixture()
+        local["RootFS"] = {"Type": "layers", "Layers": ["sha256:" + "c" * 64]}
+        with self.assertRaises(release.ReleaseError):
+            release.readback(client, "1.2.3", SHA, local)
+
+    def test_workflow_builds_without_attestations(self):
+        workflow = pathlib.Path(__file__).resolve().parents[1] / ".github/workflows/server-image.yml"
+        builds = workflow.read_text().split("docker build ")[1:]
+        self.assertEqual(len(builds), 2)
+        for build in builds:
+            invocation = build.split('tee "$RESULTS/build.log"')[0]
+            self.assertIn("--provenance=false", invocation)
+            self.assertIn("--sbom=false", invocation)
+
+    def test_retargeted_master_ancestor_tag_cannot_replace_release_event(self):
+        with tempfile.TemporaryDirectory() as directory:
+            def run(*args):
+                result = subprocess.run(["git", *args], cwd=directory, capture_output=True, text=True)
+                if result.returncode:
+                    raise release.ReleaseError("git fixture failed")
+                return result.stdout.strip()
+            run("init", "-b", "master")
+            earlier = ""
+            for message in ("earlier approved release", "later master ancestor"):
+                run("-c", "user.name=Test", "-c", "user.email=test@example.invalid", "commit", "--allow-empty", "-m", message)
+                if message.startswith("earlier"):
+                    earlier = run("rev-parse", "HEAD")
+                    run("tag", "v1.2.3")
+            later = run("rev-parse", "HEAD")
+            run("update-ref", "refs/remotes/origin/master", later)
+            run("tag", "-f", "v1.2.3", later)
+            with mock.patch.object(release, "git", side_effect=run), mock.patch.dict("os.environ", {"GITHUB_SHA": earlier}):
+                with self.assertRaisesRegex(release.ReleaseError, "event commit"):
+                    release.resolve_source(release.REPO, "release", "refs/tags/v1.2.3", "v1.2.3")
+                self.assertEqual(release.resolve_source(release.REPO, "workflow_dispatch", "refs/heads/master", "v1.2.3"), later)
+
+
+class ImmutableTagCollisions(unittest.TestCase):
+    INDEX_TYPES = ("application/vnd.oci.image.index.v1+json",
+                   "application/vnd.docker.distribution.manifest.list.v2+json")
+
+    def client(self):
+        # Exercise real Client.get, with only its transport stubbed. No login/network.
+        client = release.Client.__new__(release.Client)
+        client.github_token = "github-test-only"
+        client.registry_token = "registry-test-only"
+        return client
+
+    def test_manifest_requests_accept_all_four_image_representations(self):
+        expected = {"application/vnd.oci.image.manifest.v1+json",
+                    "application/vnd.docker.distribution.manifest.v2+json",
+                    "application/vnd.oci.image.index.v1+json",
+                    "application/vnd.docker.distribution.manifest.list.v2+json"}
+        for tag in ("1.2.3", "sha-" + SHA):
+            client = self.client()
+            path = release.registry_path("manifests/" + tag)
+            with self.subTest(tag=tag), mock.patch.object(client, "request", return_value=response(200, {})) as request:
+                self.assertEqual(client.get(path)[0], 200)
+                request.assert_called_once()
+                actual_path, headers = request.call_args.args
+                self.assertEqual(actual_path, path)
+                self.assertEqual(headers["Authorization"], "Bearer registry-test-only")
+                self.assertEqual({value.strip() for value in headers["Accept"].split(",")}, expected)
+
+    def listed_collision(self, tag):
+        client = RegistrySafety().preflight_client(tags=["old", tag])
+        client.responses[release.registry_path("manifests/" + tag)] = response(404, {"errors": [{
+            "code": "MANIFEST_UNKNOWN", "message": "OCI index found, but Accept header does not support OCI indexes"}]})
+        with self.assertRaisesRegex(release.ReleaseError, "existing immutable tag"):
+            release.preflight(client, "1.2.3", SHA)
+        self.assertEqual(client.calls, [release.REPO_API, release.PACKAGE_API, release.registry_path("tags/list")])
+
+    def test_listed_version_rejected_before_misleading_manifest_404(self):
+        self.listed_collision("1.2.3")
+
+    def test_listed_full_sha_rejected_before_misleading_manifest_404(self):
+        self.listed_collision("sha-" + SHA)
+
+    def unlisted_collision(self, media_type):
+        for tag in ("1.2.3", "sha-" + SHA):
+            # Model a tag created after tags/list: only manifest negotiation can see it.
+            responses = RegistrySafety().preflight_client(tags=["old"]).responses
+            client = self.client()
+            path = release.registry_path("manifests/" + tag)
+            negotiated = []
+
+            def request(actual_path, headers):
+                if actual_path == path:
+                    accepted = {value.strip() for value in headers["Accept"].split(",")}
+                    if media_type in accepted:
+                        negotiated.append(200)
+                        return response(200, {"schemaVersion": 2, "mediaType": media_type, "manifests": []},
+                                        {"content-type": media_type})
+                    negotiated.append(404)
+                    return response(404, {"errors": [{"code": "MANIFEST_UNKNOWN",
+                                                      "message": "Accept header does not support stored image type"}]})
+                return responses[actual_path]
+
+            with self.subTest(tag=tag), mock.patch.object(client, "request", side_effect=request) as transport:
+                with self.assertRaisesRegex(release.ReleaseError, "existing immutable tag"):
+                    release.preflight(client, "1.2.3", SHA)
+                self.assertEqual(negotiated, [200])
+                self.assertIn(mock.call(path, mock.ANY), transport.call_args_list)
+
+    def test_unlisted_oci_index_is_a_collision_for_either_tag(self):
+        self.unlisted_collision(self.INDEX_TYPES[0])
+
+    def test_unlisted_docker_manifest_list_is_a_collision_for_either_tag(self):
+        self.unlisted_collision(self.INDEX_TYPES[1])
+
+    def test_published_readback_still_rejects_both_index_types(self):
+        for media_type in self.INDEX_TYPES:
+            for tag in ("1.2.3", "sha-" + SHA):
+                fixture, local = ReadbackSafety().fixture()
+                body = encoded({"schemaVersion": 2, "mediaType": media_type, "manifests": []})
+                path = release.registry_path("manifests/" + tag)
+                fixture.responses[path] = (200, {"docker-content-digest": digest(body)}, body)
+                client = self.client()
+                with self.subTest(media_type=media_type, tag=tag), \
+                        mock.patch.object(client, "request", side_effect=lambda path, headers: fixture.responses[path]):
+                    with self.assertRaisesRegex(release.ReleaseError, "single-platform required"):
+                        release.readback(client, "1.2.3", SHA, local)
+
+
+class WorkflowBehavior(unittest.TestCase):
+    def test_build_log_pipeline_preserves_docker_failure(self):
+        workflow = pathlib.Path(__file__).resolve().parents[1] / ".github/workflows/server-image.yml"
+        lines = workflow.read_text().splitlines()
+        scripts = []
+        for start, line in enumerate(lines):
+            if line == "          set -euo pipefail":
+                end = start
+                while end < len(lines) and (not lines[end] or lines[end].startswith("          ")):
+                    end += 1
+                scripts.append("\n".join(value[10:] for value in lines[start:end]))
+        # Find build blocks by their docker command as well, so removing the
+        # pipefail line cannot cause the actual behavioral probes to disappear.
+        if not scripts:
+            for start, line in enumerate(lines):
+                if line == '          mkdir -p "$RESULTS"':
+                    end = start
+                    while end < len(lines) and (not lines[end] or lines[end].startswith("          ")):
+                        end += 1
+                    scripts.append("\n".join(value[10:] for value in lines[start:end]))
+        self.assertEqual(len(scripts), 2)
+        for script in scripts:
+            with tempfile.TemporaryDirectory() as directory:
+                env = {"PATH": "/usr/bin", "RESULTS": directory, "VERSION": "1.2.3", "SOURCE_SHA": SHA,
+                       "SDK_IMAGE": "sdk", "RUNTIME_IMAGE": "runtime", "LOCAL_IMAGE": "local"}
+                prefix = "docker() { return 17; }; python3() { printf 'UNSAFE_CONTINUATION\\n'; };\n"
+                result = subprocess.run(["bash", "-e", "-c", prefix + script],
+                                        env=env, capture_output=True, text=True)
+            self.assertEqual(result.returncode, 17, result.stdout + result.stderr)
+            self.assertNotIn("UNSAFE_CONTINUATION", result.stdout)
+
+
+class RealSourceResolution(unittest.TestCase):
+    def test_existing_tag_peels_to_commit_and_rejects_non_master_source(self):
+        with tempfile.TemporaryDirectory() as directory:
+            def run(*args):
+                result = subprocess.run(["git", *args], cwd=directory, capture_output=True, text=True)
+                if result.returncode:
+                    raise release.ReleaseError("git probe rejected")
+                return result.stdout.strip()
+            run("init", "-b", "master")
+            run("-c", "user.name=Test", "-c", "user.email=test@example.invalid", "commit", "--allow-empty", "-m", "base")
+            sha = run("rev-parse", "HEAD")
+            run("update-ref", "refs/remotes/origin/master", sha)
+            run("-c", "user.name=Test", "-c", "user.email=test@example.invalid", "tag", "-a", "v1.2.3", "-m", "approved")
+            with mock.patch.object(release, "git", side_effect=run):
+                self.assertEqual(release.resolve_source(release.REPO, "release", "refs/tags/v1.2.3", "v1.2.3", sha), sha)
+                with self.assertRaises(release.ReleaseError):
+                    release.resolve_source(release.REPO, "release", "refs/tags/v1.2.4", "v1.2.4", sha)
+                run("-c", "user.name=Test", "-c", "user.email=test@example.invalid", "commit", "--allow-empty", "-m", "unmerged")
+                run("tag", "v1.2.4")
+                unmerged_sha = run("rev-parse", "HEAD")
+                with self.assertRaises(release.ReleaseError):
+                    release.resolve_source(release.REPO, "release", "refs/tags/v1.2.4", "v1.2.4", unmerged_sha)
+
+
+if __name__ == "__main__":
+    unittest.main()