Merge pull request #14360 from tonistiigi/auth-body-close-fix

remotes/docker: close request body when authorization fails

core/remotes/docker/pusher_test.go

@@ -871,3 +871,53 @@ func TestPusherForbiddenGETFallbackProxy(t *testing.T) {
 	assert.Equal(t, samplePusherHostname, gotNsParam,
 		"GET fallback on proxy must include ?ns= query parameter")
 }
+
+// If authorizing the upload request fails, the request is never sent and
+// nobody reads the body pipe. A pending Write must be unblocked with the
+// authorization error instead of staying stuck forever.
+func TestPusherAuthorizeFailureUnblocksWrite(t *testing.T) {
+	p, reg, _, done := samplePusher(t)
+	defer done()
+
+	// Authorization of the upload PUT fails, like a bearer token refresh that
+	// fails or is cancelled after the upload was initiated.
+	authErr := errors.New("token fetch failed")
+	bodyCh := make(chan io.ReadCloser, 1)
+	p.hosts[0].Authorizer = &mockAuthorizer{
+		authorize: func(_ context.Context, r *http.Request) error {
+			if r.Method != http.MethodPut {
+				return nil
+			}
+			bodyCh <- r.Body
+			return authErr
+		},
+	}
+	reg.uploadable = true
+
+	ct := []byte("layer-content")
+	desc := ocispec.Descriptor{
+		MediaType: ocispec.MediaTypeImageLayer,
+		Digest:    digest.FromBytes(ct),
+		Size:      int64(len(ct)),
+	}
+
+	w, err := p.push(t.Context(), desc, remotes.MakeRefKey(t.Context(), desc), false)
+	require.NoError(t, err)
+
+	errCh := make(chan error, 1)
+	go func() {
+		_, err := w.Write(ct)
+		errCh <- err
+	}()
+
+	select {
+	case err := <-errCh:
+		require.ErrorIs(t, err, authErr)
+	case <-time.After(5 * time.Second):
+		// Release the blocked Write so it does not outlive the test.
+		(<-bodyCh).Close()
+		<-errCh
+		t.Fatal("Write is blocked after the upload request failed to authorize")
+	}
+	require.NoError(t, w.Close())
+}

core/remotes/docker/resolver.go

@@ -693,6 +693,13 @@ func (r *request) do(ctx context.Context) (*http.Response, error) {
 	ctx = log.WithLogger(ctx, log.G(ctx).WithField("url", r.sanitizedURL()))
 	log.G(ctx).WithFields(requestFields(req)).Debug("do request")
 	if err := r.authorize(ctx, req); err != nil {
+		// Authorization failed before the request was handed to the HTTP
+		// client, which would otherwise own and close the body. Close it
+		// here, otherwise the producer of a streaming body (e.g. the
+		// pusher's pipe) is never unblocked.
+		if req.Body != nil {
+			req.Body.Close()
+		}
 		return nil, fmt.Errorf("failed to authorize: %w", err)
 	}