Validate object lock when setting replication config. (#10200)

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-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