fix: correct fetch order for manifests and blobs with hints on media …

…type (#1209)

<!-- markdownlint-disable MD041 -->
#### What this PR does / why we need it

This PR changes the OCI fetch logic to:
1. Fetch Blobs before Manifests in case there is no media type
2. Fetch Manifests before Blobs or Blobs before Manifests if there is a
media type hint

#### Which issue(s) this PR fixes
<!--
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
-->

---------

Signed-off-by: Gergely Brautigam <[email protected]>
Co-authored-by: Gergely Brautigam <[email protected]>

api/oci/extensions/repositories/ocireg/repository.go

@@ -11,6 +11,7 @@ import (
 	"github.com/containerd/errdefs"
 	"github.com/mandelsoft/goutils/errors"
 	"github.com/mandelsoft/logging"
+	"github.com/moby/locker"
 	"oras.land/oras-go/v2/registry/remote/auth"
 	"oras.land/oras-go/v2/registry/remote/retry"
 
@@ -179,6 +180,7 @@ func (r *RepositoryImpl) getResolver(comp string) (oras.Resolver, error) {
 		Client:    authClient,
 		PlainHTTP: r.info.Scheme == "http",
 		Logger:    logger,
+		Lock:      locker.New(),
 	}), nil
 }
 

api/tech/oras/client.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/containerd/containerd/errdefs"
 	"github.com/mandelsoft/logging"
+	"github.com/moby/locker"
 	ociv1 "github.com/opencontainers/image-spec/specs-go/v1"
 	oraserr "oras.land/oras-go/v2/errdef"
 	"oras.land/oras-go/v2/registry/remote"
@@ -17,26 +18,28 @@ type ClientOptions struct {
 	Client    *auth.Client
 	PlainHTTP bool
 	Logger    logging.Logger
+	Lock      *locker.Locker
 }
 
 type Client struct {
 	client    *auth.Client
 	plainHTTP bool
 	logger    logging.Logger
+	lock      *locker.Locker
 }
 
 var _ Resolver = &Client{}
 
 func New(opts ClientOptions) *Client {
-	return &Client{client: opts.Client, plainHTTP: opts.PlainHTTP, logger: opts.Logger}
+	return &Client{client: opts.Client, plainHTTP: opts.PlainHTTP, logger: opts.Logger, lock: opts.Lock}
 }
 
 func (c *Client) Fetcher(ctx context.Context, ref string) (Fetcher, error) {
 	return &OrasFetcher{client: c.client, ref: ref, plainHTTP: c.plainHTTP}, nil
 }
 
 func (c *Client) Pusher(ctx context.Context, ref string) (Pusher, error) {
-	return &OrasPusher{client: c.client, ref: ref, plainHTTP: c.plainHTTP}, nil
+	return &OrasPusher{client: c.client, ref: ref, plainHTTP: c.plainHTTP, lock: c.lock}, nil
 }
 
 func (c *Client) Lister(ctx context.Context, ref string) (Lister, error) {

api/tech/oras/fetcher.go

@@ -5,60 +5,84 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"sync"
 
 	ociv1 "github.com/opencontainers/image-spec/specs-go/v1"
+	"oras.land/oras-go/v2/registry/remote"
 	"oras.land/oras-go/v2/registry/remote/auth"
 )
 
 type OrasFetcher struct {
 	client    *auth.Client
 	ref       string
 	plainHTTP bool
+	mu        sync.Mutex
 }
 
 func (c *OrasFetcher) Fetch(ctx context.Context, desc ociv1.Descriptor) (io.ReadCloser, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
 	src, err := createRepository(c.ref, c.client, c.plainHTTP)
 	if err != nil {
 		return nil, fmt.Errorf("failed to resolve ref %q: %w", c.ref, err)
 	}
 
-	// oras requires a Resolve to happen before a fetch because
+	// oras requires a Resolve to happen in some cases before a fetch because
 	// -1 or 0 are invalid sizes and result in a content-length mismatch error by design.
 	// This is a security consideration on ORAS' side.
 	// For more information (https://github.com/oras-project/oras-go/issues/822#issuecomment-2325622324)
-	// We explicitly call resolve on manifest first because it might be
-	// that the mediatype is not set at this point so we don't want ORAS to try to
-	// select the wrong layer to fetch from.
-	if desc.Size < 1 || desc.Digest == "" {
-		rdesc, err := src.Manifests().Resolve(ctx, desc.Digest.String())
-		if err != nil {
-			var berr error
-			rdesc, berr = src.Blobs().Resolve(ctx, desc.Digest.String())
-			if berr != nil {
-				// also display the first manifest resolve error
-				err = errors.Join(err, berr)
+	//
+	// To workaround, we resolve the correct size
+	if desc.Size < 1 {
+		if desc, err = c.resolveDescriptor(ctx, desc, src); err != nil {
+			return nil, err
+		}
+	}
 
-				return nil, fmt.Errorf("failed to resolve fetch blob %q: %w", desc.Digest.String(), err)
-			}
+	// manifest resolve succeeded return the reader directly
+	// mediatype of the descriptor should now be set to the correct type.
+	reader, err := src.Fetch(ctx, desc)
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch blob: %w", err)
+	}
 
-			reader, err := src.Blobs().Fetch(ctx, rdesc)
-			if err != nil {
-				return nil, fmt.Errorf("failed to fetch blob: %w", err)
-			}
+	return reader, nil
+}
 
-			return reader, nil
+// resolveDescriptor resolves the descriptor by fetching the blob or manifest based on the digest as a reference.
+// If the descriptor has a media type, it will be resolved directly.
+// If the descriptor has no media type, it will first try to resolve the blob, then the manifest as a fallback.
+func (c *OrasFetcher) resolveDescriptor(ctx context.Context, desc ociv1.Descriptor, src *remote.Repository) (ociv1.Descriptor, error) {
+	if desc.MediaType != "" {
+		var err error
+		// if there is a media type, resolve the descriptor directly
+		if isManifest(src.ManifestMediaTypes, desc) {
+			desc, err = src.Manifests().Resolve(ctx, desc.Digest.String())
+		} else {
+			desc, err = src.Blobs().Resolve(ctx, desc.Digest.String())
 		}
-
-		// no error
-		desc = rdesc
+		if err != nil {
+			return ociv1.Descriptor{}, fmt.Errorf("failed to resolve descriptor %q: %w", desc.Digest.String(), err)
+		}
+		return desc, nil
 	}
 
-	// manifest resolve succeeded return the reader directly
-	// mediatype of the descriptor should now be set to the correct type.
-	fetch, err := src.Fetch(ctx, desc)
+	// if there is no media type, first try the blob, then the manifest
+	// To reader: DO NOT fetch manifest first, this can result in high latency calls
+	bdesc, err := src.Blobs().Resolve(ctx, desc.Digest.String())
 	if err != nil {
-		return nil, fmt.Errorf("failed to fetch manifest: %w", err)
+		mdesc, merr := src.Manifests().Resolve(ctx, desc.Digest.String())
+		if merr != nil {
+			// also display the first manifest resolve error
+			err = errors.Join(err, merr)
+
+			return ociv1.Descriptor{}, fmt.Errorf("failed to resolve manifest %q: %w", desc.Digest.String(), err)
+		}
+		desc = mdesc
+	} else {
+		desc = bdesc
 	}
 
-	return fetch, nil
+	return desc, err
 }

api/tech/oras/manifest.go

@@ -0,0 +1,27 @@
+package oras
+
+import (
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+)
+
+// defaultManifestMediaTypes contains the default set of manifests media types.
+var defaultManifestMediaTypes = []string{
+	"application/vnd.docker.distribution.manifest.v2+json",
+	"application/vnd.docker.distribution.manifest.list.v2+json",
+	"application/vnd.oci.artifact.manifest.v1+json",
+	ocispec.MediaTypeImageManifest,
+	ocispec.MediaTypeImageIndex,
+}
+
+// isManifest determines if the given descriptor points to a manifest.
+func isManifest(manifestMediaTypes []string, desc ocispec.Descriptor) bool {
+	if len(manifestMediaTypes) == 0 {
+		manifestMediaTypes = defaultManifestMediaTypes
+	}
+	for _, mediaType := range manifestMediaTypes {
+		if desc.MediaType == mediaType {
+			return true
+		}
+	}
+	return false
+}

api/tech/oras/pusher.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	"github.com/containerd/errdefs"
+	"github.com/moby/locker"
 	ociv1 "github.com/opencontainers/image-spec/specs-go/v1"
 	"oras.land/oras-go/v2/registry"
 	"oras.land/oras-go/v2/registry/remote/auth"
@@ -16,13 +17,20 @@ type OrasPusher struct {
 	client    *auth.Client
 	ref       string
 	plainHTTP bool
+	lock      *locker.Locker
 }
 
 func (c *OrasPusher) Push(ctx context.Context, d ociv1.Descriptor, src Source) (retErr error) {
+	c.lock.Lock(c.ref)
+	defer c.lock.Unlock(c.ref)
+
 	reader, err := src.Reader()
 	if err != nil {
 		return err
 	}
+	defer func() {
+		reader.Close()
+	}()
 
 	repository, err := createRepository(c.ref, c.client, c.plainHTTP)
 	if err != nil {
@@ -60,8 +68,6 @@ func (c *OrasPusher) Push(ctx context.Context, d ociv1.Descriptor, src Source) (
 		return errdefs.ErrAlreadyExists
 	}
 
-	// We have a digest, so we use plain push for the digest.
-	// Push here decides if it's a Manifest or a Blob.
 	if err := repository.Push(ctx, d, reader); err != nil {
 		return fmt.Errorf("failed to push: %w, %s", err, c.ref)
 	}

go.mod

@@ -49,6 +49,7 @@ require (
 	github.com/mikefarah/yq/v4 v4.44.6
 	github.com/mitchellh/copystructure v1.2.0
 	github.com/mittwald/go-helm-client v0.12.15
+	github.com/moby/locker v1.0.1
 	github.com/modern-go/reflect2 v1.0.2
 	github.com/onsi/ginkgo/v2 v2.22.0
 	github.com/onsi/gomega v1.36.1
@@ -258,7 +259,6 @@ require (
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/moby/docker-image-spec v1.3.1 // indirect
-	github.com/moby/locker v1.0.1 // indirect
 	github.com/moby/spdystream v0.5.0 // indirect
 	github.com/moby/sys/capability v0.3.0 // indirect
 	github.com/moby/sys/mountinfo v0.7.2 // indirect