Merge pull request #53 from stevvooe/spool-uploads-remotely

Spool uploads remotely

docs/app.go

@@ -29,8 +29,6 @@ type App struct {
 	// services contains the main services instance for the application.
 	services *storage.Services
 
-	tokenProvider tokenProvider
-
 	layerHandler storage.LayerHandler
 
 	accessController auth.AccessController
@@ -66,8 +64,6 @@ func NewApp(configuration configuration.Configuration) *App {
 
 	app.driver = driver
 	app.services = storage.NewServices(app.driver)
-	app.tokenProvider = newHMACTokenProvider(configuration.HTTP.Secret)
-
 	authType := configuration.Auth.Type()
 
 	if authType != "" {

docs/hmac.go

@@ -0,0 +1,72 @@
+package registry
+
+import (
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"time"
+)
+
+// layerUploadState captures the state serializable state of the layer upload.
+type layerUploadState struct {
+	// name is the primary repository under which the layer will be linked.
+	Name string
+
+	// UUID identifies the upload.
+	UUID string
+
+	// offset contains the current progress of the upload.
+	Offset int64
+
+	// StartedAt is the original start time of the upload.
+	StartedAt time.Time
+}
+
+type hmacKey string
+
+// unpackUploadState unpacks and validates the layer upload state from the
+// token, using the hmacKey secret.
+func (secret hmacKey) unpackUploadState(token string) (layerUploadState, error) {
+	var state layerUploadState
+
+	tokenBytes, err := base64.URLEncoding.DecodeString(token)
+	if err != nil {
+		return state, err
+	}
+	mac := hmac.New(sha256.New, []byte(secret))
+
+	if len(tokenBytes) < mac.Size() {
+		return state, fmt.Errorf("Invalid token")
+	}
+
+	macBytes := tokenBytes[:mac.Size()]
+	messageBytes := tokenBytes[mac.Size():]
+
+	mac.Write(messageBytes)
+	if !hmac.Equal(mac.Sum(nil), macBytes) {
+		return state, fmt.Errorf("Invalid token")
+	}
+
+	if err := json.Unmarshal(messageBytes, &state); err != nil {
+		return state, err
+	}
+
+	return state, nil
+}
+
+// packUploadState packs the upload state signed with and hmac digest using
+// the hmacKey secret, encoding to url safe base64. The resulting token can be
+// used to share data with minimized risk of external tampering.
+func (secret hmacKey) packUploadState(lus layerUploadState) (string, error) {
+	mac := hmac.New(sha256.New, []byte(secret))
+	p, err := json.Marshal(lus)
+	if err != nil {
+		return "", err
+	}
+
+	mac.Write(p)
+
+	return base64.URLEncoding.EncodeToString(append(mac.Sum(nil), p...)), nil
+}

docs/hmac_test.go

@@ -1,12 +1,8 @@
 package registry
 
-import (
+import "testing"
-	"testing"
 
-	"github.com/docker/distribution/storage"
+var layerUploadStates = []layerUploadState{
-)
-
-var layerUploadStates = []storage.LayerUploadState{
 	{
 		Name:   "hello",
 		UUID:   "abcd-1234-qwer-0987",
@@ -47,15 +43,15 @@ var secrets = []string{
 // TestLayerUploadTokens constructs stateTokens from LayerUploadStates and
 // validates that the tokens can be used to reconstruct the proper upload state.
 func TestLayerUploadTokens(t *testing.T) {
-	tokenProvider := newHMACTokenProvider("supersecret")
+	secret := hmacKey("supersecret")
 
 	for _, testcase := range layerUploadStates {
-		token, err := tokenProvider.layerUploadStateToToken(testcase)
+		token, err := secret.packUploadState(testcase)
 		if err != nil {
 			t.Fatal(err)
 		}
 
-		lus, err := tokenProvider.layerUploadStateFromToken(token)
+		lus, err := secret.unpackUploadState(token)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -68,39 +64,39 @@ func TestLayerUploadTokens(t *testing.T) {
 // only if they share the same secret.
 func TestHMACValidation(t *testing.T) {
 	for _, secret := range secrets {
-		tokenProvider1 := newHMACTokenProvider(secret)
+		secret1 := hmacKey(secret)
-		tokenProvider2 := newHMACTokenProvider(secret)
+		secret2 := hmacKey(secret)
-		badTokenProvider := newHMACTokenProvider("DifferentSecret")
+		badSecret := hmacKey("DifferentSecret")
 
 		for _, testcase := range layerUploadStates {
-			token, err := tokenProvider1.layerUploadStateToToken(testcase)
+			token, err := secret1.packUploadState(testcase)
 			if err != nil {
 				t.Fatal(err)
 			}
 
-			lus, err := tokenProvider2.layerUploadStateFromToken(token)
+			lus, err := secret2.unpackUploadState(token)
 			if err != nil {
 				t.Fatal(err)
 			}
 
 			assertLayerUploadStateEquals(t, testcase, lus)
 
-			_, err = badTokenProvider.layerUploadStateFromToken(token)
+			_, err = badSecret.unpackUploadState(token)
 			if err == nil {
 				t.Fatalf("Expected token provider to fail at retrieving state from token: %s", token)
 			}
 
-			badToken, err := badTokenProvider.layerUploadStateToToken(testcase)
+			badToken, err := badSecret.packUploadState(lus)
 			if err != nil {
 				t.Fatal(err)
 			}
 
-			_, err = tokenProvider1.layerUploadStateFromToken(badToken)
+			_, err = secret1.unpackUploadState(badToken)
 			if err == nil {
 				t.Fatalf("Expected token provider to fail at retrieving state from token: %s", badToken)
 			}
 
-			_, err = tokenProvider2.layerUploadStateFromToken(badToken)
+			_, err = secret2.unpackUploadState(badToken)
 			if err == nil {
 				t.Fatalf("Expected token provider to fail at retrieving state from token: %s", badToken)
 			}
@@ -108,7 +104,7 @@ func TestHMACValidation(t *testing.T) {
 	}
 }
 
-func assertLayerUploadStateEquals(t *testing.T, expected storage.LayerUploadState, received storage.LayerUploadState) {
+func assertLayerUploadStateEquals(t *testing.T, expected layerUploadState, received layerUploadState) {
 	if expected.Name != received.Name {
 		t.Fatalf("Expected Name=%q, Received Name=%q", expected.Name, received.Name)
 	}

docs/layerupload.go

@@ -5,7 +5,7 @@ import (
 	"io"
 	"net/http"
 	"net/url"
-	"strconv"
+	"os"
 
 	"github.com/Sirupsen/logrus"
 	"github.com/docker/distribution/api/v2"
@@ -33,26 +33,57 @@ func layerUploadDispatcher(ctx *Context, r *http.Request) http.Handler {
 	if luh.UUID != "" {
 		luh.log = luh.log.WithField("uuid", luh.UUID)
 
-		state, err := ctx.tokenProvider.layerUploadStateFromToken(r.FormValue("_state"))
+		state, err := hmacKey(ctx.Config.HTTP.Secret).unpackUploadState(r.FormValue("_state"))
 		if err != nil {
 			return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-				logrus.Infof("error resolving upload: %v", err)
+				ctx.log.Infof("error resolving upload: %v", err)
-				w.WriteHeader(http.StatusInternalServerError)
+				w.WriteHeader(http.StatusBadRequest)
-				luh.Errors.Push(v2.ErrorCodeUnknown, err)
+				luh.Errors.Push(v2.ErrorCodeBlobUploadInvalid, err)
+			})
+		}
+		luh.State = state
+
+		if state.UUID != luh.UUID {
+			return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				ctx.log.Infof("mismatched uuid in upload state: %q != %q", state.UUID, luh.UUID)
+				w.WriteHeader(http.StatusBadRequest)
+				luh.Errors.Push(v2.ErrorCodeBlobUploadInvalid, err)
 			})
 		}
 
 		layers := ctx.services.Layers()
-		upload, err := layers.Resume(state)
+		upload, err := layers.Resume(luh.Name, luh.UUID)
 		if err != nil && err != storage.ErrLayerUploadUnknown {
 			return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-				logrus.Infof("error resolving upload: %v", err)
+				ctx.log.Errorf("error resolving upload: %v", err)
-				w.WriteHeader(http.StatusInternalServerError)
+				w.WriteHeader(http.StatusBadRequest)
-				luh.Errors.Push(v2.ErrorCodeUnknown, err)
+				luh.Errors.Push(v2.ErrorCodeBlobUploadUnknown, err)
 			})
 		}
-
 		luh.Upload = upload
+
+		if state.Offset > 0 {
+			// Seek the layer upload to the correct spot if it's non-zero.
+			// These error conditions should be rare and demonstrate really
+			// problems. We basically cancel the upload and tell the client to
+			// start over.
+			if nn, err := upload.Seek(luh.State.Offset, os.SEEK_SET); err != nil {
+				ctx.log.Infof("error seeking layer upload: %v", err)
+				return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					w.WriteHeader(http.StatusBadRequest)
+					luh.Errors.Push(v2.ErrorCodeBlobUploadInvalid, err)
+					upload.Cancel()
+				})
+			} else if nn != luh.State.Offset {
+				ctx.log.Infof("seek to wrong offest: %d != %d", nn, luh.State.Offset)
+				return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					w.WriteHeader(http.StatusBadRequest)
+					luh.Errors.Push(v2.ErrorCodeBlobUploadInvalid, err)
+					upload.Cancel()
+				})
+			}
+		}
+
 		handler = closeResources(handler, luh.Upload)
 	}
 
@@ -67,6 +98,8 @@ type layerUploadHandler struct {
 	UUID string
 
 	Upload storage.LayerUpload
+
+	State layerUploadState
 }
 
 // StartLayerUpload begins the layer upload process and allocates a server-
@@ -171,14 +204,30 @@ func (luh *layerUploadHandler) CancelLayerUpload(w http.ResponseWriter, r *http.
 // chunk responses. This sets the correct headers but the response status is
 // left to the caller.
 func (luh *layerUploadHandler) layerUploadResponse(w http.ResponseWriter, r *http.Request) error {
-	values := make(url.Values)
+
-	stateToken, err := luh.Context.tokenProvider.layerUploadStateToToken(storage.LayerUploadState{Name: luh.Upload.Name(), UUID: luh.Upload.UUID(), Offset: luh.Upload.Offset()})
+	offset, err := luh.Upload.Seek(0, os.SEEK_CUR)
+	if err != nil {
+		luh.log.Errorf("unable get current offset of layer upload: %v", err)
+		return err
+	}
+
+	// TODO(stevvooe): Need a better way to manage the upload state automatically.
+	luh.State.Name = luh.Name
+	luh.State.UUID = luh.Upload.UUID()
+	luh.State.Offset = offset
+	luh.State.StartedAt = luh.Upload.StartedAt()
+
+	token, err := hmacKey(luh.Config.HTTP.Secret).packUploadState(luh.State)
 	if err != nil {
 		logrus.Infof("error building upload state token: %s", err)
 		return err
 	}
-	values.Set("_state", stateToken)
+
-	uploadURL, err := luh.urlBuilder.BuildBlobUploadChunkURL(luh.Upload.Name(), luh.Upload.UUID(), values)
+	uploadURL, err := luh.urlBuilder.BuildBlobUploadChunkURL(
+		luh.Upload.Name(), luh.Upload.UUID(),
+		url.Values{
+			"_state": []string{token},
+		})
 	if err != nil {
 		logrus.Infof("error building upload url: %s", err)
 		return err
@@ -186,7 +235,7 @@ func (luh *layerUploadHandler) layerUploadResponse(w http.ResponseWriter, r *htt
 
 	w.Header().Set("Location", uploadURL)
 	w.Header().Set("Content-Length", "0")
-	w.Header().Set("Range", fmt.Sprintf("0-%d", luh.Upload.Offset()))
+	w.Header().Set("Range", fmt.Sprintf("0-%d", luh.State.Offset))
 
 	return nil
 }
@@ -198,7 +247,6 @@ var errNotReadyToComplete = fmt.Errorf("not ready to complete upload")
 func (luh *layerUploadHandler) maybeCompleteUpload(w http.ResponseWriter, r *http.Request) error {
 	// If we get a digest and length, we can finish the upload.
 	dgstStr := r.FormValue("digest") // TODO(stevvooe): Support multiple digest parameters!
-	sizeStr := r.FormValue("size")
 
 	if dgstStr == "" {
 		return errNotReadyToComplete
@@ -209,23 +257,13 @@ func (luh *layerUploadHandler) maybeCompleteUpload(w http.ResponseWriter, r *htt
 		return err
 	}
 
-	var size int64
+	luh.completeUpload(w, r, dgst)
-	if sizeStr != "" {
-		size, err = strconv.ParseInt(sizeStr, 10, 64)
-		if err != nil {
-			return err
-		}
-	} else {
-		size = -1
-	}
-
-	luh.completeUpload(w, r, size, dgst)
 	return nil
 }
 
 // completeUpload finishes out the upload with the correct response.
-func (luh *layerUploadHandler) completeUpload(w http.ResponseWriter, r *http.Request, size int64, dgst digest.Digest) {
+func (luh *layerUploadHandler) completeUpload(w http.ResponseWriter, r *http.Request, dgst digest.Digest) {
-	layer, err := luh.Upload.Finish(size, dgst)
+	layer, err := luh.Upload.Finish(dgst)
 	if err != nil {
 		luh.Errors.Push(v2.ErrorCodeUnknown, err)
 		w.WriteHeader(http.StatusInternalServerError)

docs/tokens.go

@@ -1,65 +0,0 @@
-package registry
-
-import (
-	"crypto/hmac"
-	"crypto/sha256"
-	"encoding/base64"
-	"encoding/json"
-	"fmt"
-
-	"github.com/docker/distribution/storage"
-)
-
-// tokenProvider contains methods for serializing and deserializing state from token strings.
-type tokenProvider interface {
-	// layerUploadStateFromToken retrieves the LayerUploadState for a given state token.
-	layerUploadStateFromToken(stateToken string) (storage.LayerUploadState, error)
-
-	// layerUploadStateToToken returns a token string representing the given LayerUploadState.
-	layerUploadStateToToken(layerUploadState storage.LayerUploadState) (string, error)
-}
-
-type hmacTokenProvider struct {
-	secret string
-}
-
-func newHMACTokenProvider(secret string) tokenProvider {
-	return &hmacTokenProvider{secret: secret}
-}
-
-// layerUploadStateFromToken deserializes the given HMAC stateToken and validates the prefix HMAC
-func (ts *hmacTokenProvider) layerUploadStateFromToken(stateToken string) (storage.LayerUploadState, error) {
-	var lus storage.LayerUploadState
-
-	tokenBytes, err := base64.URLEncoding.DecodeString(stateToken)
-	if err != nil {
-		return lus, err
-	}
-	mac := hmac.New(sha256.New, []byte(ts.secret))
-
-	if len(tokenBytes) < mac.Size() {
-		return lus, fmt.Errorf("Invalid token")
-	}
-
-	macBytes := tokenBytes[:mac.Size()]
-	messageBytes := tokenBytes[mac.Size():]
-
-	mac.Write(messageBytes)
-	if !hmac.Equal(mac.Sum(nil), macBytes) {
-		return lus, fmt.Errorf("Invalid token")
-	}
-
-	if err := json.Unmarshal(messageBytes, &lus); err != nil {
-		return lus, err
-	}
-
-	return lus, nil
-}
-
-// layerUploadStateToToken serializes the given LayerUploadState to JSON with an HMAC prepended
-func (ts *hmacTokenProvider) layerUploadStateToToken(lus storage.LayerUploadState) (string, error) {
-	mac := hmac.New(sha256.New, []byte(ts.secret))
-	stateJSON := fmt.Sprintf("{\"Name\": \"%s\", \"UUID\": \"%s\", \"Offset\": %d}", lus.Name, lus.UUID, lus.Offset)
-	mac.Write([]byte(stateJSON))
-	return base64.URLEncoding.EncodeToString(append(mac.Sum(nil), stateJSON...)), nil
-}