blobupload.go

Documentation: github.com/docker/distribution/registry/handlers

     1  package handlers
     2  
     3  import (
     4  	"fmt"
     5  	"net/http"
     6  	"net/url"
     7  
     8  	"github.com/distribution/reference"
     9  	"github.com/docker/distribution"
    10  	dcontext "github.com/docker/distribution/context"
    11  	"github.com/docker/distribution/registry/api/errcode"
    12  	v2 "github.com/docker/distribution/registry/api/v2"
    13  	"github.com/docker/distribution/registry/storage"
    14  	"github.com/gorilla/handlers"
    15  	"github.com/opencontainers/go-digest"
    16  )
    17  
    18  // blobUploadDispatcher constructs and returns the blob upload handler for the
    19  // given request context.
    20  func blobUploadDispatcher(ctx *Context, r *http.Request) http.Handler {
    21  	buh := &blobUploadHandler{
    22  		Context: ctx,
    23  		UUID:    getUploadUUID(ctx),
    24  	}
    25  
    26  	handler := handlers.MethodHandler{
    27  		"GET":  http.HandlerFunc(buh.GetUploadStatus),
    28  		"HEAD": http.HandlerFunc(buh.GetUploadStatus),
    29  	}
    30  
    31  	if !ctx.readOnly {
    32  		handler["POST"] = http.HandlerFunc(buh.StartBlobUpload)
    33  		handler["PATCH"] = http.HandlerFunc(buh.PatchBlobData)
    34  		handler["PUT"] = http.HandlerFunc(buh.PutBlobUploadComplete)
    35  		handler["DELETE"] = http.HandlerFunc(buh.CancelBlobUpload)
    36  	}
    37  
    38  	if buh.UUID != "" {
    39  		state, err := hmacKey(ctx.Config.HTTP.Secret).unpackUploadState(r.FormValue("_state"))
    40  		if err != nil {
    41  			return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
    42  				dcontext.GetLogger(ctx).Infof("error resolving upload: %v", err)
    43  				buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadInvalid.WithDetail(err))
    44  			})
    45  		}
    46  		buh.State = state
    47  
    48  		if state.Name != ctx.Repository.Named().Name() {
    49  			return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
    50  				dcontext.GetLogger(ctx).Infof("mismatched repository name in upload state: %q != %q", state.Name, buh.Repository.Named().Name())
    51  				buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadInvalid.WithDetail(err))
    52  			})
    53  		}
    54  
    55  		if state.UUID != buh.UUID {
    56  			return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
    57  				dcontext.GetLogger(ctx).Infof("mismatched uuid in upload state: %q != %q", state.UUID, buh.UUID)
    58  				buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadInvalid.WithDetail(err))
    59  			})
    60  		}
    61  
    62  		blobs := ctx.Repository.Blobs(buh)
    63  		upload, err := blobs.Resume(buh, buh.UUID)
    64  		if err != nil {
    65  			dcontext.GetLogger(ctx).Errorf("error resolving upload: %v", err)
    66  			if err == distribution.ErrBlobUploadUnknown {
    67  				return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
    68  					buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadUnknown.WithDetail(err))
    69  				})
    70  			}
    71  
    72  			return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
    73  				buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
    74  			})
    75  		}
    76  		buh.Upload = upload
    77  
    78  		if size := upload.Size(); size != buh.State.Offset {
    79  			defer upload.Close()
    80  			dcontext.GetLogger(ctx).Errorf("upload resumed at wrong offest: %d != %d", size, buh.State.Offset)
    81  			return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
    82  				buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadInvalid.WithDetail(err))
    83  				upload.Cancel(buh)
    84  			})
    85  		}
    86  		return closeResources(handler, buh.Upload)
    87  	}
    88  
    89  	return handler
    90  }
    91  
    92  // blobUploadHandler handles the http blob upload process.
    93  type blobUploadHandler struct {
    94  	*Context
    95  
    96  	// UUID identifies the upload instance for the current request. Using UUID
    97  	// to key blob writers since this implementation uses UUIDs.
    98  	UUID string
    99  
   100  	Upload distribution.BlobWriter
   101  
   102  	State blobUploadState
   103  }
   104  
   105  // StartBlobUpload begins the blob upload process and allocates a server-side
   106  // blob writer session, optionally mounting the blob from a separate repository.
   107  func (buh *blobUploadHandler) StartBlobUpload(w http.ResponseWriter, r *http.Request) {
   108  	var options []distribution.BlobCreateOption
   109  
   110  	fromRepo := r.FormValue("from")
   111  	mountDigest := r.FormValue("mount")
   112  
   113  	if mountDigest != "" && fromRepo != "" {
   114  		opt, err := buh.createBlobMountOption(fromRepo, mountDigest)
   115  		if opt != nil && err == nil {
   116  			options = append(options, opt)
   117  		}
   118  	}
   119  
   120  	blobs := buh.Repository.Blobs(buh)
   121  	upload, err := blobs.Create(buh, options...)
   122  
   123  	if err != nil {
   124  		if ebm, ok := err.(distribution.ErrBlobMounted); ok {
   125  			if err := buh.writeBlobCreatedHeaders(w, ebm.Descriptor); err != nil {
   126  				buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
   127  			}
   128  		} else if err == distribution.ErrUnsupported {
   129  			buh.Errors = append(buh.Errors, errcode.ErrorCodeUnsupported)
   130  		} else {
   131  			buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
   132  		}
   133  		return
   134  	}
   135  
   136  	buh.Upload = upload
   137  
   138  	if err := buh.blobUploadResponse(w, r, true); err != nil {
   139  		buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
   140  		return
   141  	}
   142  
   143  	w.Header().Set("Docker-Upload-UUID", buh.Upload.ID())
   144  	w.WriteHeader(http.StatusAccepted)
   145  }
   146  
   147  // GetUploadStatus returns the status of a given upload, identified by id.
   148  func (buh *blobUploadHandler) GetUploadStatus(w http.ResponseWriter, r *http.Request) {
   149  	if buh.Upload == nil {
   150  		buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadUnknown)
   151  		return
   152  	}
   153  
   154  	// TODO(dmcgowan): Set last argument to false in blobUploadResponse when
   155  	// resumable upload is supported. This will enable returning a non-zero
   156  	// range for clients to begin uploading at an offset.
   157  	if err := buh.blobUploadResponse(w, r, true); err != nil {
   158  		buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
   159  		return
   160  	}
   161  
   162  	w.Header().Set("Docker-Upload-UUID", buh.UUID)
   163  	w.WriteHeader(http.StatusNoContent)
   164  }
   165  
   166  // PatchBlobData writes data to an upload.
   167  func (buh *blobUploadHandler) PatchBlobData(w http.ResponseWriter, r *http.Request) {
   168  	if buh.Upload == nil {
   169  		buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadUnknown)
   170  		return
   171  	}
   172  
   173  	ct := r.Header.Get("Content-Type")
   174  	if ct != "" && ct != "application/octet-stream" {
   175  		buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(fmt.Errorf("bad Content-Type")))
   176  		// TODO(dmcgowan): encode error
   177  		return
   178  	}
   179  
   180  	// TODO(dmcgowan): support Content-Range header to seek and write range
   181  
   182  	if err := copyFullPayload(buh, w, r, buh.Upload, -1, "blob PATCH"); err != nil {
   183  		buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err.Error()))
   184  		return
   185  	}
   186  
   187  	if err := buh.blobUploadResponse(w, r, false); err != nil {
   188  		buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
   189  		return
   190  	}
   191  
   192  	w.WriteHeader(http.StatusAccepted)
   193  }
   194  
   195  // PutBlobUploadComplete takes the final request of a blob upload. The
   196  // request may include all the blob data or no blob data. Any data
   197  // provided is received and verified. If successful, the blob is linked
   198  // into the blob store and 201 Created is returned with the canonical
   199  // url of the blob.
   200  func (buh *blobUploadHandler) PutBlobUploadComplete(w http.ResponseWriter, r *http.Request) {
   201  	if buh.Upload == nil {
   202  		buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadUnknown)
   203  		return
   204  	}
   205  
   206  	dgstStr := r.FormValue("digest") // TODO(stevvooe): Support multiple digest parameters!
   207  
   208  	if dgstStr == "" {
   209  		// no digest? return error, but allow retry.
   210  		buh.Errors = append(buh.Errors, v2.ErrorCodeDigestInvalid.WithDetail("digest missing"))
   211  		return
   212  	}
   213  
   214  	dgst, err := digest.Parse(dgstStr)
   215  	if err != nil {
   216  		// no digest? return error, but allow retry.
   217  		buh.Errors = append(buh.Errors, v2.ErrorCodeDigestInvalid.WithDetail("digest parsing failed"))
   218  		return
   219  	}
   220  
   221  	if err := copyFullPayload(buh, w, r, buh.Upload, -1, "blob PUT"); err != nil {
   222  		buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err.Error()))
   223  		return
   224  	}
   225  
   226  	desc, err := buh.Upload.Commit(buh, distribution.Descriptor{
   227  		Digest: dgst,
   228  
   229  		// TODO(stevvooe): This isn't wildly important yet, but we should
   230  		// really set the mediatype. For now, we can let the backend take care
   231  		// of this.
   232  	})
   233  
   234  	if err != nil {
   235  		switch err := err.(type) {
   236  		case distribution.ErrBlobInvalidDigest:
   237  			buh.Errors = append(buh.Errors, v2.ErrorCodeDigestInvalid.WithDetail(err))
   238  		case errcode.Error:
   239  			buh.Errors = append(buh.Errors, err)
   240  		default:
   241  			switch err {
   242  			case distribution.ErrAccessDenied:
   243  				buh.Errors = append(buh.Errors, errcode.ErrorCodeDenied)
   244  			case distribution.ErrUnsupported:
   245  				buh.Errors = append(buh.Errors, errcode.ErrorCodeUnsupported)
   246  			case distribution.ErrBlobInvalidLength, distribution.ErrBlobDigestUnsupported:
   247  				buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadInvalid.WithDetail(err))
   248  			default:
   249  				dcontext.GetLogger(buh).Errorf("unknown error completing upload: %v", err)
   250  				buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
   251  			}
   252  
   253  		}
   254  
   255  		// Clean up the backend blob data if there was an error.
   256  		if err := buh.Upload.Cancel(buh); err != nil {
   257  			// If the cleanup fails, all we can do is observe and report.
   258  			dcontext.GetLogger(buh).Errorf("error canceling upload after error: %v", err)
   259  		}
   260  
   261  		return
   262  	}
   263  	if err := buh.writeBlobCreatedHeaders(w, desc); err != nil {
   264  		buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
   265  		return
   266  	}
   267  }
   268  
   269  // CancelBlobUpload cancels an in-progress upload of a blob.
   270  func (buh *blobUploadHandler) CancelBlobUpload(w http.ResponseWriter, r *http.Request) {
   271  	if buh.Upload == nil {
   272  		buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadUnknown)
   273  		return
   274  	}
   275  
   276  	w.Header().Set("Docker-Upload-UUID", buh.UUID)
   277  	if err := buh.Upload.Cancel(buh); err != nil {
   278  		dcontext.GetLogger(buh).Errorf("error encountered canceling upload: %v", err)
   279  		buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
   280  	}
   281  
   282  	w.WriteHeader(http.StatusNoContent)
   283  }
   284  
   285  // blobUploadResponse provides a standard request for uploading blobs and
   286  // chunk responses. This sets the correct headers but the response status is
   287  // left to the caller. The fresh argument is used to ensure that new blob
   288  // uploads always start at a 0 offset. This allows disabling resumable push by
   289  // always returning a 0 offset on check status.
   290  func (buh *blobUploadHandler) blobUploadResponse(w http.ResponseWriter, r *http.Request, fresh bool) error {
   291  	// TODO(stevvooe): Need a better way to manage the upload state automatically.
   292  	buh.State.Name = buh.Repository.Named().Name()
   293  	buh.State.UUID = buh.Upload.ID()
   294  	buh.Upload.Close()
   295  	buh.State.Offset = buh.Upload.Size()
   296  	buh.State.StartedAt = buh.Upload.StartedAt()
   297  
   298  	token, err := hmacKey(buh.Config.HTTP.Secret).packUploadState(buh.State)
   299  	if err != nil {
   300  		dcontext.GetLogger(buh).Infof("error building upload state token: %s", err)
   301  		return err
   302  	}
   303  
   304  	uploadURL, err := buh.urlBuilder.BuildBlobUploadChunkURL(
   305  		buh.Repository.Named(), buh.Upload.ID(),
   306  		url.Values{
   307  			"_state": []string{token},
   308  		})
   309  	if err != nil {
   310  		dcontext.GetLogger(buh).Infof("error building upload url: %s", err)
   311  		return err
   312  	}
   313  
   314  	endRange := buh.Upload.Size()
   315  	if endRange > 0 {
   316  		endRange = endRange - 1
   317  	}
   318  
   319  	w.Header().Set("Docker-Upload-UUID", buh.UUID)
   320  	w.Header().Set("Location", uploadURL)
   321  
   322  	w.Header().Set("Content-Length", "0")
   323  	w.Header().Set("Range", fmt.Sprintf("0-%d", endRange))
   324  
   325  	return nil
   326  }
   327  
   328  // mountBlob attempts to mount a blob from another repository by its digest. If
   329  // successful, the blob is linked into the blob store and 201 Created is
   330  // returned with the canonical url of the blob.
   331  func (buh *blobUploadHandler) createBlobMountOption(fromRepo, mountDigest string) (distribution.BlobCreateOption, error) {
   332  	dgst, err := digest.Parse(mountDigest)
   333  	if err != nil {
   334  		return nil, err
   335  	}
   336  
   337  	ref, err := reference.WithName(fromRepo)
   338  	if err != nil {
   339  		return nil, err
   340  	}
   341  
   342  	canonical, err := reference.WithDigest(ref, dgst)
   343  	if err != nil {
   344  		return nil, err
   345  	}
   346  
   347  	return storage.WithMountFrom(canonical), nil
   348  }
   349  
   350  // writeBlobCreatedHeaders writes the standard headers describing a newly
   351  // created blob. A 201 Created is written as well as the canonical URL and
   352  // blob digest.
   353  func (buh *blobUploadHandler) writeBlobCreatedHeaders(w http.ResponseWriter, desc distribution.Descriptor) error {
   354  	ref, err := reference.WithDigest(buh.Repository.Named(), desc.Digest)
   355  	if err != nil {
   356  		return err
   357  	}
   358  	blobURL, err := buh.urlBuilder.BuildBlobURL(ref)
   359  	if err != nil {
   360  		return err
   361  	}
   362  
   363  	w.Header().Set("Location", blobURL)
   364  	w.Header().Set("Content-Length", "0")
   365  	w.Header().Set("Docker-Content-Digest", desc.Digest.String())
   366  	w.WriteHeader(http.StatusCreated)
   367  	return nil
   368  }
   369
View as plain text