api/handlers: Implement streaming signature v4 support. (#2370)

* api/handlers: Implement streaming signature v4 support.

Fixes #2326

* tests: Add tests for quick/safe

CONTRIBUTING.md

@@ -51,7 +51,7 @@ Building Libraries
     - Push to the branch (git push origin my-new-feature)
     - Create new Pull Request
 
-* If you have additional dependencies for ``Minio``, ``Minio`` manages its depedencies using [govendor](https://github.com/kardianos/govendor)
+* If you have additional dependencies for ``Minio``, ``Minio`` manages its dependencies using [govendor](https://github.com/kardianos/govendor)
     - Run `go get foo/bar`
     - Edit your code to import foo/bar
     - Run `make pkg-add PKG=foo/bar` from top-level directory

auth-handler.go

@@ -34,40 +34,28 @@ func isRequestUnsignedPayload(r *http.Request) bool {
 
 // Verify if request has JWT.
 func isRequestJWT(r *http.Request) bool {
-	if _, ok := r.Header["Authorization"]; ok {
-		if strings.HasPrefix(r.Header.Get("Authorization"), jwtAlgorithm) {
-			return true
-		}
-	}
-	return false
+	return strings.HasPrefix(r.Header.Get("Authorization"), jwtAlgorithm)
 }
 
 // Verify if request has AWS Signature Version '4'.
 func isRequestSignatureV4(r *http.Request) bool {
-	if _, ok := r.Header["Authorization"]; ok {
-		if strings.HasPrefix(r.Header.Get("Authorization"), signV4Algorithm) {
-			return true
-		}
-	}
-	return false
+	return strings.HasPrefix(r.Header.Get("Authorization"), signV4Algorithm)
 }
 
-// Verify if request has AWS Presignature Version '4'.
+// Verify if request has AWS PreSign Version '4'.
 func isRequestPresignedSignatureV4(r *http.Request) bool {
-	if _, ok := r.URL.Query()["X-Amz-Credential"]; ok {
-		return true
-	}
-	return false
+	_, ok := r.URL.Query()["X-Amz-Credential"]
+	return ok
 }
 
 // Verify if request has AWS Post policy Signature Version '4'.
 func isRequestPostPolicySignatureV4(r *http.Request) bool {
-	if _, ok := r.Header["Content-Type"]; ok {
-		if strings.Contains(r.Header.Get("Content-Type"), "multipart/form-data") && r.Method == "POST" {
-			return true
-		}
-	}
-	return false
+	return strings.Contains(r.Header.Get("Content-Type"), "multipart/form-data") && r.Method == "POST"
+}
+
+// Verify if the request has AWS Streaming Signature Version '4'. This is only valid for 'PUT' operation.
+func isRequestSignStreamingV4(r *http.Request) bool {
+	return r.Header.Get("x-amz-content-sha256") == streamingContentSHA256 && r.Method == "PUT"
 }
 
 // Authorization type.
@@ -79,13 +67,16 @@ const (
 	authTypeAnonymous
 	authTypePresigned
 	authTypePostPolicy
+	authTypeStreamingSigned
 	authTypeSigned
 	authTypeJWT
 )
 
 // Get request authentication type.
 func getRequestAuthType(r *http.Request) authType {
-	if isRequestSignatureV4(r) {
+	if isRequestSignStreamingV4(r) {
+		return authTypeStreamingSigned
+	} else if isRequestSignatureV4(r) {
 		return authTypeSigned
 	} else if isRequestPresignedSignatureV4(r) {
 		return authTypePresigned
@@ -154,8 +145,8 @@ func isReqAuthenticated(r *http.Request) (s3Error APIErrorCode) {
 // request headers and body are used to calculate the signature validating
 // the client signature present in request.
 func checkAuth(r *http.Request) APIErrorCode {
-	authType := getRequestAuthType(r)
-	if authType != authTypePresigned && authType != authTypeSigned {
+	aType := getRequestAuthType(r)
+	if aType != authTypePresigned && aType != authTypeSigned {
 		// For all unhandled auth types return error AccessDenied.
 		return ErrAccessDenied
 	}
@@ -173,23 +164,40 @@ func setAuthHandler(h http.Handler) http.Handler {
 	return authHandler{h}
 }
 
+// List of all support S3 auth types.
+var supportedS3AuthTypes = []authType{
+	authTypeAnonymous,
+	authTypePresigned,
+	authTypeSigned,
+	authTypePostPolicy,
+	authTypeStreamingSigned,
+}
+
+// Validate if the authType is valid and supported.
+func isSupportedS3AuthType(aType authType) bool {
+	for _, a := range supportedS3AuthTypes {
+		if a == aType {
+			return true
+		}
+	}
+	return false
+}
+
 // handler for validating incoming authorization headers.
 func (a authHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	switch getRequestAuthType(r) {
-	case authTypeAnonymous, authTypePresigned, authTypeSigned, authTypePostPolicy:
-		// Let top level caller validate for anonymous and known
-		// signed requests.
+	aType := getRequestAuthType(r)
+	if isSupportedS3AuthType(aType) {
+		// Let top level caller validate for anonymous and known signed requests.
 		a.handler.ServeHTTP(w, r)
 		return
-	case authTypeJWT:
+	} else if aType == authTypeJWT {
 		// Validate Authorization header if its valid for JWT request.
 		if !isJWTReqAuthenticated(r) {
 			w.WriteHeader(http.StatusUnauthorized)
 			return
 		}
 		a.handler.ServeHTTP(w, r)
-	default:
-		writeErrorResponse(w, r, ErrSignatureVersionNotSupported, r.URL.Path)
 		return
 	}
+	writeErrorResponse(w, r, ErrSignatureVersionNotSupported, r.URL.Path)
 }

auth-handler_test.go

@@ -23,6 +23,64 @@ import (
 	"testing"
 )
 
+// Test all s3 supported auth types.
+func TestS3SupportedAuthType(t *testing.T) {
+	type testCase struct {
+		authT authType
+		pass  bool
+	}
+	// List of all valid and invalid test cases.
+	testCases := []testCase{
+		// Test 1 - supported s3 type anonymous.
+		{
+			authT: authTypeAnonymous,
+			pass:  true,
+		},
+		// Test 2 - supported s3 type presigned.
+		{
+			authT: authTypePresigned,
+			pass:  true,
+		},
+		// Test 3 - supported s3 type signed.
+		{
+			authT: authTypeSigned,
+			pass:  true,
+		},
+		// Test 4 - supported s3 type with post policy.
+		{
+			authT: authTypePostPolicy,
+			pass:  true,
+		},
+		// Test 5 - supported s3 type with streaming signed.
+		{
+			authT: authTypeStreamingSigned,
+			pass:  true,
+		},
+		// Test 6 - JWT is not supported s3 type.
+		{
+			authT: authTypeJWT,
+			pass:  false,
+		},
+		// Test 7 - unknown auth header is not supported s3 type.
+		{
+			authT: authTypeUnknown,
+			pass:  false,
+		},
+		// Test 8 - some new auth type is not supported s3 type.
+		{
+			authT: authType(7),
+			pass:  false,
+		},
+	}
+	// Validate all the test cases.
+	for i, tt := range testCases {
+		ok := isSupportedS3AuthType(tt.authT)
+		if ok != tt.pass {
+			t.Errorf("Test %d:, Expected %t, got %t", i+1, tt.pass, ok)
+		}
+	}
+}
+
 // TestIsRequestUnsignedPayload - Test validates the Unsigned payload detection logic.
 func TestIsRequestUnsignedPayload(t *testing.T) {
 	testCases := []struct {

generic-handlers.go

@@ -245,8 +245,8 @@ func (h resourceHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		bucketName = splits[0]
 		objectName = splits[1]
 	}
-	// If bucketName is present and not objectName check for bucket
-	// level resource queries.
+
+	// If bucketName is present and not objectName check for bucket level resource queries.
 	if bucketName != "" && objectName == "" {
 		if ignoreNotImplementedBucketResources(r) {
 			writeErrorResponse(w, r, ErrNotImplemented, r.URL.Path)
@@ -265,6 +265,8 @@ func (h resourceHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		writeErrorResponse(w, r, ErrNotImplemented, r.URL.Path)
 		return
 	}
+
+	// Serve HTTP.
 	h.handler.ServeHTTP(w, r)
 }
 

object-handlers.go

@@ -391,6 +391,16 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
 	}
 	/// if Content-Length is unknown/missing, deny the request
 	size := r.ContentLength
+	rAuthType := getRequestAuthType(r)
+	if rAuthType == authTypeStreamingSigned {
+		sizeStr := r.Header.Get("x-amz-decoded-content-length")
+		size, err = strconv.ParseInt(sizeStr, 10, 64)
+		if err != nil {
+			errorIf(err, "Unable to parse `x-amz-decoded-content-length` into its integer value", sizeStr)
+			writeErrorResponse(w, r, toAPIErrorCode(err), r.URL.Path)
+			return
+		}
+	}
 	if size == -1 && !contains(r.TransferEncoding, "chunked") {
 		writeErrorResponse(w, r, ErrMissingContentLength, r.URL.Path)
 		return
@@ -407,7 +417,7 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
 	metadata["md5Sum"] = hex.EncodeToString(md5Bytes)
 
 	var md5Sum string
-	switch getRequestAuthType(r) {
+	switch rAuthType {
 	default:
 		// For all unknown auth types return error.
 		writeErrorResponse(w, r, ErrAccessDenied, r.URL.Path)
@@ -420,6 +430,14 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
 		}
 		// Create anonymous object.
 		md5Sum, err = api.ObjectAPI.PutObject(bucket, object, size, r.Body, metadata)
+	case authTypeStreamingSigned:
+		// Initialize stream signature verifier.
+		reader, s3Error := newSignV4ChunkedReader(r)
+		if s3Error != ErrNone {
+			writeErrorResponse(w, r, s3Error, r.URL.Path)
+			return
+		}
+		md5Sum, err = api.ObjectAPI.PutObject(bucket, object, size, reader, metadata)
 	case authTypePresigned, authTypeSigned:
 		// Initialize signature verifier.
 		reader := newSignVerify(r)
@@ -516,6 +534,18 @@ func (api objectAPIHandlers) PutObjectPartHandler(w http.ResponseWriter, r *http
 
 	/// if Content-Length is unknown/missing, throw away
 	size := r.ContentLength
+
+	rAuthType := getRequestAuthType(r)
+	// For auth type streaming signature, we need to gather a different content length.
+	if rAuthType == authTypeStreamingSigned {
+		sizeStr := r.Header.Get("x-amz-decoded-content-length")
+		size, err = strconv.ParseInt(sizeStr, 10, 64)
+		if err != nil {
+			errorIf(err, "Unable to parse `x-amz-decoded-content-length` into its integer value", sizeStr)
+			writeErrorResponse(w, r, toAPIErrorCode(err), r.URL.Path)
+			return
+		}
+	}
 	if size == -1 {
 		writeErrorResponse(w, r, ErrMissingContentLength, r.URL.Path)
 		return
@@ -544,7 +574,7 @@ func (api objectAPIHandlers) PutObjectPartHandler(w http.ResponseWriter, r *http
 
 	var partMD5 string
 	incomingMD5 := hex.EncodeToString(md5Bytes)
-	switch getRequestAuthType(r) {
+	switch rAuthType {
 	default:
 		// For all unknown auth types return error.
 		writeErrorResponse(w, r, ErrAccessDenied, r.URL.Path)
@@ -557,6 +587,14 @@ func (api objectAPIHandlers) PutObjectPartHandler(w http.ResponseWriter, r *http
 		}
 		// No need to verify signature, anonymous request access is already allowed.
 		partMD5, err = api.ObjectAPI.PutObjectPart(bucket, object, uploadID, partID, size, r.Body, incomingMD5)
+	case authTypeStreamingSigned:
+		// Initialize stream signature verifier.
+		reader, s3Error := newSignV4ChunkedReader(r)
+		if s3Error != ErrNone {
+			writeErrorResponse(w, r, s3Error, r.URL.Path)
+			return
+		}
+		partMD5, err = api.ObjectAPI.PutObjectPart(bucket, object, uploadID, partID, size, reader, incomingMD5)
 	case authTypePresigned, authTypeSigned:
 		// Initialize signature verifier.
 		reader := newSignVerify(r)

pkg/quick/quick_test.go

@@ -19,6 +19,7 @@
 package quick_test
 
 import (
+	"encoding/json"
 	"os"
 	"testing"
 
@@ -81,6 +82,46 @@ func (s *MySuite) TestCheckData(c *C) {
 	c.Assert(err, IsNil)
 }
 
+func (s *MySuite) TestLoadFile(c *C) {
+	type myStruct struct {
+		Version  string
+		User     string
+		Password string
+		Folders  []string
+	}
+	saveMe := myStruct{}
+	_, err := quick.Load("test.json", &saveMe)
+	c.Assert(err, Not(IsNil))
+
+	file, err := os.Create("test.json")
+	c.Assert(err, IsNil)
+	c.Assert(file.Close(), IsNil)
+	_, err = quick.Load("test.json", &saveMe)
+	c.Assert(err, Not(IsNil))
+	config, err := quick.New(&saveMe)
+	c.Assert(err, IsNil)
+	err = config.Load("test-non-exist.json")
+	c.Assert(err, Not(IsNil))
+	err = config.Load("test.json")
+	c.Assert(err, Not(IsNil))
+
+	saveMe = myStruct{"1", "guest", "nopassword", []string{"Work", "Documents", "Music"}}
+	config, err = quick.New(&saveMe)
+	c.Assert(err, IsNil)
+	c.Assert(config, Not(IsNil))
+	err = config.Save("test.json")
+	c.Assert(err, IsNil)
+	saveMe1 := myStruct{}
+	_, err = quick.Load("test.json", &saveMe1)
+	c.Assert(err, IsNil)
+	c.Assert(saveMe1, DeepEquals, saveMe)
+
+	saveMe2 := myStruct{}
+	err = json.Unmarshal([]byte(config.String()), &saveMe2)
+	c.Assert(err, IsNil)
+	c.Assert(saveMe2, DeepEquals, saveMe1)
+}
+
 func (s *MySuite) TestVersion(c *C) {
 	defer os.RemoveAll("test.json")
 	type myStruct struct {

pkg/safe/safe.go

@@ -35,14 +35,14 @@ type File struct {
 
 // Write writes len(b) bytes to the temporary File.  In case of error, the temporary file is removed.
 func (file *File) Write(b []byte) (n int, err error) {
-	if file.aborted {
-		err = errors.New("write on aborted file")
-		return
-	}
 	if file.closed {
 		err = errors.New("write on closed file")
 		return
 	}
+	if file.aborted {
+		err = errors.New("write on aborted file")
+		return
+	}
 
 	defer func() {
 		if err != nil {
@@ -64,7 +64,12 @@ func (file *File) Close() (err error) {
 		}
 	}()
 
-	if file.aborted || file.closed {
+	if file.closed {
+		err = errors.New("close on closed file")
+		return
+	}
+	if file.aborted {
+		err = errors.New("close on aborted file")
 		return
 	}
 
@@ -80,7 +85,12 @@ func (file *File) Close() (err error) {
 
 // Abort aborts the temporary File by closing and removing the temporary file.
 func (file *File) Abort() (err error) {
-	if file.aborted || file.closed {
+	if file.closed {
+		err = errors.New("abort on closed file")
+		return
+	}
+	if file.aborted {
+		err = errors.New("abort on aborted file")
 		return
 	}
 

pkg/safe/safe_test.go

@@ -19,7 +19,7 @@ package safe
 import (
 	"io/ioutil"
 	"os"
-	"path/filepath"
+	"path"
 	"testing"
 
 	. "gopkg.in/check.v1"
@@ -44,26 +44,60 @@ func (s *MySuite) TearDownSuite(c *C) {
 	c.Assert(err, IsNil)
 }
 
-func (s *MySuite) TestSafe(c *C) {
-	f, err := CreateFile(filepath.Join(s.root, "testfile"))
-	c.Assert(err, IsNil)
-	_, err = os.Stat(filepath.Join(s.root, "testfile"))
-	c.Assert(err, Not(IsNil))
-	err = f.Close()
-	c.Assert(err, IsNil)
-	_, err = os.Stat(filepath.Join(s.root, "testfile"))
-	c.Assert(err, IsNil)
-	err = os.Remove(filepath.Join(s.root, "testfile"))
-	c.Assert(err, IsNil)
-}
-
 func (s *MySuite) TestSafeAbort(c *C) {
-	f, err := CreateFile(filepath.Join(s.root, "purgefile"))
+	f, err := CreateFile(path.Join(s.root, "testfile-abort"))
 	c.Assert(err, IsNil)
-	_, err = os.Stat(filepath.Join(s.root, "purgefile"))
+	_, err = os.Stat(path.Join(s.root, "testfile-abort"))
 	c.Assert(err, Not(IsNil))
 	err = f.Abort()
 	c.Assert(err, IsNil)
-	_, err = os.Stat(filepath.Join(s.root, "purgefile"))
-	c.Assert(err, Not(IsNil))
+	err = f.Close()
+	c.Assert(err.Error(), Equals, "close on aborted file")
+}
+
+func (s *MySuite) TestSafeClose(c *C) {
+	f, err := CreateFile(path.Join(s.root, "testfile-close"))
+	c.Assert(err, IsNil)
+	_, err = os.Stat(path.Join(s.root, "testfile-close"))
+	c.Assert(err, Not(IsNil))
+	err = f.Close()
+	c.Assert(err, IsNil)
+	_, err = os.Stat(path.Join(s.root, "testfile-close"))
+	c.Assert(err, IsNil)
+	err = os.Remove(path.Join(s.root, "testfile-close"))
+	c.Assert(err, IsNil)
+	err = f.Abort()
+	c.Assert(err.Error(), Equals, "abort on closed file")
+}
+
+func (s *MySuite) TestSafe(c *C) {
+	f, err := CreateFile(path.Join(s.root, "testfile-safe"))
+	c.Assert(err, IsNil)
+	_, err = os.Stat(path.Join(s.root, "testfile-safe"))
+	c.Assert(err, Not(IsNil))
+	err = f.Close()
+	c.Assert(err, IsNil)
+	_, err = f.Write([]byte("Test"))
+	c.Assert(err.Error(), Equals, "write on closed file")
+	err = f.Close()
+	c.Assert(err.Error(), Equals, "close on closed file")
+	_, err = os.Stat(path.Join(s.root, "testfile-safe"))
+	c.Assert(err, IsNil)
+	err = os.Remove(path.Join(s.root, "testfile-safe"))
+	c.Assert(err, IsNil)
+}
+
+func (s *MySuite) TestSafeAbortWrite(c *C) {
+	f, err := CreateFile(path.Join(s.root, "purgefile-abort"))
+	c.Assert(err, IsNil)
+	_, err = os.Stat(path.Join(s.root, "purgefile-abort"))
+	c.Assert(err, Not(IsNil))
+	err = f.Abort()
+	c.Assert(err, IsNil)
+	_, err = os.Stat(path.Join(s.root, "purgefile-abort"))
+	c.Assert(err, Not(IsNil))
+	err = f.Abort()
+	c.Assert(err.Error(), Equals, "abort on aborted file")
+	_, err = f.Write([]byte("Test"))
+	c.Assert(err.Error(), Equals, "write on aborted file")
 }

signature-v4.go

@@ -27,13 +27,14 @@ package main
 import (
 	"bytes"
 	"encoding/hex"
-	"github.com/minio/sha256-simd"
 	"net/http"
 	"net/url"
 	"sort"
 	"strconv"
 	"strings"
 	"time"
+
+	"github.com/minio/sha256-simd"
 )
 
 // AWS Signature Version '4' constants.
@@ -391,5 +392,7 @@ func doesSignatureMatch(hashedPayload string, r *http.Request, validateRegion bo
 	if newSignature != signV4Values.Signature {
 		return ErrSignatureDoesNotMatch
 	}
+
+	// Return error none.
 	return ErrNone
 }

streaming-signature-v4.go

@@ -0,0 +1,382 @@
+/*
+ * Minio Cloud Storage, (C) 2016 Minio, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// This file implements helper functions to validate Streaming AWS
+// Signature Version '4' authorization header.
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/hex"
+	"errors"
+	"hash"
+	"io"
+	"net/http"
+	"time"
+
+	"github.com/minio/sha256-simd"
+)
+
+// Streaming AWS Signature Version '4' constants.
+const (
+	emptySHA256            = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
+	streamingContentSHA256 = "STREAMING-AWS4-HMAC-SHA256-PAYLOAD"
+	signV4ChunkedAlgorithm = "AWS4-HMAC-SHA256-PAYLOAD"
+)
+
+// getChunkSignature - get chunk signature.
+func getChunkSignature(seedSignature string, date time.Time, hashedChunk string) string {
+	// Access credentials.
+	cred := serverConfig.GetCredential()
+
+	// Server region.
+	region := serverConfig.GetRegion()
+
+	// Calculate string to sign.
+	stringToSign := signV4ChunkedAlgorithm + "\n" +
+		date.Format(iso8601Format) + "\n" +
+		getScope(date, region) + "\n" +
+		seedSignature + "\n" +
+		emptySHA256 + "\n" +
+		hashedChunk
+
+	// Get hmac signing key.
+	signingKey := getSigningKey(cred.SecretAccessKey, date, region)
+
+	// Calculate signature.
+	newSignature := getSignature(signingKey, stringToSign)
+
+	return newSignature
+}
+
+// calculateSeedSignature - Calculate seed signature in accordance with
+//     - http://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-streaming.html
+// returns signature, error otherwise if the signature mismatches or any other
+// error while parsing and validating.
+func calculateSeedSignature(r *http.Request) (signature string, date time.Time, errCode APIErrorCode) {
+	// Access credentials.
+	cred := serverConfig.GetCredential()
+
+	// Server region.
+	region := serverConfig.GetRegion()
+
+	// Copy request.
+	req := *r
+
+	// Save authorization header.
+	v4Auth := req.Header.Get("Authorization")
+
+	// Parse signature version '4' header.
+	signV4Values, errCode := parseSignV4(v4Auth)
+	if errCode != ErrNone {
+		return "", time.Time{}, errCode
+	}
+
+	// Payload streaming.
+	payload := streamingContentSHA256
+
+	// Payload for STREAMING signature should be 'STREAMING-AWS4-HMAC-SHA256-PAYLOAD'
+	if payload != req.Header.Get("X-Amz-Content-Sha256") {
+		return "", time.Time{}, ErrContentSHA256Mismatch
+	}
+
+	// Extract all the signed headers along with its values.
+	extractedSignedHeaders := extractSignedHeaders(signV4Values.SignedHeaders, req.Header)
+
+	// Verify if the access key id matches.
+	if signV4Values.Credential.accessKey != cred.AccessKeyID {
+		return "", time.Time{}, ErrInvalidAccessKeyID
+	}
+
+	// Verify if region is valid.
+	sRegion := signV4Values.Credential.scope.region
+	// Should validate region, only if region is set. Some operations
+	// do not need region validated for example GetBucketLocation.
+	if !isValidRegion(sRegion, region) {
+		return "", time.Time{}, ErrInvalidRegion
+	}
+
+	// Extract date, if not present throw error.
+	var dateStr string
+	if dateStr = req.Header.Get(http.CanonicalHeaderKey("x-amz-date")); dateStr == "" {
+		if dateStr = r.Header.Get("Date"); dateStr == "" {
+			return "", time.Time{}, ErrMissingDateHeader
+		}
+	}
+	// Parse date header.
+	var err error
+	date, err = time.Parse(iso8601Format, dateStr)
+	if err != nil {
+		errorIf(err, "Unable to parse date", dateStr)
+		return "", time.Time{}, ErrMalformedDate
+	}
+
+	// Query string.
+	queryStr := req.URL.Query().Encode()
+
+	// Get canonical request.
+	canonicalRequest := getCanonicalRequest(extractedSignedHeaders, payload, queryStr, req.URL.Path, req.Method, req.Host)
+
+	// Get string to sign from canonical request.
+	stringToSign := getStringToSign(canonicalRequest, date, region)
+
+	// Get hmac signing key.
+	signingKey := getSigningKey(cred.SecretAccessKey, date, region)
+
+	// Calculate signature.
+	newSignature := getSignature(signingKey, stringToSign)
+
+	// Verify if signature match.
+	if newSignature != signV4Values.Signature {
+		return "", time.Time{}, ErrSignatureDoesNotMatch
+	}
+
+	// Return caculated signature.
+	return newSignature, date, ErrNone
+}
+
+const maxLineLength = 4096 // assumed <= bufio.defaultBufSize 4KiB.
+
+// lineTooLong is generated as chunk header is bigger than 4KiB.
+var errLineTooLong = errors.New("header line too long")
+
+// Malformed encoding is generated when chunk header is wrongly formed.
+var errMalformedEncoding = errors.New("malformed chunked encoding")
+
+// newSignV4ChunkedReader returns a new s3ChunkedReader that translates the data read from r
+// out of HTTP "chunked" format before returning it.
+// The s3ChunkedReader returns io.EOF when the final 0-length chunk is read.
+//
+// NewChunkedReader is not needed by normal applications. The http package
+// automatically decodes chunking when reading response bodies.
+func newSignV4ChunkedReader(req *http.Request) (io.Reader, APIErrorCode) {
+	seedSignature, seedDate, errCode := calculateSeedSignature(req)
+	if errCode != ErrNone {
+		return nil, errCode
+	}
+	return &s3ChunkedReader{
+		reader:            bufio.NewReader(req.Body),
+		seedSignature:     seedSignature,
+		seedDate:          seedDate,
+		chunkSHA256Writer: sha256.New(),
+	}, ErrNone
+}
+
+// Represents the overall state that is required for decoding a
+// AWS Signature V4 chunked reader.
+type s3ChunkedReader struct {
+	reader            *bufio.Reader
+	seedSignature     string
+	seedDate          time.Time
+	dataChunkRead     bool
+	chunkSignature    string
+	chunkSHA256Writer hash.Hash // Calculates sha256 of chunk data.
+	n                 uint64    // Unread bytes in chunk
+	err               error
+}
+
+// Read chunk reads the chunk token signature portion.
+func (cr *s3ChunkedReader) readS3ChunkHeader() {
+	// Read the first chunk line until CRLF.
+	var hexChunkSize, hexChunkSignature []byte
+	hexChunkSize, hexChunkSignature, cr.err = readChunkLine(cr.reader)
+	if cr.err != nil {
+		return
+	}
+	// <hex>;token=value - converts the hex into its uint64 form.
+	cr.n, cr.err = parseHexUint(hexChunkSize)
+	if cr.err != nil {
+		return
+	}
+	if cr.n == 0 {
+		cr.err = io.EOF
+	}
+	// is the data part already read?, set this to false.
+	cr.dataChunkRead = false
+	// Reset sha256 hasher for a fresh start.
+	cr.chunkSHA256Writer.Reset()
+	// Save the incoming chunk signature.
+	cr.chunkSignature = string(hexChunkSignature)
+}
+
+// Validate if the underlying buffer has chunk header.
+func (cr *s3ChunkedReader) s3ChunkHeaderAvailable() bool {
+	n := cr.reader.Buffered()
+	if n > 0 {
+		// Peek without seeking to look for trailing '\n'.
+		peek, _ := cr.reader.Peek(n)
+		return bytes.IndexByte(peek, '\n') >= 0
+	}
+	return false
+}
+
+// Read - implements `io.Reader`, which transparently decodes
+// the incoming AWS Signature V4 streaming signature.
+func (cr *s3ChunkedReader) Read(buf []byte) (n int, err error) {
+	for cr.err == nil {
+		if cr.n == 0 {
+			// For no chunk header available, we don't have to
+			// proceed to read again.
+			if n > 0 && !cr.s3ChunkHeaderAvailable() {
+				// We've read enough. Don't potentially block
+				// reading a new chunk header.
+				break
+			}
+			// If the chunk has been read, proceed to validate the rolling signature.
+			if cr.dataChunkRead {
+				// Calculate the hashed chunk.
+				hashedChunk := hex.EncodeToString(cr.chunkSHA256Writer.Sum(nil))
+				// Calculate the chunk signature.
+				newSignature := getChunkSignature(cr.seedSignature, cr.seedDate, hashedChunk)
+				if cr.chunkSignature != newSignature {
+					// Chunk signature doesn't match we return signature does not match.
+					cr.err = errSignatureMismatch
+					break
+				}
+				// Newly calculated signature becomes the seed for the next chunk
+				// this follows the chaining.
+				cr.seedSignature = newSignature
+			}
+			// Proceed to read the next chunk header.
+			cr.readS3ChunkHeader()
+			continue
+		}
+		// With requested buffer of zero length, no need to read further.
+		if len(buf) == 0 {
+			break
+		}
+		rbuf := buf
+		// Make sure to read only the specified payload size, stagger
+		// the rest for subsequent requests.
+		if uint64(len(rbuf)) > cr.n {
+			rbuf = rbuf[:cr.n]
+		}
+		var n0 int
+		n0, cr.err = cr.reader.Read(rbuf)
+
+		// Calculate sha256.
+		cr.chunkSHA256Writer.Write(rbuf[:n0])
+		// Set since we have read the chunk read.
+		cr.dataChunkRead = true
+
+		n += n0
+		buf = buf[n0:]
+		// Decrements the 'cr.n' for future reads.
+		cr.n -= uint64(n0)
+
+		// If we're at the end of a chunk.
+		if cr.n == 0 && cr.err == nil {
+			// Read the next two bytes to verify if they are "\r\n".
+			cr.err = checkCRLF(cr.reader)
+		}
+	}
+	// Return number of bytes read, and error if any.
+	return n, cr.err
+}
+
+// checkCRLF - check if reader only has '\r\n' CRLF character.
+// returns malformed encoding if it doesn't.
+func checkCRLF(reader io.Reader) (err error) {
+	var buf = make([]byte, 2)
+	if _, err = io.ReadFull(reader, buf[:2]); err == nil {
+		if buf[0] != '\r' || buf[1] != '\n' {
+			err = errMalformedEncoding
+		}
+	}
+	return err
+}
+
+// Read a line of bytes (up to \n) from b.
+// Give up if the line exceeds maxLineLength.
+// The returned bytes are owned by the bufio.Reader
+// so they are only valid until the next bufio read.
+func readChunkLine(b *bufio.Reader) ([]byte, []byte, error) {
+	buf, err := b.ReadSlice('\n')
+	if err != nil {
+		// We always know when EOF is coming.
+		// If the caller asked for a line, there should be a line.
+		if err == io.EOF {
+			err = io.ErrUnexpectedEOF
+		} else if err == bufio.ErrBufferFull {
+			err = errLineTooLong
+		}
+		return nil, nil, err
+	}
+	if len(buf) >= maxLineLength {
+		return nil, nil, errLineTooLong
+	}
+	// Parse s3 specific chunk extension and fetch the values.
+	hexChunkSize, hexChunkSignature := parseS3ChunkExtension(buf)
+	return hexChunkSize, hexChunkSignature, nil
+}
+
+// trimTrailingWhitespace - trim trailing white space.
+func trimTrailingWhitespace(b []byte) []byte {
+	for len(b) > 0 && isASCIISpace(b[len(b)-1]) {
+		b = b[:len(b)-1]
+	}
+	return b
+}
+
+// isASCIISpace - is ascii space?
+func isASCIISpace(b byte) bool {
+	return b == ' ' || b == '\t' || b == '\n' || b == '\r'
+}
+
+// Constant s3 chunk encoding signature.
+const s3ChunkSignatureStr = ";chunk-signature="
+
+// parses3ChunkExtension removes any s3 specific chunk-extension from buf.
+// For example,
+//     "10000;chunk-signature=..." => "10000", "chunk-signature=..."
+func parseS3ChunkExtension(buf []byte) ([]byte, []byte) {
+	buf = trimTrailingWhitespace(buf)
+	semi := bytes.Index(buf, []byte(s3ChunkSignatureStr))
+	// Chunk signature not found, return the whole buffer.
+	if semi == -1 {
+		return buf, nil
+	}
+	return buf[:semi], parseChunkSignature(buf[semi:])
+}
+
+// parseChunkSignature - parse chunk signature.
+func parseChunkSignature(chunk []byte) []byte {
+	chunkSplits := bytes.SplitN(chunk, []byte(s3ChunkSignatureStr), 2)
+	return chunkSplits[1]
+}
+
+// parse hex to uint64.
+func parseHexUint(v []byte) (n uint64, err error) {
+	for i, b := range v {
+		switch {
+		case '0' <= b && b <= '9':
+			b = b - '0'
+		case 'a' <= b && b <= 'f':
+			b = b - 'a' + 10
+		case 'A' <= b && b <= 'F':
+			b = b - 'A' + 10
+		default:
+			return 0, errors.New("invalid byte in chunk length")
+		}
+		if i == 16 {
+			return 0, errors.New("http chunk length too large")
+		}
+		n <<= 4
+		n |= uint64(b)
+	}
+	return
+}

streaming-signature-v4_test.go

@@ -0,0 +1,196 @@
+/*
+ * Minio Cloud Storage, (C) 2016 Minio, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	"io"
+	"strings"
+	"testing"
+)
+
+// Test read chunk line.
+func TestReadChunkLine(t *testing.T) {
+	type testCase struct {
+		reader         *bufio.Reader
+		expectedErr    error
+		chunkSize      []byte
+		chunkSignature []byte
+	}
+	// List of readers used.
+	readers := []io.Reader{
+		// Test - 1
+		bytes.NewReader([]byte("1000;chunk-signature=111123333333333333334444211\r\n")),
+		// Test - 2
+		bytes.NewReader([]byte("1000;")),
+		// Test - 3
+		bytes.NewReader([]byte(fmt.Sprintf("%4097d", 1))),
+		// Test - 4
+		bytes.NewReader([]byte("1000;chunk-signature=111123333333333333334444211\r\n")),
+	}
+	testCases := []testCase{
+		// Test - 1 - small bufio reader.
+		{
+			bufio.NewReaderSize(readers[0], 16),
+			errLineTooLong,
+			nil,
+			nil,
+		},
+		// Test - 2 - unexpected end of the reader.
+		{
+			bufio.NewReader(readers[1]),
+			io.ErrUnexpectedEOF,
+			nil,
+			nil,
+		},
+		// Test - 3 - line too long bigger than 4k+1
+		{
+			bufio.NewReader(readers[2]),
+			errLineTooLong,
+			nil,
+			nil,
+		},
+		// Test - 4 - parse the chunk reader properly.
+		{
+			bufio.NewReader(readers[3]),
+			nil,
+			[]byte("1000"),
+			[]byte("111123333333333333334444211"),
+		},
+	}
+	// Valid test cases for each chunk line.
+	for i, tt := range testCases {
+		chunkSize, chunkSignature, err := readChunkLine(tt.reader)
+		if err != tt.expectedErr {
+			t.Errorf("Test %d: Expected %s, got %s", i+1, tt.expectedErr, err)
+		}
+		if !bytes.Equal(chunkSize, tt.chunkSize) {
+			t.Errorf("Test %d: Expected %s, got %s", i+1, string(tt.chunkSize), string(chunkSize))
+		}
+		if !bytes.Equal(chunkSignature, tt.chunkSignature) {
+			t.Errorf("Test %d: Expected %s, got %s", i+1, string(tt.chunkSignature), string(chunkSignature))
+		}
+	}
+}
+
+// Test parsing s3 chunk extension.
+func TestParseS3ChunkExtension(t *testing.T) {
+	type testCase struct {
+		buf       []byte
+		chunkSize []byte
+		chunkSign []byte
+	}
+
+	tests := []testCase{
+		// Test - 1 valid case.
+		{
+			[]byte("10000;chunk-signature=ad80c730a21e5b8d04586a2213dd63b9a0e99e0e2307b0ade35a65485a288648"),
+			[]byte("10000"),
+			[]byte("ad80c730a21e5b8d04586a2213dd63b9a0e99e0e2307b0ade35a65485a288648"),
+		},
+		// Test - 2 no chunk extension, return same buffer.
+		{
+			[]byte("10000;"),
+			[]byte("10000;"),
+			nil,
+		},
+		// Test - 3 no chunk size, return error.
+		{
+			[]byte(";chunk-signature="),
+			nil,
+			nil,
+		},
+		// Test - 4 removes trailing slash.
+		{
+			[]byte("10000;chunk-signature=ad80c730a21e5b8d04586a2213dd63b9a0e99e0e2307b0ade35a65485a288648    \t \n"),
+			[]byte("10000"),
+			[]byte("ad80c730a21e5b8d04586a2213dd63b9a0e99e0e2307b0ade35a65485a288648"),
+		},
+	}
+	// Validate chunk extension removal.
+	for i, tt := range tests {
+		// Extract chunk size and chunk signature after parsing a standard chunk-extension format.
+		hexChunkSize, hexChunkSignature := parseS3ChunkExtension(tt.buf)
+		if !bytes.Equal(hexChunkSize, tt.chunkSize) {
+			t.Errorf("Test %d: Expected %s, got %s", i+1, string(tt.chunkSize), string(hexChunkSize))
+		}
+		if !bytes.Equal(hexChunkSignature, tt.chunkSign) {
+			t.Errorf("Test %d: Expected %s, got %s", i+1, string(tt.chunkSign), string(hexChunkSignature))
+		}
+	}
+}
+
+// Test check CRLF characters on input reader.
+func TestCheckCRLF(t *testing.T) {
+	type testCase struct {
+		reader      io.Reader
+		expectedErr error
+	}
+	tests := []testCase{
+		// Test - 1 valid buffer with CRLF.
+		{bytes.NewReader([]byte("\r\n")), nil},
+		// Test - 2 invalid buffer with no CRLF.
+		{bytes.NewReader([]byte("he")), errMalformedEncoding},
+		// Test - 3 invalid buffer with more characters.
+		{bytes.NewReader([]byte("he\r\n")), errMalformedEncoding},
+		// Test - 4 smaller buffer than expected.
+		{bytes.NewReader([]byte("h")), io.ErrUnexpectedEOF},
+	}
+	for i, tt := range tests {
+		err := checkCRLF(tt.reader)
+		if err != tt.expectedErr {
+			t.Errorf("Test %d: Expected %s, got %s this", i+1, tt.expectedErr, err)
+		}
+	}
+}
+
+// Tests parsing hex number into its uint64 decimal equivalent.
+func TestParseHexUint(t *testing.T) {
+	type testCase struct {
+		in      string
+		want    uint64
+		wantErr string
+	}
+	tests := []testCase{
+		{"x", 0, "invalid byte in chunk length"},
+		{"0000000000000000", 0, ""},
+		{"0000000000000001", 1, ""},
+		{"ffffffffffffffff", 1<<64 - 1, ""},
+		{"FFFFFFFFFFFFFFFF", 1<<64 - 1, ""},
+		{"000000000000bogus", 0, "invalid byte in chunk length"},
+		{"00000000000000000", 0, "http chunk length too large"}, // could accept if we wanted
+		{"10000000000000000", 0, "http chunk length too large"},
+		{"00000000000000001", 0, "http chunk length too large"}, // could accept if we wanted
+	}
+	for i := uint64(0); i <= 1234; i++ {
+		tests = append(tests, testCase{in: fmt.Sprintf("%x", i), want: i})
+	}
+	for _, tt := range tests {
+		got, err := parseHexUint([]byte(tt.in))
+		if tt.wantErr != "" {
+			if !strings.Contains(fmt.Sprint(err), tt.wantErr) {
+				t.Errorf("parseHexUint(%q) = %v, %v; want error %q", tt.in, got, err, tt.wantErr)
+			}
+		} else {
+			if err != nil || got != tt.want {
+				t.Errorf("parseHexUint(%q) = %v, %v; want %v", tt.in, got, err, tt.want)
+			}
+		}
+	}
+}