Enforce replica delete permissions

Also check if object lock is enabled on
destination bucket while setting replication
configuration on a object lock enabled bucket.

cmd/api-errors.go

@@ -107,6 +107,7 @@ const (
 	ErrNoSuchWebsiteConfiguration
 	ErrReplicationConfigurationNotFoundError
 	ErrReplicationDestinationNotFoundError
+	ErrReplicationDestinationMissingLock
 	ErrReplicationTargetNotFoundError
 	ErrBucketRemoteIdenticalToSource
 	ErrBucketRemoteAlreadyExists
@@ -830,6 +831,11 @@ var errorCodes = errorCodeMap{
 		Description:    "The replication destination bucket does not exist",
 		HTTPStatusCode: http.StatusNotFound,
 	},
+	ErrReplicationDestinationMissingLock: {
+		Code:           "ReplicationDestinationMissingLockError",
+		Description:    "The replication destination bucket does not have object locking enabled",
+		HTTPStatusCode: http.StatusBadRequest,
+	},
 	ErrReplicationTargetNotFoundError: {
 		Code:           "XminioAdminReplicationTargetNotFoundError",
 		Description:    "The replication target does not exist",
@@ -1909,6 +1915,8 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
 		apiErr = ErrReplicationConfigurationNotFoundError
 	case BucketReplicationDestinationNotFound:
 		apiErr = ErrReplicationDestinationNotFoundError
+	case BucketReplicationDestinationMissingLock:
+		apiErr = ErrReplicationDestinationMissingLock
 	case BucketRemoteTargetNotFound:
 		apiErr = ErrReplicationTargetNotFoundError
 	case BucketRemoteAlreadyExists:

cmd/bucket-handlers.go

@@ -429,6 +429,20 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 					continue
 				}
 			}
+			if oi, err := getObjectInfoFn(ctx, bucket, object.ObjectName, ObjectOptions{}); err == nil {
+				if oi.ReplicationStatus == replication.Replica {
+					if apiErrCode := checkRequestAuthType(ctx, r, policy.ReplicateDeleteAction, bucket, object.ObjectName); apiErrCode != ErrNone {
+						apiErr := errorCodes.ToAPIErr(apiErrCode)
+						dErrs[index] = DeleteError{
+							Code:      apiErr.Code,
+							Message:   apiErr.Description,
+							Key:       object.ObjectName,
+							VersionID: object.VersionID,
+						}
+						continue
+					}
+				}
+			}
 		}
 
 		// Avoid duplicate objects, we use map to filter them out.

cmd/bucket-replication.go

@@ -56,6 +56,14 @@ func validateReplicationDestination(ctx context.Context, bucket string, rCfg *re
 	if found, _ := clnt.BucketExists(ctx, rCfg.GetDestination().Bucket); !found {
 		return false, BucketReplicationDestinationNotFound{Bucket: rCfg.GetDestination().Bucket}
 	}
+	if ret, err := globalBucketObjectLockSys.Get(bucket); err == nil {
+		if ret.LockEnabled {
+			lock, _, _, _, err := clnt.GetObjectLockConfig(ctx, rCfg.GetDestination().Bucket)
+			if err != nil || lock != "Enabled" {
+				return false, BucketReplicationDestinationMissingLock{Bucket: rCfg.GetDestination().Bucket}
+			}
+		}
+	}
 	// validate replication ARN against target endpoint
 	c, ok := globalBucketTargetSys.arnRemotesMap[rCfg.ReplicationArn]
 	if ok {

cmd/object-api-errors.go

@@ -362,6 +362,13 @@ func (e BucketReplicationDestinationNotFound) Error() string {
 	return "Destination bucket does not exist: " + e.Bucket
 }
 
+// BucketReplicationDestinationMissingLock bucket does not have object lock enabled.
+type BucketReplicationDestinationMissingLock GenericError
+
+func (e BucketReplicationDestinationMissingLock) Error() string {
+	return "Destination bucket does not have object lock enabled: " + e.Bucket
+}
+
 // BucketRemoteTargetNotFound remote target does not exist.
 type BucketRemoteTargetNotFound GenericError
 

cmd/object-handlers.go

@@ -2734,6 +2734,17 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 			}
 		}
 	}
+	if opts.VersionID != "" {
+		if oi, err := getObjectInfo(ctx, bucket, object, opts); err == nil {
+			if oi.ReplicationStatus == replication.Replica {
+				permErr := checkRequestAuthType(ctx, r, policy.ReplicateDeleteAction, bucket, object)
+				if permErr != ErrNone {
+					writeErrorResponse(ctx, w, errorCodes.ToAPIErr(permErr), r.URL, guessIsBrowserReq(r))
+					return
+				}
+			}
+		}
+	}
 
 	if apiErr == ErrNone {
 		// http://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectDELETE.html