fix(oci): fetch artifact layers from blobs endpoint, not via Resolve

Pulling an oci:// resource resolved each layer digest through the
registry manifests endpoint, which answers 500 when the digest points
to a non-manifest blob. containerd v2.3.0+ (pulled in by buildx v0.36
and buildkit v0.32) no longer falls back to the blobs endpoint unless
manifests returned 404, so publish/pull of compose artifacts broke.
Fetch layers directly with the descriptors already listed in the
manifest instead of resolving them again.

Signed-off-by: Guillaume Lours <[email protected]>

internal/oci/resolver.go

@@ -18,6 +18,7 @@ package oci
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"net/http"
 	"net/url"
@@ -95,6 +96,40 @@ func Get(ctx context.Context, resolver remotes.Resolver, ref reference.Named) (s
 	return descriptor, content, nil
 }
 
+// GetBlob retrieves the content of a blob descriptor (e.g. an artifact layer)
+// from the repository ref belongs to. Unlike Get it doesn't Resolve the
+// digest, as the registry manifests endpoint only serves actual manifests;
+// blob content must be fetched directly from the blobs endpoint.
+func GetBlob(ctx context.Context, resolver remotes.Resolver, ref reference.Named, descriptor spec.Descriptor) ([]byte, error) {
+	fetcher, err := resolver.Fetcher(ctx, ref.String())
+	if err != nil {
+		return nil, fmt.Errorf("creating fetcher for %s: %w", ref, err)
+	}
+	fetch, err := fetcher.Fetch(ctx, descriptor)
+	if err != nil {
+		return nil, fmt.Errorf("fetching blob %s: %w", descriptor.Digest, err)
+	}
+	defer func() { _ = fetch.Close() }()
+	// bound the read by the declared size so a rogue registry can't cause
+	// unbounded allocation; the extra byte detects oversized responses.
+	content, err := io.ReadAll(io.LimitReader(fetch, descriptor.Size+1))
+	if err != nil {
+		return nil, fmt.Errorf("reading blob %s: %w", descriptor.Digest, err)
+	}
+	if int64(len(content)) != descriptor.Size {
+		return nil, fmt.Errorf("blob %s size mismatch: expected %d bytes, got %d", descriptor.Digest, descriptor.Size, len(content))
+	}
+	// GetBlob bypasses containerd's content store, so integrity must be
+	// checked here before callers write the bytes to disk.
+	if err := descriptor.Digest.Validate(); err != nil {
+		return nil, fmt.Errorf("invalid digest %s: %w", descriptor.Digest, err)
+	}
+	if actual := descriptor.Digest.Algorithm().FromBytes(content); actual != descriptor.Digest {
+		return nil, fmt.Errorf("blob digest mismatch: expected %s, got %s", descriptor.Digest, actual)
+	}
+	return content, nil
+}
+
 func Copy(ctx context.Context, resolver remotes.Resolver, image reference.Named, named reference.Named) (spec.Descriptor, error) {
 	src, desc, err := resolver.Resolve(ctx, image.String())
 	if err != nil {

internal/oci/resolver_test.go

@@ -23,7 +23,10 @@ import (
 	"sync/atomic"
 	"testing"
 
+	"github.com/distribution/reference"
 	"github.com/docker/cli/cli/config/configfile"
+	"github.com/opencontainers/go-digest"
+	spec "github.com/opencontainers/image-spec/specs-go/v1"
 	"gotest.tools/v3/assert"
 )
 
@@ -106,3 +109,107 @@ func TestNewResolver_NilTransportIsValid(t *testing.T) {
 	resolver := NewResolver(&configfile.ConfigFile{}, nil)
 	assert.Assert(t, resolver != nil, "NewResolver must return a non-nil resolver when transport is nil")
 }
+
+// TestGetBlob_FetchesFromBlobsEndpoint guards that layer content is fetched
+// straight from the blobs endpoint. Registries answer HEAD/GET on
+// manifests/<digest> with a 500 when the digest points to a non-manifest blob
+// (they try to JSON-parse it), and containerd >= 2.3 no longer falls back to
+// blobs unless the manifests endpoint returned 404 — so resolving a layer
+// digest through Resolve() breaks the pull.
+func TestGetBlob_FetchesFromBlobsEndpoint(t *testing.T) {
+	content := []byte("services:\n  test:\n    image: alpine\n")
+	dgst := digest.FromBytes(content)
+
+	var manifestHits atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case r.URL.Path == "/v2/":
+			w.WriteHeader(http.StatusOK)
+		case r.URL.Path == "/v2/test/artifact/manifests/"+dgst.String():
+			// mimic distribution: the blob exists but is not a manifest
+			manifestHits.Add(1)
+			w.WriteHeader(http.StatusInternalServerError)
+		case r.URL.Path == "/v2/test/artifact/blobs/"+dgst.String():
+			w.Header().Set("Content-Type", "application/octet-stream")
+			_, _ = w.Write(content)
+		default:
+			w.WriteHeader(http.StatusNotFound)
+		}
+	}))
+	t.Cleanup(server.Close)
+
+	host := server.Listener.Addr().String()
+	resolver := NewResolver(&configfile.ConfigFile{}, &http.Transport{}, host)
+
+	ref, err := reference.ParseDockerRef(host + "/test/artifact:latest")
+	assert.NilError(t, err)
+
+	got, err := GetBlob(t.Context(), resolver, ref, spec.Descriptor{
+		MediaType: ComposeYAMLMediaType,
+		Digest:    dgst,
+		Size:      int64(len(content)),
+	})
+	assert.NilError(t, err)
+	assert.DeepEqual(t, got, content)
+	assert.Equal(t, manifestHits.Load(), int32(0),
+		"layer fetch must not go through the manifests endpoint")
+}
+
+// blobServer serves fixed bytes on the blobs endpoint for the given digest.
+func blobServer(t *testing.T, dgst digest.Digest, served []byte) string {
+	t.Helper()
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.URL.Path {
+		case "/v2/":
+			w.WriteHeader(http.StatusOK)
+		case "/v2/test/artifact/blobs/" + dgst.String():
+			_, _ = w.Write(served)
+		default:
+			w.WriteHeader(http.StatusNotFound)
+		}
+	}))
+	t.Cleanup(server.Close)
+	return server.Listener.Addr().String()
+}
+
+// TestGetBlob_RejectsDigestMismatch guards that content served by the
+// registry is verified against the descriptor digest: GetBlob bypasses
+// containerd's content store, so nothing else checks integrity before the
+// bytes are written to disk as a Compose file.
+func TestGetBlob_RejectsDigestMismatch(t *testing.T) {
+	declared := []byte("services:\n  test:\n    image: alpine\n")
+	tampered := []byte("services:\n  test:\n    image: evil/image\n")
+	dgst := digest.FromBytes(declared)
+
+	host := blobServer(t, dgst, tampered)
+	resolver := NewResolver(&configfile.ConfigFile{}, &http.Transport{}, host)
+	ref, err := reference.ParseDockerRef(host + "/test/artifact:latest")
+	assert.NilError(t, err)
+
+	_, err = GetBlob(t.Context(), resolver, ref, spec.Descriptor{
+		MediaType: ComposeYAMLMediaType,
+		Digest:    dgst,
+		Size:      int64(len(tampered)),
+	})
+	assert.ErrorContains(t, err, "digest mismatch")
+}
+
+// TestGetBlob_RejectsSizeMismatch guards that GetBlob never reads more than
+// the declared descriptor size, so a rogue registry cannot cause unbounded
+// memory allocation.
+func TestGetBlob_RejectsSizeMismatch(t *testing.T) {
+	content := []byte("services:\n  test:\n    image: alpine\n")
+	dgst := digest.FromBytes(content)
+
+	host := blobServer(t, dgst, content)
+	resolver := NewResolver(&configfile.ConfigFile{}, &http.Transport{}, host)
+	ref, err := reference.ParseDockerRef(host + "/test/artifact:latest")
+	assert.NilError(t, err)
+
+	_, err = GetBlob(t.Context(), resolver, ref, spec.Descriptor{
+		MediaType: ComposeYAMLMediaType,
+		Digest:    dgst,
+		Size:      int64(len(content)) - 1, // registry serves more than declared
+	})
+	assert.ErrorContains(t, err, "size mismatch")
+}

pkg/remote/oci.go

@@ -209,12 +209,7 @@ func (g *ociRemoteLoader) pullComposeFiles(ctx context.Context, local string, ma
 	}
 
 	for i, layer := range manifest.Layers {
-		digested, err := reference.WithDigest(ref, layer.Digest)
-		if err != nil {
-			return err
-		}
-
-		_, content, err := oci.Get(ctx, resolver, digested)
+		content, err := oci.GetBlob(ctx, resolver, ref, layer)
 		if err != nil {
 			return err
 		}