chore: use digests instead of tags for pinned images

.github/workflows/test.yml

@@ -237,6 +237,7 @@ jobs:
 
     - name: Test cibuildwheel
       env:
+        CIBW_PLATFORM: ${{ matrix.test_select }}
         CIBW_TEST_RUNTIME: ${{ matrix.test_runtime }}
       run: uv run --no-sync bin/run_tests.py --test-select=${{ matrix.test_select || 'native' }} ${{ (runner.os == 'Linux' && runner.arch == 'X64') && '--run-podman' || '' }}
 

bin/update_docker.py

@@ -7,8 +7,9 @@
 # ]
 # ///
 
-import configparser
 import dataclasses
+from collections import defaultdict
+from functools import cache
 from pathlib import Path
 
 import requests
@@ -33,7 +34,7 @@ def __init__(self, manylinux_version: str, platforms: list[str], tag: str | None
         super().__init__(manylinux_version, platforms, image_name, tag, True)
 
 
-images = [
+IMAGES = [
     # manylinux2014 images
     PyPAImage(
         "manylinux2014",
@@ -87,50 +88,75 @@ def __init__(self, manylinux_version: str, platforms: list[str], tag: str | None
         "musllinux_1_2", ["x86_64", "i686", "aarch64", "ppc64le", "s390x", "armv7l", "riscv64"]
     ),
 ]
+# IMAGES = [
+#   PyPAImage("manylinux_2_35", ["armv7l"]),
+# ]
 
-config = configparser.ConfigParser()
 
-for image in images:
-    # get the tag name whose digest matches 'latest'
-    if image.tag is not None:
-        # image has been pinned, do not update
-        tag_name = image.tag
-    elif image.image_name.startswith("quay.io/"):
-        _, _, repository_name = image.image_name.partition("/")
-        response = requests.get(
-            f"https://quay.io/api/v1/repository/{repository_name}?includeTags=true"
-        )
-        response.raise_for_status()
-        repo_info = response.json()
-        tags_dict = repo_info["tags"]
-
-        latest_tag = tags_dict.pop("latest")
+@cache
+def get_from_quay(image_name: str, tag_name: str) -> tuple[str, str]:
+    _, _, repository_name = image_name.partition("/")
+    if tag_name == "latest":
+        url = f"https://quay.io/api/v1/repository/{repository_name}?includeTags=true"
+    else:
+        url = f"https://quay.io/api/v1/repository/{repository_name}/tag?specificTag=2026.06.12-2"
+    response = requests.get(url)
+    response.raise_for_status()
+    info = response.json()
+    if tag_name == "latest":
+        tags_dict = info["tags"]
+        tag_info = tags_dict.pop(tag_name)
         # find the tag whose manifest matches 'latest'
-        tag_name = next(
-            name
+        tag_name, digest = next(
+            (name, info["manifest_digest"])
             for (name, info) in tags_dict.items()
-            if info["manifest_digest"] == latest_tag["manifest_digest"]
+            if info["manifest_digest"] == tag_info["manifest_digest"]
         )
-    elif image.image_name.startswith("ghcr.io/"):
-        repository = image.image_name[8:]
-        response = requests.get(
-            "https://ghcr.io/token", params={"scope": f"repository:{repository}:pull"}
-        )
-        response.raise_for_status()
-        token = response.json()["token"]
+    else:
+        tags_list = info["tags"]
+        tag_info = next(tag for tag in tags_list if tag["name"] == tag_name)
+        digest = tag_info["manifest_digest"]
+
+    return tag_name, digest
+
+
+@cache
+def get_from_ghcr(image_name: str, tag_name: str) -> tuple[str, str]:
+    repository = image_name[8:]
+    response = requests.get(
+        "https://ghcr.io/token", params={"scope": f"repository:{repository}:pull"}
+    )
+    response.raise_for_status()
+    token = response.json()["token"]
+    if tag_name == "latest":
         response = requests.get(
             f"https://ghcr.io/v2/{repository}/tags/list",
             headers={"Authorization": f"Bearer {token}"},
         )
         response.raise_for_status()
-        ghcr_tags = [(Version(tag), tag) for tag in response.json()["tags"] if tag != "latest"]
+        info = response.json()
+        ghcr_tags = [(Version(tag), tag) for tag in info["tags"] if tag != "latest"]
         ghcr_tags.sort(reverse=True)
         tag_name = ghcr_tags[0][1]
-    else:
-        response = requests.get(f"https://hub.docker.com/v2/repositories/{image.image_name}/tags")
-        response.raise_for_status()
-        tags = response.json()["results"]
 
+    response = requests.head(
+        f"https://ghcr.io/v2/{repository}/manifests/{tag_name}",
+        headers={
+            "Authorization": f"Bearer {token}",
+            "Accept": "application/vnd.docker.distribution.manifest.v2+json, application/vnd.docker.distribution.manifest.list.v2+json, application/vnd.oci.image.manifest.v1+json, application/vnd.oci.image.index.v1+json, application/vnd.oci.artifact.manifest.v1+json",
+        },
+    )
+    response.raise_for_status()
+    digest = response.headers["Docker-Content-Digest"]
+    return tag_name, digest
+
+
+@cache
+def get_from_dockerhub(image_name: str, tag_name: str) -> tuple[str, str]:
+    response = requests.get(f"https://hub.docker.com/v2/repositories/{image_name}/tags")
+    response.raise_for_status()
+    tags = response.json()["results"]
+    if tag_name == "latest":
         latest_tag = next(tag for tag in tags if tag["name"] == "latest")
         # i don't know what it would mean to have multiple images per tag
         assert len(latest_tag["images"]) == 1
@@ -139,15 +165,50 @@ def __init__(self, manylinux_version: str, platforms: list[str], tag: str | None
         pinned_tag = next(
             tag for tag in tags if tag != latest_tag and tag["images"][0]["digest"] == digest
         )
-        tag_name = pinned_tag["name"]
-
-    for platform in image.platforms:
-        if not config.has_section(platform):
-            config[platform] = {}
-        suffix = ""
-        if image.use_platform_suffix:
-            suffix = f"_{platform.removeprefix('pypy_')}"
-        config[platform][image.manylinux_version] = f"{image.image_name}{suffix}:{tag_name}"
-
-with open(RESOURCES / "pinned_docker_images.cfg", "w") as f:
-    config.write(f)
+    else:
+        pinned_tag = next(tag for tag in tags if tag["name"] == tag_name)
+        digest = pinned_tag["images"][0]["digest"]
+    tag_name = pinned_tag["name"]
+    return tag_name, digest
+
+
+def main() -> None:
+    config: dict[str, dict[str, tuple[str, str, str]]] = defaultdict(dict)
+    for image in IMAGES:
+        # get the tag name whose digest matches 'latest'
+        # if image has been pinned, do not update
+        search_tag = image.tag or "latest"
+        if image.image_name.startswith("quay.io/"):
+            get_function = get_from_quay
+        elif image.image_name.startswith("ghcr.io/"):
+            get_function = get_from_ghcr
+        else:
+            get_function = get_from_dockerhub
+
+        tag_name, digest = get_function(image.image_name, search_tag)
+        for platform in image.platforms:
+            image_name = image.image_name
+            if image.use_platform_suffix:
+                image_name = f"{image_name}_{platform.removeprefix('pypy_')}"
+                _, digest = get_function(image_name, tag_name)
+            assert digest.startswith("sha256:")
+            # config[platform][image.manylinux_version] = f"{image_name}@{digest}  # {tag_name}"
+            config[platform][image.manylinux_version] = (image_name, digest, tag_name)
+
+    with (RESOURCES / "pinned_docker_images.toml").open("w") as f:
+        first = True
+        for platform, versions in config.items():
+            if first:
+                first = False
+            else:
+                f.write("\n")
+            f.write(f"[{platform}]\n")
+            for manylinux_version in versions:
+                image_name, digest, tag_name = versions[manylinux_version]
+                f.write(
+                    f'{manylinux_version} = {{ name = "{image_name}", tag = "{tag_name}", digest = "{digest}" }}\n'
+                )
+
+
+if __name__ == "__main__":
+    main()

cibuildwheel/oci_container.py

@@ -48,6 +48,7 @@
     from types import TracebackType
     from typing import IO, Self
 
+    from cibuildwheel.options import OCIImage
     from cibuildwheel.typing import PathOrStr
 
 ContainerEngineName = Literal["docker", "podman"]
@@ -231,12 +232,12 @@ class OCIContainer:
     def __init__(
         self,
         *,
-        image: str,
+        image: OCIImage,
         oci_platform: OCIPlatform,
         cwd: PathOrStr | None = None,
         engine: OCIContainerEngineConfig = DEFAULT_ENGINE,
     ):
-        if not image:
+        if not image.reference:
             msg = "Must have a non-empty image to run."
             raise ValueError(msg)
 
@@ -262,7 +263,7 @@ def _get_platform_args(self, *, oci_platform: OCIPlatform | None = None) -> tupl
                 self.engine.name,
                 "image",
                 "inspect",
-                self.image,
+                self.image.reference,
                 "--format",
                 (
                     "{{.Os}}/{{.Architecture}}/{{.Variant}}"
@@ -305,15 +306,19 @@ def __enter__(self) -> Self:
             ctr_cmd = ["uname", "-m"]
             try:
                 container_machine = call(
-                    *run_cmd, *platform_args, self.image, *ctr_cmd, capture_stdout=True
+                    *run_cmd, *platform_args, self.image.reference, *ctr_cmd, capture_stdout=True
                 ).strip()
             except subprocess.CalledProcessError:
                 if self.oci_platform == OCIPlatform.i386:
                     # The image might have been built with amd64 architecture
                     # Let's try that
                     platform_args = self._get_platform_args(oci_platform=OCIPlatform.AMD64)
                     container_machine = call(
-                        *run_cmd, *platform_args, self.image, *ctr_cmd, capture_stdout=True
+                        *run_cmd,
+                        *platform_args,
+                        self.image.reference,
+                        *ctr_cmd,
+                        capture_stdout=True,
                     ).strip()
                 else:
                     raise
@@ -323,7 +328,7 @@ def __enter__(self) -> Self:
                 call(
                     *run_cmd,
                     *platform_args,
-                    self.image,
+                    self.image.reference,
                     "linux32",
                     "/bin/true",
                     capture_stdout=True,
@@ -343,7 +348,7 @@ def __enter__(self) -> Self:
                 *network_args,
                 *platform_args,
                 *self.engine.create_args,
-                self.image,
+                self.image.reference,
                 *shell_args,
             ],
             check=True,

cibuildwheel/options.py

@@ -13,7 +13,6 @@
     "cibuildwheel.util.helpers",
     "cibuildwheel.util.packaging",
     "collections",
-    "configparser",
     "contextlib",
     "difflib",
     "packaging",
@@ -25,7 +24,6 @@
 }
 
 import collections
-import configparser
 import contextlib
 import dataclasses
 import difflib
@@ -35,6 +33,7 @@
 import textwrap
 import tomllib
 from collections.abc import Mapping, Sequence
+from functools import cached_property
 from pathlib import Path
 from typing import assert_never
 
@@ -82,6 +81,27 @@
 )
 
 
+@dataclasses.dataclass(frozen=True, kw_only=True)
+class OCIImage:
+    name: str
+    tag: str | None = None
+    digest: str | None = None
+
+    @cached_property
+    def reference(self) -> str:
+        if self.digest is not None:
+            return f"{self.name}@{self.digest}"
+        if self.tag is not None:
+            return f"{self.name}:{self.tag}"
+        return self.name
+
+    def __str__(self) -> str:
+        result = self.reference
+        if self.digest is not None and self.tag is not None:
+            result += f" (tag: {self.tag})"
+        return result
+
+
 @dataclasses.dataclass(kw_only=True)
 class CommandLineArguments:
     platform: Literal["auto", "linux", "macos", "windows"] | None
@@ -146,8 +166,8 @@ class BuildOptions:
     xbuild_tools: list[str] | None
     xbuild_files: dict[str, list[str]]
     repair_command: str
-    manylinux_images: dict[str, str] | None
-    musllinux_images: dict[str, str] | None
+    manylinux_images: dict[str, OCIImage] | None
+    musllinux_images: dict[str, OCIImage] | None
     dependency_constraints: DependencyConstraints
     test_command: str | None
     before_test: str | None
@@ -738,7 +758,7 @@ def globals(self) -> GlobalOptions:
             allow_empty=allow_empty,
         )
 
-    def _check_pinned_image(self, value: str, pinned_images: Mapping[str, str]) -> None:
+    def _check_pinned_image(self, value: str, pinned_images: Mapping[str, OCIImage]) -> None:
         error_set = {"manylinux1", "manylinux2010", "manylinux_2_24", "musllinux_1_1"}
         # Currently no warnings, next: https://github.com/pypa/manylinux/issues/1925
         warning_set: set[str] = set()
@@ -893,8 +913,8 @@ def _compute_build_options(self, identifier: str | None) -> BuildOptions:
             except ValueError:
                 build_verbosity = 0
 
-            manylinux_images: dict[str, str] = {}
-            musllinux_images: dict[str, str] = {}
+            manylinux_images: dict[str, OCIImage] = {}
+            musllinux_images: dict[str, OCIImage] = {}
             container_engine: OCIContainerEngineConfig | None = None
 
             if self.platform == "linux":
@@ -906,15 +926,19 @@ def _compute_build_options(self, identifier: str | None) -> BuildOptions:
                         f"manylinux-{build_platform}-image", ignore_empty=True
                     )
                     self._check_pinned_image(config_value, pinned_images)
-                    manylinux_images[build_platform] = pinned_images.get(config_value, config_value)
+                    manylinux_images[build_platform] = pinned_images.get(
+                        config_value, OCIImage(name=config_value)
+                    )
 
                 for build_platform in MUSLLINUX_ARCHS:
                     pinned_images = all_pinned_container_images[build_platform]
                     config_value = self.reader.get(
                         f"musllinux-{build_platform}-image", ignore_empty=True
                     )
                     self._check_pinned_image(config_value, pinned_images)
-                    musllinux_images[build_platform] = pinned_images.get(config_value, config_value)
+                    musllinux_images[build_platform] = pinned_images.get(
+                        config_value, OCIImage(name=config_value)
+                    )
 
             container_engine_str = self.reader.get(
                 "container-engine",
@@ -1116,14 +1140,18 @@ def compute_options(
 
 
 @functools.cache
-def _get_pinned_container_images() -> Mapping[str, Mapping[str, str]]:
+def _get_pinned_container_images() -> Mapping[str, Mapping[str, OCIImage]]:
     """
     This looks like a dict of dicts, e.g.
     { 'x86_64': {'manylinux1': '...', 'manylinux2010': '...', 'manylinux2014': '...'},
       'i686': {'manylinux1': '...', 'manylinux2010': '...', 'manylinux2014': '...'},
       'pypy_x86_64': {'manylinux2010': '...' }
       ... }
     """
-    all_pinned_images = configparser.ConfigParser()
-    all_pinned_images.read(resources.PINNED_DOCKER_IMAGES)
+    pinned_images = tomllib.loads(resources.PINNED_DOCKER_IMAGES.read_text())
+    all_pinned_images: dict[str, dict[str, OCIImage]] = {}
+    for platform, images in pinned_images.items():
+        all_pinned_images[platform] = {}
+        for version, image in images.items():
+            all_pinned_images[platform][version] = OCIImage(**image)
     return all_pinned_images