fix: commonPrefixes behavior in ListObjectVersions (#10286)

```
$ aws s3api --profile minio --endpoint-url http://localhost:9003 \
    list-object-versions --bucket testbucket \
    --delimiter / --prefix Veeam/Archive/

{
    "CommonPrefixes": [
        {
            "Prefix": "Veeam/Archive/003/"
        }
    ]
}
```

Also add coverage tests similar to ListObjects to
catch errors in future, skip these tests in FS mode

cmd/admin-handlers.go

@@ -1270,10 +1270,9 @@ func (a adminAPIHandlers) OBDInfoHandler(w http.ResponseWriter, r *http.Request)
 	}
 
 	deadlinedCtx, cancel := context.WithTimeout(ctx, deadline)
-
 	defer cancel()
 
-	nsLock := objectAPI.NewNSLock(deadlinedCtx, minioMetaBucket, "obd-in-progress")
+	nsLock := objectAPI.NewNSLock(ctx, minioMetaBucket, "obd-in-progress")
 	if err := nsLock.GetLock(newDynamicTimeout(deadline, deadline)); err != nil { // returns a locked lock
 		errResp(err)
 		return

cmd/admin-heal-ops.go

@@ -876,9 +876,9 @@ func (h *healSequence) healBucket(bucket string, bucketsOnly bool) error {
 		if h.object != "" {
 			// Check if an object named as the objPrefix exists,
 			// and if so heal it.
-			_, err := objectAPI.GetObjectInfo(h.ctx, bucket, h.object, ObjectOptions{})
+			oi, err := objectAPI.GetObjectInfo(h.ctx, bucket, h.object, ObjectOptions{})
 			if err == nil {
-				if err = h.healObject(bucket, h.object, ""); err != nil {
+				if err = h.healObject(bucket, h.object, oi.VersionID); err != nil {
 					return err
 				}
 			}

cmd/bucket-listobjects-handlers.go

@@ -24,12 +24,36 @@ import (
 	"strings"
 
 	"github.com/gorilla/mux"
-	"github.com/minio/minio/cmd/crypto"
 	"github.com/minio/minio/cmd/logger"
 
 	"github.com/minio/minio/pkg/bucket/policy"
+	"github.com/minio/minio/pkg/sync/errgroup"
 )
 
+func concurrentDecryptETag(ctx context.Context, objects []ObjectInfo) {
+	inParallel := func(objects []ObjectInfo) {
+		g := errgroup.WithNErrs(len(objects))
+		for index := range objects {
+			index := index
+			g.Go(func() error {
+				objects[index].ETag = objects[index].GetActualETag(nil)
+				objects[index].Size, _ = objects[index].GetActualSize()
+				return nil
+			}, index)
+		}
+		g.Wait()
+	}
+	const maxConcurrent = 500
+	for {
+		if len(objects) < maxConcurrent {
+			inParallel(objects)
+			return
+		}
+		inParallel(objects[:maxConcurrent])
+		objects = objects[maxConcurrent:]
+	}
+}
+
 // Validate all the ListObjects query arguments, returns an APIErrorCode
 // if one of the args do not meet the required conditions.
 // Special conditions required by MinIO server are as below
@@ -89,6 +113,10 @@ func (api objectAPIHandlers) ListObjectVersionsHandler(w http.ResponseWriter, r
 		return
 	}
 
+	if proxyRequestByBucket(ctx, w, r, bucket) {
+		return
+	}
+
 	listObjectVersions := objectAPI.ListObjectVersions
 
 	// Inititate a list object versions operation based on the input params.
@@ -100,16 +128,7 @@ func (api objectAPIHandlers) ListObjectVersionsHandler(w http.ResponseWriter, r
 		return
 	}
 
-	for i := range listObjectVersionsInfo.Objects {
-		if crypto.IsEncrypted(listObjectVersionsInfo.Objects[i].UserDefined) {
-			listObjectVersionsInfo.Objects[i].ETag = getDecryptedETag(r.Header, listObjectVersionsInfo.Objects[i], false)
-		}
-		listObjectVersionsInfo.Objects[i].Size, err = listObjectVersionsInfo.Objects[i].GetActualSize()
-		if err != nil {
-			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
-			return
-		}
-	}
+	concurrentDecryptETag(ctx, listObjectVersionsInfo.Objects)
 
 	response := generateListVersionsResponse(bucket, prefix, marker, versionIDMarker, delimiter, encodingType, maxkeys, listObjectVersionsInfo)
 
@@ -178,16 +197,7 @@ func (api objectAPIHandlers) ListObjectsV2MHandler(w http.ResponseWriter, r *htt
 		return
 	}
 
-	for i := range listObjectsV2Info.Objects {
-		if crypto.IsEncrypted(listObjectsV2Info.Objects[i].UserDefined) {
-			listObjectsV2Info.Objects[i].ETag = getDecryptedETag(r.Header, listObjectsV2Info.Objects[i], false)
-		}
-		listObjectsV2Info.Objects[i].Size, err = listObjectsV2Info.Objects[i].GetActualSize()
-		if err != nil {
-			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
-			return
-		}
-	}
+	concurrentDecryptETag(ctx, listObjectsV2Info.Objects)
 
 	// The next continuation token has id@node_index format to optimize paginated listing
 	nextContinuationToken := listObjectsV2Info.NextContinuationToken
@@ -264,16 +274,7 @@ func (api objectAPIHandlers) ListObjectsV2Handler(w http.ResponseWriter, r *http
 		return
 	}
 
-	for i := range listObjectsV2Info.Objects {
-		if crypto.IsEncrypted(listObjectsV2Info.Objects[i].UserDefined) {
-			listObjectsV2Info.Objects[i].ETag = getDecryptedETag(r.Header, listObjectsV2Info.Objects[i], false)
-		}
-		listObjectsV2Info.Objects[i].Size, err = listObjectsV2Info.Objects[i].GetActualSize()
-		if err != nil {
-			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
-			return
-		}
-	}
+	concurrentDecryptETag(ctx, listObjectsV2Info.Objects)
 
 	// The next continuation token has id@node_index format to optimize paginated listing
 	nextContinuationToken := listObjectsV2Info.NextContinuationToken
@@ -396,16 +397,7 @@ func (api objectAPIHandlers) ListObjectsV1Handler(w http.ResponseWriter, r *http
 		return
 	}
 
-	for i := range listObjectsInfo.Objects {
-		if crypto.IsEncrypted(listObjectsInfo.Objects[i].UserDefined) {
-			listObjectsInfo.Objects[i].ETag = getDecryptedETag(r.Header, listObjectsInfo.Objects[i], false)
-		}
-		listObjectsInfo.Objects[i].Size, err = listObjectsInfo.Objects[i].GetActualSize()
-		if err != nil {
-			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
-			return
-		}
-	}
+	concurrentDecryptETag(ctx, listObjectsInfo.Objects)
 
 	response := generateListObjectsV1Response(bucket, prefix, marker, delimiter, encodingType, maxKeys, listObjectsInfo)
 

cmd/bucket-policy.go

@@ -68,6 +68,12 @@ func getConditionValues(r *http.Request, lc string, username string, claims map[
 	principalType := "Anonymous"
 	if username != "" {
 		principalType = "User"
+		if len(claims) > 0 {
+			principalType = "AssumedRole"
+		}
+		if username == globalActiveCred.AccessKey {
+			principalType = "Account"
+		}
 	}
 
 	vid := r.URL.Query().Get("versionId")
@@ -143,6 +149,10 @@ func getConditionValues(r *http.Request, lc string, username string, claims map[
 	for k, v := range claims {
 		vStr, ok := v.(string)
 		if ok {
+			// Special case for AD/LDAP STS users
+			if k == ldapUser {
+				args[ldapUserPolicyVariable] = []string{vStr}
+			}
 			args[k] = []string{vStr}
 		}
 	}

cmd/config-current.go

@@ -455,28 +455,32 @@ func lookupConfigs(s config.Config, setDriveCount int) {
 	for _, l := range loggerCfg.HTTP {
 		if l.Enabled {
 			// Enable http logging
-			logger.AddTarget(
+			if err = logger.AddTarget(
 				http.New(http.WithEndpoint(l.Endpoint),
 					http.WithAuthToken(l.AuthToken),
 					http.WithUserAgent(loggerUserAgent),
 					http.WithLogKind(string(logger.All)),
 					http.WithTransport(NewGatewayHTTPTransport()),
 				),
-			)
+			); err != nil {
+				logger.LogIf(ctx, fmt.Errorf("Unable to initialize console HTTP target: %w", err))
+			}
 		}
 	}
 
 	for _, l := range loggerCfg.Audit {
 		if l.Enabled {
 			// Enable http audit logging
-			logger.AddAuditTarget(
+			if err = logger.AddAuditTarget(
 				http.New(http.WithEndpoint(l.Endpoint),
 					http.WithAuthToken(l.AuthToken),
 					http.WithUserAgent(loggerUserAgent),
 					http.WithLogKind(string(logger.All)),
 					http.WithTransport(NewGatewayHTTPTransport()),
 				),
-			)
+			); err != nil {
+				logger.LogIf(ctx, fmt.Errorf("Unable to initialize audit HTTP target: %w", err))
+			}
 		}
 	}
 

cmd/config/storageclass/storage-class.go

@@ -226,7 +226,7 @@ func LookupConfig(kvs config.KVS, drivesPerSet int) (cfg Config, err error) {
 	cfg.RRS.Parity = defaultRRSParity
 
 	if err = config.CheckValidKeys(config.StorageClassSubSys, kvs, DefaultKVS); err != nil {
-		return cfg, err
+		return Config{}, err
 	}
 
 	ssc := env.Get(StandardEnv, kvs.Get(ClassStandard))
@@ -235,7 +235,7 @@ func LookupConfig(kvs config.KVS, drivesPerSet int) (cfg Config, err error) {
 	if ssc != "" {
 		cfg.Standard, err = parseStorageClass(ssc)
 		if err != nil {
-			return cfg, err
+			return Config{}, err
 		}
 	}
 	if cfg.Standard.Parity == 0 {
@@ -245,7 +245,7 @@ func LookupConfig(kvs config.KVS, drivesPerSet int) (cfg Config, err error) {
 	if rrsc != "" {
 		cfg.RRS, err = parseStorageClass(rrsc)
 		if err != nil {
-			return cfg, err
+			return Config{}, err
 		}
 	}
 	if cfg.RRS.Parity == 0 {
@@ -255,7 +255,7 @@ func LookupConfig(kvs config.KVS, drivesPerSet int) (cfg Config, err error) {
 	// 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 err = validateParity(cfg.Standard.Parity, cfg.RRS.Parity, drivesPerSet); err != nil {
-		return cfg, err
+		return Config{}, err
 	}
 
 	return cfg, nil

cmd/consolelogger.go

@@ -117,6 +117,11 @@ func (sys *HTTPConsoleLoggerSys) Subscribe(subCh chan interface{}, doneCh <-chan
 	sys.pubsub.Subscribe(subCh, doneCh, filter)
 }
 
+// Validate if HTTPConsoleLoggerSys is valid, always returns nil right now
+func (sys *HTTPConsoleLoggerSys) Validate() error {
+	return nil
+}
+
 // Send log message 'e' to console and publish to console
 // log pubsub system
 func (sys *HTTPConsoleLoggerSys) Send(e interface{}, logKind string) error {

cmd/crypto/header.go

@@ -18,10 +18,10 @@ import (
 	"bytes"
 	"crypto/md5"
 	"encoding/base64"
-	"encoding/json"
 	"net/http"
 	"strings"
 
+	jsoniter "github.com/json-iterator/go"
 	xhttp "github.com/minio/minio/cmd/http"
 )
 
@@ -148,6 +148,7 @@ func (s3KMS) ParseHTTP(h http.Header) (string, interface{}, error) {
 	contextStr, ok := h[SSEKmsContext]
 	if ok {
 		var context map[string]interface{}
+		var json = jsoniter.ConfigCompatibleWithStandardLibrary
 		if err := json.Unmarshal([]byte(contextStr[0]), &context); err != nil {
 			return "", nil, err
 		}

cmd/crypto/kes.go

@@ -18,7 +18,6 @@ import (
 	"bytes"
 	"crypto/tls"
 	"crypto/x509"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -30,10 +29,13 @@ import (
 	"strings"
 	"time"
 
+	jsoniter "github.com/json-iterator/go"
 	xhttp "github.com/minio/minio/cmd/http"
 	xnet "github.com/minio/minio/pkg/net"
 )
 
+var json = jsoniter.ConfigCompatibleWithStandardLibrary
+
 // ErrKESKeyExists is the error returned a KES server
 // when a master key does exist.
 var ErrKESKeyExists = NewKESError(http.StatusBadRequest, "key does already exist")

cmd/data-update-tracker_test.go

@@ -42,6 +42,10 @@ type testingLogger struct {
 	t  testLoggerI
 }
 
+func (t *testingLogger) Validate() error {
+	return nil
+}
+
 func (t *testingLogger) Send(entry interface{}, errKind string) error {
 	t.mu.Lock()
 	defer t.mu.Unlock()

cmd/disk-cache-backend.go

@@ -417,7 +417,7 @@ func (c *diskCache) Stat(ctx context.Context, bucket, object string) (oi ObjectI
 func (c *diskCache) statCachedMeta(ctx context.Context, cacheObjPath string) (meta *cacheMeta, partial bool, numHits int, err error) {
 
 	cLock := c.NewNSLockFn(ctx, cacheObjPath)
-	if err = cLock.GetRLock(globalObjectTimeout); err != nil {
+	if err = cLock.GetRLock(globalOperationTimeout); err != nil {
 		return
 	}
 
@@ -499,7 +499,7 @@ func (c *diskCache) statCache(ctx context.Context, cacheObjPath string) (meta *c
 func (c *diskCache) SaveMetadata(ctx context.Context, bucket, object string, meta map[string]string, actualSize int64, rs *HTTPRangeSpec, rsFileName string, incHitsOnly bool) error {
 	cachedPath := getCacheSHADir(c.dir, bucket, object)
 	cLock := c.NewNSLockFn(ctx, cachedPath)
-	if err := cLock.GetLock(globalObjectTimeout); err != nil {
+	if err := cLock.GetLock(globalOperationTimeout); err != nil {
 		return err
 	}
 	defer cLock.Unlock()
@@ -665,7 +665,7 @@ func (c *diskCache) Put(ctx context.Context, bucket, object string, data io.Read
 	}
 	cachePath := getCacheSHADir(c.dir, bucket, object)
 	cLock := c.NewNSLockFn(ctx, cachePath)
-	if err := cLock.GetLock(globalObjectTimeout); err != nil {
+	if err := cLock.GetLock(globalOperationTimeout); err != nil {
 		return err
 	}
 	defer cLock.Unlock()
@@ -871,7 +871,7 @@ func (c *diskCache) bitrotReadFromCache(ctx context.Context, filePath string, of
 func (c *diskCache) Get(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, opts ObjectOptions) (gr *GetObjectReader, numHits int, err error) {
 	cacheObjPath := getCacheSHADir(c.dir, bucket, object)
 	cLock := c.NewNSLockFn(ctx, cacheObjPath)
-	if err := cLock.GetRLock(globalObjectTimeout); err != nil {
+	if err := cLock.GetRLock(globalOperationTimeout); err != nil {
 		return nil, numHits, err
 	}
 
@@ -935,7 +935,7 @@ func (c *diskCache) Get(ctx context.Context, bucket, object string, rs *HTTPRang
 // Deletes the cached object
 func (c *diskCache) delete(ctx context.Context, cacheObjPath string) (err error) {
 	cLock := c.NewNSLockFn(ctx, cacheObjPath)
-	if err := cLock.GetLock(globalObjectTimeout); err != nil {
+	if err := cLock.GetLock(globalOperationTimeout); err != nil {
 		return err
 	}
 	defer cLock.Unlock()

cmd/disk-cache.go

@@ -326,7 +326,9 @@ func (c *cacheObjects) GetObjectNInfo(ctx context.Context, bucket, object string
 			// avoid cache overwrite if another background routine filled cache
 			if err != nil || oi.ETag != bReader.ObjInfo.ETag {
 				// use a new context to avoid locker prematurely timing out operation when the GetObjectNInfo returns.
-				dcache.Put(context.Background(), bucket, object, bReader, bReader.ObjInfo.Size, rs, ObjectOptions{UserDefined: getMetadata(bReader.ObjInfo)}, false)
+				dcache.Put(GlobalContext, bucket, object, bReader, bReader.ObjInfo.Size, rs, ObjectOptions{
+					UserDefined: getMetadata(bReader.ObjInfo),
+				}, false)
 				return
 			}
 		}()
@@ -337,7 +339,11 @@ func (c *cacheObjects) GetObjectNInfo(ctx context.Context, bucket, object string
 	pipeReader, pipeWriter := io.Pipe()
 	teeReader := io.TeeReader(bkReader, pipeWriter)
 	go func() {
-		putErr := dcache.Put(ctx, bucket, object, io.LimitReader(pipeReader, bkReader.ObjInfo.Size), bkReader.ObjInfo.Size, nil, ObjectOptions{UserDefined: getMetadata(bkReader.ObjInfo)}, false)
+		putErr := dcache.Put(ctx, bucket, object,
+			io.LimitReader(pipeReader, bkReader.ObjInfo.Size),
+			bkReader.ObjInfo.Size, nil, ObjectOptions{
+				UserDefined: getMetadata(bkReader.ObjInfo),
+			}, false)
 		// close the write end of the pipe, so the error gets
 		// propagated to getObjReader
 		pipeWriter.CloseWithError(putErr)

cmd/encryption-v1.go

@@ -605,12 +605,14 @@ func getDecryptedETag(headers http.Header, objInfo ObjectInfo, copySource bool)
 	if crypto.IsMultiPart(objInfo.UserDefined) {
 		return objInfo.ETag
 	}
+
 	if crypto.SSECopy.IsRequested(headers) {
 		key, err = crypto.SSECopy.ParseHTTP(headers)
 		if err != nil {
 			return objInfo.ETag
 		}
 	}
+
 	// As per AWS S3 Spec, ETag for SSE-C encrypted objects need not be MD5Sum of the data.
 	// Since server side copy with same source and dest just replaces the ETag, we save
 	// encrypted content MD5Sum as ETag for both SSE-C and SSE-S3, we standardize the ETag

cmd/erasure-zones.go

@@ -22,6 +22,7 @@ import (
 	"io"
 	"math/rand"
 	"net/http"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -457,12 +458,12 @@ func (z *erasureZones) GetObjectNInfo(ctx context.Context, bucket, object string
 		lock := z.NewNSLock(ctx, bucket, object)
 		switch lockType {
 		case writeLock:
-			if err = lock.GetLock(globalObjectTimeout); err != nil {
+			if err = lock.GetLock(globalOperationTimeout); err != nil {
 				return nil, err
 			}
 			nsUnlocker = lock.Unlock
 		case readLock:
-			if err = lock.GetRLock(globalObjectTimeout); err != nil {
+			if err = lock.GetRLock(globalOperationTimeout); err != nil {
 				return nil, err
 			}
 			nsUnlocker = lock.RUnlock
@@ -491,7 +492,7 @@ func (z *erasureZones) GetObjectNInfo(ctx context.Context, bucket, object string
 func (z *erasureZones) GetObject(ctx context.Context, bucket, object string, startOffset int64, length int64, writer io.Writer, etag string, opts ObjectOptions) error {
 	// Lock the object before reading.
 	lk := z.NewNSLock(ctx, bucket, object)
-	if err := lk.GetRLock(globalObjectTimeout); err != nil {
+	if err := lk.GetRLock(globalOperationTimeout); err != nil {
 		return err
 	}
 	defer lk.RUnlock()
@@ -515,7 +516,7 @@ func (z *erasureZones) GetObject(ctx context.Context, bucket, object string, sta
 func (z *erasureZones) GetObjectInfo(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err error) {
 	// Lock the object before reading.
 	lk := z.NewNSLock(ctx, bucket, object)
-	if err := lk.GetRLock(globalObjectTimeout); err != nil {
+	if err := lk.GetRLock(globalOperationTimeout); err != nil {
 		return ObjectInfo{}, err
 	}
 	defer lk.RUnlock()
@@ -543,7 +544,7 @@ func (z *erasureZones) GetObjectInfo(ctx context.Context, bucket, object string,
 func (z *erasureZones) PutObject(ctx context.Context, bucket string, object string, data *PutObjReader, opts ObjectOptions) (ObjectInfo, error) {
 	// Lock the object.
 	lk := z.NewNSLock(ctx, bucket, object)
-	if err := lk.GetLock(globalObjectTimeout); err != nil {
+	if err := lk.GetLock(globalOperationTimeout); err != nil {
 		return ObjectInfo{}, err
 	}
 	defer lk.Unlock()
@@ -624,7 +625,7 @@ func (z *erasureZones) CopyObject(ctx context.Context, srcBucket, srcObject, dst
 	cpSrcDstSame := isStringEqual(pathJoin(srcBucket, srcObject), pathJoin(dstBucket, dstObject))
 	if !cpSrcDstSame {
 		lk := z.NewNSLock(ctx, dstBucket, dstObject)
-		if err := lk.GetLock(globalObjectTimeout); err != nil {
+		if err := lk.GetLock(globalOperationTimeout); err != nil {
 			return objInfo, err
 		}
 		defer lk.Unlock()
@@ -1731,7 +1732,7 @@ func (z *erasureZones) ListBuckets(ctx context.Context) (buckets []BucketInfo, e
 func (z *erasureZones) ReloadFormat(ctx context.Context, dryRun bool) error {
 	// Acquire lock on format.json
 	formatLock := z.NewNSLock(ctx, minioMetaBucket, formatConfigFile)
-	if err := formatLock.GetRLock(globalHealingTimeout); err != nil {
+	if err := formatLock.GetRLock(globalOperationTimeout); err != nil {
 		return err
 	}
 	defer formatLock.RUnlock()
@@ -1747,7 +1748,7 @@ func (z *erasureZones) ReloadFormat(ctx context.Context, dryRun bool) error {
 func (z *erasureZones) HealFormat(ctx context.Context, dryRun bool) (madmin.HealResultItem, error) {
 	// Acquire lock on format.json
 	formatLock := z.NewNSLock(ctx, minioMetaBucket, formatConfigFile)
-	if err := formatLock.GetLock(globalHealingTimeout); err != nil {
+	if err := formatLock.GetLock(globalOperationTimeout); err != nil {
 		return madmin.HealResultItem{}, err
 	}
 	defer formatLock.Unlock()
@@ -1950,14 +1951,14 @@ func (z *erasureZones) HealObject(ctx context.Context, bucket, object, versionID
 	lk := z.NewNSLock(ctx, bucket, object)
 	if bucket == minioMetaBucket {
 		// For .minio.sys bucket heals we should hold write locks.
-		if err := lk.GetLock(globalHealingTimeout); err != nil {
+		if err := lk.GetLock(globalOperationTimeout); err != nil {
 			return madmin.HealResultItem{}, err
 		}
 		defer lk.Unlock()
 	} else {
 		// Lock the object before healing. Use read lock since healing
 		// will only regenerate parts & xl.meta of outdated disks.
-		if err := lk.GetRLock(globalHealingTimeout); err != nil {
+		if err := lk.GetRLock(globalOperationTimeout); err != nil {
 			return madmin.HealResultItem{}, err
 		}
 		defer lk.RUnlock()
@@ -2063,6 +2064,8 @@ func (z *erasureZones) Health(ctx context.Context, opts HealthOptions) HealthRes
 		}
 	}
 
+	reqInfo := (&logger.ReqInfo{}).AppendTags("maintenance", strconv.FormatBool(opts.Maintenance))
+
 	for zoneIdx := range erasureSetUpCount {
 		parityDrives := globalStorageClass.GetParityForSC(storageclass.STANDARD)
 		diskCount := z.zones[zoneIdx].drivesPerSet
@@ -2076,8 +2079,9 @@ func (z *erasureZones) Health(ctx context.Context, opts HealthOptions) HealthRes
 		}
 		for setIdx := range erasureSetUpCount[zoneIdx] {
 			if erasureSetUpCount[zoneIdx][setIdx] < writeQuorum {
-				logger.LogIf(ctx, fmt.Errorf("Write quorum lost on zone: %d, set: %d, expected write quorum: %d",
-					zoneIdx, setIdx, writeQuorum))
+				logger.LogIf(logger.SetReqInfo(ctx, reqInfo),
+					fmt.Errorf("Write quorum may be lost on zone: %d, set: %d, expected write quorum: %d",
+						zoneIdx, setIdx, writeQuorum))
 				return HealthResult{
 					Healthy:     false,
 					ZoneID:      zoneIdx,
@@ -2101,14 +2105,14 @@ func (z *erasureZones) Health(ctx context.Context, opts HealthOptions) HealthRes
 	// is not taken down for maintenance
 	aggHealStateResult, err := getAggregatedBackgroundHealState(ctx, true)
 	if err != nil {
-		logger.LogIf(ctx, fmt.Errorf("Unable to verify global heal status: %w", err))
+		logger.LogIf(logger.SetReqInfo(ctx, reqInfo), fmt.Errorf("Unable to verify global heal status: %w", err))
 		return HealthResult{
 			Healthy: false,
 		}
 	}
 
 	if len(aggHealStateResult.HealDisks) > 0 {
-		logger.LogIf(ctx, fmt.Errorf("Total drives to be healed %d", len(aggHealStateResult.HealDisks)))
+		logger.LogIf(logger.SetReqInfo(ctx, reqInfo), fmt.Errorf("Total drives to be healed %d", len(aggHealStateResult.HealDisks)))
 	}
 
 	healthy := len(aggHealStateResult.HealDisks) == 0

cmd/fs-v1-multipart.go

@@ -707,7 +707,7 @@ func (fs *FSObjects) CompleteMultipartUpload(ctx context.Context, bucket string,
 
 	// Hold write lock on the object.
 	destLock := fs.NewNSLock(ctx, bucket, object)
-	if err = destLock.GetLock(globalObjectTimeout); err != nil {
+	if err = destLock.GetLock(globalOperationTimeout); err != nil {
 		return oi, err
 	}
 	defer destLock.Unlock()

cmd/fs-v1.go

@@ -601,7 +601,7 @@ func (fs *FSObjects) CopyObject(ctx context.Context, srcBucket, srcObject, dstBu
 
 	if !cpSrcDstSame {
 		objectDWLock := fs.NewNSLock(ctx, dstBucket, dstObject)
-		if err := objectDWLock.GetLock(globalObjectTimeout); err != nil {
+		if err := objectDWLock.GetLock(globalOperationTimeout); err != nil {
 			return oi, err
 		}
 		defer objectDWLock.Unlock()
@@ -691,12 +691,12 @@ func (fs *FSObjects) GetObjectNInfo(ctx context.Context, bucket, object string,
 		lock := fs.NewNSLock(ctx, bucket, object)
 		switch lockType {
 		case writeLock:
-			if err = lock.GetLock(globalObjectTimeout); err != nil {
+			if err = lock.GetLock(globalOperationTimeout); err != nil {
 				return nil, err
 			}
 			nsUnlocker = lock.Unlock
 		case readLock:
-			if err = lock.GetRLock(globalObjectTimeout); err != nil {
+			if err = lock.GetRLock(globalOperationTimeout); err != nil {
 				return nil, err
 			}
 			nsUnlocker = lock.RUnlock
@@ -782,7 +782,7 @@ func (fs *FSObjects) GetObject(ctx context.Context, bucket, object string, offse
 
 	// Lock the object before reading.
 	lk := fs.NewNSLock(ctx, bucket, object)
-	if err := lk.GetRLock(globalObjectTimeout); err != nil {
+	if err := lk.GetRLock(globalOperationTimeout); err != nil {
 		logger.LogIf(ctx, err)
 		return err
 	}
@@ -1006,7 +1006,7 @@ func (fs *FSObjects) getObjectInfo(ctx context.Context, bucket, object string) (
 func (fs *FSObjects) getObjectInfoWithLock(ctx context.Context, bucket, object string) (oi ObjectInfo, e error) {
 	// Lock the object before reading.
 	lk := fs.NewNSLock(ctx, bucket, object)
-	if err := lk.GetRLock(globalObjectTimeout); err != nil {
+	if err := lk.GetRLock(globalOperationTimeout); err != nil {
 		return oi, err
 	}
 	defer lk.RUnlock()
@@ -1044,7 +1044,7 @@ func (fs *FSObjects) GetObjectInfo(ctx context.Context, bucket, object string, o
 	oi, err := fs.getObjectInfoWithLock(ctx, bucket, object)
 	if err == errCorruptedFormat || err == io.EOF {
 		lk := fs.NewNSLock(ctx, bucket, object)
-		if err = lk.GetLock(globalObjectTimeout); err != nil {
+		if err = lk.GetLock(globalOperationTimeout); err != nil {
 			return oi, toObjectErr(err, bucket, object)
 		}
 
@@ -1095,7 +1095,7 @@ func (fs *FSObjects) PutObject(ctx context.Context, bucket string, object string
 
 	// Lock the object.
 	lk := fs.NewNSLock(ctx, bucket, object)
-	if err := lk.GetLock(globalObjectTimeout); err != nil {
+	if err := lk.GetLock(globalOperationTimeout); err != nil {
 		logger.LogIf(ctx, err)
 		return objInfo, err
 	}

cmd/gateway-main.go

@@ -180,6 +180,11 @@ func StartGateway(ctx *cli.Context, gw Gateway) {
 	globalRootCAs, err = certs.GetRootCAs(globalCertsCADir.Get())
 	logger.FatalIf(err, "Failed to read root CAs (%v)", err)
 
+	// Add the global public crts as part of global root CAs
+	for _, publicCrt := range globalPublicCerts {
+		globalRootCAs.AddCert(publicCrt)
+	}
+
 	// Register root CAs for remote ENVs
 	env.RegisterGlobalCAs(globalRootCAs)
 

cmd/globals.go

@@ -208,9 +208,7 @@ var (
 	globalDomainNames []string      // Root domains for virtual host style requests
 	globalDomainIPs   set.StringSet // Root domain IP address(s) for a distributed MinIO deployment
 
-	globalObjectTimeout    = newDynamicTimeout( /*1*/ 10*time.Minute /*10*/, 600*time.Second) // timeout for Object API related ops
-	globalOperationTimeout = newDynamicTimeout(10*time.Minute /*30*/, 600*time.Second)        // default timeout for general ops
-	globalHealingTimeout   = newDynamicTimeout(30*time.Minute /*1*/, 30*time.Minute)          // timeout for healing related ops
+	globalOperationTimeout = newDynamicTimeout(10*time.Minute, 5*time.Minute) // default timeout for general ops
 
 	globalBucketObjectLockSys *BucketObjectLockSys
 	globalBucketQuotaSys      *BucketQuotaSys

cmd/iam-object-store.go

@@ -151,7 +151,7 @@ func (iamOS *IAMObjectStore) migrateUsersConfigToV1(ctx context.Context, isSTS b
 		cred.AccessKey = user
 		u := newUserIdentity(cred)
 		if err := iamOS.saveIAMConfig(u, identityPath); err != nil {
-			logger.LogIf(context.Background(), err)
+			logger.LogIf(ctx, err)
 			return err
 		}
 
@@ -216,7 +216,7 @@ func (iamOS *IAMObjectStore) saveIAMConfig(item interface{}, path string) error
 			return err
 		}
 	}
-	return saveConfig(context.Background(), iamOS.objAPI, path, data)
+	return saveConfig(GlobalContext, iamOS.objAPI, path, data)
 }
 
 func (iamOS *IAMObjectStore) loadIAMConfig(item interface{}, path string) error {
@@ -586,7 +586,7 @@ func listIAMConfigItems(ctx context.Context, objAPI ObjectLayer, pathPrefix stri
 
 		marker := ""
 		for {
-			lo, err := objAPI.ListObjects(context.Background(),
+			lo, err := objAPI.ListObjects(ctx,
 				minioMetaBucket, pathPrefix, marker, SlashSeparator, maxObjectList)
 			if err != nil {
 				select {

cmd/iam.go

@@ -449,10 +449,13 @@ func (sys *IAMSys) Init(ctx context.Context, objAPI ObjectLayer) {
 	rquorum := InsufficientReadQuorum{}
 	wquorum := InsufficientWriteQuorum{}
 
+	// allocate dynamic timeout once before the loop
+	iamLockTimeout := newDynamicTimeout(3*time.Second, 5*time.Second)
+
 	for range retry.NewTimerWithJitter(retryCtx, time.Second, 5*time.Second, retry.MaxJitter) {
 		// let one of the server acquire the lock, if not let them timeout.
 		// which shall be retried again by this loop.
-		if err := txnLk.GetLock(newDynamicTimeout(3*time.Second, 5*time.Second)); err != nil {
+		if err := txnLk.GetLock(iamLockTimeout); err != nil {
 			logger.Info("Waiting for all MinIO IAM sub-system to be initialized.. trying to acquire lock")
 			continue
 		}
@@ -1640,7 +1643,7 @@ func (sys *IAMSys) IsAllowedServiceAccount(args iampolicy.Args, parent string) b
 	subPolicy, err := iampolicy.ParseConfig(bytes.NewReader([]byte(spolicyStr)))
 	if err != nil {
 		// Log any error in input session policy config.
-		logger.LogIf(context.Background(), err)
+		logger.LogIf(GlobalContext, err)
 		return false
 	}
 
@@ -1783,7 +1786,7 @@ func (sys *IAMSys) IsAllowedSTS(args iampolicy.Args) bool {
 		subPolicy, err := iampolicy.ParseConfig(bytes.NewReader([]byte(spolicyStr)))
 		if err != nil {
 			// Log any error in input session policy config.
-			logger.LogIf(context.Background(), err)
+			logger.LogIf(GlobalContext, err)
 			return false
 		}
 
@@ -1807,7 +1810,7 @@ func (sys *IAMSys) IsAllowed(args iampolicy.Args) bool {
 	if globalPolicyOPA != nil {
 		ok, err := globalPolicyOPA.IsAllowed(args)
 		if err != nil {
-			logger.LogIf(context.Background(), err)
+			logger.LogIf(GlobalContext, err)
 		}
 		return ok
 	}

cmd/lock-rest-server.go

@@ -355,7 +355,7 @@ func registerLockRESTHandlers(router *mux.Router, endpointZones EndpointZones) {
 			}
 
 			subrouter := router.PathPrefix(path.Join(lockRESTPrefix, endpoint.Path)).Subrouter()
-			subrouter.Methods(http.MethodPost).Path(lockRESTVersionPrefix + lockRESTMethodHealth).HandlerFunc(httpTraceHdrs(lockServer.HealthHandler)).Queries(queries...)
+			subrouter.Methods(http.MethodPost).Path(lockRESTVersionPrefix + lockRESTMethodHealth).HandlerFunc(httpTraceHdrs(lockServer.HealthHandler))
 			subrouter.Methods(http.MethodPost).Path(lockRESTVersionPrefix + lockRESTMethodLock).HandlerFunc(httpTraceHdrs(lockServer.LockHandler)).Queries(queries...)
 			subrouter.Methods(http.MethodPost).Path(lockRESTVersionPrefix + lockRESTMethodRLock).HandlerFunc(httpTraceHdrs(lockServer.RLockHandler)).Queries(queries...)
 			subrouter.Methods(http.MethodPost).Path(lockRESTVersionPrefix + lockRESTMethodUnlock).HandlerFunc(httpTraceHdrs(lockServer.UnlockHandler)).Queries(queries...)

cmd/logger/audit.go

@@ -125,15 +125,6 @@ func (lrw *ResponseWriter) Size() int {
 	return lrw.bytesWritten
 }
 
-// AuditTargets is the list of enabled audit loggers
-var AuditTargets = []Target{}
-
-// AddAuditTarget adds a new audit logger target to the
-// list of enabled loggers
-func AddAuditTarget(t Target) {
-	AuditTargets = append(AuditTargets, t)
-}
-
 // AuditLog - logs audit logs to all audit targets.
 func AuditLog(w http.ResponseWriter, r *http.Request, api string, reqClaims map[string]interface{}, filterKeys ...string) {
 	// Fast exit if there is not audit target configured

cmd/logger/target/console/console.go

@@ -32,6 +32,11 @@ import (
 // in plain or json format to the standard output.
 type Target struct{}
 
+// Validate - validate if the tty can be written to
+func (c *Target) Validate() error {
+	return nil
+}
+
 // Send log message 'e' to console
 func (c *Target) Send(e interface{}, logKind string) error {
 	entry, ok := e.(log.Entry)

cmd/logger/target/http/http.go

@@ -18,12 +18,16 @@ package http
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"net/http"
 	"strings"
+	"time"
 
 	xhttp "github.com/minio/minio/cmd/http"
+	"github.com/minio/minio/cmd/logger"
 )
 
 // Target implements logger.Target and sends the json
@@ -45,6 +49,47 @@ type Target struct {
 	client    http.Client
 }
 
+// Validate validate the http target
+func (h *Target) Validate() error {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, h.endpoint, strings.NewReader(`{}`))
+	if err != nil {
+		return err
+	}
+
+	req.Header.Set(xhttp.ContentType, "application/json")
+
+	// Set user-agent to indicate MinIO release
+	// version to the configured log endpoint
+	req.Header.Set("User-Agent", h.userAgent)
+
+	if h.authToken != "" {
+		req.Header.Set("Authorization", h.authToken)
+	}
+
+	resp, err := h.client.Do(req)
+	if err != nil {
+		return err
+	}
+
+	// Drain any response.
+	xhttp.DrainBody(resp.Body)
+
+	if resp.StatusCode != http.StatusOK {
+		switch resp.StatusCode {
+		case http.StatusForbidden:
+			return fmt.Errorf("%s returned '%s', please check if your auth token is correctly set",
+				h.endpoint, resp.Status)
+		}
+		return fmt.Errorf("%s returned '%s', please check your endpoint configuration",
+			h.endpoint, resp.Status)
+	}
+
+	return nil
+}
+
 func (h *Target) startHTTPLogger() {
 	// Create a routine which sends json logs received
 	// from an internal channel.
@@ -55,8 +100,11 @@ func (h *Target) startHTTPLogger() {
 				continue
 			}
 
-			req, err := http.NewRequest(http.MethodPost, h.endpoint, bytes.NewReader(logJSON))
+			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			req, err := http.NewRequestWithContext(ctx, http.MethodPost,
+				h.endpoint, bytes.NewReader(logJSON))
 			if err != nil {
+				cancel()
 				continue
 			}
 			req.Header.Set(xhttp.ContentType, "application/json")
@@ -70,13 +118,26 @@ func (h *Target) startHTTPLogger() {
 			}
 
 			resp, err := h.client.Do(req)
+			cancel()
 			if err != nil {
-				h.client.CloseIdleConnections()
+				logger.LogIf(ctx, fmt.Errorf("%s returned '%w', please check your endpoint configuration\n",
+					h.endpoint, err))
 				continue
 			}
 
 			// Drain any response.
 			xhttp.DrainBody(resp.Body)
+
+			if resp.StatusCode != http.StatusOK {
+				switch resp.StatusCode {
+				case http.StatusForbidden:
+					logger.LogIf(ctx, fmt.Errorf("%s returned '%s', please check if your auth token is correctly set",
+						h.endpoint, resp.Status))
+				default:
+					logger.LogIf(ctx, fmt.Errorf("%s returned '%s', please check your endpoint configuration",
+						h.endpoint, resp.Status))
+				}
+			}
 		}
 	}()
 }

cmd/logger/targets.go

@@ -20,14 +20,33 @@ package logger
 // a single log entry and Send it to the log target
 //   e.g. Send the log to a http server
 type Target interface {
+	Validate() error
 	Send(entry interface{}, errKind string) error
 }
 
 // Targets is the set of enabled loggers
 var Targets = []Target{}
 
+// AuditTargets is the list of enabled audit loggers
+var AuditTargets = []Target{}
+
+// AddAuditTarget adds a new audit logger target to the
+// list of enabled loggers
+func AddAuditTarget(t Target) error {
+	if err := t.Validate(); err != nil {
+		return err
+	}
+
+	AuditTargets = append(AuditTargets, t)
+	return nil
+}
+
 // AddTarget adds a new logger target to the
 // list of enabled loggers
-func AddTarget(t Target) {
+func AddTarget(t Target) error {
+	if err := t.Validate(); err != nil {
+		return err
+	}
 	Targets = append(Targets, t)
+	return nil
 }

cmd/namespace-lock.go

@@ -28,6 +28,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/minio/minio/cmd/config/storageclass"
 	"github.com/minio/minio/cmd/logger"
 	"github.com/minio/minio/pkg/dsync"
 	"github.com/minio/minio/pkg/lsync"
@@ -147,7 +148,18 @@ func (di *distLockInstance) GetLock(timeout *dynamicTimeout) (timedOutErr error)
 	lockSource := getSource(2)
 	start := UTCNow()
 
-	if !di.rwMutex.GetLock(di.ctx, di.opsID, lockSource, timeout.Timeout()) {
+	// Lockers default to standard storage class always, why because
+	// we always dictate storage tolerance in terms of standard
+	// storage class be it number of drives or a multiplicative
+	// of number of nodes, defaulting lockers to this value
+	// simply means that locking is always similar in behavior
+	// and effect with erasure coded drive tolerance.
+	tolerance := globalStorageClass.GetParityForSC(storageclass.STANDARD)
+
+	if !di.rwMutex.GetLock(di.ctx, di.opsID, lockSource, dsync.Options{
+		Timeout:   timeout.Timeout(),
+		Tolerance: tolerance,
+	}) {
 		timeout.LogFailure()
 		return OperationTimedOut{}
 	}
@@ -164,7 +176,14 @@ func (di *distLockInstance) Unlock() {
 func (di *distLockInstance) GetRLock(timeout *dynamicTimeout) (timedOutErr error) {
 	lockSource := getSource(2)
 	start := UTCNow()
-	if !di.rwMutex.GetRLock(di.ctx, di.opsID, lockSource, timeout.Timeout()) {
+
+	// Lockers default to standard storage class always.
+	tolerance := globalStorageClass.GetParityForSC(storageclass.STANDARD)
+
+	if !di.rwMutex.GetRLock(di.ctx, di.opsID, lockSource, dsync.Options{
+		Timeout:   timeout.Timeout(),
+		Tolerance: tolerance,
+	}) {
 		timeout.LogFailure()
 		return OperationTimedOut{}
 	}

cmd/object-api-listobjects_test.go

@@ -1,5 +1,5 @@
 /*
- * MinIO Cloud Storage, (C) 2015-2016, 2017 MinIO, Inc.
+ * MinIO Cloud Storage, (C) 2015-2020 MinIO, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -405,7 +405,7 @@ func testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler) {
 			Prefixes: []string{"newzen/"},
 		},
 		// ListObjectsResult-29.
-		// Marker is set to "Asia/India/Karnataka/Bangalore/Koramangala/pics" in the testCase and delimeter set, (testCase 61).
+		// Marker is set to "Asia/India/Karnataka/Bangalore/Koramangala/pics" in the testCase and delimiter set, (testCase 61).
 		{
 			IsTruncated: false,
 			Objects: []ObjectInfo{
@@ -448,7 +448,9 @@ func testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler) {
 				{Name: "temporary/0/"},
 			},
 		},
-		// ListObjectsResult-34 Listing with marker > last object should return empty
+		// ListObjectsResult-34:
+		//    * Listing with marker > last object should return empty
+		//    * Listing an object with a trailing slash and '/' delimiter
 		{
 			IsTruncated: false,
 			Objects:     []ObjectInfo{},
@@ -460,7 +462,7 @@ func testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler) {
 		bucketName string
 		prefix     string
 		marker     string
-		delimeter  string
+		delimiter  string
 		maxKeys    int32
 		// Expected output of ListObjects.
 		result ListObjectsInfo
@@ -550,7 +552,7 @@ func testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler) {
 		{"test-bucket-list-object", "Asia/India/", "", "", 10, resultCases[23], nil, true},
 		{"test-bucket-list-object", "Asia", "", "", 10, resultCases[24], nil, true},
 		// Tests with prefix and delimiter (55-57).
-		// With delimeter the code should not recurse into the sub-directories of prefix Dir.
+		// With delimiter the code should not recurse into the sub-directories of prefix Dir.
 		{"test-bucket-list-object", "Asia", "", SlashSeparator, 10, resultCases[25], nil, true},
 		{"test-bucket-list-object", "new", "", SlashSeparator, 10, resultCases[26], nil, true},
 		{"test-bucket-list-object", "Asia/India/", "", SlashSeparator, 10, resultCases[27], nil, true},
@@ -569,13 +571,15 @@ func testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler) {
 		{"test-bucket-empty-dir", "", "temporary/", "", 10, resultCases[33], nil, true},
 		// Test listing with marker > last object such that response should be empty (65)
 		{"test-bucket-single-object", "", "A/C", "", 1000, resultCases[34], nil, true},
+		// Test listing an object with a trailing slash and a slash delimiter (66)
+		{"test-bucket-list-object", "Asia-maps.png/", "", "/", 1000, resultCases[34], nil, true},
 	}
 
 	for i, testCase := range testCases {
 		testCase := testCase
 		t.Run(fmt.Sprintf("%s-Test%d", instanceType, i+1), func(t *testing.T) {
 			result, err := obj.ListObjects(context.Background(), testCase.bucketName,
-				testCase.prefix, testCase.marker, testCase.delimeter, int(testCase.maxKeys))
+				testCase.prefix, testCase.marker, testCase.delimiter, int(testCase.maxKeys))
 			if err != nil && testCase.shouldPass {
 				t.Errorf("Test %d: %s:  Expected to pass, but failed with: <ERROR> %s", i+1, instanceType, err.Error())
 			}
@@ -636,7 +640,632 @@ func testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler) {
 			// Take ListObject treeWalk go-routine to completion, if available in the treewalk pool.
 			if result.IsTruncated {
 				_, err = obj.ListObjects(context.Background(), testCase.bucketName,
-					testCase.prefix, result.NextMarker, testCase.delimeter, 1000)
+					testCase.prefix, result.NextMarker, testCase.delimiter, 1000)
+				if err != nil {
+					t.Fatal(err)
+				}
+			}
+		})
+	}
+}
+
+// Wrapper for calling ListObjectVersions tests for both Erasure multiple disks and single node setup.
+func TestListObjectVersions(t *testing.T) {
+	ExecObjectLayerTest(t, testListObjectVersions)
+}
+
+// Unit test for ListObjectVersions
+func testListObjectVersions(obj ObjectLayer, instanceType string, t1 TestErrHandler) {
+	t, _ := t1.(*testing.T)
+	testBuckets := []string{
+		// This bucket is used for testing ListObject operations.
+		"test-bucket-list-object",
+		// This bucket will be tested with empty directories
+		"test-bucket-empty-dir",
+		// Will not store any objects in this bucket,
+		// Its to test ListObjects on an empty bucket.
+		"empty-bucket",
+		// Listing the case where the marker > last object.
+		"test-bucket-single-object",
+	}
+	for _, bucket := range testBuckets {
+		err := obj.MakeBucketWithLocation(context.Background(), bucket, BucketOptions{})
+		if err != nil {
+			t.Fatalf("%s : %s", instanceType, err.Error())
+		}
+	}
+
+	var err error
+	testObjects := []struct {
+		parentBucket string
+		name         string
+		content      string
+		meta         map[string]string
+	}{
+		{testBuckets[0], "Asia-maps.png", "asis-maps", map[string]string{"content-type": "image/png"}},
+		{testBuckets[0], "Asia/India/India-summer-photos-1", "contentstring", nil},
+		{testBuckets[0], "Asia/India/Karnataka/Bangalore/Koramangala/pics", "contentstring", nil},
+		{testBuckets[0], "newPrefix0", "newPrefix0", nil},
+		{testBuckets[0], "newPrefix1", "newPrefix1", nil},
+		{testBuckets[0], "newzen/zen/recurse/again/again/again/pics", "recurse", nil},
+		{testBuckets[0], "obj0", "obj0", nil},
+		{testBuckets[0], "obj1", "obj1", nil},
+		{testBuckets[0], "obj2", "obj2", nil},
+		{testBuckets[1], "obj1", "obj1", nil},
+		{testBuckets[1], "obj2", "obj2", nil},
+		{testBuckets[1], "temporary/0/", "", nil},
+		{testBuckets[3], "A/B", "contentstring", nil},
+	}
+	for _, object := range testObjects {
+		md5Bytes := md5.Sum([]byte(object.content))
+		_, err = obj.PutObject(context.Background(), object.parentBucket, object.name, mustGetPutObjReader(t, bytes.NewBufferString(object.content),
+			int64(len(object.content)), hex.EncodeToString(md5Bytes[:]), ""), ObjectOptions{UserDefined: object.meta})
+		if err != nil {
+			t.Fatalf("%s : %s", instanceType, err.Error())
+		}
+
+	}
+
+	// Formualting the result data set to be expected from ListObjects call inside the tests,
+	// This will be used in testCases and used for asserting the correctness of ListObjects output in the tests.
+
+	resultCases := []ListObjectsInfo{
+		// ListObjectsResult-0.
+		// Testing for listing all objects in the bucket, (testCase 20,21,22).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "Asia-maps.png"},
+				{Name: "Asia/India/India-summer-photos-1"},
+				{Name: "Asia/India/Karnataka/Bangalore/Koramangala/pics"},
+				{Name: "newPrefix0"},
+				{Name: "newPrefix1"},
+				{Name: "newzen/zen/recurse/again/again/again/pics"},
+				{Name: "obj0"},
+				{Name: "obj1"},
+				{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-1.
+		// Used for asserting the truncated case, (testCase 23).
+		{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				{Name: "Asia-maps.png"},
+				{Name: "Asia/India/India-summer-photos-1"},
+				{Name: "Asia/India/Karnataka/Bangalore/Koramangala/pics"},
+				{Name: "newPrefix0"},
+				{Name: "newPrefix1"},
+			},
+		},
+		// ListObjectsResult-2.
+		// (TestCase 24).
+		{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				{Name: "Asia-maps.png"},
+				{Name: "Asia/India/India-summer-photos-1"},
+				{Name: "Asia/India/Karnataka/Bangalore/Koramangala/pics"},
+				{Name: "newPrefix0"},
+			},
+		},
+		// ListObjectsResult-3.
+		// (TestCase 25).
+		{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				{Name: "Asia-maps.png"},
+				{Name: "Asia/India/India-summer-photos-1"},
+				{Name: "Asia/India/Karnataka/Bangalore/Koramangala/pics"},
+			},
+		},
+		// ListObjectsResult-4.
+		// Again used for truncated case.
+		// (TestCase 26).
+		{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				{Name: "Asia-maps.png"},
+			},
+		},
+		// ListObjectsResult-5.
+		// Used for Asserting prefixes.
+		// Used for test case with prefix "new", (testCase 27-29).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "newPrefix0"},
+				{Name: "newPrefix1"},
+				{Name: "newzen/zen/recurse/again/again/again/pics"},
+			},
+		},
+		// ListObjectsResult-6.
+		// Used for Asserting prefixes.
+		// Used for test case with prefix = "obj", (testCase 30).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "obj0"},
+				{Name: "obj1"},
+				{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-7.
+		// Used for Asserting prefixes and truncation.
+		// Used for test case with prefix = "new" and maxKeys = 1, (testCase 31).
+		{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				{Name: "newPrefix0"},
+			},
+		},
+		// ListObjectsResult-8.
+		// Used for Asserting prefixes.
+		// Used for test case with prefix = "obj" and maxKeys = 2, (testCase 32).
+		{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				{Name: "obj0"},
+				{Name: "obj1"},
+			},
+		},
+		// ListObjectsResult-9.
+		// Used for asserting the case with marker, but without prefix.
+		//marker is set to "newPrefix0" in the testCase, (testCase 33).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "newPrefix1"},
+				{Name: "newzen/zen/recurse/again/again/again/pics"},
+				{Name: "obj0"},
+				{Name: "obj1"},
+				{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-10.
+		//marker is set to "newPrefix1" in the testCase, (testCase 34).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "newzen/zen/recurse/again/again/again/pics"},
+				{Name: "obj0"},
+				{Name: "obj1"},
+				{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-11.
+		//marker is set to "obj0" in the testCase, (testCase 35).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "obj1"},
+				{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-12.
+		// Marker is set to "obj1" in the testCase, (testCase 36).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-13.
+		// Marker is set to "man" in the testCase, (testCase37).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "newPrefix0"},
+				{Name: "newPrefix1"},
+				{Name: "newzen/zen/recurse/again/again/again/pics"},
+				{Name: "obj0"},
+				{Name: "obj1"},
+				{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-14.
+		// Marker is set to "Abc" in the testCase, (testCase 39).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "Asia-maps.png"},
+				{Name: "Asia/India/India-summer-photos-1"},
+				{Name: "Asia/India/Karnataka/Bangalore/Koramangala/pics"},
+				{Name: "newPrefix0"},
+				{Name: "newPrefix1"},
+				{Name: "newzen/zen/recurse/again/again/again/pics"},
+				{Name: "obj0"},
+				{Name: "obj1"},
+				{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-15.
+		// Marker is set to "Asia/India/India-summer-photos-1" in the testCase, (testCase 40).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "Asia/India/Karnataka/Bangalore/Koramangala/pics"},
+				{Name: "newPrefix0"},
+				{Name: "newPrefix1"},
+				{Name: "newzen/zen/recurse/again/again/again/pics"},
+				{Name: "obj0"},
+				{Name: "obj1"},
+				{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-16.
+		// Marker is set to "Asia/India/Karnataka/Bangalore/Koramangala/pics" in the testCase, (testCase 41).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "newPrefix0"},
+				{Name: "newPrefix1"},
+				{Name: "newzen/zen/recurse/again/again/again/pics"},
+				{Name: "obj0"},
+				{Name: "obj1"},
+				{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-17.
+		// Used for asserting the case with marker, without prefix but with truncation.
+		// Marker =  "newPrefix0" & maxKeys = 3 in the testCase, (testCase42).
+		// Output truncated to 3 values.
+		{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				{Name: "newPrefix1"},
+				{Name: "newzen/zen/recurse/again/again/again/pics"},
+				{Name: "obj0"},
+			},
+		},
+		// ListObjectsResult-18.
+		// Marker = "newPrefix1" & maxkeys = 1 in the testCase, (testCase43).
+		// Output truncated to 1 value.
+		{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				{Name: "newzen/zen/recurse/again/again/again/pics"},
+			},
+		},
+		// ListObjectsResult-19.
+		// Marker = "obj0" & maxKeys = 1 in the testCase, (testCase44).
+		// Output truncated to 1 value.
+		{
+			IsTruncated: true,
+			Objects: []ObjectInfo{
+				{Name: "obj1"},
+			},
+		},
+		// ListObjectsResult-20.
+		// Marker = "obj0" & prefix = "obj" in the testCase, (testCase 45).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "obj1"},
+				{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-21.
+		// Marker = "obj1" & prefix = "obj" in the testCase, (testCase 46).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "obj2"},
+			},
+		},
+		// ListObjectsResult-22.
+		// Marker = "newPrefix0" & prefix = "new" in the testCase,, (testCase 47).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "newPrefix1"},
+				{Name: "newzen/zen/recurse/again/again/again/pics"},
+			},
+		},
+		// ListObjectsResult-23.
+		// Prefix is set to "Asia/India/" in the testCase, and delimiter is not set (testCase 55).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "Asia/India/India-summer-photos-1"},
+				{Name: "Asia/India/Karnataka/Bangalore/Koramangala/pics"},
+			},
+		},
+
+		// ListObjectsResult-24.
+		// Prefix is set to "Asia" in the testCase, and delimiter is not set (testCase 56).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "Asia-maps.png"},
+				{Name: "Asia/India/India-summer-photos-1"},
+				{Name: "Asia/India/Karnataka/Bangalore/Koramangala/pics"},
+			},
+		},
+
+		// ListObjectsResult-25.
+		// Prefix is set to "Asia" in the testCase, and delimiter is set (testCase 57).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "Asia-maps.png"},
+			},
+			Prefixes: []string{"Asia/"},
+		},
+		// ListObjectsResult-26.
+		// prefix = "new" and delimiter is set in the testCase.(testCase 58).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "newPrefix0"},
+				{Name: "newPrefix1"},
+			},
+			Prefixes: []string{"newzen/"},
+		},
+		// ListObjectsResult-27.
+		// Prefix is set to "Asia/India/" in the testCase, and delimiter is set to forward slash '/' (testCase 59).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "Asia/India/India-summer-photos-1"},
+			},
+			Prefixes: []string{"Asia/India/Karnataka/"},
+		},
+		// ListObjectsResult-28.
+		// Marker is set to "Asia/India/India-summer-photos-1" and delimiter set in the testCase, (testCase 60).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "newPrefix0"},
+				{Name: "newPrefix1"},
+				{Name: "obj0"},
+				{Name: "obj1"},
+				{Name: "obj2"},
+			},
+			Prefixes: []string{"newzen/"},
+		},
+		// ListObjectsResult-29.
+		// Marker is set to "Asia/India/Karnataka/Bangalore/Koramangala/pics" in the testCase and delimiter set, (testCase 61).
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "newPrefix0"},
+				{Name: "newPrefix1"},
+				{Name: "obj0"},
+				{Name: "obj1"},
+				{Name: "obj2"},
+			},
+			Prefixes: []string{"newzen/"},
+		},
+		// ListObjectsResult-30.
+		// Prefix and Delimiter is set to '/', (testCase 62).
+		{
+			IsTruncated: false,
+			Objects:     []ObjectInfo{},
+		},
+		// ListObjectsResult-31 Empty directory, recursive listing
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "obj1"},
+				{Name: "obj2"},
+				{Name: "temporary/0/"},
+			},
+		},
+		// ListObjectsResult-32 Empty directory, non recursive listing
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "obj1"},
+				{Name: "obj2"},
+			},
+			Prefixes: []string{"temporary/"},
+		},
+		// ListObjectsResult-33 Listing empty directory only
+		{
+			IsTruncated: false,
+			Objects: []ObjectInfo{
+				{Name: "temporary/0/"},
+			},
+		},
+		// ListObjectsResult-34:
+		//    * Listing with marker > last object should return empty
+		//    * Listing an object with a trailing slash and '/' delimiter
+		{
+			IsTruncated: false,
+			Objects:     []ObjectInfo{},
+		},
+	}
+
+	testCases := []struct {
+		// Inputs to ListObjects.
+		bucketName string
+		prefix     string
+		marker     string
+		delimiter  string
+		maxKeys    int32
+		// Expected output of ListObjects.
+		result ListObjectsInfo
+		err    error
+		// Flag indicating whether the test is expected to pass or not.
+		shouldPass bool
+	}{
+		// Test cases with invalid bucket names ( Test number 1-4).
+		{".test", "", "", "", 0, ListObjectsInfo{}, BucketNotFound{Bucket: ".test"}, false},
+		{"Test", "", "", "", 0, ListObjectsInfo{}, BucketNotFound{Bucket: "Test"}, false},
+		{"---", "", "", "", 0, ListObjectsInfo{}, BucketNotFound{Bucket: "---"}, false},
+		{"ad", "", "", "", 0, ListObjectsInfo{}, BucketNotFound{Bucket: "ad"}, false},
+		// Using an existing file for bucket name, but its not a directory (5).
+		{"simple-file.txt", "", "", "", 0, ListObjectsInfo{}, BucketNotFound{Bucket: "simple-file.txt"}, false},
+		// Valid bucket names, but they donot exist (6-8).
+		{"volatile-bucket-1", "", "", "", 0, ListObjectsInfo{}, BucketNotFound{Bucket: "volatile-bucket-1"}, false},
+		{"volatile-bucket-2", "", "", "", 0, ListObjectsInfo{}, BucketNotFound{Bucket: "volatile-bucket-2"}, false},
+		{"volatile-bucket-3", "", "", "", 0, ListObjectsInfo{}, BucketNotFound{Bucket: "volatile-bucket-3"}, false},
+		// Testing for failure cases with both perfix and marker (9).
+		// The prefix and marker combination to be valid it should satisfy strings.HasPrefix(marker, prefix).
+		{"test-bucket-list-object", "asia", "europe-object", "", 0, ListObjectsInfo{}, fmt.Errorf("Invalid combination of marker '%s' and prefix '%s'", "europe-object", "asia"), false},
+		// Setting a non-existing directory to be prefix (10-11).
+		{"empty-bucket", "europe/france/", "", "", 1, ListObjectsInfo{}, nil, true},
+		{"empty-bucket", "africa/tunisia/", "", "", 1, ListObjectsInfo{}, nil, true},
+		// Testing on empty bucket, that is, bucket without any objects in it (12).
+		{"empty-bucket", "", "", "", 0, ListObjectsInfo{}, nil, true},
+		// Setting maxKeys to negative value (13-14).
+		{"empty-bucket", "", "", "", -1, ListObjectsInfo{}, nil, true},
+		{"empty-bucket", "", "", "", 1, ListObjectsInfo{}, nil, true},
+		// Setting maxKeys to a very large value (15).
+		{"empty-bucket", "", "", "", 111100000, ListObjectsInfo{}, nil, true},
+		// Testing for all 10 objects in the bucket (16).
+		{"test-bucket-list-object", "", "", "", 10, resultCases[0], nil, true},
+		//Testing for negative value of maxKey, this should set maxKeys to listObjectsLimit (17).
+		{"test-bucket-list-object", "", "", "", -1, resultCases[0], nil, true},
+		// Testing for very large value of maxKey, this should set maxKeys to listObjectsLimit (18).
+		{"test-bucket-list-object", "", "", "", 1234567890, resultCases[0], nil, true},
+		// Testing for trancated value (19-22).
+		{"test-bucket-list-object", "", "", "", 5, resultCases[1], nil, true},
+		{"test-bucket-list-object", "", "", "", 4, resultCases[2], nil, true},
+		{"test-bucket-list-object", "", "", "", 3, resultCases[3], nil, true},
+		{"test-bucket-list-object", "", "", "", 1, resultCases[4], nil, true},
+		// Testing with prefix (23-26).
+		{"test-bucket-list-object", "new", "", "", 3, resultCases[5], nil, true},
+		{"test-bucket-list-object", "new", "", "", 4, resultCases[5], nil, true},
+		{"test-bucket-list-object", "new", "", "", 5, resultCases[5], nil, true},
+		{"test-bucket-list-object", "obj", "", "", 3, resultCases[6], nil, true},
+		// Testing with prefix and truncation (27-28).
+		{"test-bucket-list-object", "new", "", "", 1, resultCases[7], nil, true},
+		{"test-bucket-list-object", "obj", "", "", 2, resultCases[8], nil, true},
+		// Testing with marker, but without prefix and truncation (29-33).
+		{"test-bucket-list-object", "", "newPrefix0", "", 6, resultCases[9], nil, true},
+		{"test-bucket-list-object", "", "newPrefix1", "", 5, resultCases[10], nil, true},
+		{"test-bucket-list-object", "", "obj0", "", 4, resultCases[11], nil, true},
+		{"test-bucket-list-object", "", "obj1", "", 2, resultCases[12], nil, true},
+		{"test-bucket-list-object", "", "man", "", 11, resultCases[13], nil, true},
+		// Marker being set to a value which is greater than and all object names when sorted (34).
+		// Expected to send an empty response in this case.
+		{"test-bucket-list-object", "", "zen", "", 10, ListObjectsInfo{}, nil, true},
+		// Marker being set to a value which is lesser than and all object names when sorted (35).
+		// Expected to send all the objects in the bucket in this case.
+		{"test-bucket-list-object", "", "Abc", "", 10, resultCases[14], nil, true},
+		// Marker is to a hierarhical value (36-37).
+		{"test-bucket-list-object", "", "Asia/India/India-summer-photos-1", "", 10, resultCases[15], nil, true},
+		{"test-bucket-list-object", "", "Asia/India/Karnataka/Bangalore/Koramangala/pics", "", 10, resultCases[16], nil, true},
+		// Testing with marker and truncation, but no prefix (38-40).
+		{"test-bucket-list-object", "", "newPrefix0", "", 3, resultCases[17], nil, true},
+		{"test-bucket-list-object", "", "newPrefix1", "", 1, resultCases[18], nil, true},
+		{"test-bucket-list-object", "", "obj0", "", 1, resultCases[19], nil, true},
+		// Testing with both marker and prefix, but without truncation (41-43).
+		// The valid combination of marker and prefix should satisfy strings.HasPrefix(marker, prefix).
+		{"test-bucket-list-object", "obj", "obj0", "", 2, resultCases[20], nil, true},
+		{"test-bucket-list-object", "obj", "obj1", "", 1, resultCases[21], nil, true},
+		{"test-bucket-list-object", "new", "newPrefix0", "", 2, resultCases[22], nil, true},
+		// Testing with maxKeys set to 0 (44-50).
+		// The parameters have to valid.
+		{"test-bucket-list-object", "", "obj1", "", 0, ListObjectsInfo{}, nil, true},
+		{"test-bucket-list-object", "", "obj0", "", 0, ListObjectsInfo{}, nil, true},
+		{"test-bucket-list-object", "new", "", "", 0, ListObjectsInfo{}, nil, true},
+		{"test-bucket-list-object", "obj", "", "", 0, ListObjectsInfo{}, nil, true},
+		{"test-bucket-list-object", "obj", "obj0", "", 0, ListObjectsInfo{}, nil, true},
+		{"test-bucket-list-object", "obj", "obj1", "", 0, ListObjectsInfo{}, nil, true},
+		{"test-bucket-list-object", "new", "newPrefix0", "", 0, ListObjectsInfo{}, nil, true},
+		// Tests on hierarchical key names as prefix.
+		// Without delimteter the code should recurse into the prefix Dir.
+		// Tests with prefix, but without delimiter (51-52).
+		{"test-bucket-list-object", "Asia/India/", "", "", 10, resultCases[23], nil, true},
+		{"test-bucket-list-object", "Asia", "", "", 10, resultCases[24], nil, true},
+		// Tests with prefix and delimiter (53-55).
+		// With delimiter the code should not recurse into the sub-directories of prefix Dir.
+		{"test-bucket-list-object", "Asia", "", SlashSeparator, 10, resultCases[25], nil, true},
+		{"test-bucket-list-object", "new", "", SlashSeparator, 10, resultCases[26], nil, true},
+		{"test-bucket-list-object", "Asia/India/", "", SlashSeparator, 10, resultCases[27], nil, true},
+		// Test with marker set as hierarhical value and with delimiter. (56-57)
+		{"test-bucket-list-object", "", "Asia/India/India-summer-photos-1", SlashSeparator, 10, resultCases[28], nil, true},
+		{"test-bucket-list-object", "", "Asia/India/Karnataka/Bangalore/Koramangala/pics", SlashSeparator, 10, resultCases[29], nil, true},
+		// Test with prefix and delimiter set to '/'. (58)
+		{"test-bucket-list-object", SlashSeparator, "", SlashSeparator, 10, resultCases[30], nil, true},
+		// Test with invalid prefix (59)
+		{"test-bucket-list-object", "\\", "", SlashSeparator, 10, ListObjectsInfo{}, nil, true},
+		// Test listing an empty directory in recursive mode (60)
+		{"test-bucket-empty-dir", "", "", "", 10, resultCases[31], nil, true},
+		// Test listing an empty directory in a non recursive mode (61)
+		{"test-bucket-empty-dir", "", "", SlashSeparator, 10, resultCases[32], nil, true},
+		// Test listing a directory which contains an empty directory (62)
+		{"test-bucket-empty-dir", "", "temporary/", "", 10, resultCases[33], nil, true},
+		// Test listing with marker > last object such that response should be empty (63)
+		{"test-bucket-single-object", "", "A/C", "", 1000, resultCases[34], nil, true},
+		// Test listing an object with a trailing slash and a slash delimiter (64)
+		{"test-bucket-list-object", "Asia-maps.png/", "", "/", 1000, resultCases[34], nil, true},
+	}
+
+	for i, testCase := range testCases {
+		testCase := testCase
+		t.Run(fmt.Sprintf("%s-Test%d", instanceType, i+1), func(t *testing.T) {
+			result, err := obj.ListObjectVersions(context.Background(), testCase.bucketName,
+				testCase.prefix, testCase.marker, "", testCase.delimiter, int(testCase.maxKeys))
+			if _, ok := err.(NotImplemented); ok {
+				// Not implemented should be skipped
+				t.Skip()
+			}
+			if err != nil && testCase.shouldPass {
+				t.Errorf("%s:  Expected to pass, but failed with: <ERROR> %s", instanceType, err.Error())
+			}
+			if err == nil && !testCase.shouldPass {
+				t.Errorf("%s: Expected to fail with <ERROR> \"%s\", but passed instead", instanceType, testCase.err.Error())
+			}
+			// Failed as expected, but does it fail for the expected reason.
+			if err != nil && !testCase.shouldPass {
+				if !strings.Contains(err.Error(), testCase.err.Error()) {
+					t.Errorf("%s: Expected to fail with error \"%s\", but instead failed with error \"%s\" instead", instanceType, testCase.err.Error(), err.Error())
+				}
+			}
+			// Since there are cases for which ListObjects fails, this is
+			// necessary. Test passes as expected, but the output values
+			// are verified for correctness here.
+			if err == nil && testCase.shouldPass {
+				// The length of the expected ListObjectsResult.Objects
+				// should match in both expected result from test cases
+				// and in the output. On failure calling t.Fatalf,
+				// otherwise it may lead to index out of range error in
+				// assertion following this.
+				if len(testCase.result.Objects) != len(result.Objects) {
+					t.Fatalf("%s: Expected number of object in the result to be '%d', but found '%d' objects instead", instanceType, len(testCase.result.Objects), len(result.Objects))
+				}
+				for j := 0; j < len(testCase.result.Objects); j++ {
+					if testCase.result.Objects[j].Name != result.Objects[j].Name {
+						t.Errorf("%s: Expected object name to be \"%s\", but found \"%s\" instead", instanceType, testCase.result.Objects[j].Name, result.Objects[j].Name)
+					}
+					// FIXME: we should always check for ETag
+					if result.Objects[j].ETag == "" && !strings.HasSuffix(result.Objects[j].Name, SlashSeparator) {
+						t.Errorf("%s: Expected ETag to be not empty, but found empty instead (%v)", instanceType, result.Objects[j].Name)
+					}
+
+				}
+
+				if len(testCase.result.Prefixes) != len(result.Prefixes) {
+					fmt.Println(testCase, testCase.result.Prefixes, result.Prefixes)
+					t.Fatalf("%s: Expected number of prefixes in the result to be '%d', but found '%d' prefixes instead", instanceType, len(testCase.result.Prefixes), len(result.Prefixes))
+				}
+				for j := 0; j < len(testCase.result.Prefixes); j++ {
+					if testCase.result.Prefixes[j] != result.Prefixes[j] {
+						t.Errorf("%s: Expected prefix name to be \"%s\", but found \"%s\" instead", instanceType, testCase.result.Prefixes[j], result.Prefixes[j])
+					}
+				}
+
+				if testCase.result.IsTruncated != result.IsTruncated {
+					t.Errorf("%s: Expected IsTruncated flag to be %v, but instead found it to be %v", instanceType, testCase.result.IsTruncated, result.IsTruncated)
+				}
+
+				if testCase.result.IsTruncated && result.NextMarker == "" {
+					t.Errorf("%s: Expected NextContinuationToken to contain a string since listing is truncated, but instead found it to be empty", instanceType)
+				}
+
+				if !testCase.result.IsTruncated && result.NextMarker != "" {
+					t.Errorf("%s: Expected NextContinuationToken to be empty since listing is not truncated, but instead found `%v`", instanceType, result.NextMarker)
+				}
+
+			}
+			// Take ListObject treeWalk go-routine to completion, if available in the treewalk pool.
+			if result.IsTruncated {
+				_, err = obj.ListObjectVersions(context.Background(), testCase.bucketName,
+					testCase.prefix, result.NextMarker, "", testCase.delimiter, 1000)
 				if err != nil {
 					t.Fatal(err)
 				}

cmd/object-api-utils.go

@@ -400,6 +400,15 @@ func (o ObjectInfo) IsCompressedOK() (bool, error) {
 	return true, fmt.Errorf("unknown compression scheme: %s", scheme)
 }
 
+// GetActualETag - returns the actual etag of the stored object
+// decrypts SSE objects.
+func (o ObjectInfo) GetActualETag(h http.Header) string {
+	if !crypto.IsEncrypted(o.UserDefined) {
+		return o.ETag
+	}
+	return getDecryptedETag(h, o, false)
+}
+
 // GetActualSize - returns the actual size of the stored object
 func (o ObjectInfo) GetActualSize() (int64, error) {
 	if crypto.IsEncrypted(o.UserDefined) {

cmd/object-handlers.go

@@ -1261,7 +1261,7 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 	response := generateCopyObjectResponse(objInfo.ETag, objInfo.ModTime)
 	encodedSuccessResponse := encodeResponse(response)
 	if mustReplicate(ctx, r, dstBucket, dstObject, objInfo.UserDefined, objInfo.ReplicationStatus.String()) {
-		defer replicateObject(context.Background(), dstBucket, dstObject, objInfo.VersionID, objectAPI, &eventArgs{
+		defer replicateObject(GlobalContext, dstBucket, dstObject, objInfo.VersionID, objectAPI, &eventArgs{
 			EventName:    event.ObjectCreatedCopy,
 			BucketName:   dstBucket,
 			Object:       objInfo,
@@ -1577,7 +1577,7 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
 		}
 	}
 	if mustReplicate(ctx, r, bucket, object, metadata, "") {
-		defer replicateObject(context.Background(), bucket, object, objInfo.VersionID, objectAPI, &eventArgs{
+		defer replicateObject(GlobalContext, bucket, object, objInfo.VersionID, objectAPI, &eventArgs{
 			EventName:    event.ObjectCreatedPut,
 			BucketName:   bucket,
 			Object:       objInfo,
@@ -2652,7 +2652,7 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite
 
 	setPutObjHeaders(w, objInfo, false)
 	if mustReplicate(ctx, r, bucket, object, objInfo.UserDefined, objInfo.ReplicationStatus.String()) {
-		defer replicateObject(context.Background(), bucket, object, objInfo.VersionID, objectAPI, &eventArgs{
+		defer replicateObject(GlobalContext, bucket, object, objInfo.VersionID, objectAPI, &eventArgs{
 			EventName:    event.ObjectCreatedCompleteMultipartUpload,
 			BucketName:   bucket,
 			Object:       objInfo,

cmd/peer-rest-server.go

@@ -182,7 +182,7 @@ func (s *peerRESTServer) DeleteServiceAccountHandler(w http.ResponseWriter, r *h
 		return
 	}
 
-	if err := globalIAMSys.DeleteServiceAccount(context.Background(), accessKey); err != nil {
+	if err := globalIAMSys.DeleteServiceAccount(r.Context(), accessKey); err != nil {
 		s.writeErrorResponse(w, err)
 		return
 	}

cmd/server-main.go

@@ -122,6 +122,11 @@ func serverHandleCmdArgs(ctx *cli.Context) {
 	globalRootCAs, err = certs.GetRootCAs(globalCertsCADir.Get())
 	logger.FatalIf(err, "Failed to read root CAs (%v)", err)
 
+	// Add the global public crts as part of global root CAs
+	for _, publicCrt := range globalPublicCerts {
+		globalRootCAs.AddCert(publicCrt)
+	}
+
 	// Register root CAs for remote ENVs
 	env.RegisterGlobalCAs(globalRootCAs)
 
@@ -224,6 +229,9 @@ func initSafeMode(ctx context.Context, newObject ObjectLayer) (err error) {
 		initAutoHeal(ctx, newObject)
 	}
 
+	// allocate dynamic timeout once before the loop
+	configLockTimeout := newDynamicTimeout(3*time.Second, 5*time.Second)
+
 	// ****  WARNING ****
 	// Migrating to encrypted backend should happen before initialization of any
 	// sub-systems, make sure that we do not move the above codeblock elsewhere.
@@ -239,7 +247,7 @@ func initSafeMode(ctx context.Context, newObject ObjectLayer) (err error) {
 	for range retry.NewTimer(retryCtx) {
 		// let one of the server acquire the lock, if not let them timeout.
 		// which shall be retried again by this loop.
-		if err = txnLk.GetLock(newDynamicTimeout(3*time.Second, 3*time.Second)); err != nil {
+		if err = txnLk.GetLock(configLockTimeout); err != nil {
 			logger.Info("Waiting for all MinIO sub-systems to be initialized.. trying to acquire lock")
 			continue
 		}

cmd/server-startup-msg.go

@@ -94,13 +94,13 @@ func printStartupSafeModeMessage(apiEndpoints []string, err error) {
 	if color.IsTerminal() && !globalCLIContext.Anonymous {
 		logStartupMessage(color.RedBold("\nCommand-line Access: ") + mcAdminQuickStartGuide)
 		if runtime.GOOS == globalWindowsOSName {
-			mcMessage := fmt.Sprintf("> mc.exe config host add %s %s %s %s --api s3v4", alias,
+			mcMessage := fmt.Sprintf("> mc.exe alias set %s %s %s %s --api s3v4", alias,
 				endPoint, cred.AccessKey, cred.SecretKey)
 			logStartupMessage(fmt.Sprintf(getFormatStr(len(mcMessage), 3), mcMessage))
 			mcMessage = "> mc.exe admin config --help"
 			logStartupMessage(fmt.Sprintf(getFormatStr(len(mcMessage), 3), mcMessage))
 		} else {
-			mcMessage := fmt.Sprintf("$ mc config host add %s %s %s %s --api s3v4", alias,
+			mcMessage := fmt.Sprintf("$ mc alias set %s %s %s %s --api s3v4", alias,
 				endPoint, cred.AccessKey, cred.SecretKey)
 			logStartupMessage(fmt.Sprintf(getFormatStr(len(mcMessage), 3), mcMessage))
 			mcMessage = "$ mc admin config --help"
@@ -233,11 +233,11 @@ func printCLIAccessMsg(endPoint string, alias string) {
 	if color.IsTerminal() && !globalCLIContext.Anonymous {
 		logStartupMessage(color.Blue("\nCommand-line Access: ") + mcQuickStartGuide)
 		if runtime.GOOS == globalWindowsOSName {
-			mcMessage := fmt.Sprintf("$ mc.exe config host add %s %s %s %s", alias,
+			mcMessage := fmt.Sprintf("$ mc.exe alias set %s %s %s %s", alias,
 				endPoint, cred.AccessKey, cred.SecretKey)
 			logStartupMessage(fmt.Sprintf(getFormatStr(len(mcMessage), 3), mcMessage))
 		} else {
-			mcMessage := fmt.Sprintf("$ mc config host add %s %s %s %s", alias,
+			mcMessage := fmt.Sprintf("$ mc alias set %s %s %s %s", alias,
 				endPoint, cred.AccessKey, cred.SecretKey)
 			logStartupMessage(fmt.Sprintf(getFormatStr(len(mcMessage), 3), mcMessage))
 		}

cmd/storage-rest-client.go

@@ -478,7 +478,7 @@ func (client *storageRESTClient) WalkVersions(volume, dirPath, marker string, re
 			if gerr := decoder.Decode(&fi); gerr != nil {
 				// Upon error return
 				if gerr != io.EOF {
-					logger.LogIf(context.Background(), gerr)
+					logger.LogIf(GlobalContext, gerr)
 				}
 				return
 			}

cmd/sts-handlers.go

@@ -61,7 +61,8 @@ const (
 	parentClaim = "parent"
 
 	// LDAP claim keys
-	ldapUser = "ldapUser"
+	ldapUser               = "ldapUser"
+	ldapUserPolicyVariable = "ldap:user"
 )
 
 // stsAPIHandlers implements and provides http handlers for AWS STS API.

cmd/web-handlers.go

@@ -1130,9 +1130,6 @@ func (web *webAPIHandlers) Upload(w http.ResponseWriter, r *http.Request) {
 	// Ensure that metadata does not contain sensitive information
 	crypto.RemoveSensitiveEntries(metadata)
 
-	retentionRequested := objectlock.IsObjectLockRetentionRequested(r.Header)
-	legalHoldRequested := objectlock.IsObjectLockLegalHoldRequested(r.Header)
-
 	putObject := objectAPI.PutObject
 	getObjectInfo := objectAPI.GetObjectInfo
 	if web.CacheAPI() != nil {
@@ -1140,20 +1137,15 @@ func (web *webAPIHandlers) Upload(w http.ResponseWriter, r *http.Request) {
 		getObjectInfo = web.CacheAPI().GetObjectInfo
 	}
 
-	if retentionRequested || legalHoldRequested {
-		// enforce object retention rules
-		retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, getObjectInfo, retPerms, holdPerms)
-		if s3Err == ErrNone && retentionMode != "" {
-			opts.UserDefined[xhttp.AmzObjectLockMode] = string(retentionMode)
-			opts.UserDefined[xhttp.AmzObjectLockRetainUntilDate] = retentionDate.UTC().Format(iso8601TimeFormat)
-		}
-		if s3Err == ErrNone && legalHold.Status != "" {
-			opts.UserDefined[xhttp.AmzObjectLockLegalHold] = string(legalHold.Status)
-		}
-		if s3Err != ErrNone {
-			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL, guessIsBrowserReq(r))
-			return
-		}
+	// enforce object retention rules
+	retentionMode, retentionDate, _, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, getObjectInfo, retPerms, holdPerms)
+	if s3Err != ErrNone {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL, guessIsBrowserReq(r))
+		return
+	}
+	if retentionMode != "" {
+		opts.UserDefined[xhttp.AmzObjectLockMode] = string(retentionMode)
+		opts.UserDefined[xhttp.AmzObjectLockRetainUntilDate] = retentionDate.UTC().Format(iso8601TimeFormat)
 	}
 
 	objInfo, err := putObject(GlobalContext, bucket, object, pReader, opts)
@@ -1173,7 +1165,7 @@ func (web *webAPIHandlers) Upload(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 	if mustReplicate {
-		defer replicateObject(context.Background(), bucket, object, objInfo.VersionID, objectAPI, &eventArgs{
+		defer replicateObject(GlobalContext, bucket, object, objInfo.VersionID, objectAPI, &eventArgs{
 			EventName:    event.ObjectCreatedPut,
 			BucketName:   bucket,
 			Object:       objInfo,

cmd/xl-storage.go

@@ -873,6 +873,14 @@ func (s *xlStorage) WalkVersions(volume, dirPath, marker string, recursive bool,
 		return nil, err
 	}
 
+	// Fast exit track to check if we are listing an object with
+	// a trailing slash, this will avoid to list the object content.
+	if HasSuffix(dirPath, SlashSeparator) {
+		if st, err := os.Stat(pathJoin(volumeDir, dirPath, xlStorageFormatFile)); err == nil && st.Mode().IsRegular() {
+			return nil, errFileNotFound
+		}
+	}
+
 	// buffer channel matches the S3 ListObjects implementation
 	ch = make(chan FileInfoVersions, maxObjectList)
 	go func() {
@@ -894,6 +902,8 @@ func (s *xlStorage) WalkVersions(volume, dirPath, marker string, recursive bool,
 			var fiv FileInfoVersions
 			if HasSuffix(walkResult.entry, SlashSeparator) {
 				fiv = FileInfoVersions{
+					Volume: volume,
+					Name:   walkResult.entry,
 					Versions: []FileInfo{
 						{
 							Volume: volume,
@@ -949,6 +959,14 @@ func (s *xlStorage) Walk(volume, dirPath, marker string, recursive bool, endWalk
 		return nil, err
 	}
 
+	// Fast exit track to check if we are listing an object with
+	// a trailing slash, this will avoid to list the object content.
+	if HasSuffix(dirPath, SlashSeparator) {
+		if st, err := os.Stat(pathJoin(volumeDir, dirPath, xlStorageFormatFile)); err == nil && st.Mode().IsRegular() {
+			return nil, errFileNotFound
+		}
+	}
+
 	// buffer channel matches the S3 ListObjects implementation
 	ch = make(chan FileInfo, maxObjectList)
 	go func() {

docs/gateway/azure.md

@@ -27,7 +27,7 @@ MinIO Gateway comes with an embedded web based object browser. Point your web br
 
 ### Configure `mc`
 ```
-mc config host add myazure http://gateway-ip:9000 azureaccountname azureaccountkey
+mc alias set myazure http://gateway-ip:9000 azureaccountname azureaccountkey
 ```
 
 ### List containers on Microsoft Azure

docs/gateway/gcs.md

@@ -53,7 +53,7 @@ MinIO Client is a command-line tool called `mc` that provides UNIX-like commands
 Use the following command to configure the gateway:
 
 ```sh
-mc config host add mygcs http://gateway-ip:9000 minioaccesskey miniosecretkey
+mc alias set mygcs http://gateway-ip:9000 minioaccesskey miniosecretkey
 ```
 
 ### 3.2 List Containers on GCS

docs/gateway/hdfs.md

@@ -42,7 +42,7 @@ docker run -p 9000:9000 \
 ### Configure `mc`
 
 ```
-mc config host add myhdfs http://gateway-ip:9000 access_key secret_key
+mc alias set myhdfs http://gateway-ip:9000 access_key secret_key
 ```
 
 ### List buckets on hdfs

docs/gateway/nas.md

@@ -37,7 +37,7 @@ MinIO Gateway comes with an embedded web based object browser. Point your web br
 ### Configure `mc`
 
 ```
-mc config host add mynas http://gateway-ip:9000 access_key secret_key
+mc alias set mynas http://gateway-ip:9000 access_key secret_key
 ```
 
 ### List buckets on nas

docs/multi-user/README.md

@@ -109,10 +109,135 @@ mc admin group list myminio
 
 ### 8. Configure `mc`
 ```
-mc config host add myminio-newuser http://localhost:9000 newuser newuser123 --api s3v4
+mc alias set myminio-newuser http://localhost:9000 newuser newuser123 --api s3v4
 mc cat myminio-newuser/my-bucketname/my-objectname
 ```
 
+### Policy Variables
+You can use policy variables in the *Resource* element and in string comparisons in the *Condition* element.
+
+You can use a policy variable in the Resource element, but only in the resource portion of the ARN. This portion of the ARN appears after the 5th colon (:). You can't use a variable to replace parts of the ARN before the 5th colon, such as the service or account. The following policy might be attached to a group. It gives each of the users in the group full programmatic access to a user-specific object (their own "home directory") in MinIO.
+
+```
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Action": ["s3:ListBucket"],
+      "Effect": "Allow",
+      "Resource": ["arn:aws:s3:::mybucket"],
+      "Condition": {"StringLike": {"s3:prefix": ["${aws:username}/*"]}}
+    },
+    {
+      "Action": [
+        "s3:GetObject",
+        "s3:PutObject"
+      ],
+      "Effect": "Allow",
+      "Resource": ["arn:aws:s3:::mybucket/${aws:username}/*"]
+    }
+  ]
+}
+```
+
+If the user is authenticating using an STS credential which was authorized from OpenID connect we allow all `jwt:*` variables specified in the JWT specification, custom `jwt:*` or extensions are not supported.
+
+List of policy variables for OpenID based STS.
+```
+"jwt:sub"
+"jwt:iss"
+"jwt:aud"
+"jwt:jti"
+"jwt:upn"
+"jwt:name"
+"jwt:groups"
+"jwt:given_name"
+"jwt:family_name"
+"jwt:middle_name"
+"jwt:nickname"
+"jwt:preferred_username"
+"jwt:profile"
+"jwt:picture"
+"jwt:website"
+"jwt:email"
+"jwt:gender"
+"jwt:birthdate"
+"jwt:phone_number"
+"jwt:address"
+"jwt:scope"
+"jwt:client_id"
+```
+
+Following example shows OpenID users with full programmatic access to a OpenID user-specific directory (their own "home directory") in MinIO.
+```
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Action": ["s3:ListBucket"],
+      "Effect": "Allow",
+      "Resource": ["arn:aws:s3:::mybucket"],
+      "Condition": {"StringLike": {"s3:prefix": ["${jwt:preferred_username}/*"]}}
+    },
+    {
+      "Action": [
+        "s3:GetObject",
+        "s3:PutObject"
+      ],
+      "Effect": "Allow",
+      "Resource": ["arn:aws:s3:::mybucket/${jwt:preferred_username}/*"]
+    }
+  ]
+}
+```
+
+If the user is authenticating using an STS credential which was authorized from AD/LDAP we allow `ldap:*` variables, currently only supports `ldap:user`. Following example shows LDAP users full programmatic access to a LDAP user-specific directory (their own "home directory") in MinIO.
+```
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Action": ["s3:ListBucket"],
+      "Effect": "Allow",
+      "Resource": ["arn:aws:s3:::mybucket"],
+      "Condition": {"StringLike": {"s3:prefix": ["${ldap:user}/*"]}}
+    },
+    {
+      "Action": [
+        "s3:GetObject",
+        "s3:PutObject"
+      ],
+      "Effect": "Allow",
+      "Resource": ["arn:aws:s3:::mybucket/${ldap:user}/*"]
+    }
+  ]
+}
+```
+
+#### Common information available in all requests
+
+- *aws:CurrentTime* - This can be used for conditions that check the date and time.
+- *aws:EpochTime* - This is the date in epoch or Unix time, for use with date/time conditions.
+- *aws:PrincipalType* - This value indicates whether the principal is an account (Root credential), user (MinIO user), or assumed role (STS)
+- *aws:SecureTransport* - This is a Boolean value that represents whether the request was sent over TLS.
+- *aws:SourceIp* - This is the requester's IP address, for use with IP address conditions. If running behind Nginx like proxies, MinIO preserve's the source IP.
+
+```
+{
+  "Version": "2012-10-17",
+  "Statement": {
+    "Effect": "Allow",
+    "Action": "s3:ListBucket*",
+    "Resource": "arn:aws:s3:::mybucket",
+    "Condition": {"IpAddress": {"aws:SourceIp": "203.0.113.0/24"}}
+  }
+}
+```
+
+- *aws:UserAgent* - This value is a string that contains information about the requester's client application. This string is generated by the client and can be unreliable. You can only use this context key from `mc` or other MinIO SDKs which standardize the User-Agent string.
+- *aws:username* - This is a string containing the friendly name of the current user, this value would point to STS temporary credential in `AssumeRole`ed requests, instead use `jwt:preferred_username` in case of OpenID connect and `ldap:user` in case of AD/LDAP connect. *aws:userid* is an alias to *aws:username* in MinIO.
+
+
 ## Explore Further
 - [MinIO Client Complete Guide](https://docs.min.io/docs/minio-client-complete-guide)
 - [MinIO STS Quickstart Guide](https://docs.min.io/docs/minio-sts-quickstart-guide)

docs/multi-user/admin/README.md

@@ -62,7 +62,7 @@ This admin user will then be allowed to perform create/delete user operations vi
 
 ### 3. Configure `mc` and create another user user1 with attached policy user1policy
 ```
-mc config host add myminio-admin1 http://localhost:9000 admin1 admin123 --api s3v4
+mc alias set myminio-admin1 http://localhost:9000 admin1 admin123 --api s3v4
 
 mc admin user add myminio-admin1 user1 user123
 mc admin policy add myminio-admin1 user1policy ~/user1policy.json

docs/orchestration/docker-compose/docker-compose.yaml

@@ -5,7 +5,7 @@ version: '3.7'
 # 9001 through 9004.
 services:
   minio1:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-16T18-39-38Z
     volumes:
       - data1-1:/data1
       - data1-2:/data2
@@ -22,7 +22,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-16T18-39-38Z
     volumes:
       - data2-1:/data1
       - data2-2:/data2
@@ -39,7 +39,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-16T18-39-38Z
     volumes:
       - data3-1:/data1
       - data3-2:/data2
@@ -56,7 +56,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-16T18-39-38Z
     volumes:
       - data4-1:/data1
       - data4-2:/data2

docs/orchestration/docker-swarm/docker-compose-secrets.yaml

@@ -2,7 +2,7 @@ version: '3.7'
 
 services:
   minio1:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-16T18-39-38Z
     hostname: minio1
     volumes:
       - minio1-data:/export
@@ -29,7 +29,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-16T18-39-38Z
     hostname: minio2
     volumes:
       - minio2-data:/export
@@ -56,7 +56,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-16T18-39-38Z
     hostname: minio3
     volumes:
       - minio3-data:/export
@@ -83,7 +83,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-16T18-39-38Z
     hostname: minio4
     volumes:
       - minio4-data:/export

docs/orchestration/docker-swarm/docker-compose.yaml

@@ -2,7 +2,7 @@ version: '3.7'
 
 services:
   minio1:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-16T18-39-38Z
     hostname: minio1
     volumes:
       - minio1-data:/export
@@ -33,7 +33,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-16T18-39-38Z
     hostname: minio2
     volumes:
       - minio2-data:/export
@@ -64,7 +64,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-16T18-39-38Z
     hostname: minio3
     volumes:
       - minio3-data:/export
@@ -95,7 +95,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-16T18-39-38Z
     hostname: minio4
     volumes:
       - minio4-data:/export

docs/select/README.md

@@ -3,13 +3,26 @@ Traditional retrieval of objects is always as whole entities, i.e GetObject for
 
 You can use the Select API to query objects with following features:
 
-- CSV, JSON and Parquet - Objects must be in CSV, JSON, or Parquet format.
+- Objects must be in CSV, JSON, or Parquet(*) format. 
 - UTF-8 is the only encoding type the Select API supports.
 - GZIP or BZIP2 - CSV and JSON files can be compressed using GZIP or BZIP2. The Select API supports columnar compression for Parquet using GZIP, Snappy, LZ4. Whole object compression is not supported for Parquet objects.
 - Server-side encryption - The Select API supports querying objects that are protected with server-side encryption.
 
 Type inference and automatic conversion of values is performed based on the context when the value is un-typed (such as when reading CSV data). If present, the CAST function overrides automatic conversion.
 
+The [mc sql](https://docs.min.io/docs/minio-client-complete-guide.html#sql) command can be used for executing queries using the command line. 
+
+(*) Parquet is disabled on the MinIO server by default. See below how to enable it.
+
+## Enabling Parquet Format
+
+Parquet is DISABLED by default since hostile crafted input can easily crash the server.
+
+If you are in a controlled environment where it is safe to assume no hostile content can be uploaded to your cluster you can safely enable Parquet.
+To enable Parquet set the environment variable `MINIO_API_SELECT_PARQUET=on`.
+
+# Example using Python API 
+
 ## 1. Prerequisites
 - Install MinIO Server from [here](http://docs.min.io/docs/minio-quickstart-guide).
 - Familiarity with AWS S3 API.

docs/zh_CN/bucket/retention/README.md

@@ -0,0 +1,60 @@
+# 对象锁定指南 [![Slack](https://slack.min.io/slack?type=svg)](https://slack.min.io)
+
+MinIO服务器允许针对特定的对象一次写入，多次读取 (WORM),或者通过使用默认的对象锁定配置设置存储桶，该配置将默认的保留模式和保留期限应用于所有对象。这将使存储桶中的对象不可改变，也就是说，直到存储桶的对象锁定配置或对象保留中指定的时间到期后，才允许删除版本。
+
+对象锁定要求在创建存储桶时启用锁定，而且还会自动在存储桶上启用版本控制。此外，可以在存储桶上配置默认保留期限和保留模式，以应用于该存储桶中创建的对象。
+
+和保留期限无关，一个对象也可以被依法保留。依法保留的对象将一直有效（不可变），直到通过API调用删除依法保留为止。
+
+## 开始使用
+
+### 1. 前置条件
+
+- 安装 MinIO - [MinIO快速入门指南](https://docs.min.io/cn/minio-quickstart-guide)
+- 安装 `awscli` - [Installing AWS Command Line Interface](https://docs.aws.amazon.com/zh_cn/cli/latest/userguide/cli-chap-install.html)
+
+### 2. 设置存储桶WORM配置
+
+通过设置对象锁定配置可以启用存储桶的一次写入，多次读取 (WORM)模式。此配置将应用于存储桶中的现有对象和新对象。下面的例子设置了`Governance`模式，并且`mybucket`里的所有对象自创建时起，保留一天的时间。
+
+```sh
+$ awscli s3api put-object-lock-configuration --bucket mybucket --object-lock-configuration 'ObjectLockEnabled=\"Enabled\",Rule={DefaultRetention={Mode=\"GOVERNANCE\",Days=1}}'
+```
+
+### 设置对象锁定
+
+PutObject API 可以通过`x-amz-object-lock-mode`和`x-amz-object-lock-retain-until-date`请求头设置每一个对象的保留模式和保留期限。这要比存储桶上设置的对象锁定配置优先级高。
+
+```sh
+aws s3api put-object --bucket testbucket --key lockme --object-lock-mode GOVERNANCE --object-lock-retain-until-date "2019-11-20"  --body /etc/issue
+```
+
+请参阅 https://docs.aws.amazon.com/zh_cn/AmazonS3/latest/dev/object-lock-overview.html 以获取有关对象锁定、governance bypass所需权限的AWS S3规范。
+
+### 设置依法保留对象
+
+PutObject API 可以通过 `x-amz-object-lock-legal-hold` 请求头设置依法保留.
+
+```sh
+aws s3api put-object --bucket testbucket --key legalhold --object-lock-legal-hold-status ON --body /etc/issue
+```
+
+请参阅 https://docs.aws.amazon.com/zh_cn/AmazonS3/latest/dev/object-lock-overview.html 以获取有关对象锁定、指定依法保留所需的权限。
+
+## 概念
+- 如果一个对象被依法保留,除非明确删除各个版本ID的依法保留，否则无法删除该对象。 DeleteObjectVersion() 会失败。
+- 在 `Compliance` 模式中, 在各个版本ID的保留期限到期之前，任何人都无法删除对象。如果用户具有必需的governance bypass权限，则可以在`Compliance`模式下延长对象的保留期限。
+- 一旦在一个存储桶上设置对象锁定设置后
+  - 新对象会自动继承存储桶对象锁定配置的保留设置
+  - 上传对象时可以选择是否设置保留头信息
+  - 会对对象调用PutObjectRetention API
+- 如果不需要系统时间来设置保留日期，则可以将*MINIO_NTP_SERVER*环境变量设置为远程NTP server endpoint。
+- **对象锁定功能仅在纠删码和分布式纠删码模式下可用**。
+
+## 进一步探索
+
+- [使用`mc`](https://docs.min.io/cn/minio-client-quickstart-guide)
+- [使用`aws-cli`](https://docs.min.io/cn/aws-cli-with-minio)
+- [使用`s3cmd`](https://docs.min.io/cn/s3cmd-with-minio)
+- [使用`minio-go`](https://docs.min.io/cn/golang-client-quickstart-guide)
+- [MinIO文档](https://docs.min.io/cn)

docs/zh_CN/gateway/azure.md

@@ -26,7 +26,7 @@ MinIO Gateway配有嵌入式网络对象浏览器。 将您的Web浏览器指向
 
 ### 配置 `mc`
 ```
-mc config host add myazure http://gateway-ip:9000 azureaccountname azureaccountkey
+mc alias set myazure http://gateway-ip:9000 azureaccountname azureaccountkey
 ```
 
 ### 列出微软Azure上的容器

docs/zh_CN/gateway/gcs.md

@@ -41,7 +41,7 @@ MinIO Gateway配有嵌入式网络对象浏览器。 将您的Web浏览器指向
 
 ### 配置  `mc`
 ```
-mc config host add mygcs http://gateway-ip:9000 minioaccesskey miniosecretkey
+mc alias set mygcs http://gateway-ip:9000 minioaccesskey miniosecretkey
 ```
 
 ### 列出GCS上的容器

docs/zh_CN/gateway/nas.md

@@ -26,7 +26,7 @@ minio gateway nas /shared/nasvol
 
 ### 设置`mc`
 ```
-mc config host add mynas http://gateway-ip:9000 access_key secret_key
+mc alias set mynas http://gateway-ip:9000 access_key secret_key
 ```
 
 ### 列举nas上的存储桶

go.mod

@@ -83,6 +83,7 @@ require (
 	golang.org/x/crypto v0.0.0-20200709230013-948cd5f35899
 	golang.org/x/net v0.0.0-20200707034311-ab3426394381
 	golang.org/x/sys v0.0.0-20200625212154-ddb9806d33ae
+	golang.org/x/tools v0.0.0-20200814172026-c4923e618c08 // indirect
 	google.golang.org/api v0.5.0
 	gopkg.in/jcmturner/gokrb5.v7 v7.3.0
 	gopkg.in/ldap.v3 v3.0.3

go.sum

@@ -447,6 +447,7 @@ github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2 h1:eY9dn8+vbi4tKz5Qo6v2eYzo7kUS51QINcR5jNpbZS8=
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 go.etcd.io/bbolt v1.3.5 h1:XAzx9gjCb0Rxj7EoqcClPD1d5ZBxZJk0jbuoPHenBt0=
 go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
 go.etcd.io/etcd/v3 v3.3.0-rc.0.0.20200707003333-58bb8ae09f8e h1:HZQLoe71Q24wVyDrGBRcVuogx32U+cPlcm/WoSLUI6c=
@@ -487,6 +488,8 @@ golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHl
 golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc=
 golang.org/x/mod v0.2.0 h1:KU7oHjnv3XNWfa5COkzUifxZmxp1TyI7ImMXqFxLwvQ=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.3.0 h1:RM4zey1++hCTbCVQfnWeKs9/IEsaBLA8vTkd0WVtmH4=
+golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -500,6 +503,7 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191112182307-2180aed22343/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20200707034311-ab3426394381 h1:VXak5I6aEWmAXeQjA+QSZzlgNrpq9mjcfDemuexIKsU=
 golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -512,6 +516,8 @@ golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e h1:vcxGaoTs7kV8m5Np9uUNQin4BrLOthgV7252N8V+FwY=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208 h1:qwRHBd0NqMbJxfbotnDhm2ByMI1Shq4Y6oRJo21SGJA=
+golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -560,6 +566,8 @@ golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtn
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20200425043458-8463f397d07c h1:iHhCR0b26amDCiiO+kBguKZom9aMF+NrFxh9zeKR/XU=
 golang.org/x/tools v0.0.0-20200425043458-8463f397d07c/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
+golang.org/x/tools v0.0.0-20200814172026-c4923e618c08 h1:sfBQLM20fzeXhOixVQirwEbuW4PGStP773EXQpsBB6E=
+golang.org/x/tools v0.0.0-20200814172026-c4923e618c08/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=

pkg/bucket/policy/condition/key.go

@@ -137,6 +137,7 @@ var AllSupportedKeys = append([]Key{
 	AWSPrincipalType,
 	AWSUserID,
 	AWSUsername,
+	LDAPUser,
 	// Add new supported condition keys.
 }, JWTKeys...)
 
@@ -152,6 +153,7 @@ var CommonKeys = append([]Key{
 	AWSUserID,
 	AWSUsername,
 	S3XAmzContentSha256,
+	LDAPUser,
 }, JWTKeys...)
 
 func substFuncFromValues(values map[string][]string) func(string) string {
@@ -199,6 +201,8 @@ func (key Key) Name() string {
 		return strings.TrimPrefix(keyString, "aws:")
 	} else if strings.HasPrefix(keyString, "jwt:") {
 		return strings.TrimPrefix(keyString, "jwt:")
+	} else if strings.HasPrefix(keyString, "ldap:") {
+		return strings.TrimPrefix(keyString, "ldap:")
 	}
 	return strings.TrimPrefix(keyString, "s3:")
 }

pkg/bucket/policy/condition/ldap.go

@@ -0,0 +1,22 @@
+/*
+ * MinIO Cloud Storage, (C) 2020 MinIO, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package condition
+
+const (
+	// LDAPUser - LDAP username, in MinIO this value is equal to your authenticating LDAP user.
+	LDAPUser Key = "ldap:user"
+)

pkg/disk/root_disk_unix.go

@@ -31,7 +31,7 @@ func IsRootDisk(diskPath string) (bool, error) {
 	if err != nil {
 		return false, err
 	}
-	rootInfo, err := os.Stat("/")
+	rootInfo, err := os.Stat("/etc/hosts")
 	if err != nil {
 		return false, err
 	}

pkg/dsync/drwmutex.go

@@ -86,7 +86,15 @@ func NewDRWMutex(clnt *Dsync, names ...string) *DRWMutex {
 func (dm *DRWMutex) Lock(id, source string) {
 
 	isReadLock := false
-	dm.lockBlocking(context.Background(), drwMutexInfinite, id, source, isReadLock)
+	dm.lockBlocking(context.Background(), id, source, isReadLock, Options{
+		Timeout: drwMutexInfinite,
+	})
+}
+
+// Options lock options.
+type Options struct {
+	Timeout   time.Duration
+	Tolerance int
 }
 
 // GetLock tries to get a write lock on dm before the timeout elapses.
@@ -94,10 +102,10 @@ func (dm *DRWMutex) Lock(id, source string) {
 // If the lock is already in use, the calling go routine
 // blocks until either the mutex becomes available and return success or
 // more time has passed than the timeout value and return false.
-func (dm *DRWMutex) GetLock(ctx context.Context, id, source string, timeout time.Duration) (locked bool) {
+func (dm *DRWMutex) GetLock(ctx context.Context, id, source string, opts Options) (locked bool) {
 
 	isReadLock := false
-	return dm.lockBlocking(ctx, timeout, id, source, isReadLock)
+	return dm.lockBlocking(ctx, id, source, isReadLock, opts)
 }
 
 // RLock holds a read lock on dm.
@@ -107,7 +115,9 @@ func (dm *DRWMutex) GetLock(ctx context.Context, id, source string, timeout time
 func (dm *DRWMutex) RLock(id, source string) {
 
 	isReadLock := true
-	dm.lockBlocking(context.Background(), drwMutexInfinite, id, source, isReadLock)
+	dm.lockBlocking(context.Background(), id, source, isReadLock, Options{
+		Timeout: drwMutexInfinite,
+	})
 }
 
 // GetRLock tries to get a read lock on dm before the timeout elapses.
@@ -116,10 +126,10 @@ func (dm *DRWMutex) RLock(id, source string) {
 // Otherwise the calling go routine blocks until either the mutex becomes
 // available and return success or more time has passed than the timeout
 // value and return false.
-func (dm *DRWMutex) GetRLock(ctx context.Context, id, source string, timeout time.Duration) (locked bool) {
+func (dm *DRWMutex) GetRLock(ctx context.Context, id, source string, opts Options) (locked bool) {
 
 	isReadLock := true
-	return dm.lockBlocking(ctx, timeout, id, source, isReadLock)
+	return dm.lockBlocking(ctx, id, source, isReadLock, opts)
 }
 
 // lockBlocking will try to acquire either a read or a write lock
@@ -127,10 +137,10 @@ func (dm *DRWMutex) GetRLock(ctx context.Context, id, source string, timeout tim
 // The function will loop using a built-in timing randomized back-off
 // algorithm until either the lock is acquired successfully or more
 // time has elapsed than the timeout value.
-func (dm *DRWMutex) lockBlocking(ctx context.Context, timeout time.Duration, id, source string, isReadLock bool) (locked bool) {
+func (dm *DRWMutex) lockBlocking(ctx context.Context, id, source string, isReadLock bool, opts Options) (locked bool) {
 	restClnts := dm.clnt.GetLockersFn()
 
-	retryCtx, cancel := context.WithTimeout(ctx, timeout)
+	retryCtx, cancel := context.WithTimeout(ctx, opts.Timeout)
 
 	defer cancel()
 
@@ -140,7 +150,7 @@ func (dm *DRWMutex) lockBlocking(ctx context.Context, timeout time.Duration, id,
 		locks := make([]string, len(restClnts))
 
 		// Try to acquire the lock.
-		locked = lock(retryCtx, dm.clnt, &locks, id, source, isReadLock, dm.Names...)
+		locked = lock(retryCtx, dm.clnt, &locks, id, source, isReadLock, opts.Tolerance, dm.Names...)
 		if !locked {
 			continue
 		}
@@ -167,10 +177,29 @@ func (dm *DRWMutex) lockBlocking(ctx context.Context, timeout time.Duration, id,
 }
 
 // lock tries to acquire the distributed lock, returning true or false.
-func lock(ctx context.Context, ds *Dsync, locks *[]string, id, source string, isReadLock bool, lockNames ...string) bool {
+func lock(ctx context.Context, ds *Dsync, locks *[]string, id, source string, isReadLock bool, tolerance int, lockNames ...string) bool {
 
 	restClnts := ds.GetLockersFn()
 
+	// Tolerance is not set, defaults to half of the locker clients.
+	if tolerance == 0 {
+		tolerance = len(restClnts) / 2
+	}
+
+	// Quorum is effectively = total clients subtracted with tolerance limit
+	quorum := len(restClnts) - tolerance
+	if !isReadLock {
+		// In situations for write locks, as a special case
+		// to avoid split brains we make sure to acquire
+		// quorum + 1 when tolerance is exactly half of the
+		// total locker clients.
+		if quorum == tolerance {
+			quorum++
+		}
+	}
+
+	tolerance = len(restClnts) - quorum
+
 	// Create buffered channel of size equal to total number of nodes.
 	ch := make(chan Granted, len(restClnts))
 	defer close(ch)
@@ -217,7 +246,7 @@ func lock(ctx context.Context, ds *Dsync, locks *[]string, id, source string, is
 		}(index, isReadLock, c)
 	}
 
-	quorum := false
+	quorumMet := false
 
 	wg.Add(1)
 	go func(isReadLock bool) {
@@ -232,9 +261,6 @@ func lock(ctx context.Context, ds *Dsync, locks *[]string, id, source string, is
 		done := false
 		timeout := time.After(DRWMutexAcquireTimeout)
 
-		dquorumReads := (len(restClnts) + 1) / 2
-		dquorum := dquorumReads + 1
-
 		for ; i < len(restClnts); i++ { // Loop until we acquired all locks
 
 			select {
@@ -244,8 +270,7 @@ func lock(ctx context.Context, ds *Dsync, locks *[]string, id, source string, is
 					(*locks)[grant.index] = grant.lockUID
 				} else {
 					locksFailed++
-					if !isReadLock && locksFailed > len(restClnts)-dquorum ||
-						isReadLock && locksFailed > len(restClnts)-dquorumReads {
+					if locksFailed > tolerance {
 						// We know that we are not going to get the lock anymore,
 						// so exit out and release any locks that did get acquired
 						done = true
@@ -258,7 +283,7 @@ func lock(ctx context.Context, ds *Dsync, locks *[]string, id, source string, is
 				done = true
 				// timeout happened, maybe one of the nodes is slow, count
 				// number of locks to check whether we have quorum or not
-				if !quorumMet(locks, isReadLock, dquorum, dquorumReads) {
+				if !checkQuorumMet(locks, quorum) {
 					log("Quorum not met after timeout\n")
 					releaseAll(ds, locks, isReadLock, restClnts, lockNames...)
 				} else {
@@ -272,7 +297,7 @@ func lock(ctx context.Context, ds *Dsync, locks *[]string, id, source string, is
 		}
 
 		// Count locks in order to determine whether we have quorum or not
-		quorum = quorumMet(locks, isReadLock, dquorum, dquorumReads)
+		quorumMet = checkQuorumMet(locks, quorum)
 
 		// Signal that we have the quorum
 		wg.Done()
@@ -292,12 +317,11 @@ func lock(ctx context.Context, ds *Dsync, locks *[]string, id, source string, is
 
 	wg.Wait()
 
-	return quorum
+	return quorumMet
 }
 
-// quorumMet determines whether we have acquired the required quorum of underlying locks or not
-func quorumMet(locks *[]string, isReadLock bool, quorum, quorumReads int) bool {
-
+// checkQuorumMet determines whether we have acquired the required quorum of underlying locks or not
+func checkQuorumMet(locks *[]string, quorum int) bool {
 	count := 0
 	for _, uid := range *locks {
 		if isLocked(uid) {
@@ -305,14 +329,7 @@ func quorumMet(locks *[]string, isReadLock bool, quorum, quorumReads int) bool {
 		}
 	}
 
-	var metQuorum bool
-	if isReadLock {
-		metQuorum = count >= quorumReads
-	} else {
-		metQuorum = count >= quorum
-	}
-
-	return metQuorum
+	return count >= quorum
 }
 
 // releaseAll releases all locks that are marked as locked

pkg/dsync/drwmutex_test.go

@@ -36,12 +36,12 @@ func testSimpleWriteLock(t *testing.T, duration time.Duration) (locked bool) {
 
 	drwm := NewDRWMutex(ds, "simplelock")
 
-	if !drwm.GetRLock(context.Background(), id, source, time.Second) {
+	if !drwm.GetRLock(context.Background(), id, source, Options{Timeout: time.Second}) {
 		panic("Failed to acquire read lock")
 	}
 	// fmt.Println("1st read lock acquired, waiting...")
 
-	if !drwm.GetRLock(context.Background(), id, source, time.Second) {
+	if !drwm.GetRLock(context.Background(), id, source, Options{Timeout: time.Second}) {
 		panic("Failed to acquire read lock")
 	}
 	// fmt.Println("2nd read lock acquired, waiting...")
@@ -59,7 +59,7 @@ func testSimpleWriteLock(t *testing.T, duration time.Duration) (locked bool) {
 	}()
 
 	// fmt.Println("Trying to acquire write lock, waiting...")
-	locked = drwm.GetLock(context.Background(), id, source, duration)
+	locked = drwm.GetLock(context.Background(), id, source, Options{Timeout: duration})
 	if locked {
 		// fmt.Println("Write lock acquired, waiting...")
 		time.Sleep(time.Second)
@@ -93,7 +93,7 @@ func testDualWriteLock(t *testing.T, duration time.Duration) (locked bool) {
 	drwm := NewDRWMutex(ds, "duallock")
 
 	// fmt.Println("Getting initial write lock")
-	if !drwm.GetLock(context.Background(), id, source, time.Second) {
+	if !drwm.GetLock(context.Background(), id, source, Options{Timeout: time.Second}) {
 		panic("Failed to acquire initial write lock")
 	}
 
@@ -104,7 +104,7 @@ func testDualWriteLock(t *testing.T, duration time.Duration) (locked bool) {
 	}()
 
 	// fmt.Println("Trying to acquire 2nd write lock, waiting...")
-	locked = drwm.GetLock(context.Background(), id, source, duration)
+	locked = drwm.GetLock(context.Background(), id, source, Options{Timeout: duration})
 	if locked {
 		// fmt.Println("2nd write lock acquired, waiting...")
 		time.Sleep(time.Second)
@@ -139,7 +139,7 @@ func TestDualWriteLockTimedOut(t *testing.T) {
 
 // Borrowed from rwmutex_test.go
 func parallelReader(ctx context.Context, m *DRWMutex, clocked, cunlock, cdone chan bool) {
-	if m.GetRLock(ctx, id, source, time.Second) {
+	if m.GetRLock(ctx, id, source, Options{Timeout: time.Second}) {
 		clocked <- true
 		<-cunlock
 		m.RUnlock()
@@ -182,7 +182,7 @@ func TestParallelReaders(t *testing.T) {
 // Borrowed from rwmutex_test.go
 func reader(rwm *DRWMutex, numIterations int, activity *int32, cdone chan bool) {
 	for i := 0; i < numIterations; i++ {
-		if rwm.GetRLock(context.Background(), id, source, time.Second) {
+		if rwm.GetRLock(context.Background(), id, source, Options{Timeout: time.Second}) {
 			n := atomic.AddInt32(activity, 1)
 			if n < 1 || n >= 10000 {
 				panic(fmt.Sprintf("wlock(%d)\n", n))
@@ -199,7 +199,7 @@ func reader(rwm *DRWMutex, numIterations int, activity *int32, cdone chan bool)
 // Borrowed from rwmutex_test.go
 func writer(rwm *DRWMutex, numIterations int, activity *int32, cdone chan bool) {
 	for i := 0; i < numIterations; i++ {
-		if rwm.GetLock(context.Background(), id, source, time.Second) {
+		if rwm.GetLock(context.Background(), id, source, Options{Timeout: time.Second}) {
 			n := atomic.AddInt32(activity, 10000)
 			if n != 10000 {
 				panic(fmt.Sprintf("wlock(%d)\n", n))

pkg/dsync/dsync_test.go

@@ -63,7 +63,7 @@ func TestMain(m *testing.M) {
 
 	rand.Seed(time.Now().UTC().UnixNano())
 
-	nodes := make([]string, 4) // list of node IP addrs or hostname with ports.
+	nodes := make([]string, 5) // list of node IP addrs or hostname with ports.
 	for i := range nodes {
 		nodes[i] = fmt.Sprintf("127.0.0.1:%d", i+12345)
 	}

pkg/licverifier/verifier.go

@@ -32,20 +32,21 @@ type LicenseVerifier struct {
 
 // LicenseInfo holds customer metadata present in the license key.
 type LicenseInfo struct {
-	Email           string `json:"sub"`         // Email of the license key requestor
-	TeamName        string `json:"teamName"`    // Subnet team name
-	AccountID       int64  `json:"accountId"`   // Subnet account id
-	StorageCapacity int64  `json:"capacity"`    // Storage capacity used in TB
-	ServiceType     string `json:"serviceType"` // Subnet service type
+	Email           string // Email of the license key requestor
+	TeamName        string // Subnet team name
+	AccountID       int64  // Subnet account id
+	StorageCapacity int64  // Storage capacity used in TB
+	ServiceType     string // Subnet service type
 }
 
-// Valid checks if customer metadata from the license key is valid.
-func (li *LicenseInfo) Valid() error {
-	if li.AccountID <= 0 {
-		return errors.New("Invalid accountId in claims")
-	}
-	return nil
-}
+// license key JSON field names
+const (
+	accountID   = "accountId"
+	sub         = "sub"
+	teamName    = "teamName"
+	capacity    = "capacity"
+	serviceType = "serviceType"
+)
 
 // NewLicenseVerifier returns an initialized license verifier with the given
 // ECDSA public key in PEM format.
@@ -59,14 +60,49 @@ func NewLicenseVerifier(pemBytes []byte) (*LicenseVerifier, error) {
 	}, nil
 }
 
+// toLicenseInfo extracts LicenseInfo from claims. It returns an error if any of
+// the claim values are invalid.
+func toLicenseInfo(claims jwt.MapClaims) (LicenseInfo, error) {
+	accID, ok := claims[accountID].(float64)
+	if !ok || ok && accID <= 0 {
+		return LicenseInfo{}, errors.New("Invalid accountId in claims")
+	}
+	email, ok := claims[sub].(string)
+	if !ok {
+		return LicenseInfo{}, errors.New("Invalid email in claims")
+	}
+	tName, ok := claims[teamName].(string)
+	if !ok {
+		return LicenseInfo{}, errors.New("Invalid team name in claims")
+	}
+	storageCap, ok := claims[capacity].(float64)
+	if !ok {
+		return LicenseInfo{}, errors.New("Invalid storage capacity in claims")
+	}
+	sType, ok := claims[serviceType].(string)
+	if !ok {
+		return LicenseInfo{}, errors.New("Invalid service type in claims")
+	}
+	return LicenseInfo{
+		Email:           email,
+		TeamName:        tName,
+		AccountID:       int64(accID),
+		StorageCapacity: int64(storageCap),
+		ServiceType:     sType,
+	}, nil
+
+}
+
 // Verify verifies the license key and validates the claims present in it.
 func (lv *LicenseVerifier) Verify(license string) (LicenseInfo, error) {
-	var licInfo LicenseInfo
-	_, err := jwt.ParseWithClaims(license, &licInfo, func(token *jwt.Token) (interface{}, error) {
+	token, err := jwt.ParseWithClaims(license, &jwt.MapClaims{}, func(token *jwt.Token) (interface{}, error) {
 		return lv.ecPubKey, nil
 	})
 	if err != nil {
 		return LicenseInfo{}, fmt.Errorf("Failed to verify license: %s", err)
 	}
-	return licInfo, nil
+	if claims, ok := token.Claims.(*jwt.MapClaims); ok && token.Valid {
+		return toLicenseInfo(*claims)
+	}
+	return LicenseInfo{}, errors.New("Invalid claims found in license")
 }

pkg/licverifier/verifier_test.go

@@ -19,8 +19,18 @@ package licverifier
 import (
 	"fmt"
 	"testing"
+	"time"
+
+	"github.com/dgrijalva/jwt-go"
 )
 
+// at fixes the jwt.TimeFunc at t and calls f in that context.
+func at(t time.Time, f func()) {
+	jwt.TimeFunc = func() time.Time { return t }
+	f()
+	jwt.TimeFunc = time.Now
+}
+
 // TestLicenseVerify tests the license key verification process with a valid and
 // an invalid key.
 func TestLicenseVerify(t *testing.T) {
@@ -48,18 +58,22 @@ mr/cKCUyBL7rcAvg0zNq1vcSrUSGlAmY3SEDCu3GOKnjG/U4E7+p957ocWSV+mQU
 	}
 
 	for i, tc := range testCases {
-		licInfo, err := lv.Verify(tc.lic)
-		if err != nil && tc.shouldPass {
-			t.Fatalf("%d: Expected license to pass verification but failed with %s", i+1, err)
-		}
-		if err == nil {
-			if !tc.shouldPass {
-				t.Fatalf("%d: Expected license to fail verification but passed", i+1)
+		// Fixing the jwt.TimeFunc at 2020-08-05 22:17:43 +0000 UTC to
+		// ensure that the license JWT doesn't expire ever.
+		at(time.Unix(int64(1596665863), 0), func() {
+			licInfo, err := lv.Verify(tc.lic)
+			if err != nil && tc.shouldPass {
+				t.Fatalf("%d: Expected license to pass verification but failed with %s", i+1, err)
 			}
-			if tc.expectedLicInfo != licInfo {
-				t.Fatalf("%d: Expected license info %v but got %v", i+1, tc.expectedLicInfo, licInfo)
+			if err == nil {
+				if !tc.shouldPass {
+					t.Fatalf("%d: Expected license to fail verification but passed", i+1)
+				}
+				if tc.expectedLicInfo != licInfo {
+					t.Fatalf("%d: Expected license info %v but got %v", i+1, tc.expectedLicInfo, licInfo)
+				}
 			}
-		}
+		})
 	}
 }
 

pkg/s3select/parquet/reader.go

@@ -17,6 +17,7 @@
 package parquet
 
 import (
+	"fmt"
 	"io"
 
 	"github.com/bcicen/jstream"
@@ -34,6 +35,12 @@ type Reader struct {
 
 // Read - reads single record.
 func (r *Reader) Read(dst sql.Record) (rec sql.Record, rerr error) {
+	defer func() {
+		if rec := recover(); rec != nil {
+			rerr = fmt.Errorf("panic reading parquet record: %v", rec)
+		}
+	}()
+
 	parquetRecord, err := r.reader.Read()
 	if err != nil {
 		if err != io.EOF {
@@ -92,7 +99,12 @@ func (r *Reader) Close() error {
 }
 
 // NewReader - creates new Parquet reader using readerFunc callback.
-func NewReader(getReaderFunc func(offset, length int64) (io.ReadCloser, error), args *ReaderArgs) (*Reader, error) {
+func NewReader(getReaderFunc func(offset, length int64) (io.ReadCloser, error), args *ReaderArgs) (r *Reader, err error) {
+	defer func() {
+		if rec := recover(); rec != nil {
+			err = fmt.Errorf("panic reading parquet header: %v", rec)
+		}
+	}()
 	reader, err := parquetgo.NewReader(getReaderFunc, nil)
 	if err != nil {
 		if err != io.EOF {

pkg/s3select/select.go

@@ -26,6 +26,7 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"os"
 	"strings"
 	"sync"
 
@@ -334,6 +335,9 @@ func (s3Select *S3Select) Open(getReader func(offset, length int64) (io.ReadClos
 		}
 		return nil
 	case parquetFormat:
+		if !strings.EqualFold(os.Getenv("MINIO_API_SELECT_PARQUET"), "on") {
+			return errors.New("parquet format parsing not enabled on server")
+		}
 		var err error
 		s3Select.recordReader, err = parquet.NewReader(getReader, &s3Select.Input.ParquetArgs)
 		return err

pkg/s3select/select_test.go

@@ -925,6 +925,8 @@ func TestJSONInput(t *testing.T) {
 }
 
 func TestParquetInput(t *testing.T) {
+	os.Setenv("MINIO_API_SELECT_PARQUET", "on")
+	defer os.Setenv("MINIO_API_SELECT_PARQUET", "off")
 
 	var testTable = []struct {
 		requestXML     []byte