Add tests for regular and streaming v4 PutObject Handler (#2618)

cmd/object-handlers_test.go

@@ -174,6 +174,193 @@ func testAPIGetOjectHandler(obj ObjectLayer, instanceType string, t TestErrHandl
 	}
 }
 
+// Wrapper for calling PutObject API handler tests using streaming signature v4 for both XL multiple disks and FS single drive setup.
+func TestAPIPutObjectStreamSigV4Handler(t *testing.T) {
+	ExecObjectLayerTest(t, testAPIPutObjectStreamSigV4Handler)
+}
+
+func testAPIPutObjectStreamSigV4Handler(obj ObjectLayer, instanceType string, t TestErrHandler) {
+	// get random bucket name.
+	bucketName := getRandomBucketName()
+	objectName := "test-object"
+	// Create bucket.
+	err := obj.MakeBucket(bucketName)
+	if err != nil {
+		// failed to create newbucket, abort.
+		t.Fatalf("%s : %s", instanceType, err)
+	}
+	// Register the API end points with XL/FS object layer.
+	// Registering only the GetObject handler.
+	apiRouter := initTestAPIEndPoints(obj, []string{"PutObject"})
+	// initialize the server and obtain the credentials and root.
+	// credentials are necessary to sign the HTTP request.
+	rootPath, err := newTestConfig("us-east-1")
+	if err != nil {
+		t.Fatalf("Init Test config failed")
+	}
+	// remove the root folder after the test ends.
+	defer removeAll(rootPath)
+
+	credentials := serverConfig.GetCredential()
+
+	bytesDataLen := 65 * 1024
+	bytesData := bytes.Repeat([]byte{'a'}, bytesDataLen)
+
+	// byte data for PutObject.
+	// test cases with inputs and expected result for GetObject.
+	testCases := []struct {
+		bucketName string
+		objectName string
+		data       []byte
+		dataLen    int
+		// expected output.
+		expectedContent    []byte // expected response body.
+		expectedRespStatus int    // expected response status body.
+	}{
+		// Test case - 1.
+		// Fetching the entire object and validating its contents.
+		{
+			bucketName:         bucketName,
+			objectName:         objectName,
+			data:               bytesData,
+			dataLen:            len(bytesData),
+			expectedContent:    []byte{},
+			expectedRespStatus: http.StatusOK,
+		},
+	}
+	// Iterating over the cases, fetching the object validating the response.
+	for i, testCase := range testCases {
+		// initialize HTTP NewRecorder, this records any mutations to response writer inside the handler.
+		rec := httptest.NewRecorder()
+		// construct HTTP request for Put Object end point.
+		req, err := newTestStreamingSignedRequest("PUT",
+			getPutObjectURL("", testCase.bucketName, testCase.objectName),
+			int64(testCase.dataLen), 64*1024, bytes.NewReader(testCase.data),
+			credentials.AccessKeyID, credentials.SecretAccessKey)
+		if err != nil {
+			t.Fatalf("Test %d: Failed to create HTTP request for Put Object: <ERROR> %v", i+1, err)
+		}
+		// Since `apiRouter` satisfies `http.Handler` it has a ServeHTTP to execute the logic of the handler.
+		// Call the ServeHTTP to execute the handler,`func (api objectAPIHandlers) GetObjectHandler`  handles the request.
+		apiRouter.ServeHTTP(rec, req)
+		// Assert the response code with the expected status.
+		if rec.Code != testCase.expectedRespStatus {
+			t.Fatalf("Case %d: Expected the response status to be `%d`, but instead found `%d`", i+1, testCase.expectedRespStatus, rec.Code)
+		}
+		// read the response body.
+		actualContent, err := ioutil.ReadAll(rec.Body)
+		if err != nil {
+			t.Fatalf("Test %d: %s: Failed parsing response body: <ERROR> %v", i+1, instanceType, err)
+		}
+		// Verify whether the bucket obtained object is same as the one inserted.
+		if !bytes.Equal(testCase.expectedContent, actualContent) {
+			t.Errorf("Test %d: %s: Object content differs from expected value.: %s", i+1, instanceType, string(actualContent))
+		}
+
+		buffer := new(bytes.Buffer)
+		err = obj.GetObject(testCase.bucketName, testCase.objectName, 0, int64(bytesDataLen), buffer)
+		if err != nil {
+			t.Fatalf("Test %d: %s: Failed to fetch the copied object: <ERROR> %s", i+1, instanceType, err)
+		}
+		if !bytes.Equal(bytesData, buffer.Bytes()) {
+			t.Errorf("Test %d: %s: Data Mismatch: Data fetched back from the uploaded object doesn't match the original one.", i+1, instanceType)
+		}
+		buffer.Reset()
+	}
+}
+
+// Wrapper for calling PutObject API handler tests for both XL multiple disks and FS single drive setup.
+func TestAPIPutObjectHandler(t *testing.T) {
+	ExecObjectLayerTest(t, testAPIPutObjectHandler)
+}
+
+func testAPIPutObjectHandler(obj ObjectLayer, instanceType string, t TestErrHandler) {
+	// get random bucket name.
+	bucketName := getRandomBucketName()
+	objectName := "test-object"
+	// Create bucket.
+	err := obj.MakeBucket(bucketName)
+	if err != nil {
+		// failed to create newbucket, abort.
+		t.Fatalf("%s : %s", instanceType, err)
+	}
+	// Register the API end points with XL/FS object layer.
+	// Registering only the GetObject handler.
+	apiRouter := initTestAPIEndPoints(obj, []string{"PutObject"})
+	// initialize the server and obtain the credentials and root.
+	// credentials are necessary to sign the HTTP request.
+	rootPath, err := newTestConfig("us-east-1")
+	if err != nil {
+		t.Fatalf("Init Test config failed")
+	}
+	// remove the root folder after the test ends.
+	defer removeAll(rootPath)
+
+	credentials := serverConfig.GetCredential()
+
+	// byte data for PutObject.
+	bytesData := generateBytesData(6 * 1024 * 1024)
+
+	// test cases with inputs and expected result for GetObject.
+	testCases := []struct {
+		bucketName string
+		objectName string
+		data       []byte
+		dataLen    int
+		// expected output.
+		expectedContent    []byte // expected response body.
+		expectedRespStatus int    // expected response status body.
+	}{
+		// Test case - 1.
+		// Fetching the entire object and validating its contents.
+		{
+			bucketName:         bucketName,
+			objectName:         objectName,
+			data:               bytesData,
+			dataLen:            len(bytesData),
+			expectedContent:    []byte{},
+			expectedRespStatus: http.StatusOK,
+		},
+	}
+	// Iterating over the cases, fetching the object validating the response.
+	for i, testCase := range testCases {
+		// initialize HTTP NewRecorder, this records any mutations to response writer inside the handler.
+		rec := httptest.NewRecorder()
+		// construct HTTP request for Get Object end point.
+		req, err := newTestSignedRequest("PUT", getPutObjectURL("", testCase.bucketName, testCase.objectName),
+			int64(testCase.dataLen), bytes.NewReader(testCase.data), credentials.AccessKeyID, credentials.SecretAccessKey)
+		if err != nil {
+			t.Fatalf("Test %d: Failed to create HTTP request for Put Object: <ERROR> %v", i+1, err)
+		}
+		// Since `apiRouter` satisfies `http.Handler` it has a ServeHTTP to execute the logic of the handler.
+		// Call the ServeHTTP to execute the handler,`func (api objectAPIHandlers) GetObjectHandler`  handles the request.
+		apiRouter.ServeHTTP(rec, req)
+		// Assert the response code with the expected status.
+		if rec.Code != testCase.expectedRespStatus {
+			t.Fatalf("Case %d: Expected the response status to be `%d`, but instead found `%d`", i+1, testCase.expectedRespStatus, rec.Code)
+		}
+		// read the response body.
+		actualContent, err := ioutil.ReadAll(rec.Body)
+		if err != nil {
+			t.Fatalf("Test %d: %s: Failed parsing response body: <ERROR> %v", i+1, instanceType, err)
+		}
+		// Verify whether the bucket obtained object is same as the one inserted.
+		if !bytes.Equal(testCase.expectedContent, actualContent) {
+			t.Errorf("Test %d: %s: Object content differs from expected value.: %s", i+1, instanceType, string(actualContent))
+		}
+
+		buffer := new(bytes.Buffer)
+		err = obj.GetObject(testCase.bucketName, testCase.objectName, 0, int64(len(bytesData)), buffer)
+		if err != nil {
+			t.Fatalf("Test %d: %s: Failed to fetch the copied object: <ERROR> %s", i+1, instanceType, err)
+		}
+		if !bytes.Equal(bytesData, buffer.Bytes()) {
+			t.Errorf("Test %d: %s: Data Mismatch: Data fetched back from the uploaded object doesn't match the original one.", i+1, instanceType)
+		}
+		buffer.Reset()
+	}
+}
+
 // Wrapper for calling Copy Object API handler tests for both XL multiple disks and single node setup.
 func TestAPICopyObjectHandler(t *testing.T) {
 	ExecObjectLayerTest(t, testAPICopyObjectHandler)

cmd/server_utils_test.go

@@ -18,6 +18,7 @@ package cmd
 
 import (
 	"encoding/xml"
+	"fmt"
 	"io/ioutil"
 	"net"
 	"net/http"
@@ -65,6 +66,40 @@ var ignoredHeaders = map[string]bool{
 	"User-Agent":     true,
 }
 
+// Headers to ignore in streaming v4
+var ignoredStreamingHeaders = map[string]bool{
+	"Authorization": true,
+	"Content-Type":  true,
+	"Content-Md5":   true,
+	"User-Agent":    true,
+}
+
+// calculateSignedChunkLength - calculates the length of chunk metadata
+func calculateSignedChunkLength(chunkDataSize int64) int64 {
+	return int64(len(fmt.Sprintf("%x", chunkDataSize))) +
+		17 + // ";chunk-signature="
+		64 + // e.g. "f2ca1bb6c7e907d06dafe4687e579fce76b37e4e93b7605022da52e6ccc26fd2"
+		2 + // CRLF
+		chunkDataSize +
+		2 // CRLF
+}
+
+// calculateSignedChunkLength - calculates the length of the overall stream (data + metadata)
+func calculateStreamContentLength(dataLen, chunkSize int64) int64 {
+	if dataLen <= 0 {
+		return 0
+	}
+	chunksCount := int64(dataLen / chunkSize)
+	remainingBytes := int64(dataLen % chunkSize)
+	streamLen := int64(0)
+	streamLen += chunksCount * calculateSignedChunkLength(chunkSize)
+	if remainingBytes > 0 {
+		streamLen += calculateSignedChunkLength(remainingBytes)
+	}
+	streamLen += calculateSignedChunkLength(0)
+	return streamLen
+}
+
 // Ask the kernel for a free open port.
 func getFreePort() int {
 	addr, err := net.ResolveTCPAddr("tcp", "localhost:0")

cmd/test-utils_test.go

@@ -267,6 +267,208 @@ func (testServer TestServer) Stop() {
 	testServer.Server.Close()
 }
 
+// Sign given request using Signature V4.
+func signStreamingRequest(req *http.Request, accessKey, secretKey string) (string, error) {
+	// Get hashed payload.
+	hashedPayload := req.Header.Get("x-amz-content-sha256")
+	if hashedPayload == "" {
+		return "", fmt.Errorf("Invalid hashed payload.")
+	}
+
+	currTime := time.Now().UTC()
+	// Set x-amz-date.
+	req.Header.Set("x-amz-date", currTime.Format(iso8601Format))
+
+	// Get header map.
+	headerMap := make(map[string][]string)
+	for k, vv := range req.Header {
+		// If request header key is not in ignored headers, then add it.
+		if _, ok := ignoredStreamingHeaders[http.CanonicalHeaderKey(k)]; !ok {
+			headerMap[strings.ToLower(k)] = vv
+		}
+	}
+
+	// Get header keys.
+	headers := []string{"host"}
+	for k := range headerMap {
+		headers = append(headers, k)
+	}
+	sort.Strings(headers)
+
+	// Get canonical headers.
+	var buf bytes.Buffer
+	for _, k := range headers {
+		buf.WriteString(k)
+		buf.WriteByte(':')
+		switch {
+		case k == "host":
+			buf.WriteString(req.URL.Host)
+			fallthrough
+		default:
+			for idx, v := range headerMap[k] {
+				if idx > 0 {
+					buf.WriteByte(',')
+				}
+				buf.WriteString(v)
+			}
+			buf.WriteByte('\n')
+		}
+	}
+	canonicalHeaders := buf.String()
+
+	// Get signed headers.
+	signedHeaders := strings.Join(headers, ";")
+
+	// Get canonical query string.
+	req.URL.RawQuery = strings.Replace(req.URL.Query().Encode(), "+", "%20", -1)
+
+	// Get canonical URI.
+	canonicalURI := getURLEncodedName(req.URL.Path)
+
+	// Get canonical request.
+	// canonicalRequest =
+	//  <HTTPMethod>\n
+	//  <CanonicalURI>\n
+	//  <CanonicalQueryString>\n
+	//  <CanonicalHeaders>\n
+	//  <SignedHeaders>\n
+	//  <HashedPayload>
+	//
+	canonicalRequest := strings.Join([]string{
+		req.Method,
+		canonicalURI,
+		req.URL.RawQuery,
+		canonicalHeaders,
+		signedHeaders,
+		hashedPayload,
+	}, "\n")
+
+	// Get scope.
+	scope := strings.Join([]string{
+		currTime.Format(yyyymmdd),
+		"us-east-1",
+		"s3",
+		"aws4_request",
+	}, "/")
+
+	stringToSign := "AWS4-HMAC-SHA256" + "\n" + currTime.Format(iso8601Format) + "\n"
+	stringToSign = stringToSign + scope + "\n"
+	stringToSign = stringToSign + hex.EncodeToString(sum256([]byte(canonicalRequest)))
+
+	date := sumHMAC([]byte("AWS4"+secretKey), []byte(currTime.Format(yyyymmdd)))
+	region := sumHMAC(date, []byte("us-east-1"))
+	service := sumHMAC(region, []byte("s3"))
+	signingKey := sumHMAC(service, []byte("aws4_request"))
+
+	signature := hex.EncodeToString(sumHMAC(signingKey, []byte(stringToSign)))
+
+	// final Authorization header
+	parts := []string{
+		"AWS4-HMAC-SHA256" + " Credential=" + accessKey + "/" + scope,
+		"SignedHeaders=" + signedHeaders,
+		"Signature=" + signature,
+	}
+	auth := strings.Join(parts, ", ")
+	req.Header.Set("Authorization", auth)
+
+	return signature, nil
+}
+
+// Returns new HTTP request object.
+func newTestStreamingRequest(method, urlStr string, dataLength, chunkSize int64, body io.ReadSeeker) (*http.Request, error) {
+	if method == "" {
+		method = "POST"
+	}
+
+	req, err := http.NewRequest(method, urlStr, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	if body == nil {
+		// this is added to avoid panic during ioutil.ReadAll(req.Body).
+		// th stack trace can be found here  https://github.com/minio/minio/pull/2074 .
+		// This is very similar to https://github.com/golang/go/issues/7527.
+		req.Body = ioutil.NopCloser(bytes.NewReader([]byte("")))
+	}
+
+	contentLength := calculateStreamContentLength(dataLength, chunkSize)
+
+	req.Header.Set("x-amz-content-sha256", "STREAMING-AWS4-HMAC-SHA256-PAYLOAD")
+	req.Header.Set("content-encoding", "aws-chunked")
+	req.Header.Set("x-amz-storage-class", "REDUCED_REDUNDANCY")
+
+	req.Header.Set("x-amz-decoded-content-length", strconv.FormatInt(dataLength, 10))
+	req.Header.Set("content-length", strconv.FormatInt(contentLength, 10))
+
+	// Seek back to beginning.
+	body.Seek(0, 0)
+	// Add body
+	req.Body = ioutil.NopCloser(body)
+	req.ContentLength = contentLength
+
+	return req, nil
+}
+
+// Returns new HTTP request object signed with streaming signature v4.
+func newTestStreamingSignedRequest(method, urlStr string, contentLength, chunkSize int64, body io.ReadSeeker, accessKey, secretKey string) (*http.Request, error) {
+	req, err := newTestStreamingRequest(method, urlStr, contentLength, chunkSize, body)
+	if err != nil {
+		return nil, err
+	}
+
+	signature, err := signStreamingRequest(req, accessKey, secretKey)
+	if err != nil {
+		return nil, err
+	}
+
+	var stream []byte
+	var buffer []byte
+	body.Seek(0, 0)
+	for {
+		buffer = make([]byte, chunkSize)
+		n, err := body.Read(buffer)
+		if err != nil && err != io.EOF {
+			return nil, err
+		}
+
+		currTime := time.Now().UTC()
+		// Get scope.
+		scope := strings.Join([]string{
+			currTime.Format(yyyymmdd),
+			"us-east-1",
+			"s3",
+			"aws4_request",
+		}, "/")
+
+		stringToSign := "AWS4-HMAC-SHA256-PAYLOAD" + "\n"
+		stringToSign = stringToSign + currTime.Format(iso8601Format) + "\n"
+		stringToSign = stringToSign + scope + "\n"
+		stringToSign = stringToSign + signature + "\n"
+		stringToSign = stringToSign + "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + "\n" // hex(sum256(""))
+		stringToSign = stringToSign + hex.EncodeToString(sum256(buffer[:n]))
+
+		date := sumHMAC([]byte("AWS4"+secretKey), []byte(currTime.Format(yyyymmdd)))
+		region := sumHMAC(date, []byte("us-east-1"))
+		service := sumHMAC(region, []byte("s3"))
+		signingKey := sumHMAC(service, []byte("aws4_request"))
+
+		signature = hex.EncodeToString(sumHMAC(signingKey, []byte(stringToSign)))
+
+		stream = append(stream, []byte(fmt.Sprintf("%x", n)+";chunk-signature="+signature+"\r\n")...)
+		stream = append(stream, buffer[:n]...)
+		stream = append(stream, []byte("\r\n")...)
+
+		if n <= 0 {
+			break
+		}
+
+	}
+
+	req.Body = ioutil.NopCloser(bytes.NewReader(stream))
+	return req, nil
+}
+
 // Sign given request using Signature V4.
 func signRequest(req *http.Request, accessKey, secretKey string) error {
 	// Get hashed payload.