Merge pull request #50 from leoleoasd/suport-chunked-upload

Support for chunk upload signatures

chunk.go

@@ -0,0 +1,71 @@
+package gofakes3
+
+import (
+	"fmt"
+	"io"
+	"io/ioutil"
+)
+
+type chunkedReader struct {
+	inner         io.Reader
+	chunkRemain   int
+	notFirstChunk bool
+}
+
+func newChunkedReader(inner io.Reader) *chunkedReader {
+	return &chunkedReader{
+		inner:         inner,
+		chunkRemain:   0,
+		notFirstChunk: false,
+	}
+}
+
+func (r *chunkedReader) Read(p []byte) (n int, err error) {
+	sizeToRead := len(p)
+	for sizeToRead > 0 {
+		if r.chunkRemain > sizeToRead {
+			r.chunkRemain -= sizeToRead
+			// read sizeToRead bytes from inner reader
+			// to p, start from n.
+			// n is bytes already read.
+			innerN, err := r.inner.Read(p[n : n+sizeToRead])
+			sizeToRead -= innerN
+			n += innerN
+			if err != nil {
+				return n, err
+			}
+		} else if r.chunkRemain > 0 {
+			// read until this chunk ends
+			innerN, err := r.inner.Read(p[n : n+r.chunkRemain])
+			r.chunkRemain -= innerN
+			n += innerN
+			sizeToRead -= innerN
+			if err != nil {
+				return n, err
+			}
+		} else {
+			if !r.notFirstChunk {
+				// Is first chunk.
+				r.notFirstChunk = true
+			} else {
+				// skip last chunk's b"\r\n"
+				_, err = io.CopyN(ioutil.Discard, r.inner, 2)
+				if err != nil {
+					return n, err
+				}
+			}
+			// read next chunk header
+			chunkSize := 0
+			_, err = fmt.Fscanf(r.inner, "%x;", &chunkSize)
+			if err != nil {
+				return n, err
+			}
+			r.chunkRemain = chunkSize
+			_, err = io.CopyN(ioutil.Discard, r.inner, 16+64+2) // "chunk-signature=" + sizeOfHash + "\r\n"
+			if err != nil {
+				return n, err
+			}
+		}
+	}
+	return n, nil
+}

chunk_test.go

@@ -0,0 +1,116 @@
+package gofakes3
+
+import (
+	"errors"
+	"github.com/stretchr/testify/assert"
+	"io"
+	"io/ioutil"
+	"strings"
+	"testing"
+)
+
+func TestChunkedUploadSuccess(t *testing.T) {
+	// From https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-streaming.html#example-signature-calculations-streaming
+	// actual data is (65536 + 1024) * 'a'
+	// divided into 3 chunks.
+
+	// first chunk
+	payload := "10000;chunk-signature=ad80c730a21e5b8d04586a2213dd63b9a0e99e0e2307b0ade35a65485a288648\r\n"
+	payload += strings.Repeat("a", 65536)
+	payload += "\r\n"
+
+	// second chunk
+	payload += "400;chunk-signature=0055627c9e194cb4542bae2aa5492e3c1575bbb81b612b7d234b86a503ef5497\r\n"
+	payload += strings.Repeat("a", 1024)
+	payload += "\r\n"
+
+	// third chunk, with empty chunk-data representing end of request
+	payload += "0;chunk-signature=b6c6ea8a5354eaf15b3cb7646744f4275b71ea724fed81ceb9323e279d449df9\r\n\r\n"
+
+	inner := strings.NewReader(payload)
+	chunkedReader := newChunkedReader(inner)
+	buf, err := ioutil.ReadAll(chunkedReader)
+	assert.Equal(t, nil, err)
+	assert.Equal(t, string(buf), strings.Repeat("a", 65536+1024))
+}
+
+type errReader struct{}
+
+func (errReader) Read(p []byte) (n int, err error) {
+	return 0, errors.New("err")
+}
+
+func TestChunkedUploadFail(t *testing.T) {
+	chunkedReader := newChunkedReader(errReader{})
+	buf, err := ioutil.ReadAll(chunkedReader)
+	assert.Equal(t, errors.New("err"), err)
+	assert.Equal(t, "", string(buf))
+
+	chunkedReader = newChunkedReader(io.MultiReader(
+		strings.NewReader("10000;chunk-signature=ad80c730a21e5b8d04586a2213dd63b9a0e99e0e2307b0ade35a65485a288648\r\n"),
+		errReader{},
+	))
+	buf, err = ioutil.ReadAll(chunkedReader)
+	assert.Equal(t, errors.New("err"), err)
+	assert.Equal(t, "", string(buf))
+
+	chunkedReader = newChunkedReader(io.MultiReader(
+		strings.NewReader("incorrect_data"),
+		errReader{},
+	))
+	buf, err = ioutil.ReadAll(chunkedReader)
+	assert.Equal(t, errors.New("expected integer"), err)
+	assert.Equal(t, "", string(buf))
+
+	payload := "10000;chunk-signature=ad80c730a21e5b8d04586a2213dd63b9a0e99e0e2307b0ade35a65485a288648\r\n"
+	payload += strings.Repeat("a", 200)
+	chunkedReader = newChunkedReader(io.MultiReader(
+		strings.NewReader(payload),
+		errReader{},
+	))
+	buf, err = ioutil.ReadAll(chunkedReader)
+	assert.Equal(t, errors.New("err"), err)
+	assert.Equal(t, strings.Repeat("a", 200), string(buf))
+
+	payload = "10000;chunk-signature=ad80c730a21e5b8d04586a2213dd63b9a0e99e0e2307b0ade35a65485a288648\r\n"
+	payload += strings.Repeat("a", 1024+100)
+	chunkedReader = newChunkedReader(io.MultiReader(
+		strings.NewReader(payload),
+		errReader{},
+	))
+	buf = make([]byte, 1024)
+	n, err := chunkedReader.Read(buf)
+	assert.Equal(t, nil, err)
+	assert.Equal(t, strings.Repeat("a", 1024), string(buf[:n]))
+	assert.Equal(t, 1024, n)
+
+	buf = make([]byte, 65536)
+	n, err = chunkedReader.Read(buf)
+	assert.Equal(t, errors.New("err"), err)
+	assert.Equal(t, strings.Repeat("a", 100), string(buf[:n]))
+	assert.Equal(t, 100, n)
+
+	payload = "10000;chunk-signature=ad80c730a21e5b8d04586a2213dd63b9a0e99e0e2307b0ade35a65485a288648\r\n"
+	payload += strings.Repeat("a", 65536)
+	chunkedReader = newChunkedReader(io.MultiReader(
+		strings.NewReader(payload),
+		errReader{},
+	))
+	buf = make([]byte, 65536+1024)
+	n, err = io.ReadFull(chunkedReader, buf)
+	assert.Equal(t, errors.New("err"), err)
+	assert.Equal(t, strings.Repeat("a", 65536), string(buf[:n]))
+	assert.Equal(t, 65536, n)
+
+	payload = "10000;chunk-signature=ad80c"
+	chunkedReader = newChunkedReader(io.MultiReader(
+		strings.NewReader(payload),
+		errReader{},
+	))
+	buf = make([]byte, 65536+1024)
+	n, err = io.ReadFull(chunkedReader, buf)
+	assert.Equal(t, errors.New("err"), err)
+	assert.Equal(t, "", string(buf[:n]))
+	assert.Equal(t, 0, n)
+
+}

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46
 	github.com/shabbyrobe/gocovmerge v0.0.0-20180507124511-f6ea450bfb63
 	github.com/spf13/afero v1.2.1
-	github.com/stretchr/testify v1.3.0 // indirect
+	github.com/stretchr/testify v1.3.0
 	golang.org/x/net v0.0.0-20190310074541-c10a0554eabf // indirect
 	golang.org/x/sys v0.0.0-20190310054646-10058d7d4faa // indirect
 	golang.org/x/tools v0.0.0-20190308174544-00c44ba9c14f

gofakes3.go

@@ -23,7 +23,7 @@ import (
 //
 // Logic is delegated to other components, like Backend or uploader.
 type GoFakeS3 struct {
-	requestID               uint64
+	requestID uint64
 
 	storage   Backend
 	versioned VersionedBackend
@@ -568,9 +568,22 @@ func (g *GoFakeS3) createObject(bucket, object string, w http.ResponseWriter, r
 		}
 	}
 
+	var reader io.Reader
+
+	if sha, ok := meta["X-Amz-Content-Sha256"]; ok && sha == "STREAMING-AWS4-HMAC-SHA256-PAYLOAD" {
+		reader = newChunkedReader(r.Body)
+		size, err = strconv.ParseInt(meta["X-Amz-Decoded-Content-Length"], 10, 64)
+		if err != nil {
+			w.WriteHeader(http.StatusBadRequest) // XXX: no code for this, according to s3tests
+			return nil
+		}
+	} else {
+		reader = r.Body
+	}
+
 	// hashingReader is still needed to get the ETag even if integrityCheck
 	// is set to false:
-	rdr, err := newHashingReader(r.Body, md5Base64)
+	rdr, err := newHashingReader(reader, md5Base64)
 	defer r.Body.Close()
 	if err != nil {
 		return err