fs: Make sure to validate bucket first in PutObject() (#4427)

Currently even when bucket doesn't exist we wrongly
return success, when an object is a directory prefix with
 '/' as suffix and is of size 0.

This PR fixes this behavior.

cmd/fs-v1.go

@@ -605,9 +605,15 @@ func (fs fsObjects) parentDirIsObject(bucket, parent string) bool {
 // for future object operations.
 func (fs fsObjects) PutObject(bucket string, object string, size int64, data io.Reader, metadata map[string]string, sha256sum string) (objInfo ObjectInfo, retErr error) {
 	var err error
-	// This is a special case with size as '0' and object ends with
+
-	// a slash separator, we treat it like a valid operation and
+	// Validate if bucket name is valid and exists.
-	// return success.
+	if _, err = fs.statBucketDir(bucket); err != nil {
+		return ObjectInfo{}, toObjectErr(err, bucket)
+	}
+
+	// This is a special case with size as '0' and object ends
+	// with a slash separator, we treat it like a valid operation
+	// and return success.
 	if isObjectDir(object, size) {
 		// Check if an object is present as one of the parent dir.
 		if fs.parentDirIsObject(bucket, path.Dir(object)) {
@@ -625,10 +631,6 @@ func (fs fsObjects) PutObject(bucket string, object string, size int64, data io.
 		return ObjectInfo{}, toObjectErr(traceError(errFileAccessDenied), bucket, object)
 	}
 
-	if _, err = fs.statBucketDir(bucket); err != nil {
-		return ObjectInfo{}, toObjectErr(err, bucket)
-	}
-
 	// No metadata is set, allocate a new one.
 	if metadata == nil {
 		metadata = make(map[string]string)

cmd/fs-v1_test.go

@@ -523,6 +523,7 @@ func TestFSGetBucketInfo(t *testing.T) {
 	}
 }
 
+// Tests FS backend put object behavior.
 func TestFSPutObject(t *testing.T) {
 	// Prepare for tests
 	disk := filepath.Join(globalTestTmpDir, "minio-"+nextSuffix())
@@ -535,14 +536,34 @@ func TestFSPutObject(t *testing.T) {
 	if err := obj.MakeBucket(bucketName); err != nil {
 		t.Fatal(err)
 	}
+
 	sha256sum := ""
-	_, err := obj.PutObject(bucketName, objectName, int64(len("abcd")), bytes.NewReader([]byte("abcd")), nil, sha256sum)
+
+	// With a regular object.
+	_, err := obj.PutObject(bucketName+"non-existent", objectName, int64(len("abcd")), bytes.NewReader([]byte("abcd")), nil, sha256sum)
+	if err == nil {
+		t.Fatal("Unexpected should fail here, bucket doesn't exist")
+	}
+	if _, ok := errorCause(err).(BucketNotFound); !ok {
+		t.Fatalf("Expected error type BucketNotFound, got %#v", err)
+	}
+
+	// With a directory object.
+	_, err = obj.PutObject(bucketName+"non-existent", objectName+"/", int64(0), bytes.NewReader([]byte("")), nil, sha256sum)
+	if err == nil {
+		t.Fatal("Unexpected should fail here, bucket doesn't exist")
+	}
+	if _, ok := errorCause(err).(BucketNotFound); !ok {
+		t.Fatalf("Expected error type BucketNotFound, got %#v", err)
+	}
+
+	_, err = obj.PutObject(bucketName, objectName, int64(len("abcd")), bytes.NewReader([]byte("abcd")), nil, sha256sum)
 	if err != nil {
 		t.Fatal(err)
 	}
 	_, err = obj.PutObject(bucketName, objectName+"/1", int64(len("abcd")), bytes.NewReader([]byte("abcd")), nil, sha256sum)
 	if err == nil {
-		t.Fatal("Unexpected should fail here, backned corruption occurred")
+		t.Fatal("Unexpected should fail here, backend corruption occurred")
 	}
 	if nerr, ok := errorCause(err).(PrefixAccessDenied); !ok {
 		t.Fatalf("Expected PrefixAccessDenied, got %#v", err)