d8fcbeeb71
The registry uses partial Named values which the named parsers no longer support. To allow the registry service to continue to operate without canonicalization, switch to use WithName. In the future, the registry should start using fully canonical values on the backend and WithName should no longer support creating partial values. Signed-off-by: Derek McGowan <derek@mcgstyle.net> (github: dmcgowan)
368 lines
12 KiB
Go
368 lines
12 KiB
Go
package handlers
|
|
|
|
import (
|
|
"fmt"
|
|
"net/http"
|
|
"net/url"
|
|
|
|
"github.com/docker/distribution"
|
|
ctxu "github.com/docker/distribution/context"
|
|
"github.com/docker/distribution/reference"
|
|
"github.com/docker/distribution/registry/api/errcode"
|
|
"github.com/docker/distribution/registry/api/v2"
|
|
"github.com/docker/distribution/registry/storage"
|
|
"github.com/gorilla/handlers"
|
|
"github.com/opencontainers/go-digest"
|
|
)
|
|
|
|
// blobUploadDispatcher constructs and returns the blob upload handler for the
|
|
// given request context.
|
|
func blobUploadDispatcher(ctx *Context, r *http.Request) http.Handler {
|
|
buh := &blobUploadHandler{
|
|
Context: ctx,
|
|
UUID: getUploadUUID(ctx),
|
|
}
|
|
|
|
handler := handlers.MethodHandler{
|
|
"GET": http.HandlerFunc(buh.GetUploadStatus),
|
|
"HEAD": http.HandlerFunc(buh.GetUploadStatus),
|
|
}
|
|
|
|
if !ctx.readOnly {
|
|
handler["POST"] = http.HandlerFunc(buh.StartBlobUpload)
|
|
handler["PATCH"] = http.HandlerFunc(buh.PatchBlobData)
|
|
handler["PUT"] = http.HandlerFunc(buh.PutBlobUploadComplete)
|
|
handler["DELETE"] = http.HandlerFunc(buh.CancelBlobUpload)
|
|
}
|
|
|
|
if buh.UUID != "" {
|
|
state, err := hmacKey(ctx.Config.HTTP.Secret).unpackUploadState(r.FormValue("_state"))
|
|
if err != nil {
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
ctxu.GetLogger(ctx).Infof("error resolving upload: %v", err)
|
|
buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadInvalid.WithDetail(err))
|
|
})
|
|
}
|
|
buh.State = state
|
|
|
|
if state.Name != ctx.Repository.Named().Name() {
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
ctxu.GetLogger(ctx).Infof("mismatched repository name in upload state: %q != %q", state.Name, buh.Repository.Named().Name())
|
|
buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadInvalid.WithDetail(err))
|
|
})
|
|
}
|
|
|
|
if state.UUID != buh.UUID {
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
ctxu.GetLogger(ctx).Infof("mismatched uuid in upload state: %q != %q", state.UUID, buh.UUID)
|
|
buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadInvalid.WithDetail(err))
|
|
})
|
|
}
|
|
|
|
blobs := ctx.Repository.Blobs(buh)
|
|
upload, err := blobs.Resume(buh, buh.UUID)
|
|
if err != nil {
|
|
ctxu.GetLogger(ctx).Errorf("error resolving upload: %v", err)
|
|
if err == distribution.ErrBlobUploadUnknown {
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadUnknown.WithDetail(err))
|
|
})
|
|
}
|
|
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
|
|
})
|
|
}
|
|
buh.Upload = upload
|
|
|
|
if size := upload.Size(); size != buh.State.Offset {
|
|
defer upload.Close()
|
|
ctxu.GetLogger(ctx).Errorf("upload resumed at wrong offest: %d != %d", size, buh.State.Offset)
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadInvalid.WithDetail(err))
|
|
upload.Cancel(buh)
|
|
})
|
|
}
|
|
return closeResources(handler, buh.Upload)
|
|
}
|
|
|
|
return handler
|
|
}
|
|
|
|
// blobUploadHandler handles the http blob upload process.
|
|
type blobUploadHandler struct {
|
|
*Context
|
|
|
|
// UUID identifies the upload instance for the current request. Using UUID
|
|
// to key blob writers since this implementation uses UUIDs.
|
|
UUID string
|
|
|
|
Upload distribution.BlobWriter
|
|
|
|
State blobUploadState
|
|
}
|
|
|
|
// StartBlobUpload begins the blob upload process and allocates a server-side
|
|
// blob writer session, optionally mounting the blob from a separate repository.
|
|
func (buh *blobUploadHandler) StartBlobUpload(w http.ResponseWriter, r *http.Request) {
|
|
var options []distribution.BlobCreateOption
|
|
|
|
fromRepo := r.FormValue("from")
|
|
mountDigest := r.FormValue("mount")
|
|
|
|
if mountDigest != "" && fromRepo != "" {
|
|
opt, err := buh.createBlobMountOption(fromRepo, mountDigest)
|
|
if opt != nil && err == nil {
|
|
options = append(options, opt)
|
|
}
|
|
}
|
|
|
|
blobs := buh.Repository.Blobs(buh)
|
|
upload, err := blobs.Create(buh, options...)
|
|
|
|
if err != nil {
|
|
if ebm, ok := err.(distribution.ErrBlobMounted); ok {
|
|
if err := buh.writeBlobCreatedHeaders(w, ebm.Descriptor); err != nil {
|
|
buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
|
|
}
|
|
} else if err == distribution.ErrUnsupported {
|
|
buh.Errors = append(buh.Errors, errcode.ErrorCodeUnsupported)
|
|
} else {
|
|
buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
|
|
}
|
|
return
|
|
}
|
|
|
|
buh.Upload = upload
|
|
|
|
if err := buh.blobUploadResponse(w, r, true); err != nil {
|
|
buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
|
|
return
|
|
}
|
|
|
|
w.Header().Set("Docker-Upload-UUID", buh.Upload.ID())
|
|
w.WriteHeader(http.StatusAccepted)
|
|
}
|
|
|
|
// GetUploadStatus returns the status of a given upload, identified by id.
|
|
func (buh *blobUploadHandler) GetUploadStatus(w http.ResponseWriter, r *http.Request) {
|
|
if buh.Upload == nil {
|
|
buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadUnknown)
|
|
return
|
|
}
|
|
|
|
// TODO(dmcgowan): Set last argument to false in blobUploadResponse when
|
|
// resumable upload is supported. This will enable returning a non-zero
|
|
// range for clients to begin uploading at an offset.
|
|
if err := buh.blobUploadResponse(w, r, true); err != nil {
|
|
buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
|
|
return
|
|
}
|
|
|
|
w.Header().Set("Docker-Upload-UUID", buh.UUID)
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
// PatchBlobData writes data to an upload.
|
|
func (buh *blobUploadHandler) PatchBlobData(w http.ResponseWriter, r *http.Request) {
|
|
if buh.Upload == nil {
|
|
buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadUnknown)
|
|
return
|
|
}
|
|
|
|
ct := r.Header.Get("Content-Type")
|
|
if ct != "" && ct != "application/octet-stream" {
|
|
buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(fmt.Errorf("Bad Content-Type")))
|
|
// TODO(dmcgowan): encode error
|
|
return
|
|
}
|
|
|
|
// TODO(dmcgowan): support Content-Range header to seek and write range
|
|
|
|
if err := copyFullPayload(w, r, buh.Upload, buh, "blob PATCH", &buh.Errors); err != nil {
|
|
// copyFullPayload reports the error if necessary
|
|
return
|
|
}
|
|
|
|
if err := buh.blobUploadResponse(w, r, false); err != nil {
|
|
buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
|
|
return
|
|
}
|
|
|
|
w.WriteHeader(http.StatusAccepted)
|
|
}
|
|
|
|
// PutBlobUploadComplete takes the final request of a blob upload. The
|
|
// request may include all the blob data or no blob data. Any data
|
|
// provided is received and verified. If successful, the blob is linked
|
|
// into the blob store and 201 Created is returned with the canonical
|
|
// url of the blob.
|
|
func (buh *blobUploadHandler) PutBlobUploadComplete(w http.ResponseWriter, r *http.Request) {
|
|
if buh.Upload == nil {
|
|
buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadUnknown)
|
|
return
|
|
}
|
|
|
|
dgstStr := r.FormValue("digest") // TODO(stevvooe): Support multiple digest parameters!
|
|
|
|
if dgstStr == "" {
|
|
// no digest? return error, but allow retry.
|
|
buh.Errors = append(buh.Errors, v2.ErrorCodeDigestInvalid.WithDetail("digest missing"))
|
|
return
|
|
}
|
|
|
|
dgst, err := digest.Parse(dgstStr)
|
|
if err != nil {
|
|
// no digest? return error, but allow retry.
|
|
buh.Errors = append(buh.Errors, v2.ErrorCodeDigestInvalid.WithDetail("digest parsing failed"))
|
|
return
|
|
}
|
|
|
|
if err := copyFullPayload(w, r, buh.Upload, buh, "blob PUT", &buh.Errors); err != nil {
|
|
// copyFullPayload reports the error if necessary
|
|
return
|
|
}
|
|
|
|
desc, err := buh.Upload.Commit(buh, distribution.Descriptor{
|
|
Digest: dgst,
|
|
|
|
// TODO(stevvooe): This isn't wildly important yet, but we should
|
|
// really set the mediatype. For now, we can let the backend take care
|
|
// of this.
|
|
})
|
|
|
|
if err != nil {
|
|
switch err := err.(type) {
|
|
case distribution.ErrBlobInvalidDigest:
|
|
buh.Errors = append(buh.Errors, v2.ErrorCodeDigestInvalid.WithDetail(err))
|
|
case errcode.Error:
|
|
buh.Errors = append(buh.Errors, err)
|
|
default:
|
|
switch err {
|
|
case distribution.ErrAccessDenied:
|
|
buh.Errors = append(buh.Errors, errcode.ErrorCodeDenied)
|
|
case distribution.ErrUnsupported:
|
|
buh.Errors = append(buh.Errors, errcode.ErrorCodeUnsupported)
|
|
case distribution.ErrBlobInvalidLength, distribution.ErrBlobDigestUnsupported:
|
|
buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadInvalid.WithDetail(err))
|
|
default:
|
|
ctxu.GetLogger(buh).Errorf("unknown error completing upload: %v", err)
|
|
buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
|
|
}
|
|
|
|
}
|
|
|
|
// Clean up the backend blob data if there was an error.
|
|
if err := buh.Upload.Cancel(buh); err != nil {
|
|
// If the cleanup fails, all we can do is observe and report.
|
|
ctxu.GetLogger(buh).Errorf("error canceling upload after error: %v", err)
|
|
}
|
|
|
|
return
|
|
}
|
|
if err := buh.writeBlobCreatedHeaders(w, desc); err != nil {
|
|
buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
|
|
return
|
|
}
|
|
}
|
|
|
|
// CancelBlobUpload cancels an in-progress upload of a blob.
|
|
func (buh *blobUploadHandler) CancelBlobUpload(w http.ResponseWriter, r *http.Request) {
|
|
if buh.Upload == nil {
|
|
buh.Errors = append(buh.Errors, v2.ErrorCodeBlobUploadUnknown)
|
|
return
|
|
}
|
|
|
|
w.Header().Set("Docker-Upload-UUID", buh.UUID)
|
|
if err := buh.Upload.Cancel(buh); err != nil {
|
|
ctxu.GetLogger(buh).Errorf("error encountered canceling upload: %v", err)
|
|
buh.Errors = append(buh.Errors, errcode.ErrorCodeUnknown.WithDetail(err))
|
|
}
|
|
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
// blobUploadResponse provides a standard request for uploading blobs and
|
|
// chunk responses. This sets the correct headers but the response status is
|
|
// left to the caller. The fresh argument is used to ensure that new blob
|
|
// uploads always start at a 0 offset. This allows disabling resumable push by
|
|
// always returning a 0 offset on check status.
|
|
func (buh *blobUploadHandler) blobUploadResponse(w http.ResponseWriter, r *http.Request, fresh bool) error {
|
|
// TODO(stevvooe): Need a better way to manage the upload state automatically.
|
|
buh.State.Name = buh.Repository.Named().Name()
|
|
buh.State.UUID = buh.Upload.ID()
|
|
buh.Upload.Close()
|
|
buh.State.Offset = buh.Upload.Size()
|
|
buh.State.StartedAt = buh.Upload.StartedAt()
|
|
|
|
token, err := hmacKey(buh.Config.HTTP.Secret).packUploadState(buh.State)
|
|
if err != nil {
|
|
ctxu.GetLogger(buh).Infof("error building upload state token: %s", err)
|
|
return err
|
|
}
|
|
|
|
uploadURL, err := buh.urlBuilder.BuildBlobUploadChunkURL(
|
|
buh.Repository.Named(), buh.Upload.ID(),
|
|
url.Values{
|
|
"_state": []string{token},
|
|
})
|
|
if err != nil {
|
|
ctxu.GetLogger(buh).Infof("error building upload url: %s", err)
|
|
return err
|
|
}
|
|
|
|
endRange := buh.Upload.Size()
|
|
if endRange > 0 {
|
|
endRange = endRange - 1
|
|
}
|
|
|
|
w.Header().Set("Docker-Upload-UUID", buh.UUID)
|
|
w.Header().Set("Location", uploadURL)
|
|
|
|
w.Header().Set("Content-Length", "0")
|
|
w.Header().Set("Range", fmt.Sprintf("0-%d", endRange))
|
|
|
|
return nil
|
|
}
|
|
|
|
// mountBlob attempts to mount a blob from another repository by its digest. If
|
|
// successful, the blob is linked into the blob store and 201 Created is
|
|
// returned with the canonical url of the blob.
|
|
func (buh *blobUploadHandler) createBlobMountOption(fromRepo, mountDigest string) (distribution.BlobCreateOption, error) {
|
|
dgst, err := digest.Parse(mountDigest)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
ref, err := reference.WithName(fromRepo)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
canonical, err := reference.WithDigest(ref, dgst)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return storage.WithMountFrom(canonical), nil
|
|
}
|
|
|
|
// writeBlobCreatedHeaders writes the standard headers describing a newly
|
|
// created blob. A 201 Created is written as well as the canonical URL and
|
|
// blob digest.
|
|
func (buh *blobUploadHandler) writeBlobCreatedHeaders(w http.ResponseWriter, desc distribution.Descriptor) error {
|
|
ref, err := reference.WithDigest(buh.Repository.Named(), desc.Digest)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
blobURL, err := buh.urlBuilder.BuildBlobURL(ref)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
w.Header().Set("Location", blobURL)
|
|
w.Header().Set("Content-Length", "0")
|
|
w.Header().Set("Docker-Content-Digest", desc.Digest.String())
|
|
w.WriteHeader(http.StatusCreated)
|
|
return nil
|
|
}
|