Fix storage class related issues (#5322)

- Add storage class metadata validation for request header
- Change storage class header values to be consistent with AWS S3
- Refactor internal method to take only the reqd argument

cmd/api-errors.go

@@ -164,6 +164,10 @@ const (
 	ErrOperationTimedOut
 	ErrPartsSizeUnequal
 	ErrInvalidRequest
+
+	// Minio storage class error codes
+	ErrInvalidStorageClass
+
 	// Add new extended error codes here.
 	// Please open a https://github.com/minio/minio/issues before adding
 	// new error codes here.
@@ -194,6 +198,11 @@ var errorCodeResponse = map[APIErrorCode]APIError{
 		Description:    "Unknown metadata directive.",
 		HTTPStatusCode: http.StatusBadRequest,
 	},
+	ErrInvalidStorageClass: {
+		Code:           "InvalidStorageClass",
+		Description:    "Invalid storage class.",
+		HTTPStatusCode: http.StatusBadRequest,
+	},
 	ErrInvalidRequestBody: {
 		Code:           "InvalidArgument",
 		Description:    "Body shouldn't be set for this request.",

cmd/common-main.go

@@ -119,27 +119,27 @@ func handleCommonEnvVars() {
 		var err error
 
 		// Check for environment variables and parse into storageClass struct
-		if ssc := os.Getenv(standardStorageClass); ssc != "" {
+		if ssc := os.Getenv(standardStorageClassEnv); ssc != "" {
 			globalStandardStorageClass, err = parseStorageClass(ssc)
-			fatalIf(err, "Invalid value set in environment variable %s.", standardStorageClass)
+			fatalIf(err, "Invalid value set in environment variable %s.", standardStorageClassEnv)
 		}
 
-		if rrsc := os.Getenv(reducedRedundancyStorageClass); rrsc != "" {
+		if rrsc := os.Getenv(reducedRedundancyStorageClassEnv); rrsc != "" {
 			globalRRStorageClass, err = parseStorageClass(rrsc)
-			fatalIf(err, "Invalid value set in environment variable %s.", reducedRedundancyStorageClass)
+			fatalIf(err, "Invalid value set in environment variable %s.", reducedRedundancyStorageClassEnv)
 		}
 
 		// Validation is done after parsing both the storage classes. This is needed because we need one
 		// storage class value to deduce the correct value of the other storage class.
 		if globalRRStorageClass.Scheme != "" {
 			err := validateRRSParity(globalRRStorageClass.Parity, globalStandardStorageClass.Parity)
-			fatalIf(err, "Invalid value set in environment variable %s.", reducedRedundancyStorageClass)
+			fatalIf(err, "Invalid value set in environment variable %s.", reducedRedundancyStorageClassEnv)
 			globalIsStorageClass = true
 		}
 
 		if globalStandardStorageClass.Scheme != "" {
 			err := validateSSParity(globalStandardStorageClass.Parity, globalRRStorageClass.Parity)
-			fatalIf(err, "Invalid value set in environment variable %s.", standardStorageClass)
+			fatalIf(err, "Invalid value set in environment variable %s.", standardStorageClassEnv)
 			globalIsStorageClass = true
 		}
 	}

cmd/gateway-handlers.go

@@ -186,6 +186,14 @@ func (api gatewayAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Re
 	bucket = vars["bucket"]
 	object = vars["object"]
 
+	// Validate storage class metadata if present
+	if _, ok := r.Header[amzStorageClassCanonical]; ok {
+		if !isValidStorageClassMeta(r.Header.Get(amzStorageClassCanonical)) {
+			writeErrorResponse(w, ErrInvalidStorageClass, r.URL)
+			return
+		}
+	}
+
 	// TODO: we should validate the object name here
 
 	// Get Content-Md5 sent by client and verify if valid

cmd/object-handlers.go

@@ -477,6 +477,14 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
 	bucket := vars["bucket"]
 	object := vars["object"]
 
+	// Validate storage class metadata if present
+	if _, ok := r.Header[amzStorageClassCanonical]; ok {
+		if !isValidStorageClassMeta(r.Header.Get(amzStorageClassCanonical)) {
+			writeErrorResponse(w, ErrInvalidStorageClass, r.URL)
+			return
+		}
+	}
+
 	// Get Content-Md5 sent by client and verify if valid
 	md5Bytes, err := checkValidMD5(r.Header.Get("Content-Md5"))
 	if err != nil {
@@ -659,6 +667,14 @@ func (api objectAPIHandlers) NewMultipartUploadHandler(w http.ResponseWriter, r
 		return
 	}
 
+	// Validate storage class metadata if present
+	if _, ok := r.Header[amzStorageClassCanonical]; ok {
+		if !isValidStorageClassMeta(r.Header.Get(amzStorageClassCanonical)) {
+			writeErrorResponse(w, ErrInvalidStorageClass, r.URL)
+			return
+		}
+	}
+
 	if IsSSECustomerRequest(r.Header) { // handle SSE-C requests
 		// SSE-C is not implemented for multipart operations yet
 		writeErrorResponse(w, ErrNotImplemented, r.URL)

cmd/object-handlers_test.go

@@ -801,6 +801,8 @@ func testAPIPutObjectHandler(obj ObjectLayer, instanceType, bucketName string, a
 	copySourceHeader.Set("X-Amz-Copy-Source", "somewhere")
 	invalidMD5Header := http.Header{}
 	invalidMD5Header.Set("Content-Md5", "42")
+	inalidStorageClassHeader := http.Header{}
+	inalidStorageClassHeader.Set(amzStorageClass, "INVALID")
 
 	addCustomHeaders := func(req *http.Request, customHeaders http.Header) {
 		for k, values := range customHeaders {
@@ -895,6 +897,18 @@ func testAPIPutObjectHandler(obj ObjectLayer, instanceType, bucketName string, a
 			fault:              MissingContentLength,
 			expectedRespStatus: http.StatusLengthRequired,
 		},
+		// Test case - 7.
+		// Test Case with invalid header key X-Amz-Storage-Class
+		{
+			bucketName:         bucketName,
+			objectName:         objectName,
+			headers:            inalidStorageClassHeader,
+			data:               bytesData,
+			dataLen:            len(bytesData),
+			accessKey:          credentials.AccessKey,
+			secretKey:          credentials.SecretKey,
+			expectedRespStatus: http.StatusBadRequest,
+		},
 	}
 	// Iterating over the cases, fetching the object validating the response.
 	for i, testCase := range testCases {

cmd/storage-class.go

@@ -26,10 +26,16 @@ import (
 const (
 	// metadata entry for storage class
 	amzStorageClass = "x-amz-storage-class"
+	// Canonical metadata entry for storage class
+	amzStorageClassCanonical = "X-Amz-Storage-Class"
 	// Reduced redundancy storage class
-	reducedRedundancyStorageClass = "MINIO_STORAGE_CLASS_RRS"
+	reducedRedundancyStorageClass = "REDUCED_REDUNDANCY"
 	// Standard storage class
-	standardStorageClass = "MINIO_STORAGE_CLASS_STANDARD"
+	standardStorageClass = "STANDARD"
+	// Reduced redundancy storage class environment variable
+	reducedRedundancyStorageClassEnv = "MINIO_STORAGE_CLASS_RRS"
+	// Standard storage class environment variable
+	standardStorageClassEnv = "MINIO_STORAGE_CLASS_STANDARD"
 	// Supported storage class scheme is EC
 	supportedStorageClassScheme = "EC"
 	// Minimum parity disks
@@ -48,6 +54,12 @@ type storageClassConfig struct {
 	RRS      string `json:"rrs"`
 }
 
+// Validate if storage class in metadata
+// Only Standard and RRS Storage classes are supported
+func isValidStorageClassMeta(sc string) bool {
+	return sc == reducedRedundancyStorageClass || sc == standardStorageClass
+}
+
 // Parses given storageClassEnv and returns a storageClass structure.
 // Supported Storage Class format is "Scheme:Number of parity disks".
 // Currently only supported scheme is "EC".
@@ -142,8 +154,7 @@ func validateSSParity(ssParity, rrsParity int) (err error) {
 // -- Default for Reduced Redundancy Storage class is, parity = 2 and data = N-Parity
 // -- Default for Standard Storage class is, parity = N/2, data = N/2
 // If storage class is not present in metadata, default value is data = N/2, parity = N/2
-func getDrivesCount(sc string, disks []StorageAPI) (data, parity int) {
-	totalDisks := len(disks)
+func getRedundancyCount(sc string, totalDisks int) (data, parity int) {
 	parity = totalDisks / 2
 	switch sc {
 	case reducedRedundancyStorageClass:

cmd/storage-class_test.go

@@ -152,11 +152,11 @@ func testValidateSSParity(obj ObjectLayer, instanceType string, dirs []string, t
 	}
 }
 
-func TestGetDrivesCount(t *testing.T) {
-	ExecObjectLayerTestWithDirs(t, testGetDrivesCount)
+func TestRedundancyCount(t *testing.T) {
+	ExecObjectLayerTestWithDirs(t, testGetRedundancyCount)
 }
 
-func testGetDrivesCount(obj ObjectLayer, instanceType string, dirs []string, t TestErrHandler) {
+func testGetRedundancyCount(obj ObjectLayer, instanceType string, dirs []string, t TestErrHandler) {
 	// Reset global storage class flags
 	resetGlobalStorageEnvs()
 	xl := obj.(*xlObjects)
@@ -183,7 +183,7 @@ func testGetDrivesCount(obj ObjectLayer, instanceType string, dirs []string, t T
 		if tt.name == 5 {
 			globalStandardStorageClass.Parity = 6
 		}
-		data, parity := getDrivesCount(tt.sc, tt.disks)
+		data, parity := getRedundancyCount(tt.sc, len(tt.disks))
 		if data != tt.expectedData {
 			t.Errorf("Test %d, Expected data disks %d, got %d", tt.name, tt.expectedData, data)
 			return
@@ -353,3 +353,25 @@ func testObjectQuorumFromMeta(obj ObjectLayer, instanceType string, dirs []strin
 		}
 	}
 }
+
+// Test isValidStorageClassMeta method with valid and invalid inputs
+func TestIsValidStorageClassMeta(t *testing.T) {
+	tests := []struct {
+		name int
+		sc   string
+		want bool
+	}{
+		{1, "STANDARD", true},
+		{2, "REDUCED_REDUNDANCY", true},
+		{3, "", false},
+		{4, "INVALID", false},
+		{5, "123", false},
+		{6, "MINIO_STORAGE_CLASS_RRS", false},
+		{7, "MINIO_STORAGE_CLASS_STANDARD", false},
+	}
+	for _, tt := range tests {
+		if got := isValidStorageClassMeta(tt.sc); got != tt.want {
+			t.Errorf("Test %d, Expected Storage Class to be %t, got %t", tt.name, tt.want, got)
+		}
+	}
+}

cmd/xl-v1-multipart.go

@@ -501,7 +501,7 @@ func (xl xlObjects) ListMultipartUploads(bucket, object, keyMarker, uploadIDMark
 // operation(s) on the object.
 func (xl xlObjects) newMultipartUpload(bucket string, object string, meta map[string]string) (string, error) {
 
-	dataBlocks, parityBlocks := getDrivesCount(meta[amzStorageClass], xl.storageDisks)
+	dataBlocks, parityBlocks := getRedundancyCount(meta[amzStorageClass], len(xl.storageDisks))
 
 	xlMeta := newXLMetaV1(object, dataBlocks, parityBlocks)
 

cmd/xl-v1-object.go

@@ -508,7 +508,7 @@ func (xl xlObjects) PutObject(bucket string, object string, data *hash.Reader, m
 		}
 	}
 	// Get parity and data drive count based on storage class metadata
-	dataDrives, parityDrives := getDrivesCount(metadata[amzStorageClass], xl.storageDisks)
+	dataDrives, parityDrives := getRedundancyCount(metadata[amzStorageClass], len(xl.storageDisks))
 
 	// we now know the number of blocks this object needs for data and parity.
 	// writeQuorum is dataBlocks + 1

cmd/xl-v1.go

@@ -246,10 +246,10 @@ func getStorageInfo(disks []StorageAPI) StorageInfo {
 	storageInfo.Backend.OnlineDisks = onlineDisks
 	storageInfo.Backend.OfflineDisks = offlineDisks
 
-	_, scParity := getDrivesCount(standardStorageClass, disks)
+	_, scParity := getRedundancyCount(standardStorageClass, len(disks))
 	storageInfo.Backend.standardSCParity = scParity
 
-	_, rrSCparity := getDrivesCount(reducedRedundancyStorageClass, disks)
+	_, rrSCparity := getRedundancyCount(reducedRedundancyStorageClass, len(disks))
 	storageInfo.Backend.rrSCParity = rrSCparity
 
 	return storageInfo