fs: Verify if parent is an object before i/o. (#4304)

PutObject() needs to verify and fail.

Fixes #4301

cmd/fs-v1.go

@@ -25,6 +25,7 @@ import (
 	"io/ioutil"
 	"os"
 	"os/signal"
+	"path"
 	"path/filepath"
 	"sort"
 	"syscall"
@@ -473,7 +474,6 @@ func (fs fsObjects) CopyObject(srcBucket, srcObject, dstBucket, dstObject string
 // startOffset indicates the starting read location of the object.
 // length indicates the total length of the object.
 func (fs fsObjects) GetObject(bucket, object string, offset int64, length int64, writer io.Writer) (err error) {
-	// This is a special case with object whose name ends with
 	if err = checkGetObjArgs(bucket, object); err != nil {
 		return err
 	}
@@ -580,6 +580,25 @@ func (fs fsObjects) GetObjectInfo(bucket, object string) (ObjectInfo, error) {
 	return fs.getObjectInfo(bucket, object)
 }
 
+// This function does the following check, suppose
+// object is "a/b/c/d", stat makes sure that objects ""a/b/c""
+// "a/b" and "a" do not exist.
+func (fs fsObjects) parentDirIsObject(bucket, parent string) bool {
+	var isParentDirObject func(string) bool
+	isParentDirObject = func(p string) bool {
+		if p == "." {
+			return false
+		}
+		if _, err := fsStatFile(pathJoin(fs.fsPath, bucket, p)); err == nil {
+			// If there is already a file at prefix "p" return error.
+			return true
+		}
+		// Check if there is a file as one of the parent paths.
+		return isParentDirObject(path.Dir(p))
+	}
+	return isParentDirObject(parent)
+}
+
 // PutObject - creates an object upon reading from the input stream
 // until EOF, writes data directly to configured filesystem path.
 // Additionally writes `fs.json` which carries the necessary metadata
@@ -590,6 +609,10 @@ func (fs fsObjects) PutObject(bucket string, object string, size int64, data io.
 	// 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)) {
+			return ObjectInfo{}, toObjectErr(traceError(errFileAccessDenied), bucket, object)
+		}
 		return dirObjectInfo(bucket, object, size, metadata), nil
 	}
 
@@ -597,6 +620,11 @@ func (fs fsObjects) PutObject(bucket string, object string, size int64, data io.
 		return ObjectInfo{}, err
 	}
 
+	// Check if an object is present as one of the parent dir.
+	if fs.parentDirIsObject(bucket, path.Dir(object)) {
+		return ObjectInfo{}, toObjectErr(traceError(errFileAccessDenied), bucket, object)
+	}
+
 	if _, err = fs.statBucketDir(bucket); err != nil {
 		return ObjectInfo{}, toObjectErr(err, bucket)
 	}