Add missing validation for replication API conditions (#10114)

cmd/admin-bucket-handlers.go

@@ -135,6 +135,10 @@ func (a adminAPIHandlers) SetBucketReplicationTargetHandler(w http.ResponseWrite
 		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
 		return
 	}
+	if !globalIsErasure {
+		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL)
+		return
+	}
 	// Turn off replication if disk crawl is unavailable.
 	if env.Get(envDataUsageCrawlConf, config.EnableOn) == config.EnableOff {
 		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrBucketReplicationDisabledError), r.URL)
@@ -214,8 +218,7 @@ func (a adminAPIHandlers) GetBucketReplicationTargetsHandler(w http.ResponseWrit
 	if !target.Empty() {
 		var creds auth.Credentials
 		creds.AccessKey = target.Credentials.AccessKey
-		tgt = madmin.BucketReplicationTarget{Endpoint: target.Endpoint, TargetBucket: target.TargetBucket, Credentials: &creds}
-
+		tgt = madmin.BucketReplicationTarget{Endpoint: target.Endpoint, TargetBucket: target.TargetBucket, Credentials: &creds, Arn: target.Arn}
 	}
 	data, err := json.Marshal(tgt)
 	if err != nil {

cmd/api-errors.go

@@ -108,7 +108,7 @@ const (
 	ErrReplicationConfigurationNotFoundError
 	ErrReplicationDestinationNotFoundError
 	ErrReplicationTargetNotFoundError
-
+	ErrReplicationTargetNotVersionedError
 	ErrReplicationNeedsVersioningError
 	ErrReplicationBucketNeedsVersioningError
 	ErrBucketReplicationDisabledError
@@ -830,6 +830,11 @@ var errorCodes = errorCodeMap{
 		Description:    "The replication target does not exist",
 		HTTPStatusCode: http.StatusNotFound,
 	},
+	ErrReplicationTargetNotVersionedError: {
+		Code:           "ReplicationTargetNotVersionedError",
+		Description:    "The replication target does not have versioning enabled",
+		HTTPStatusCode: http.StatusNotFound,
+	},
 	ErrReplicationNeedsVersioningError: {
 		Code:           "InvalidRequest",
 		Description:    "Versioning must be 'Enabled' on the bucket to apply a replication configuration",
@@ -1876,6 +1881,8 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
 		apiErr = ErrReplicationDestinationNotFoundError
 	case BucketReplicationTargetNotFound:
 		apiErr = ErrReplicationTargetNotFoundError
+	case BucketReplicationTargetNotVersioned:
+		apiErr = ErrReplicationTargetNotVersionedError
 	case BucketQuotaExceeded:
 		apiErr = ErrAdminBucketQuotaExceeded
 	case *event.ErrInvalidEventName:

cmd/bucket-handlers.go

@@ -1036,7 +1036,10 @@ func (api objectAPIHandlers) PutBucketObjectLockConfigHandler(w http.ResponseWri
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL, guessIsBrowserReq(r))
 		return
 	}
-
+	if !globalIsErasure {
+		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL)
+		return
+	}
 	if s3Error := checkRequestAuthType(ctx, r, policy.PutBucketObjectLockConfigurationAction, bucket, ""); s3Error != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL, guessIsBrowserReq(r))
 		return
@@ -1238,6 +1241,10 @@ func (api objectAPIHandlers) PutBucketReplicationConfigHandler(w http.ResponseWr
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL, guessIsBrowserReq(r))
 		return
 	}
+	if !globalIsErasure {
+		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL)
+		return
+	}
 	// Turn off replication if disk crawl is unavailable.
 	if env.Get(envDataUsageCrawlConf, config.EnableOn) == config.EnableOff {
 		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrBucketReplicationDisabledError), r.URL)

cmd/bucket-replication.go

@@ -32,6 +32,7 @@ import (
 	xhttp "github.com/minio/minio/cmd/http"
 	"github.com/minio/minio/cmd/logger"
 	"github.com/minio/minio/pkg/bucket/replication"
+	"github.com/minio/minio/pkg/bucket/versioning"
 	"github.com/minio/minio/pkg/event"
 	iampolicy "github.com/minio/minio/pkg/iam/policy"
 	"github.com/minio/minio/pkg/madmin"
@@ -84,6 +85,10 @@ func (sys *BucketReplicationSys) SetTarget(ctx context.Context, bucket string, t
 	if !ok {
 		return BucketReplicationDestinationNotFound{Bucket: tgt.TargetBucket}
 	}
+	vcfg, err := clnt.GetBucketVersioning(ctx, tgt.TargetBucket)
+	if err != nil || vcfg.Status != string(versioning.Enabled) {
+		return BucketReplicationTargetNotVersioned{Bucket: tgt.TargetBucket}
+	}
 	sys.Lock()
 	sys.targetsMap[bucket] = clnt
 	sys.targetsARNMap[tgt.URL()] = tgt.Arn

cmd/object-api-errors.go

@@ -369,6 +369,13 @@ func (e BucketReplicationTargetNotFound) Error() string {
 	return "Replication target not found: " + e.Bucket
 }
 
+// BucketReplicationTargetNotVersioned replication target does not have versioning enabled.
+type BucketReplicationTargetNotVersioned GenericError
+
+func (e BucketReplicationTargetNotVersioned) Error() string {
+	return "Replication target does not have versioning enabled: " + e.Bucket
+}
+
 /// Bucket related errors.
 
 // BucketNameInvalid - bucketname provided is invalid.