Merge branch 'master' into support_ssd_volume

6 files changed, 109 insertions, 15 deletions

diff --git a/weed/s3api/auth_credentials.go b/weed/s3api/auth_credentials.go
index 93544b75e..b8af6381a 100644
--- a/weed/s3api/auth_credentials.go
+++ b/weed/s3api/auth_credentials.go
@@ -3,6 +3,7 @@ package s3api
 import (
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/s3api/s3_constants"
 	"io/ioutil"
 	"net/http"
 
@@ -185,7 +186,6 @@ func (iam *IdentityAccessManagement) authRequest(r *http.Request, action Action)
 		return identity, s3err.ErrNotImplemented
 	}
 
-	glog.V(3).Infof("auth error: %v", s3Err)
 	if s3Err != s3err.ErrNone {
 		return identity, s3Err
 	}
@@ -202,6 +202,44 @@ func (iam *IdentityAccessManagement) authRequest(r *http.Request, action Action)
 
 }
 
+func (iam *IdentityAccessManagement) authUser(r *http.Request) (*Identity, s3err.ErrorCode) {
+	var identity *Identity
+	var s3Err s3err.ErrorCode
+	var found bool
+	switch getRequestAuthType(r) {
+	case authTypeStreamingSigned:
+		return identity, s3err.ErrNone
+	case authTypeUnknown:
+		glog.V(3).Infof("unknown auth type")
+		return identity, s3err.ErrAccessDenied
+	case authTypePresignedV2, authTypeSignedV2:
+		glog.V(3).Infof("v2 auth type")
+		identity, s3Err = iam.isReqAuthenticatedV2(r)
+	case authTypeSigned, authTypePresigned:
+		glog.V(3).Infof("v4 auth type")
+		identity, s3Err = iam.reqSignatureV4Verify(r)
+	case authTypePostPolicy:
+		glog.V(3).Infof("post policy auth type")
+		return identity, s3err.ErrNone
+	case authTypeJWT:
+		glog.V(3).Infof("jwt auth type")
+		return identity, s3err.ErrNotImplemented
+	case authTypeAnonymous:
+		identity, found = iam.lookupAnonymous()
+		if !found {
+			return identity, s3err.ErrAccessDenied
+		}
+	default:
+		return identity, s3err.ErrNotImplemented
+	}
+
+	glog.V(3).Infof("auth error: %v", s3Err)
+	if s3Err != s3err.ErrNone {
+		return identity, s3Err
+	}
+	return identity, s3err.ErrNone
+}
+
 func (identity *Identity) canDo(action Action, bucket string) bool {
 	if identity.isAdmin() {
 		return true
@@ -215,10 +253,14 @@ func (identity *Identity) canDo(action Action, bucket string) bool {
 		return false
 	}
 	limitedByBucket := string(action) + ":" + bucket
+	adminLimitedByBucket := s3_constants.ACTION_ADMIN + ":" + bucket
 	for _, a := range identity.Actions {
 		if string(a) == limitedByBucket {
 			return true
 		}
+		if string(a) == adminLimitedByBucket {
+			return true
+		}
 	}
 	return false
 }
diff --git a/weed/s3api/http/header.go b/weed/s3api/http/header.go
index a960d2370..6614b0af0 100644
--- a/weed/s3api/http/header.go
+++ b/weed/s3api/http/header.go
@@ -31,6 +31,6 @@ const (
 
 // Non-Standard S3 HTTP request constants
 const (
-	AmzIdentityId = "x-amz-identity-id"
-	AmzIsAdmin    = "x-amz-is-admin" // only set to http request header as a context
+	AmzIdentityId = "s3-identity-id"
+	AmzIsAdmin    = "s3-is-admin" // only set to http request header as a context
 )
diff --git a/weed/s3api/s3api_bucket_handlers.go b/weed/s3api/s3api_bucket_handlers.go
index 00b7382cc..338f82668 100644
--- a/weed/s3api/s3api_bucket_handlers.go
+++ b/weed/s3api/s3api_bucket_handlers.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/xml"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/s3api/s3_constants"
 	"math"
 	"net/http"
 	"time"
@@ -26,6 +27,16 @@ type ListAllMyBucketsResult struct {
 
 func (s3a *S3ApiServer) ListBucketsHandler(w http.ResponseWriter, r *http.Request) {
 
+	var identity *Identity
+	var s3Err s3err.ErrorCode
+	if s3a.iam.isEnabled() {
+		identity, s3Err = s3a.iam.authUser(r)
+		if s3Err != s3err.ErrNone {
+			writeErrorResponse(w, s3Err, r.URL)
+			return
+		}
+	}
+
 	var response ListAllMyBucketsResult
 
 	entries, _, err := s3a.list(s3a.option.BucketsPath, "", "", false, math.MaxInt32)
@@ -40,7 +51,7 @@ func (s3a *S3ApiServer) ListBucketsHandler(w http.ResponseWriter, r *http.Reques
 	var buckets []*s3.Bucket
 	for _, entry := range entries {
 		if entry.IsDirectory {
-			if !s3a.hasAccess(r, entry) {
+			if identity != nil && !identity.canDo(s3_constants.ACTION_ADMIN, entry.Name) {
 				continue
 			}
 			buckets = append(buckets, &s3.Bucket{
diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go
index 401e2f96c..c9e124328 100644
--- a/weed/s3api/s3api_object_handlers.go
+++ b/weed/s3api/s3api_object_handlers.go
@@ -8,6 +8,7 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"sort"
 	"strings"
 
 	"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
@@ -175,16 +176,15 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 	var deletedObjects []ObjectIdentifier
 	var deleteErrors []DeleteError
 
+	directoriesWithDeletion := make(map[string]int)
+
 	s3a.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
+		// delete file entries
 		for _, object := range deleteObjects.Objects {
-			response, _ := s3a.listFilerEntries(bucket, object.ObjectName, 1, "", "/")
-			if len(response.Contents) != 0 && strings.HasSuffix(object.ObjectName, "/") {
-				continue
-			}
 
 			lastSeparator := strings.LastIndex(object.ObjectName, "/")
-			parentDirectoryPath, entryName, isDeleteData, isRecursive := "/", object.ObjectName, true, true
+			parentDirectoryPath, entryName, isDeleteData, isRecursive := "/", object.ObjectName, true, false
 			if lastSeparator > 0 && lastSeparator+1 < len(object.ObjectName) {
 				entryName = object.ObjectName[lastSeparator+1:]
 				parentDirectoryPath = "/" + object.ObjectName[:lastSeparator]
@@ -193,8 +193,10 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 
 			err := doDeleteEntry(client, parentDirectoryPath, entryName, isDeleteData, isRecursive)
 			if err == nil {
+				directoriesWithDeletion[parentDirectoryPath]++
 				deletedObjects = append(deletedObjects, object)
 			} else {
+				delete(directoriesWithDeletion, parentDirectoryPath)
 				deleteErrors = append(deleteErrors, DeleteError{
 					Code:    "",
 					Message: err.Error(),
@@ -202,6 +204,12 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 				})
 			}
 		}
+
+		// purge empty folders, only checking folders with deletions
+		for len(directoriesWithDeletion) > 0 {
+			directoriesWithDeletion = doDeleteEmptyDirectories(client, directoriesWithDeletion)
+		}
+
 		return nil
 	})
 
@@ -215,6 +223,26 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 
 }
 
+func doDeleteEmptyDirectories(client filer_pb.SeaweedFilerClient, directoriesWithDeletion map[string]int) (newDirectoriesWithDeletion map[string]int){
+	var allDirs []string
+	for dir, _ := range directoriesWithDeletion {
+		allDirs = append(allDirs, dir)
+	}
+	sort.Slice(allDirs, func(i, j int) bool {
+		return len(allDirs[i]) > len(allDirs[j])
+	})
+	newDirectoriesWithDeletion = make(map[string]int)
+	for _, dir := range allDirs {
+		parentDir, dirName := util.FullPath(dir).DirAndName()
+		if err := doDeleteEntry(client, parentDir, dirName, false, false); err != nil {
+			glog.V(4).Infof("directory %s has %d deletion but still not empty: %v", dir, directoriesWithDeletion[dir], err)
+		} else {
+			newDirectoriesWithDeletion[parentDir]++
+		}
+	}
+	return
+}
+
 var passThroughHeaders = []string{
 	"response-cache-control",
 	"response-content-disposition",
@@ -268,8 +296,10 @@ func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, des
 	defer util.CloseResponse(resp)
 
 	if (resp.ContentLength == -1 || resp.StatusCode == 404) && !strings.HasSuffix(destUrl, "/") {
-		writeErrorResponse(w, s3err.ErrNoSuchKey, r.URL)
-		return
+		if r.Method != "DELETE" {
+			writeErrorResponse(w, s3err.ErrNoSuchKey, r.URL)
+			return
+		}
 	}
 
 	responseFn(resp, w)
diff --git a/weed/s3api/s3api_objects_list_handlers.go b/weed/s3api/s3api_objects_list_handlers.go
index 0af099967..2d36c6ec9 100644
--- a/weed/s3api/s3api_objects_list_handlers.go
+++ b/weed/s3api/s3api_objects_list_handlers.go
@@ -71,7 +71,7 @@ func (s3a *S3ApiServer) ListObjectsV2Handler(w http.ResponseWriter, r *http.Requ
 		ContinuationToken:     continuationToken,
 		Delimiter:             response.Delimiter,
 		IsTruncated:           response.IsTruncated,
-		KeyCount:              len(response.Contents),
+		KeyCount:              len(response.Contents) + len(response.CommonPrefixes),
 		MaxKeys:               response.MaxKeys,
 		NextContinuationToken: response.NextMarker,
 		Prefix:                response.Prefix,
@@ -264,8 +264,10 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d
 					}
 				} else {
 					var isEmpty bool
-					if isEmpty, err = s3a.isDirectoryAllEmpty(client, dir, entry.Name); err != nil {
-						glog.Errorf("check empty folder %s: %v", dir, err)
+					if !s3a.option.AllowEmptyFolder {
+						if isEmpty, err = s3a.isDirectoryAllEmpty(client, dir, entry.Name); err != nil {
+							glog.Errorf("check empty folder %s: %v", dir, err)
+						}
 					}
 					if !isEmpty {
 						eachEntryFn(dir, entry)
@@ -310,13 +312,17 @@ func getListObjectsV1Args(values url.Values) (prefix, marker, delimiter string,
 
 func (s3a *S3ApiServer) isDirectoryAllEmpty(filerClient filer_pb.SeaweedFilerClient, parentDir, name string) (isEmpty bool, err error) {
 	// println("+ isDirectoryAllEmpty", dir, name)
+	glog.V(4).Infof("+ isEmpty %s/%s", parentDir, name)
+	defer glog.V(4).Infof("- isEmpty %s/%s %v", parentDir, name, isEmpty)
 	var fileCounter int
 	var subDirs []string
 	currentDir := parentDir + "/" + name
 	var startFrom string
 	var isExhausted bool
+	var foundEntry bool
 	for fileCounter == 0 && !isExhausted && err == nil {
 		err = filer_pb.SeaweedList(filerClient, currentDir, "", func(entry *filer_pb.Entry, isLast bool) error {
+			foundEntry = true
 			if entry.IsDirectory {
 				subDirs = append(subDirs, entry.Name)
 			} else {
@@ -324,8 +330,12 @@ func (s3a *S3ApiServer) isDirectoryAllEmpty(filerClient filer_pb.SeaweedFilerCli
 			}
 			startFrom = entry.Name
 			isExhausted = isExhausted || isLast
+			glog.V(4).Infof("    * %s/%s isLast: %t", currentDir, startFrom, isLast)
 			return nil
 		}, startFrom, false, 8)
+		if !foundEntry {
+			break
+		}
 	}
 
 	if err != nil {
diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go
index 93e2bb575..4993104ae 100644
--- a/weed/s3api/s3api_server.go
+++ b/weed/s3api/s3api_server.go
@@ -20,6 +20,7 @@ type S3ApiServerOption struct {
 	DomainName       string
 	BucketsPath      string
 	GrpcDialOption   grpc.DialOption
+	AllowEmptyFolder bool
 }
 
 type S3ApiServer struct {
@@ -128,7 +129,7 @@ func (s3a *S3ApiServer) registerRouter(router *mux.Router) {
 	}
 
 	// ListBuckets
-	apiRouter.Methods("GET").Path("/").HandlerFunc(track(s3a.iam.Auth(s3a.ListBucketsHandler, ACTION_ADMIN), "LIST"))
+	apiRouter.Methods("GET").Path("/").HandlerFunc(track(s3a.ListBucketsHandler, "LIST"))
 
 	// NotFound
 	apiRouter.NotFoundHandler = http.HandlerFunc(notFoundHandler)