fix: support object-remaining-retention-days policy condition (#9259)

This PR also tries to simplify the approach taken in
object-locking implementation by preferential treatment
given towards full validation.

This in-turn has fixed couple of bugs related to
how policy should have been honored when ByPassGovernance
is provided.

Simplifies code a bit, but also duplicates code intentionally
for clarity due to complex nature of object locking
implementation.

cmd/auth-handler.go

@@ -26,12 +26,15 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"strconv"
 	"strings"
+	"time"
 
 	xhttp "github.com/minio/minio/cmd/http"
 	xjwt "github.com/minio/minio/cmd/jwt"
 	"github.com/minio/minio/cmd/logger"
 	"github.com/minio/minio/pkg/auth"
+	objectlock "github.com/minio/minio/pkg/bucket/object/lock"
 	"github.com/minio/minio/pkg/bucket/policy"
 	"github.com/minio/minio/pkg/hash"
 	iampolicy "github.com/minio/minio/pkg/iam/policy"
@@ -464,6 +467,106 @@ func (a authHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	writeErrorResponse(r.Context(), w, errorCodes.ToAPIErr(ErrSignatureVersionNotSupported), r.URL, guessIsBrowserReq(r))
 }
 
+func validateSignature(atype authType, r *http.Request) (auth.Credentials, bool, map[string]interface{}, APIErrorCode) {
+	var cred auth.Credentials
+	var owner bool
+	var s3Err APIErrorCode
+	switch atype {
+	case authTypeUnknown, authTypeStreamingSigned:
+		return cred, owner, nil, ErrSignatureVersionNotSupported
+	case authTypeSignedV2, authTypePresignedV2:
+		if s3Err = isReqAuthenticatedV2(r); s3Err != ErrNone {
+			return cred, owner, nil, s3Err
+		}
+		cred, owner, s3Err = getReqAccessKeyV2(r)
+	case authTypePresigned, authTypeSigned:
+		region := globalServerRegion
+		if s3Err = isReqAuthenticated(GlobalContext, r, region, serviceS3); s3Err != ErrNone {
+			return cred, owner, nil, s3Err
+		}
+		cred, owner, s3Err = getReqAccessKeyV4(r, region, serviceS3)
+	}
+	if s3Err != ErrNone {
+		return cred, owner, nil, s3Err
+	}
+
+	claims, s3Err := checkClaimsFromToken(r, cred)
+	if s3Err != ErrNone {
+		return cred, owner, nil, s3Err
+	}
+
+	return cred, owner, claims, ErrNone
+}
+
+func isPutRetentionAllowed(bucketName, objectName string, retDays int, retDate time.Time, retMode objectlock.RetMode, byPassSet bool, r *http.Request, cred auth.Credentials, owner bool, claims map[string]interface{}) (s3Err APIErrorCode) {
+	var retSet bool
+	if cred.AccessKey == "" {
+		conditions := getConditionValues(r, "", "", nil)
+		conditions["object-lock-mode"] = []string{string(retMode)}
+		conditions["object-lock-retain-until-date"] = []string{retDate.Format(time.RFC3339)}
+		if retDays > 0 {
+			conditions["object-lock-remaining-retention-days"] = []string{strconv.Itoa(retDays)}
+		}
+		if retMode == objectlock.RetGovernance && byPassSet {
+			byPassSet = globalPolicySys.IsAllowed(policy.Args{
+				AccountName:     cred.AccessKey,
+				Action:          policy.Action(policy.BypassGovernanceRetentionAction),
+				BucketName:      bucketName,
+				ConditionValues: conditions,
+				IsOwner:         false,
+				ObjectName:      objectName,
+			})
+		}
+		if globalPolicySys.IsAllowed(policy.Args{
+			AccountName:     cred.AccessKey,
+			Action:          policy.Action(policy.PutObjectRetentionAction),
+			BucketName:      bucketName,
+			ConditionValues: conditions,
+			IsOwner:         false,
+			ObjectName:      objectName,
+		}) {
+			retSet = true
+		}
+		if byPassSet || retSet {
+			return ErrNone
+		}
+		return ErrAccessDenied
+	}
+
+	conditions := getConditionValues(r, "", cred.AccessKey, claims)
+	conditions["object-lock-mode"] = []string{string(retMode)}
+	conditions["object-lock-retain-until-date"] = []string{retDate.Format(time.RFC3339)}
+	if retDays > 0 {
+		conditions["object-lock-remaining-retention-days"] = []string{strconv.Itoa(retDays)}
+	}
+	if retMode == objectlock.RetGovernance && byPassSet {
+		byPassSet = globalIAMSys.IsAllowed(iampolicy.Args{
+			AccountName:     cred.AccessKey,
+			Action:          policy.BypassGovernanceRetentionAction,
+			BucketName:      bucketName,
+			ObjectName:      objectName,
+			ConditionValues: conditions,
+			IsOwner:         owner,
+			Claims:          claims,
+		})
+	}
+	if globalIAMSys.IsAllowed(iampolicy.Args{
+		AccountName:     cred.AccessKey,
+		Action:          policy.PutObjectRetentionAction,
+		BucketName:      bucketName,
+		ConditionValues: conditions,
+		ObjectName:      objectName,
+		IsOwner:         owner,
+		Claims:          claims,
+	}) {
+		retSet = true
+	}
+	if byPassSet || retSet {
+		return ErrNone
+	}
+	return ErrAccessDenied
+}
+
 // isPutActionAllowed - check if PUT operation is allowed on the resource, this
 // call verifies bucket policies and IAM policies, supports multi user
 // checks etc.

cmd/bucket-handlers.go

@@ -399,11 +399,14 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 			}
 			continue
 		}
-		govBypassPerms := checkRequestAuthType(ctx, r, policy.BypassGovernanceRetentionAction, bucket, object.ObjectName)
-		if _, err := enforceRetentionBypassForDelete(ctx, r, bucket, object.ObjectName, getObjectInfoFn, govBypassPerms); err != ErrNone {
-			dErrs[index] = err
-			continue
+
+		if _, ok := globalBucketObjectLockConfig.Get(bucket); ok {
+			if err := enforceRetentionBypassForDelete(ctx, r, bucket, object.ObjectName, getObjectInfoFn); err != ErrNone {
+				dErrs[index] = err
+				continue
+			}
 		}
+
 		// Avoid duplicate objects, we use map to filter them out.
 		if _, ok := objectsToDelete[object.ObjectName]; !ok {
 			objectsToDelete[object.ObjectName] = index

cmd/disk-cache.go

@@ -239,7 +239,7 @@ func (c *cacheObjects) GetObjectNInfo(ctx context.Context, bucket, object string
 	// skip cache for objects with locks
 	objRetention := objectlock.GetObjectRetentionMeta(objInfo.UserDefined)
 	legalHold := objectlock.GetObjectLegalHoldMeta(objInfo.UserDefined)
-	if objRetention.Mode != objectlock.Invalid || legalHold.Status != "" {
+	if objRetention.Mode.Valid() || legalHold.Status.Valid() {
 		c.cacheStats.incMiss()
 		return c.GetObjectNInfoFn(ctx, bucket, object, rs, h, lockType, opts)
 	}
@@ -614,7 +614,7 @@ func (c *cacheObjects) PutObject(ctx context.Context, bucket, object string, r *
 	// skip cache for objects with locks
 	objRetention := objectlock.GetObjectRetentionMeta(opts.UserDefined)
 	legalHold := objectlock.GetObjectLegalHoldMeta(opts.UserDefined)
-	if objRetention.Mode != objectlock.Invalid || legalHold.Status != "" {
+	if objRetention.Mode.Valid() || legalHold.Status.Valid() {
 		dcache.Delete(ctx, bucket, object)
 		return putObjectFn(ctx, bucket, object, r, opts)
 	}

cmd/object-handlers.go

@@ -1017,27 +1017,29 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 		srcInfo.UserDefined[xhttp.AmzObjectTagging] = tags
 	}
 
-	getObjectInfo := objectAPI.GetObjectInfo
-	if api.CacheAPI() != nil {
-		getObjectInfo = api.CacheAPI().GetObjectInfo
-	}
 	srcInfo.UserDefined = objectlock.FilterObjectLockMetadata(srcInfo.UserDefined, true, true)
 	retPerms := isPutActionAllowed(getRequestAuthType(r), dstBucket, dstObject, r, iampolicy.PutObjectRetentionAction)
 	holdPerms := isPutActionAllowed(getRequestAuthType(r), dstBucket, dstObject, r, iampolicy.PutObjectLegalHoldAction)
 
+	getObjectInfo := objectAPI.GetObjectInfo
+	if api.CacheAPI() != nil {
+		getObjectInfo = api.CacheAPI().GetObjectInfo
+	}
+
 	// apply default bucket configuration/governance headers for dest side.
 	retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, dstBucket, dstObject, getObjectInfo, retPerms, holdPerms)
-	if s3Err == ErrNone && retentionMode != "" {
+	if s3Err == ErrNone && retentionMode.Valid() {
 		srcInfo.UserDefined[xhttp.AmzObjectLockMode] = string(retentionMode)
 		srcInfo.UserDefined[xhttp.AmzObjectLockRetainUntilDate] = retentionDate.UTC().Format(time.RFC3339)
 	}
-	if s3Err == ErrNone && legalHold.Status != "" {
+	if s3Err == ErrNone && legalHold.Status.Valid() {
 		srcInfo.UserDefined[xhttp.AmzObjectLockLegalHold] = string(legalHold.Status)
 	}
 	if s3Err != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL, guessIsBrowserReq(r))
 		return
 	}
+
 	// Store the preserved compression metadata.
 	for k, v := range compressMetadata {
 		srcInfo.UserDefined[k] = v
@@ -1327,20 +1329,25 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
 		writeErrorResponseHeadersOnly(w, toAPIError(ctx, err))
 		return
 	}
+
+	if api.CacheAPI() != nil {
+		putObject = api.CacheAPI().PutObject
+	}
+
+	retPerms := isPutActionAllowed(getRequestAuthType(r), bucket, object, r, iampolicy.PutObjectRetentionAction)
+	holdPerms := isPutActionAllowed(getRequestAuthType(r), bucket, object, r, iampolicy.PutObjectLegalHoldAction)
+
 	getObjectInfo := objectAPI.GetObjectInfo
 	if api.CacheAPI() != nil {
 		getObjectInfo = api.CacheAPI().GetObjectInfo
-		putObject = api.CacheAPI().PutObject
 	}
-	retPerms := isPutActionAllowed(rAuthType, bucket, object, r, iampolicy.PutObjectRetentionAction)
-	holdPerms := isPutActionAllowed(getRequestAuthType(r), bucket, object, r, iampolicy.PutObjectLegalHoldAction)
 
 	retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, getObjectInfo, retPerms, holdPerms)
-	if s3Err == ErrNone && retentionMode != "" {
+	if s3Err == ErrNone && retentionMode.Valid() {
 		metadata[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode)
 		metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = retentionDate.UTC().Format(time.RFC3339)
 	}
-	if s3Err == ErrNone && legalHold.Status != "" {
+	if s3Err == ErrNone && legalHold.Status.Valid() {
 		metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status)
 	}
 	if s3Err != ErrNone {
@@ -1471,6 +1478,14 @@ func (api objectAPIHandlers) NewMultipartUploadHandler(w http.ResponseWriter, r
 		return
 	}
 
+	// Deny if WORM is enabled
+	if globalWORMEnabled {
+		if _, err := objectAPI.GetObjectInfo(ctx, bucket, object, opts); err == nil {
+			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMethodNotAllowed), r.URL, guessIsBrowserReq(r))
+			return
+		}
+	}
+
 	// Validate storage class metadata if present
 	if sc := r.Header.Get(xhttp.AmzStorageClass); sc != "" {
 		if !storageclass.IsValid(sc) {
@@ -1499,21 +1514,28 @@ func (api objectAPIHandlers) NewMultipartUploadHandler(w http.ResponseWriter, r
 		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
 		return
 	}
+
 	retPerms := isPutActionAllowed(getRequestAuthType(r), bucket, object, r, iampolicy.PutObjectRetentionAction)
 	holdPerms := isPutActionAllowed(getRequestAuthType(r), bucket, object, r, iampolicy.PutObjectLegalHoldAction)
 
-	retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, objectAPI.GetObjectInfo, retPerms, holdPerms)
-	if s3Err == ErrNone && retentionMode != "" {
+	getObjectInfo := objectAPI.GetObjectInfo
+	if api.CacheAPI() != nil {
+		getObjectInfo = api.CacheAPI().GetObjectInfo
+	}
+
+	retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, getObjectInfo, retPerms, holdPerms)
+	if s3Err == ErrNone && retentionMode.Valid() {
 		metadata[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode)
 		metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = retentionDate.UTC().Format(time.RFC3339)
 	}
-	if s3Err == ErrNone && legalHold.Status != "" {
+	if s3Err == ErrNone && legalHold.Status.Valid() {
 		metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status)
 	}
 	if s3Err != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL, guessIsBrowserReq(r))
 		return
 	}
+
 	// We need to preserve the encryption headers set in EncryptRequest,
 	// so we do not want to override them, copy them instead.
 	for k, v := range encMetadata {
@@ -2339,22 +2361,6 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite
 		return
 	}
 
-	// Reject retention or governance headers if set, CompleteMultipartUpload spec
-	// does not use these headers, and should not be passed down to checkPutObjectLockAllowed
-	if objectlock.IsObjectLockRequested(r.Header) || objectlock.IsObjectLockGovernanceBypassSet(r.Header) {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidRequest), r.URL, guessIsBrowserReq(r))
-		return
-	}
-
-	// Enforce object lock governance in case a competing upload finalized first.
-	retPerms := isPutActionAllowed(getRequestAuthType(r), bucket, object, r, iampolicy.PutObjectRetentionAction)
-	holdPerms := isPutActionAllowed(getRequestAuthType(r), bucket, object, r, iampolicy.PutObjectLegalHoldAction)
-
-	if _, _, _, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, objectAPI.GetObjectInfo, retPerms, holdPerms); s3Err != ErrNone {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL, guessIsBrowserReq(r))
-		return
-	}
-
 	// Get upload id.
 	uploadID, _, _, _, s3Error := getObjectResources(r.URL.Query())
 	if s3Error != ErrNone {
@@ -2375,6 +2381,36 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidPartOrder), r.URL, guessIsBrowserReq(r))
 		return
 	}
+
+	// Reject retention or governance headers if set, CompleteMultipartUpload spec
+	// does not use these headers, and should not be passed down to checkPutObjectLockAllowed
+	if objectlock.IsObjectLockRequested(r.Header) || objectlock.IsObjectLockGovernanceBypassSet(r.Header) {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidRequest), r.URL, guessIsBrowserReq(r))
+		return
+	}
+
+	// Deny if global WORM is enabled
+	if globalWORMEnabled {
+		opts, err := getOpts(ctx, r, bucket, object)
+		if err != nil {
+			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
+			return
+		}
+		if _, err := objectAPI.GetObjectInfo(ctx, bucket, object, opts); err == nil {
+			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMethodNotAllowed), r.URL, guessIsBrowserReq(r))
+			return
+		}
+	}
+
+	// Enforce object lock governance in case a competing upload finalized first.
+	retPerms := isPutActionAllowed(getRequestAuthType(r), bucket, object, r, iampolicy.PutObjectRetentionAction)
+	holdPerms := isPutActionAllowed(getRequestAuthType(r), bucket, object, r, iampolicy.PutObjectLegalHoldAction)
+
+	if _, _, _, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, objectAPI.GetObjectInfo, retPerms, holdPerms); s3Err != ErrNone {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL, guessIsBrowserReq(r))
+		return
+	}
+
 	var objectEncryptionKey []byte
 	var opts ObjectOptions
 	var isEncrypted, ssec bool
@@ -2546,12 +2582,6 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 		getObjectInfo = api.CacheAPI().GetObjectInfo
 	}
 
-	govBypassPerms := checkRequestAuthType(ctx, r, policy.BypassGovernanceRetentionAction, bucket, object)
-	if _, err := enforceRetentionBypassForDelete(ctx, r, bucket, object, getObjectInfo, govBypassPerms); err != ErrNone {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(err), r.URL, guessIsBrowserReq(r))
-		return
-	}
-
 	if globalDNSConfig != nil {
 		_, err := globalDNSConfig.Get(bucket)
 		if err != nil {
@@ -2560,16 +2590,41 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 		}
 	}
 
-	// http://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectDELETE.html
-	if err := deleteObject(ctx, objectAPI, api.CacheAPI(), bucket, object, r); err != nil {
-		switch err.(type) {
-		case BucketNotFound:
-			// When bucket doesn't exist specially handle it.
+	// Deny if global WORM is enabled
+	if globalWORMEnabled {
+		opts, err := getOpts(ctx, r, bucket, object)
+		if err != nil {
 			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
 			return
 		}
-		// Ignore delete object errors while replying to client, since we are suppposed to reply only 204.
+		if _, err := objectAPI.GetObjectInfo(ctx, bucket, object, opts); err == nil {
+			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMethodNotAllowed), r.URL, guessIsBrowserReq(r))
+			return
+		}
 	}
+
+	apiErr := ErrNone
+	if _, ok := globalBucketObjectLockConfig.Get(bucket); ok {
+		apiErr = enforceRetentionBypassForDelete(ctx, r, bucket, object, getObjectInfo)
+		if apiErr != ErrNone && apiErr != ErrNoSuchKey {
+			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(apiErr), r.URL, guessIsBrowserReq(r))
+			return
+		}
+	}
+
+	if apiErr == ErrNone {
+		// http://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectDELETE.html
+		if err := deleteObject(ctx, objectAPI, api.CacheAPI(), bucket, object, r); err != nil {
+			switch err.(type) {
+			case BucketNotFound:
+				// When bucket doesn't exist specially handle it.
+				writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
+				return
+			}
+			// Ignore delete object errors while replying to client, since we are suppposed to reply only 204.
+		}
+	}
+
 	writeSuccessNoContent(w)
 }
 
@@ -2695,6 +2750,11 @@ func (api objectAPIHandlers) GetObjectLegalHoldHandler(w http.ResponseWriter, r
 		getObjectInfo = api.CacheAPI().GetObjectInfo
 	}
 
+	if _, ok := globalBucketObjectLockConfig.Get(bucket); !ok {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidBucketObjectLockConfiguration), r.URL, guessIsBrowserReq(r))
+		return
+	}
+
 	opts, err := getOpts(ctx, r, bucket, object)
 	if err != nil {
 		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
@@ -2707,15 +2767,12 @@ func (api objectAPIHandlers) GetObjectLegalHoldHandler(w http.ResponseWriter, r
 		return
 	}
 
-	if _, ok := globalBucketObjectLockConfig.Get(bucket); !ok {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidBucketObjectLockConfiguration), r.URL, guessIsBrowserReq(r))
-		return
-	}
 	legalHold := objectlock.GetObjectLegalHoldMeta(objInfo.UserDefined)
 	if legalHold.IsEmpty() {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrNoSuchObjectLockConfiguration), r.URL, guessIsBrowserReq(r))
 		return
 	}
+
 	writeSuccessResponseXML(w, encodeResponse(legalHold))
 	// Notify object legal hold accessed via a GET request.
 	sendEvent(eventArgs{
@@ -2753,8 +2810,9 @@ func (api objectAPIHandlers) PutObjectRetentionHandler(w http.ResponseWriter, r
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL, guessIsBrowserReq(r))
 		return
 	}
-	// Check permissions to perform this governance operation
-	if s3Err := checkRequestAuthType(ctx, r, policy.PutObjectRetentionAction, bucket, object); s3Err != ErrNone {
+
+	cred, owner, claims, s3Err := validateSignature(getRequestAuthType(r), r)
+	if s3Err != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL, guessIsBrowserReq(r))
 		return
 	}
@@ -2763,11 +2821,17 @@ func (api objectAPIHandlers) PutObjectRetentionHandler(w http.ResponseWriter, r
 		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
 		return
 	}
+
 	if !hasContentMD5(r.Header) {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMissingContentMD5), r.URL, guessIsBrowserReq(r))
 		return
 	}
 
+	if globalWORMEnabled {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrObjectLocked), r.URL, guessIsBrowserReq(r))
+		return
+	}
+
 	if _, ok := globalBucketObjectLockConfig.Get(bucket); !ok {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidBucketObjectLockConfiguration), r.URL, guessIsBrowserReq(r))
 		return
@@ -2780,13 +2844,13 @@ func (api objectAPIHandlers) PutObjectRetentionHandler(w http.ResponseWriter, r
 		writeErrorResponse(ctx, w, apiErr, r.URL, guessIsBrowserReq(r))
 		return
 	}
+
 	getObjectInfo := objectAPI.GetObjectInfo
 	if api.CacheAPI() != nil {
 		getObjectInfo = api.CacheAPI().GetObjectInfo
 	}
 
-	govBypassPerms := isPutActionAllowed(getRequestAuthType(r), bucket, object, r, policy.BypassGovernanceRetentionAction)
-	objInfo, s3Err := enforceRetentionBypassForPut(ctx, r, bucket, object, getObjectInfo, govBypassPerms, objRetention)
+	objInfo, s3Err := enforceRetentionBypassForPut(ctx, r, bucket, object, getObjectInfo, objRetention, cred, owner, claims)
 	if s3Err != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL, guessIsBrowserReq(r))
 		return
@@ -2794,7 +2858,7 @@ func (api objectAPIHandlers) PutObjectRetentionHandler(w http.ResponseWriter, r
 
 	objInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] = string(objRetention.Mode)
 	objInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = objRetention.RetainUntilDate.UTC().Format(time.RFC3339)
-	objInfo.metadataOnly = true
+	objInfo.metadataOnly = true // Perform only metadata updates.
 	if _, err = objectAPI.CopyObject(ctx, bucket, object, bucket, object, objInfo, ObjectOptions{}, ObjectOptions{}); err != nil {
 		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
 		return

cmd/object-lock.go

@@ -20,62 +20,164 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"math"
 	"net/http"
 	"path"
 
 	"github.com/minio/minio/cmd/logger"
+	"github.com/minio/minio/pkg/auth"
 	objectlock "github.com/minio/minio/pkg/bucket/object/lock"
+	"github.com/minio/minio/pkg/bucket/policy"
 )
 
+// Similar to enforceRetentionBypassForDelete but for WebUI
+func enforceRetentionBypassForDeleteWeb(ctx context.Context, r *http.Request, bucket, object string, getObjectInfoFn GetObjectInfoFn) APIErrorCode {
+	opts, err := getOpts(ctx, r, bucket, object)
+	if err != nil {
+		return toAPIErrorCode(ctx, err)
+	}
+
+	oi, err := getObjectInfoFn(ctx, bucket, object, opts)
+	if err != nil {
+		return toAPIErrorCode(ctx, err)
+	}
+
+	lhold := objectlock.GetObjectLegalHoldMeta(oi.UserDefined)
+	if lhold.Status.Valid() && lhold.Status == objectlock.LegalHoldOn {
+		return ErrObjectLocked
+	}
+
+	ret := objectlock.GetObjectRetentionMeta(oi.UserDefined)
+	if ret.Mode.Valid() {
+		switch ret.Mode {
+		case objectlock.RetCompliance:
+			// In compliance mode, a protected object version can't be overwritten
+			// or deleted by any user, including the root user in your AWS account.
+			// When an object is locked in compliance mode, its retention mode can't
+			// be changed, and its retention period can't be shortened. Compliance mode
+			// ensures that an object version can't be overwritten or deleted for the
+			// duration of the retention period.
+			t, err := objectlock.UTCNowNTP()
+			if err != nil {
+				logger.LogIf(ctx, err)
+				return ErrObjectLocked
+			}
+
+			if !ret.RetainUntilDate.Before(t) {
+				return ErrObjectLocked
+			}
+			return ErrNone
+		case objectlock.RetGovernance:
+			// In governance mode, users can't overwrite or delete an object
+			// version or alter its lock settings unless they have special
+			// permissions. With governance mode, you protect objects against
+			// being deleted by most users, but you can still grant some users
+			// permission to alter the retention settings or delete the object
+			// if necessary. You can also use governance mode to test retention-period
+			// settings before creating a compliance-mode retention period.
+			// To override or remove governance-mode retention settings, a
+			// user must have the s3:BypassGovernanceRetention permission
+			// and must explicitly include x-amz-bypass-governance-retention:true
+			// as a request header with any request that requires overriding
+			// governance mode.
+			byPassSet := objectlock.IsObjectLockGovernanceBypassSet(r.Header)
+			if !byPassSet {
+				t, err := objectlock.UTCNowNTP()
+				if err != nil {
+					logger.LogIf(ctx, err)
+					return ErrObjectLocked
+				}
+
+				if !ret.RetainUntilDate.Before(t) {
+					return ErrObjectLocked
+				}
+				return ErrNone
+			}
+		}
+	}
+	return ErrNone
+}
+
 // enforceRetentionBypassForDelete enforces whether an existing object under governance can be deleted
 // with governance bypass headers set in the request.
 // Objects under site wide WORM can never be overwritten.
 // For objects in "Governance" mode, overwrite is allowed if a) object retention date is past OR
 // governance bypass headers are set and user has governance bypass permissions.
 // Objects in "Compliance" mode can be overwritten only if retention date is past.
-func enforceRetentionBypassForDelete(ctx context.Context, r *http.Request, bucket, object string, getObjectInfoFn GetObjectInfoFn, govBypassPerm APIErrorCode) (oi ObjectInfo, s3Err APIErrorCode) {
-	if globalWORMEnabled {
-		return oi, ErrObjectLocked
-	}
-	var err error
-	var opts ObjectOptions
-	opts, err = getOpts(ctx, r, bucket, object)
+func enforceRetentionBypassForDelete(ctx context.Context, r *http.Request, bucket, object string, getObjectInfoFn GetObjectInfoFn) APIErrorCode {
+	opts, err := getOpts(ctx, r, bucket, object)
 	if err != nil {
-		return oi, toAPIErrorCode(ctx, err)
+		return toAPIErrorCode(ctx, err)
 	}
-	oi, err = getObjectInfoFn(ctx, bucket, object, opts)
+
+	oi, err := getObjectInfoFn(ctx, bucket, object, opts)
 	if err != nil {
-		// ignore case where object no longer exists
-		if toAPIError(ctx, err).Code == "NoSuchKey" {
-			oi.UserDefined = map[string]string{}
-			return oi, ErrNone
-		}
-		return oi, toAPIErrorCode(ctx, err)
+		return toAPIErrorCode(ctx, err)
 	}
-	ret := objectlock.GetObjectRetentionMeta(oi.UserDefined)
+
 	lhold := objectlock.GetObjectLegalHoldMeta(oi.UserDefined)
-	if lhold.Status == objectlock.ON {
-		return oi, ErrObjectLocked
+	if lhold.Status.Valid() && lhold.Status == objectlock.LegalHoldOn {
+		return ErrObjectLocked
 	}
-	// Here bucket does not support object lock
-	if ret.Mode == objectlock.Invalid {
-		return oi, ErrNone
+
+	ret := objectlock.GetObjectRetentionMeta(oi.UserDefined)
+	if ret.Mode.Valid() {
+		switch ret.Mode {
+		case objectlock.RetCompliance:
+			// In compliance mode, a protected object version can't be overwritten
+			// or deleted by any user, including the root user in your AWS account.
+			// When an object is locked in compliance mode, its retention mode can't
+			// be changed, and its retention period can't be shortened. Compliance mode
+			// ensures that an object version can't be overwritten or deleted for the
+			// duration of the retention period.
+			t, err := objectlock.UTCNowNTP()
+			if err != nil {
+				logger.LogIf(ctx, err)
+				return ErrObjectLocked
+			}
+
+			if !ret.RetainUntilDate.Before(t) {
+				return ErrObjectLocked
+			}
+			return ErrNone
+		case objectlock.RetGovernance:
+			// In governance mode, users can't overwrite or delete an object
+			// version or alter its lock settings unless they have special
+			// permissions. With governance mode, you protect objects against
+			// being deleted by most users, but you can still grant some users
+			// permission to alter the retention settings or delete the object
+			// if necessary. You can also use governance mode to test retention-period
+			// settings before creating a compliance-mode retention period.
+			// To override or remove governance-mode retention settings, a
+			// user must have the s3:BypassGovernanceRetention permission
+			// and must explicitly include x-amz-bypass-governance-retention:true
+			// as a request header with any request that requires overriding
+			// governance mode.
+			//
+			byPassSet := objectlock.IsObjectLockGovernanceBypassSet(r.Header)
+			if !byPassSet {
+				t, err := objectlock.UTCNowNTP()
+				if err != nil {
+					logger.LogIf(ctx, err)
+					return ErrObjectLocked
+				}
+
+				if !ret.RetainUntilDate.Before(t) {
+					return ErrObjectLocked
+				}
+				return ErrNone
+			}
+			// https://docs.aws.amazon.com/AmazonS3/latest/dev/object-lock-overview.html#object-lock-retention-modes
+			// If you try to delete objects protected by governance mode and have s3:BypassGovernanceRetention
+			// or s3:GetBucketObjectLockConfiguration permissions, the operation will succeed.
+			govBypassPerms1 := checkRequestAuthType(ctx, r, policy.BypassGovernanceRetentionAction, bucket, object)
+			govBypassPerms2 := checkRequestAuthType(ctx, r, policy.GetBucketObjectLockConfigurationAction, bucket, object)
+			if govBypassPerms1 != ErrNone && govBypassPerms2 != ErrNone {
+				return ErrAccessDenied
+			}
+		}
 	}
-	if ret.Mode != objectlock.Compliance && ret.Mode != objectlock.Governance {
-		return oi, ErrUnknownWORMModeDirective
-	}
-	t, err := objectlock.UTCNowNTP()
-	if err != nil {
-		logger.LogIf(ctx, err)
-		return oi, ErrObjectLocked
-	}
-	if ret.RetainUntilDate.Before(t) {
-		return oi, ErrNone
-	}
-	if objectlock.IsObjectLockGovernanceBypassSet(r.Header) && ret.Mode == objectlock.Governance && govBypassPerm == ErrNone {
-		return oi, ErrNone
-	}
-	return oi, ErrObjectLocked
+	return ErrNone
 }
 
 // enforceRetentionBypassForPut enforces whether an existing object under governance can be overwritten
@@ -84,66 +186,68 @@ func enforceRetentionBypassForDelete(ctx context.Context, r *http.Request, bucke
 // For objects in "Governance" mode, overwrite is allowed if a) object retention date is past OR
 // governance bypass headers are set and user has governance bypass permissions.
 // Objects in compliance mode can be overwritten only if retention date is being extended. No mode change is permitted.
-func enforceRetentionBypassForPut(ctx context.Context, r *http.Request, bucket, object string, getObjectInfoFn GetObjectInfoFn, govBypassPerm APIErrorCode, objRetention *objectlock.ObjectRetention) (oi ObjectInfo, s3Err APIErrorCode) {
-	if globalWORMEnabled {
-		return oi, ErrObjectLocked
+func enforceRetentionBypassForPut(ctx context.Context, r *http.Request, bucket, object string, getObjectInfoFn GetObjectInfoFn, objRetention *objectlock.ObjectRetention, cred auth.Credentials, owner bool, claims map[string]interface{}) (ObjectInfo, APIErrorCode) {
+	byPassSet := objectlock.IsObjectLockGovernanceBypassSet(r.Header)
+	opts, err := getOpts(ctx, r, bucket, object)
+	if err != nil {
+		return ObjectInfo{}, toAPIErrorCode(ctx, err)
 	}
 
-	var err error
-	var opts ObjectOptions
-	opts, err = getOpts(ctx, r, bucket, object)
+	oi, err := getObjectInfoFn(ctx, bucket, object, opts)
 	if err != nil {
 		return oi, toAPIErrorCode(ctx, err)
 	}
-	oi, err = getObjectInfoFn(ctx, bucket, object, opts)
-	if err != nil {
-		// ignore case where object no longer exists
-		if toAPIError(ctx, err).Code == "NoSuchKey" {
-			oi.UserDefined = map[string]string{}
-			return oi, ErrNone
-		}
-		return oi, toAPIErrorCode(ctx, err)
-	}
 
-	ret := objectlock.GetObjectRetentionMeta(oi.UserDefined)
-	// no retention metadata on object
-	if ret.Mode == objectlock.Invalid {
-		if _, isWORMBucket := globalBucketObjectLockConfig.Get(bucket); !isWORMBucket {
-			return oi, ErrInvalidBucketObjectLockConfiguration
-		}
-		return oi, ErrNone
-	}
 	t, err := objectlock.UTCNowNTP()
 	if err != nil {
 		logger.LogIf(ctx, err)
 		return oi, ErrObjectLocked
 	}
 
-	if ret.Mode == objectlock.Compliance {
-		// Compliance retention mode cannot be changed and retention period cannot be shortened as per
-		// https://docs.aws.amazon.com/AmazonS3/latest/dev/object-lock-overview.html#object-lock-retention-modes
-		if objRetention.Mode != objectlock.Compliance || objRetention.RetainUntilDate.Before(ret.RetainUntilDate.Time) {
-			return oi, ErrObjectLocked
-		}
-		if objRetention.RetainUntilDate.Before(t) {
-			return oi, ErrInvalidRetentionDate
-		}
-		return oi, ErrNone
-	}
+	// Pass in relative days from current time, to additionally to verify "object-lock-remaining-retention-days" policy if any.
+	days := int(math.Ceil(math.Abs(objRetention.RetainUntilDate.Sub(t).Hours()) / 24))
 
-	if ret.Mode == objectlock.Governance {
-		if !objectlock.IsObjectLockGovernanceBypassSet(r.Header) {
-			if objRetention.RetainUntilDate.Before(t) {
-				return oi, ErrInvalidRetentionDate
+	ret := objectlock.GetObjectRetentionMeta(oi.UserDefined)
+	if ret.Mode.Valid() {
+		// Retention has expired you may change whatever you like.
+		if ret.RetainUntilDate.Before(t) {
+			perm := isPutRetentionAllowed(bucket, object,
+				days, objRetention.RetainUntilDate.Time,
+				objRetention.Mode, byPassSet, r, cred,
+				owner, claims)
+			return oi, perm
+		}
+
+		switch ret.Mode {
+		case objectlock.RetGovernance:
+			govPerm := isPutRetentionAllowed(bucket, object, days,
+				objRetention.RetainUntilDate.Time, objRetention.Mode,
+				byPassSet, r, cred, owner, claims)
+			// Governance mode retention period cannot be shortened, if x-amz-bypass-governance is not set.
+			if !byPassSet {
+				if objRetention.Mode != objectlock.RetGovernance || objRetention.RetainUntilDate.Before((ret.RetainUntilDate.Time)) {
+					return oi, ErrObjectLocked
+				}
 			}
-			if objRetention.RetainUntilDate.Before((ret.RetainUntilDate.Time)) {
+			return oi, govPerm
+		case objectlock.RetCompliance:
+			// Compliance retention mode cannot be changed or shortened.
+			// https://docs.aws.amazon.com/AmazonS3/latest/dev/object-lock-overview.html#object-lock-retention-modes
+			if objRetention.Mode != objectlock.RetCompliance || objRetention.RetainUntilDate.Before((ret.RetainUntilDate.Time)) {
 				return oi, ErrObjectLocked
 			}
-			return oi, ErrNone
+			compliancePerm := isPutRetentionAllowed(bucket, object,
+				days, objRetention.RetainUntilDate.Time, objRetention.Mode,
+				false, r, cred, owner, claims)
+			return oi, compliancePerm
 		}
-		return oi, govBypassPerm
-	}
-	return oi, ErrNone
+		return oi, ErrNone
+	} // No pre-existing retention metadata present.
+
+	perm := isPutRetentionAllowed(bucket, object,
+		days, objRetention.RetainUntilDate.Time,
+		objRetention.Mode, byPassSet, r, cred, owner, claims)
+	return oi, perm
 }
 
 // checkPutObjectLockAllowed enforces object retention policy and legal hold policy
@@ -156,16 +260,23 @@ func enforceRetentionBypassForPut(ctx context.Context, r *http.Request, bucket,
 // For objects in "Compliance" mode, retention date cannot be shortened, and mode cannot be altered.
 // For objects with legal hold header set, the s3:PutObjectLegalHold permission is expected to be set
 // Both legal hold and retention can be applied independently on an object
-func checkPutObjectLockAllowed(ctx context.Context, r *http.Request, bucket, object string, getObjectInfoFn GetObjectInfoFn, retentionPermErr, legalHoldPermErr APIErrorCode) (objectlock.Mode, objectlock.RetentionDate, objectlock.ObjectLegalHold, APIErrorCode) {
-	var mode objectlock.Mode
+func checkPutObjectLockAllowed(ctx context.Context, r *http.Request, bucket, object string, getObjectInfoFn GetObjectInfoFn, retentionPermErr, legalHoldPermErr APIErrorCode) (objectlock.RetMode, objectlock.RetentionDate, objectlock.ObjectLegalHold, APIErrorCode) {
+	var mode objectlock.RetMode
 	var retainDate objectlock.RetentionDate
 	var legalHold objectlock.ObjectLegalHold
 
-	retentionCfg, isWORMBucket := globalBucketObjectLockConfig.Get(bucket)
-
 	retentionRequested := objectlock.IsObjectLockRetentionRequested(r.Header)
 	legalHoldRequested := objectlock.IsObjectLockLegalHoldRequested(r.Header)
 
+	retentionCfg, isWORMBucket := globalBucketObjectLockConfig.Get(bucket)
+	if !isWORMBucket {
+		if legalHoldRequested || retentionRequested {
+			return mode, retainDate, legalHold, ErrInvalidBucketObjectLockConfiguration
+		}
+		// If this not a WORM enabled bucket, we should return right here.
+		return mode, retainDate, legalHold, ErrNone
+	}
+
 	var objExists bool
 	opts, err := getOpts(ctx, r, bucket, object)
 	if err != nil {
@@ -177,33 +288,30 @@ func checkPutObjectLockAllowed(ctx context.Context, r *http.Request, bucket, obj
 		logger.LogIf(ctx, err)
 		return mode, retainDate, legalHold, ErrObjectLocked
 	}
+
 	if objInfo, err := getObjectInfoFn(ctx, bucket, object, opts); err == nil {
 		objExists = true
 		r := objectlock.GetObjectRetentionMeta(objInfo.UserDefined)
-		if globalWORMEnabled || ((r.Mode == objectlock.Compliance) && r.RetainUntilDate.After(t)) {
+		if globalWORMEnabled || ((r.Mode == objectlock.RetCompliance) && r.RetainUntilDate.After(t)) {
 			return mode, retainDate, legalHold, ErrObjectLocked
 		}
 		mode = r.Mode
 		retainDate = r.RetainUntilDate
 		legalHold = objectlock.GetObjectLegalHoldMeta(objInfo.UserDefined)
 		// Disallow overwriting an object on legal hold
-		if legalHold.Status == "ON" {
+		if legalHold.Status == objectlock.LegalHoldOn {
 			return mode, retainDate, legalHold, ErrObjectLocked
 		}
 	}
+
 	if legalHoldRequested {
-		if !isWORMBucket {
-			return mode, retainDate, legalHold, ErrInvalidBucketObjectLockConfiguration
-		}
 		var lerr error
 		if legalHold, lerr = objectlock.ParseObjectLockLegalHoldHeaders(r.Header); lerr != nil {
 			return mode, retainDate, legalHold, toAPIErrorCode(ctx, err)
 		}
 	}
+
 	if retentionRequested {
-		if !isWORMBucket {
-			return mode, retainDate, legalHold, ErrInvalidBucketObjectLockConfiguration
-		}
 		legalHold, err := objectlock.ParseObjectLockLegalHoldHeaders(r.Header)
 		if err != nil {
 			return mode, retainDate, legalHold, toAPIErrorCode(ctx, err)
@@ -215,9 +323,6 @@ func checkPutObjectLockAllowed(ctx context.Context, r *http.Request, bucket, obj
 		if objExists && retainDate.After(t) {
 			return mode, retainDate, legalHold, ErrObjectLocked
 		}
-		if rMode == objectlock.Invalid {
-			return mode, retainDate, legalHold, toAPIErrorCode(ctx, objectlock.ErrObjectLockInvalidHeaders)
-		}
 		if retentionPermErr != ErrNone {
 			return mode, retainDate, legalHold, retentionPermErr
 		}
@@ -241,7 +346,7 @@ func checkPutObjectLockAllowed(ctx context.Context, r *http.Request, bucket, obj
 			// inherit retention from bucket configuration
 			return retentionCfg.Mode, objectlock.RetentionDate{Time: t.Add(retentionCfg.Validity)}, legalHold, ErrNone
 		}
-		return objectlock.Mode(""), objectlock.RetentionDate{}, legalHold, ErrNone
+		return "", objectlock.RetentionDate{}, legalHold, ErrNone
 	}
 	return mode, retainDate, legalHold, ErrNone
 }

cmd/policy.go

@@ -20,10 +20,10 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
-	"fmt"
 	"net/http"
 	"net/url"
 	"path"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -146,7 +146,7 @@ func NewPolicySys() *PolicySys {
 	}
 }
 
-func getConditionValues(request *http.Request, locationConstraint string, username string, claims map[string]interface{}) map[string][]string {
+func getConditionValues(r *http.Request, lc string, username string, claims map[string]interface{}) map[string][]string {
 	currTime := UTCNow()
 
 	principalType := "Anonymous"
@@ -156,22 +156,21 @@ func getConditionValues(request *http.Request, locationConstraint string, userna
 
 	args := map[string][]string{
 		"CurrentTime":     {currTime.Format(time.RFC3339)},
-		"EpochTime":       {fmt.Sprintf("%d", currTime.Unix())},
+		"EpochTime":       {strconv.FormatInt(currTime.Unix(), 10)},
+		"SecureTransport": {strconv.FormatBool(r.TLS != nil)},
+		"SourceIp":        {handlers.GetSourceIP(r)},
+		"UserAgent":       {r.UserAgent()},
+		"Referer":         {r.Referer()},
 		"principaltype":   {principalType},
-		"SecureTransport": {fmt.Sprintf("%t", request.TLS != nil)},
-		"SourceIp":        {handlers.GetSourceIP(request)},
-		"UserAgent":       {request.UserAgent()},
-		"Referer":         {request.Referer()},
 		"userid":          {username},
 		"username":        {username},
 	}
 
-	if locationConstraint != "" {
-		args["LocationConstraint"] = []string{locationConstraint}
+	if lc != "" {
+		args["LocationConstraint"] = []string{lc}
 	}
 
-	// TODO: support object-lock-remaining-retention-days
-	cloneHeader := request.Header.Clone()
+	cloneHeader := r.Header.Clone()
 
 	for _, objLock := range []string{
 		xhttp.AmzObjectLockMode,
@@ -193,7 +192,7 @@ func getConditionValues(request *http.Request, locationConstraint string, userna
 	}
 
 	var cloneURLValues = url.Values{}
-	for k, v := range request.URL.Query() {
+	for k, v := range r.URL.Query() {
 		cloneURLValues[k] = v
 	}
 

cmd/web-handlers.go

@@ -588,11 +588,17 @@ func (web *webAPIHandlers) RemoveObject(r *http.Request, args *RemoveObjectArgs,
 	if objectAPI == nil {
 		return toJSONError(ctx, errServerNotInitialized)
 	}
-	listObjects := objectAPI.ListObjects
+
 	getObjectInfo := objectAPI.GetObjectInfo
 	if web.CacheAPI() != nil {
 		getObjectInfo = web.CacheAPI().GetObjectInfo
 	}
+
+	deleteObjects := objectAPI.DeleteObjects
+	if web.CacheAPI() != nil {
+		deleteObjects = web.CacheAPI().DeleteObjects
+	}
+
 	claims, owner, authErr := webRequestAuthenticate(r)
 	if authErr != nil {
 		if authErr == errNoAuthToken {
@@ -688,6 +694,17 @@ next:
 				}) {
 					govBypassPerms = ErrNone
 				}
+				if globalIAMSys.IsAllowed(iampolicy.Args{
+					AccountName:     claims.AccessKey,
+					Action:          iampolicy.GetBucketObjectLockConfigurationAction,
+					BucketName:      args.BucketName,
+					ConditionValues: getConditionValues(r, "", claims.AccessKey, claims.Map()),
+					IsOwner:         owner,
+					ObjectName:      objectName,
+					Claims:          claims.Map(),
+				}) {
+					govBypassPerms = ErrNone
+				}
 			}
 			if authErr == errNoAuthToken {
 				// Check if object is allowed to be deleted anonymously
@@ -711,12 +728,44 @@ next:
 				}) {
 					govBypassPerms = ErrNone
 				}
+
+				// Check if object is allowed to be deleted anonymously
+				if globalPolicySys.IsAllowed(policy.Args{
+					Action:          policy.GetBucketObjectLockConfigurationAction,
+					BucketName:      args.BucketName,
+					ConditionValues: getConditionValues(r, "", "", nil),
+					IsOwner:         false,
+					ObjectName:      objectName,
+				}) {
+					govBypassPerms = ErrNone
+				}
 			}
-			if _, err := enforceRetentionBypassForDelete(ctx, r, args.BucketName, objectName, getObjectInfo, govBypassPerms); err != ErrNone {
+			if govBypassPerms != ErrNone {
 				return toJSONError(ctx, errAccessDenied)
 			}
-			if err = deleteObject(ctx, objectAPI, web.CacheAPI(), args.BucketName, objectName, r); err != nil {
-				break next
+
+			apiErr := ErrNone
+			// Deny if global WORM is enabled
+			if globalWORMEnabled {
+				opts, err := getOpts(ctx, r, args.BucketName, objectName)
+				if err != nil {
+					apiErr = toAPIErrorCode(ctx, err)
+				} else {
+					if _, err := getObjectInfo(ctx, args.BucketName, objectName, opts); err == nil {
+						apiErr = ErrMethodNotAllowed
+					}
+				}
+			}
+			if _, ok := globalBucketObjectLockConfig.Get(args.BucketName); ok && (apiErr == ErrNone) {
+				apiErr = enforceRetentionBypassForDeleteWeb(ctx, r, args.BucketName, objectName, getObjectInfo)
+				if apiErr != ErrNone && apiErr != ErrNoSuchKey {
+					return toJSONError(ctx, errAccessDenied)
+				}
+			}
+			if apiErr == ErrNone {
+				if err = deleteObject(ctx, objectAPI, web.CacheAPI(), args.BucketName, objectName, r); err != nil {
+					break next
+				}
 			}
 			continue
 		}
@@ -746,23 +795,34 @@ next:
 			}
 		}
 
-		// For directories, list the contents recursively and remove.
-		marker := ""
+		// Allocate new results channel to receive ObjectInfo.
+		objInfoCh := make(chan ObjectInfo)
+
+		// Walk through all objects
+		if err = objectAPI.Walk(ctx, args.BucketName, objectName, objInfoCh); err != nil {
+			break next
+		}
+
 		for {
-			var lo ListObjectsInfo
-			lo, err = listObjects(ctx, args.BucketName, objectName, marker, "", maxObjectList)
-			if err != nil {
+			var objects []string
+			for obj := range objInfoCh {
+				if len(objects) == maxObjectList {
+					// Reached maximum delete requests, attempt a delete for now.
+					break
+				}
+				objects = append(objects, obj.Name)
+			}
+
+			// Nothing to do.
+			if len(objects) == 0 {
 				break next
 			}
-			marker = lo.NextMarker
-			for _, obj := range lo.Objects {
-				err = deleteObject(ctx, objectAPI, web.CacheAPI(), args.BucketName, obj.Name, r)
-				if err != nil {
-					break next
-				}
-			}
-			if !lo.IsTruncated {
-				break
+
+			// Deletes a list of objects.
+			_, err = deleteObjects(ctx, args.BucketName, objects)
+			if err != nil {
+				logger.LogIf(ctx, err)
+				break next
 			}
 		}
 	}
@@ -1097,27 +1157,30 @@ func (web *webAPIHandlers) Upload(w http.ResponseWriter, r *http.Request) {
 	// Ensure that metadata does not contain sensitive information
 	crypto.RemoveSensitiveEntries(metadata)
 
-	getObjectInfo := objectAPI.GetObjectInfo
-	if web.CacheAPI() != nil {
-		getObjectInfo = web.CacheAPI().GetObjectInfo
-	}
-	// 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(time.RFC3339)
-	}
-	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
-	}
+	retentionRequested := objectlock.IsObjectLockRetentionRequested(r.Header)
+	legalHoldRequested := objectlock.IsObjectLockLegalHoldRequested(r.Header)
 
 	putObject := objectAPI.PutObject
+	getObjectInfo := objectAPI.GetObjectInfo
 	if web.CacheAPI() != nil {
 		putObject = web.CacheAPI().PutObject
+		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(time.RFC3339)
+		}
+		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
+		}
 	}
 
 	objInfo, err := putObject(context.Background(), bucket, object, pReader, opts)
@@ -1462,13 +1525,12 @@ func (web *webAPIHandlers) DownloadZip(w http.ResponseWriter, r *http.Request) {
 		writeWebErrorResponse(w, errInvalidBucketName)
 		return
 	}
+
 	getObjectNInfo := objectAPI.GetObjectNInfo
 	if web.CacheAPI() != nil {
 		getObjectNInfo = web.CacheAPI().GetObjectNInfo
 	}
 
-	listObjects := objectAPI.ListObjects
-
 	archive := zip.NewWriter(w)
 	defer archive.Close()
 
@@ -1541,29 +1603,24 @@ func (web *webAPIHandlers) DownloadZip(w http.ResponseWriter, r *http.Request) {
 			// If not a directory, compress the file and write it to response.
 			err := zipit(pathJoin(args.Prefix, object))
 			if err != nil {
+				logger.LogIf(ctx, err)
 				return
 			}
 			continue
 		}
 
-		// For directories, list the contents recursively and write the objects as compressed
-		// date to the response writer.
-		marker := ""
-		for {
-			lo, err := listObjects(ctx, args.BucketName, pathJoin(args.Prefix, object), marker, "",
-				maxObjectList)
-			if err != nil {
-				return
-			}
-			marker = lo.NextMarker
-			for _, obj := range lo.Objects {
-				err = zipit(obj.Name)
-				if err != nil {
-					return
-				}
-			}
-			if !lo.IsTruncated {
-				break
+		objInfoCh := make(chan ObjectInfo)
+
+		// Walk through all objects
+		if err := objectAPI.Walk(ctx, args.BucketName, pathJoin(args.Prefix, object), objInfoCh); err != nil {
+			logger.LogIf(ctx, err)
+			continue
+		}
+
+		for obj := range objInfoCh {
+			if err := zipit(obj.Name); err != nil {
+				logger.LogIf(ctx, err)
+				continue
 			}
 		}
 	}