Fix retention enforcement in Compliance mode (#8556)

In compliance mode, the retention date can be extended with 
governance bypass permissions

cmd/bucket-handlers.go

@@ -392,7 +392,7 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 			continue
 		}
 		govBypassPerms := checkRequestAuthType(ctx, r, policy.BypassGovernanceRetentionAction, bucket, object.ObjectName)
-		if _, err := checkGovernanceBypassAllowed(ctx, r, bucket, object.ObjectName, getObjectInfoFn, govBypassPerms); err != ErrNone {
+		if _, err := enforceRetentionBypassForDelete(ctx, r, bucket, object.ObjectName, getObjectInfoFn, govBypassPerms); err != ErrNone {
 			dErrs[index] = err
 			continue
 		}

cmd/object-handlers.go

@@ -2405,7 +2405,7 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 	}
 
 	govBypassPerms := checkRequestAuthType(ctx, r, policy.BypassGovernanceRetentionAction, bucket, object)
-	if _, err := checkGovernanceBypassAllowed(ctx, r, bucket, object, getObjectInfo, govBypassPerms); err != ErrNone {
+	if _, err := enforceRetentionBypassForDelete(ctx, r, bucket, object, getObjectInfo, govBypassPerms); err != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(err), r.URL, guessIsBrowserReq(r))
 		return
 	}
@@ -2536,6 +2536,11 @@ func (api objectAPIHandlers) PutObjectRetentionHandler(w http.ResponseWriter, r
 		return
 	}
 
+	if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
+		return
+	}
+
 	// Get Content-Md5 sent by client and verify if valid
 	md5Bytes, err := checkValidMD5(r.Header)
 	if err != nil {
@@ -2547,25 +2552,24 @@ func (api objectAPIHandlers) PutObjectRetentionHandler(w http.ResponseWriter, r
 		return
 	}
 
+	objRetention, err := parseObjectRetention(r.Body)
+	if err != nil {
+		apiErr := errorCodes.ToAPIErr(ErrMalformedXML)
+		apiErr.Description = err.Error()
+		writeErrorResponse(ctx, w, apiErr, r.URL, guessIsBrowserReq(r))
+		return
+	}
 	getObjectInfo := objectAPI.GetObjectInfo
 	if api.CacheAPI() != nil {
 		getObjectInfo = api.CacheAPI().GetObjectInfo
 	}
 
-	govBypassPerms := checkRequestAuthType(ctx, r, policy.BypassGovernanceRetentionAction, bucket, object)
-	objInfo, s3Err := checkGovernanceBypassAllowed(ctx, r, bucket, object, getObjectInfo, govBypassPerms)
+	govBypassPerms := isPutActionAllowed(getRequestAuthType(r), bucket, object, r, policy.BypassGovernanceRetentionAction)
+	objInfo, s3Err := enforceRetentionBypassForPut(ctx, r, bucket, object, getObjectInfo, govBypassPerms, objRetention)
 	if s3Err != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL, guessIsBrowserReq(r))
 		return
 	}
-
-	objRetention, err := parseObjectRetention(r.Body)
-	if err != nil {
-		apiErr := errorCodes.ToAPIErr(ErrMalformedXML)
-		apiErr.Description = err.Error()
-		writeErrorResponse(ctx, w, apiErr, r.URL, guessIsBrowserReq(r))
-		return
-	}
 	// verify Content-MD5 sum of request body if this header set
 	if len(md5Bytes) > 0 {
 		data, err := xml.Marshal(objRetention)

cmd/object-lock.go

@@ -43,7 +43,7 @@ const (
 	Compliance RetentionMode = "COMPLIANCE"
 
 	// Invalid - invalid retention mode.
-	Invalid RetentionMode = "Invalid"
+	Invalid RetentionMode = ""
 )
 
 func parseRetentionMode(modeStr string) (mode RetentionMode) {
@@ -387,6 +387,7 @@ func parseObjectLockRetentionHeaders(h http.Header) (rmode RetentionMode, r Rete
 func getObjectRetentionMeta(meta map[string]string) ObjectRetention {
 	var mode RetentionMode
 	var retainTill RetentionDate
+
 	if modeStr, ok := meta[strings.ToLower(xhttp.AmzObjectLockMode)]; ok {
 		mode = parseRetentionMode(modeStr)
 	}
@@ -398,12 +399,16 @@ func getObjectRetentionMeta(meta map[string]string) ObjectRetention {
 	return ObjectRetention{Mode: mode, RetainUntilDate: retainTill}
 }
 
-// checkGovernanceBypassAllowed enforces whether an existing object under governance can be overwritten
+// enforceRetentionBypassForDelete enforces whether an existing object under governance can be deleted
 // with governance bypass headers set in the request.
-// Objects under site wide WORM or those in "Compliance" mode can never be overwritten.
+// 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.
-func checkGovernanceBypassAllowed(ctx context.Context, r *http.Request, bucket, object string, getObjectInfoFn GetObjectInfoFn, govBypassPerm APIErrorCode) (oi ObjectInfo, s3Err APIErrorCode) {
+// 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)
@@ -420,31 +425,93 @@ func checkGovernanceBypassAllowed(ctx context.Context, r *http.Request, bucket,
 		return oi, toAPIErrorCode(ctx, err)
 	}
 	ret := getObjectRetentionMeta(oi.UserDefined)
-	if globalWORMEnabled || ret.Mode == Compliance {
+
+	// Here bucket does not support object lock
+	if ret.Mode == Invalid {
+		return oi, ErrNone
+	}
+	if ret.Mode != Compliance && ret.Mode != Governance {
+		return oi, ErrUnknownWORMModeDirective
+	}
+	t, err := UTCNowNTP()
+	if err != nil {
+		logger.LogIf(ctx, err)
 		return oi, ErrObjectLocked
 	}
-	// Here bucket does not support object lock
-	if ret.Mode == Invalid && isObjectLockGovernanceBypassSet(r.Header) {
-		return oi, ErrInvalidBucketObjectLockConfiguration
+	if ret.RetainUntilDate.Before(t) {
+		return oi, ErrNone
+	}
+	if isObjectLockGovernanceBypassSet(r.Header) && ret.Mode == Governance && govBypassPerm == ErrNone {
+		return oi, ErrNone
 	}
-	if ret.Mode == Compliance {
 	return oi, ErrObjectLocked
 }
-	if ret.Mode == Governance {
-		if !isObjectLockGovernanceBypassSet(r.Header) {
+
+// enforceRetentionBypassForPut enforces whether an existing object under governance can be overwritten
+// with governance bypass headers set in the request.
+// Objects under site wide WORM cannot 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 being extended. No mode change is permitted.
+func enforceRetentionBypassForPut(ctx context.Context, r *http.Request, bucket, object string, getObjectInfoFn GetObjectInfoFn, govBypassPerm APIErrorCode, objRetention *ObjectRetention) (oi ObjectInfo, s3Err APIErrorCode) {
+	if globalWORMEnabled {
+		return oi, ErrObjectLocked
+	}
+
+	var err error
+	var opts ObjectOptions
+	opts, err = getOpts(ctx, r, bucket, object)
+	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 := getObjectRetentionMeta(oi.UserDefined)
+	// no retention metadata on object
+	if ret.Mode == Invalid {
+		_, isWORMBucket := isWORMEnabled(bucket)
+		if !isWORMBucket {
+			return oi, ErrInvalidBucketObjectLockConfiguration
+		}
+		return oi, ErrNone
+	}
 	t, err := UTCNowNTP()
 	if err != nil {
 		logger.LogIf(ctx, err)
 		return oi, ErrObjectLocked
 	}
-			if ret.RetainUntilDate.After(t) {
+
+	if ret.Mode == 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 != Compliance || objRetention.RetainUntilDate.Before(ret.RetainUntilDate.Time) {
 			return oi, ErrObjectLocked
 		}
+		if objRetention.RetainUntilDate.Before(t) {
+			return oi, ErrInvalidRetentionDate
+		}
 		return oi, ErrNone
 	}
-		if govBypassPerm != ErrNone {
-			return oi, ErrAccessDenied
+
+	if ret.Mode == Governance {
+		if !isObjectLockGovernanceBypassSet(r.Header) {
+			if objRetention.RetainUntilDate.Before(t) {
+				return oi, ErrInvalidRetentionDate
+			}
+			if objRetention.RetainUntilDate.Before((ret.RetainUntilDate.Time)) {
+				return oi, ErrObjectLocked
+			}
+			return oi, ErrNone
 		}
+		return oi, govBypassPerm
 	}
 	return oi, ErrNone
 }
@@ -453,11 +520,13 @@ func checkGovernanceBypassAllowed(ctx context.Context, r *http.Request, bucket,
 // See https://docs.aws.amazon.com/AmazonS3/latest/dev/object-lock-managing.html for the spec.
 // For non-existing objects with object retention headers set, this method returns ErrNone if bucket has
 // locking enabled and user has requisite permissions (s3:PutObjectRetention)
-// If object exists on object store, if retention mode is "Compliance" or site wide WORM enabled -this method
+// If object exists on object store and site wide WORM enabled - this method
 // returns an error. For objects in "Governance" mode, overwrite is allowed if the retention date has expired.
+// For objects in "Compliance" mode, retention date cannot be shortened, and mode cannot be altered.
 func checkPutObjectRetentionAllowed(ctx context.Context, r *http.Request, bucket, object string, getObjectInfoFn GetObjectInfoFn, retentionPermErr APIErrorCode) (RetentionMode, RetentionDate, APIErrorCode) {
 	var mode RetentionMode
 	var retainDate RetentionDate
+
 	retention, isWORMBucket := isWORMEnabled(bucket)
 
 	retentionRequested := isObjectLockRequested(r.Header)
@@ -501,7 +570,11 @@ func checkPutObjectRetentionAllowed(ctx context.Context, r *http.Request, bucket
 		}
 		return rMode, rDate, ErrNone
 	}
-	if !retentionRequested && isWORMBucket && !retention.IsEmpty() {
+
+	if !retentionRequested && isWORMBucket {
+		if retention.IsEmpty() && (mode == Compliance || mode == Governance) {
+			return mode, retainDate, ErrObjectLocked
+		}
 		if retentionPermErr != ErrNone {
 			return mode, retainDate, retentionPermErr
 		}

cmd/web-handlers.go

@@ -710,7 +710,7 @@ next:
 					govBypassPerms = ErrNone
 				}
 			}
-			if _, err := checkGovernanceBypassAllowed(ctx, r, args.BucketName, objectName, getObjectInfo, govBypassPerms); err != ErrNone {
+			if _, err := enforceRetentionBypassForDelete(ctx, r, args.BucketName, objectName, getObjectInfo, govBypassPerms); err != ErrNone {
 				return toJSONError(ctx, errAccessDenied)
 			}
 			if err = deleteObject(ctx, objectAPI, web.CacheAPI(), args.BucketName, objectName, r); err != nil {

docs/retention/README.md

@@ -32,13 +32,18 @@ object locking and permissions required for object retention and governance bypa
 ### 3. Note
 
 - When global WORM is enabled by `MINIO_WORM` environment variable or `worm` field in configuration file supersedes bucket level WORM and `PUT object lock configuration` REST API is disabled.
-- global WORM and objects in `Compliance` mode can never be overwritten
+- In global WORM mode objects can never be overwritten
+- In `Compliance` mode, objects cannot be overwritten or deleted by anyone until retention period
+is expired. If user has requisite governance bypass permissions, an object's retention date can
+be extended in `Compliance` mode.
 - Currently `Governance` mode does not allow overwriting an existing object as versioning is not
-available in MinIO. To that extent `Governance` mode is similar to `Compliance`. However,
-if user has requisite `Governance` bypass permissions, an object in `Governance` mode can be overwritten.
+available in MinIO. However, if user has requisite `Governance` bypass permissions, an object in `Governance` mode can be overwritten.
 - Once object lock configuration is set to a bucket, new objects inherit the retention settings of the bucket object lock configuration (if set) or the retention headers set in the PUT request
 or set with PutObjectRetention API call
 
+- MINIO_NTP_SERVER environment variable can be set to remote NTP server endpoint if system time
+is not desired for setting retention dates.
+
 ## Explore Further
 
 - [Use `mc` with MinIO Server](https://docs.min.io/docs/minio-client-quickstart-guide)