Only send the ECR pull secret for images that need it

libs/python/cua-sandbox/cua_sandbox/transport/fleet_cloud.py

@@ -325,6 +325,23 @@ def service_url(self, sandbox: Any, service: str) -> str:
         return f"{self._base_url}/api/svc/{sandbox.namespace}/{sandbox.name}-{service}/"
 
 
+_ECR_HOST_SUFFIX = ".amazonaws.com"
+_ECR_HOST_MARKER = ".dkr.ecr."
+
+
+def _needs_ecr_pull_secret(image: "str | None") -> bool:
+    """True only for the account's private ECR, which the secret authenticates.
+
+    Public registries (public.ecr.aws, ghcr.io, quay.io, Docker Hub) are pulled
+    anonymously; attaching a credential for them makes the gateway enforce its
+    private-registry allowlist against an image that never needed one.
+    """
+    if not image:
+        return False
+    host = image.split("/", 1)[0]
+    return _ECR_HOST_MARKER in host and host.endswith(_ECR_HOST_SUFFIX)
+
+
 class FleetCloudTransport(FleetTransport):
     """Provision image-backed pools or claim pre-created pools through Fleet."""
 
@@ -564,17 +581,24 @@ def _template_request(self) -> CreateTemplateRequest:
             .build()
             for name, port in service_ports.items()
         ]
+        container_disk_image = cloud_registry_image(self._image)
         vm_template_builder = (
             VmTemplateBuilder()
-            .container_disk_image(cloud_registry_image(self._image))
-            .image_pull_secret("ecr-credentials")
+            .container_disk_image(container_disk_image)
             .probes(
                 PreservedJson.from_json(
                     json.dumps({"readinessProbe": {"tcpSocket": {"port": self._server_port}}})
                 )
             )
             .services(services)
         )
+        # The pull secret authenticates the account's private ECR and nothing else.
+        # Attaching it to a public image is not merely redundant: the gateway's
+        # admission policy reads its presence as "enforce the ECR allowlist", so a
+        # public image the cluster can pull anonymously was refused outright.
+        if _needs_ecr_pull_secret(container_disk_image):
+            vm_template_builder = vm_template_builder.image_pull_secret("ecr-credentials")
+
         # Windows guest disks are built UEFI-only (see registry/qemu_builder.py), and the
         # Fleet schema defaults firmware to BIOS, so a Windows image left at the default
         # boots SeaBIOS against a GPT/ESP disk and never reaches the readiness probe.

libs/python/cua-sandbox/tests/test_pull_secret.py

@@ -0,0 +1,59 @@
+"""The ECR pull secret must only ride on images that need it.
+
+Attaching `imagePullSecret: ecr-credentials` to a public image is not merely
+redundant — the gateway's admission policy reads its presence as "enforce the
+private-registry allowlist", so a public image the cluster can pull anonymously
+was refused with `403 k8s request is not allowed`. Proven with identical request
+bodies differing only in the image: ghcr.io -> 403, public.ecr.aws -> 201.
+"""
+
+import pytest
+from cua_sandbox import Image
+from cua_sandbox.transport.fleet_cloud import (
+    FleetCloudTransport,
+    _needs_ecr_pull_secret,
+)
+
+PRIVATE = "296062593712.dkr.ecr.us-west-2.amazonaws.com/cua-server-windows:main-bac7daa3"
+PUBLIC_ECR = "public.ecr.aws/k5j5w0x5/cua-windows-2022:main-bac7daa3"
+
+
+@pytest.mark.parametrize(
+    "ref,expected",
+    [
+        (PRIVATE, True),
+        ("123456789012.dkr.ecr.eu-central-1.amazonaws.com/x:1", True),
+        (PUBLIC_ECR, False),
+        ("ghcr.io/trycua/minecraft-workspace:latest", False),
+        ("quay.io/org/image:tag", False),
+        ("ubuntu:22.04", False),
+        (None, False),
+        ("", False),
+    ],
+)
+def test_only_private_ecr_needs_the_secret(ref, expected):
+    assert _needs_ecr_pull_secret(ref) is expected
+
+
+def test_public_image_template_carries_no_pull_secret():
+    """A public image must not be forced into the allowlist branch."""
+    request = FleetCloudTransport(image=Image.windows(), name="demo")._template_request()
+    template = request.spec.vm_template
+    assert template.container_disk_image == PUBLIC_ECR
+    assert not getattr(template, "image_pull_secret", None)
+
+
+def test_private_registry_image_still_carries_the_secret():
+    request = FleetCloudTransport(
+        image=Image.from_registry(PRIVATE), name="demo"
+    )._template_request()
+    assert request.spec.vm_template.image_pull_secret == "ecr-credentials"
+
+
+def test_a_public_non_ecr_registry_is_usable():
+    """The case the Minecraft guide needs: a containerDisk pushed to ghcr.io."""
+    ref = "ghcr.io/trycua/minecraft-workspace:latest"
+    request = FleetCloudTransport(image=Image.from_registry(ref), name="demo")._template_request()
+    template = request.spec.vm_template
+    assert template.container_disk_image == ref
+    assert not getattr(template, "image_pull_secret", None)