xl/CreateFile: handle errFileNameTooLong error properly (#1523)

When errFileNameTooLong error is returned from posix, xl.CreateFile()
treats the error specially by returning the same error immediately.

Fixes #1501

object-errors.go

@@ -58,6 +58,13 @@ func toObjectErr(err error, params ...string) error {
 				Object: params[1],
 			}
 		}
+	case errFileNameTooLong:
+		if len(params) >= 2 {
+			return ObjectNameInvalid{
+				Bucket: params[0],
+				Object: params[1],
+			}
+		}
 	case io.ErrUnexpectedEOF, io.ErrShortWrite:
 		return IncompleteBody{}
 	}

posix.go

@@ -20,6 +20,7 @@ import (
 	"io"
 	"os"
 	slashpath "path"
+	"runtime"
 	"strings"
 	"syscall"
 
@@ -38,6 +39,31 @@ type fsStorage struct {
 	minFreeDisk int64
 }
 
+// checkPathLength - returns error if given path name length more than 255
+func checkPathLength(pathName string) error {
+	// For MS Windows, the maximum path length is 255
+	if runtime.GOOS == "windows" {
+		if len(pathName) > 255 {
+			return errFileNameTooLong
+		}
+
+		return nil
+	}
+
+	// For non-windows system, check each path segment length is > 255
+	for len(pathName) > 0 && pathName != "." && pathName != "/" {
+		dir, file := slashpath.Dir(pathName), slashpath.Base(pathName)
+
+		if len(file) > 255 {
+			return errFileNameTooLong
+		}
+
+		pathName = dir
+	}
+
+	return nil
+}
+
 // isDirEmpty - returns whether given directory is empty or not.
 func isDirEmpty(dirname string) bool {
 	f, err := os.Open(dirname)
@@ -67,6 +93,9 @@ func newPosix(diskPath string) (StorageAPI, error) {
 		log.Error("Disk cannot be empty")
 		return nil, errInvalidArgument
 	}
+	if err := checkPathLength(diskPath); err != nil {
+		return nil, err
+	}
 	st, err := os.Stat(diskPath)
 	if err != nil {
 		log.WithFields(logrus.Fields{
@@ -93,6 +122,9 @@ func newPosix(diskPath string) (StorageAPI, error) {
 
 // checkDiskFree verifies if disk path has sufficient minium free disk space.
 func checkDiskFree(diskPath string, minFreeDisk int64) (err error) {
+	if err = checkPathLength(diskPath); err != nil {
+		return err
+	}
 	di, err := disk.GetInfo(diskPath)
 	if err != nil {
 		log.WithFields(logrus.Fields{
@@ -138,6 +170,9 @@ func removeDuplicateVols(volsInfo []VolInfo) []VolInfo {
 
 // gets all the unique directories from diskPath.
 func getAllUniqueVols(dirPath string) ([]VolInfo, error) {
+	if err := checkPathLength(dirPath); err != nil {
+		return nil, err
+	}
 	entries, err := readDir(dirPath)
 	if err != nil {
 		log.WithFields(logrus.Fields{
@@ -181,6 +216,9 @@ func (s fsStorage) getVolumeDir(volume string) (string, error) {
 	if !isValidVolname(volume) {
 		return "", errInvalidArgument
 	}
+	if err := checkPathLength(volume); err != nil {
+		return "", err
+	}
 	volumeDir := pathJoin(s.diskPath, volume)
 	_, err := os.Stat(volumeDir)
 	if err == nil {
@@ -395,6 +433,9 @@ func (s fsStorage) ReadFile(volume string, path string, offset int64) (readClose
 	}
 
 	filePath := pathJoin(volumeDir, path)
+	if err = checkPathLength(filePath); err != nil {
+		return nil, err
+	}
 	file, err := os.Open(filePath)
 	if err != nil {
 		if os.IsNotExist(err) {
@@ -451,6 +492,9 @@ func (s fsStorage) CreateFile(volume, path string) (writeCloser io.WriteCloser,
 		return nil, err
 	}
 	filePath := pathJoin(volumeDir, path)
+	if err = checkPathLength(filePath); err != nil {
+		return nil, err
+	}
 	// Verify if the file already exists and is not of regular type.
 	var st os.FileInfo
 	if st, err = os.Stat(filePath); err == nil {
@@ -485,6 +529,9 @@ func (s fsStorage) StatFile(volume, path string) (file FileInfo, err error) {
 	}
 
 	filePath := slashpath.Join(volumeDir, path)
+	if err = checkPathLength(filePath); err != nil {
+		return FileInfo{}, err
+	}
 	st, err := os.Stat(filePath)
 	if err != nil {
 		log.WithFields(logrus.Fields{
@@ -577,6 +624,9 @@ func (s fsStorage) DeleteFile(volume, path string) error {
 	// Following code is needed so that we retain "/" suffix if any in
 	// path argument.
 	filePath := pathJoin(volumeDir, path)
+	if err = checkPathLength(filePath); err != nil {
+		return err
+	}
 
 	// Delete file and delete parent directory as well if its empty.
 	return deleteFile(volumeDir, filePath)

server_xl_test.go

@@ -19,6 +19,7 @@ package main
 import (
 	"bytes"
 	"crypto/md5"
+	"fmt"
 	"io"
 	"io/ioutil"
 	"os"
@@ -678,6 +679,25 @@ func (s *MyAPIXLSuite) TestPutObject(c *C) {
 	c.Assert(response.StatusCode, Equals, http.StatusOK)
 }
 
+func (s *MyAPIXLSuite) TestPutObjectLongName(c *C) {
+	request, err := s.newRequest("PUT", testAPIXLServer.URL+"/put-object-long-name", 0, nil)
+	c.Assert(err, IsNil)
+
+	client := http.Client{}
+	response, err := client.Do(request)
+	c.Assert(err, IsNil)
+	c.Assert(response.StatusCode, Equals, http.StatusOK)
+
+	buffer := bytes.NewReader([]byte("hello world"))
+	longObjName := fmt.Sprintf("%0256d", 1)
+	request, err = s.newRequest("PUT", testAPIXLServer.URL+"/put-object-long-name/"+longObjName, int64(buffer.Len()), buffer)
+	c.Assert(err, IsNil)
+
+	response, err = client.Do(request)
+	c.Assert(err, IsNil)
+	c.Assert(response.StatusCode, Equals, http.StatusNotFound)
+}
+
 func (s *MyAPIXLSuite) TestListBuckets(c *C) {
 	request, err := s.newRequest("GET", testAPIXLServer.URL+"/", 0, nil)
 	c.Assert(err, IsNil)

storage-errors.go

@@ -27,6 +27,9 @@ var errDiskNotFound = errors.New("disk not found")
 // errFileNotFound - cannot find the file.
 var errFileNotFound = errors.New("file not found")
 
+// errFileNameTooLong - given file name is too long than supported length.
+var errFileNameTooLong = errors.New("file name too long")
+
 // errVolumeExists - cannot create same volume again.
 var errVolumeExists = errors.New("volume already exists")
 

xl-erasure-v1-createfile.go

@@ -105,6 +105,14 @@ func (xl XL) writeErasure(volume, path string, reader *io.PipeReader, wcloser *w
 				"volume": volume,
 				"path":   path,
 			}).Errorf("CreateFile failed with %s", err)
+
+			// treat errFileNameTooLong specially
+			if err == errFileNameTooLong {
+				xl.cleanupCreateFileOps(volume, path, append(writers, metadataWriters...)...)
+				reader.CloseWithError(err)
+				return
+			}
+
 			createFileError++
 
 			// We can safely allow CreateFile errors up to len(xl.storageDisks) - xl.writeQuorum