Merge pull request #11709 from fiam/alberto/fix-pusher-authorizer-error-on-mount

Fix cross-repo mount fallback after authorization failure

core/remotes/docker/pusher.go

@@ -178,17 +178,21 @@ func (p dockerPusher) push(ctx context.Context, desc ocispec.Descriptor, ref str
 			// query and send the request again.
 			resp, err = preq.doWithRetries(pctx, nil)
 			if err != nil {
-				return nil, err
+				if !errors.Is(err, ErrInvalidAuthorization) {
+					return nil, fmt.Errorf("pushing with mount from %s: %w", fromRepo, err)
+				}
+				log.G(ctx).Debugf("failed to push with mount from repository %s: %v", fromRepo, err)
 			}
+			if resp != nil {
+				switch resp.StatusCode {
+				case http.StatusUnauthorized:
+					log.G(ctx).Debugf("failed to mount from repository %s, not authorized", fromRepo)
 
-			switch resp.StatusCode {
-			case http.StatusUnauthorized:
-				log.G(ctx).Debugf("failed to mount from repository %s", fromRepo)
-
-				resp.Body.Close()
-				resp = nil
-			case http.StatusCreated:
-				mountedFrom = path.Join(p.refspec.Hostname(), fromRepo)
+					resp.Body.Close()
+					resp = nil
+				case http.StatusCreated:
+					mountedFrom = path.Join(p.refspec.Hostname(), fromRepo)
+				}
 			}
 		}
 

core/remotes/docker/pusher_test.go

@@ -28,7 +28,10 @@ import (
 	"net/url"
 	"reflect"
 	"regexp"
+	"slices"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/containerd/containerd/v2/core/content"
@@ -40,6 +43,7 @@ import (
 	ocispecv "github.com/opencontainers/image-spec/specs-go"
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestGetManifestPath(t *testing.T) {
@@ -173,6 +177,136 @@ func TestPusherErrReset(t *testing.T) {
 	assert.NoError(t, err)
 }
 
+func TestPusherInvalidAuthorizationOnMount(t *testing.T) {
+	t.Parallel()
+	// Simulate trying to mount a private repo that we cannot access to
+
+	isCrossRepoMount := func(r *http.Request) bool {
+		return r.URL.Query().Get("mount") != ""
+	}
+
+	testCases := []struct {
+		name  string
+		setup func(t *testing.T, p *dockerPusher, reg *uploadableMockRegistry, triggered func())
+	}{
+		{
+			name: "Authorizer.Authorize error",
+			setup: func(t *testing.T, p *dockerPusher, _ *uploadableMockRegistry, triggered func()) {
+				p.hosts[0].Authorizer = &mockAuthorizer{
+					authorize: func(ctx context.Context, r *http.Request) error {
+						// When trying to authorize the request to mount, return an error
+						// to force a fallback
+						if isCrossRepoMount(r) {
+							triggered()
+							return ErrInvalidAuthorization
+						}
+						return nil
+					},
+				}
+			},
+		},
+		{
+			name: "Authorizer.AddResponses error",
+			setup: func(t *testing.T, p *dockerPusher, reg *uploadableMockRegistry, triggered func()) {
+
+				reg.defaultHandlerFunc = func(w http.ResponseWriter, r *http.Request) bool {
+					if isCrossRepoMount(r) && r.Header.Get("Authorization") == "" {
+						w.Header().Set("WWW-Authenticate", "Bearer realm=localhost")
+						w.WriteHeader(http.StatusUnauthorized)
+						return true
+					}
+					return false
+				}
+
+				var allResp []*http.Response
+				var mu sync.Mutex
+				p.hosts[0].Authorizer = &mockAuthorizer{
+					authorize: func(ctc context.Context, r *http.Request) error {
+						mu.Lock()
+						defer mu.Unlock()
+						hasFirstResp := slices.ContainsFunc(allResp, func(resp *http.Response) bool {
+							return resp.Request.URL.String() == r.URL.String()
+						})
+						if hasFirstResp {
+							r.Header.Add("Authorization", "Bearer test")
+						}
+						return nil
+					},
+					addResponses: func(ctx context.Context, resp []*http.Response) error {
+						mu.Lock()
+						defer mu.Unlock()
+						// When trying to add a response to the request to mount, return an error
+						// to force a fallback
+						allResp = append(allResp, resp...)
+						last := resp[len(resp)-1]
+						if isCrossRepoMount(last.Request) {
+							triggered()
+							return ErrInvalidAuthorization
+						}
+						return nil
+					},
+				}
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			//t.Parallel()
+
+			p, reg, _, done := samplePusher(t)
+			defer done()
+
+			var triggered atomic.Bool
+
+			p.object = "layer@sha256:9f7d2a1b5c8d6eeadfbb8e786ce21958a1b6cc0b7c263c9d3e4eaf6f24c3a1bd"
+
+			reg.uploadable = true
+			tc.setup(t, &p, reg, func() { triggered.Store(true) })
+
+			ct := []byte("layer-bytes")
+
+			desc := ocispec.Descriptor{
+				MediaType: ocispec.MediaTypeImageLayer,
+				Digest:    digest.FromBytes(ct),
+				Size:      int64(len(ct)),
+				Annotations: map[string]string{
+					distributionSourceLabelKey(samplePusherHostname): samplePusherHostname + "/anotherrepository:latest",
+				},
+			}
+
+			w, err := p.push(context.Background(), desc, remotes.MakeRefKey(context.Background(), desc), false)
+			require.NoError(t, err)
+
+			_, err = w.Write(ct)
+			assert.NoError(t, err)
+			err = w.Commit(context.Background(), desc.Size, desc.Digest)
+			assert.NoError(t, err)
+
+			assert.True(t, triggered.Load(), "error return was not triggered")
+		})
+	}
+}
+
+type mockAuthorizer struct {
+	authorize    func(ctx context.Context, r *http.Request) error
+	addResponses func(ctx context.Context, resp []*http.Response) error
+}
+
+func (a *mockAuthorizer) Authorize(ctx context.Context, r *http.Request) error {
+	if a.authorize == nil {
+		return nil
+	}
+	return a.authorize(ctx, r)
+}
+
+func (a *mockAuthorizer) AddResponses(ctx context.Context, resp []*http.Response) error {
+	if a.addResponses == nil {
+		return nil
+	}
+	return a.addResponses(ctx, resp)
+}
+
 func tryUpload(ctx context.Context, t *testing.T, p dockerPusher, layerContent []byte) error {
 	desc := ocispec.Descriptor{
 		MediaType: ocispec.MediaTypeImageLayerGzip,
@@ -235,6 +369,11 @@ func tryUpload(ctx context.Context, t *testing.T, p dockerPusher, layerContent [
 	return nil
 }
 
+const (
+	samplePusherHostname = "example.com"
+	samplePusherLocator  = samplePusherHostname + "/samplerepository:latest"
+)
+
 func samplePusher(t *testing.T) (dockerPusher, *uploadableMockRegistry, StatusTrackLocker, func()) {
 	reg := &uploadableMockRegistry{
 		availableContents: make([]string, 0),
@@ -248,7 +387,7 @@ func samplePusher(t *testing.T) (dockerPusher, *uploadableMockRegistry, StatusTr
 	return dockerPusher{
 		dockerBase: &dockerBase{
 			refspec: reference.Spec{
-				Locator: "example.com/samplerepository:latest",
+				Locator: samplePusherLocator,
 			},
 			repository: "samplerepository",
 			hosts: []RegistryHost{
@@ -271,12 +410,13 @@ var blobUploadRegexp = regexp.MustCompile(`/([a-z0-9]+)/blobs/uploads/(.*)`)
 
 // uploadableMockRegistry provides minimal registry APIs which are enough to serve requests from dockerPusher.
 type uploadableMockRegistry struct {
-	availableContents []string
-	uploadable        bool
-	putHandlerFunc    func(w http.ResponseWriter, r *http.Request) bool
-	locationPrefix    string
-	username          string
-	secret            string
+	availableContents  []string
+	uploadable         bool
+	putHandlerFunc     func(w http.ResponseWriter, r *http.Request) bool
+	defaultHandlerFunc func(w http.ResponseWriter, r *http.Request) bool
+	locationPrefix     string
+	username           string
+	secret             string
 }
 
 func (u *uploadableMockRegistry) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -289,11 +429,14 @@ func (u *uploadableMockRegistry) ServeHTTP(w http.ResponseWriter, r *http.Reques
 		}
 	}
 	if r.Method == http.MethodPut && u.putHandlerFunc != nil {
-		// if true return the response witout calling default handler
+		// if true return the response without calling default handler
 		if u.putHandlerFunc(w, r) {
 			return
 		}
 	}
+	if u.defaultHandlerFunc != nil && u.defaultHandlerFunc(w, r) {
+		return
+	}
 	u.defaultHandler(w, r)
 }