From c68910005b940b623f0d1516b707c8fb29b08123 Mon Sep 17 00:00:00 2001
From: Harshavardhana <harsha@minio.io>
Date: Sat, 15 Oct 2022 11:58:31 -0700
Subject: [PATCH] validate bucket before attempting batch replication (#15861)

---
 cmd/api-errors.go           |  6 ++++
 cmd/batch-handlers.go       | 60 ++++++++++++++++++++++++++++++++++++-
 cmd/erasure-server-pool.go  |  4 +++
 cmd/erasure-single-drive.go |  9 +++++-
 cmd/object-api-datatypes.go |  3 ++
 5 files changed, 80 insertions(+), 2 deletions(-)

diff --git a/cmd/api-errors.go b/cmd/api-errors.go
index a59b0a810..a99151e26 100644
--- a/cmd/api-errors.go
+++ b/cmd/api-errors.go
@@ -2228,6 +2228,12 @@ func toAPIError(ctx context.Context, err error) APIError {
 		// their internal error types. This code is only
 		// useful with gateway implementations.
 		switch e := err.(type) {
+		case batchReplicationJobError:
+			apiErr = APIError{
+				Code:           e.Code,
+				Description:    e.Description,
+				HTTPStatusCode: e.HTTPStatusCode,
+			}
 		case InvalidArgument:
 			apiErr = APIError{
 				Code:           "InvalidArgument",
diff --git a/cmd/batch-handlers.go b/cmd/batch-handlers.go
index 3dd1fa261..7e0fc575c 100644
--- a/cmd/batch-handlers.go
+++ b/cmd/batch-handlers.go
@@ -217,6 +217,8 @@ type BatchJobReplicateV1 struct {
 	Flags      BatchJobReplicateFlags  `yaml:"flags" json:"flags"`
 	Target     BatchJobReplicateTarget `yaml:"target" json:"target"`
 	Source     BatchJobReplicateSource `yaml:"source" json:"source"`
+
+	clnt *miniogo.Core `msg:"-"`
 }
 
 // BatchJobRequest this is an internal data structure not for external consumption.
@@ -640,6 +642,17 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 	return nil
 }
 
+//msgp:ignore batchReplicationJobError
+type batchReplicationJobError struct {
+	Code           string
+	Description    string
+	HTTPStatusCode int
+}
+
+func (e batchReplicationJobError) Error() string {
+	return e.Description
+}
+
 // Validate validates the job definition input
 func (r *BatchJobReplicateV1) Validate(ctx context.Context, o ObjectLayer) error {
 	if r == nil {
@@ -654,8 +667,15 @@ func (r *BatchJobReplicateV1) Validate(ctx context.Context, o ObjectLayer) error
 		return errInvalidArgument
 	}
 
-	_, err := o.GetBucketInfo(ctx, r.Source.Bucket, BucketOptions{})
+	info, err := o.GetBucketInfo(ctx, r.Source.Bucket, BucketOptions{})
 	if err != nil {
+		if isErrBucketNotFound(err) {
+			return batchReplicationJobError{
+				Code:           "NoSuchSourceBucket",
+				Description:    "The specified source bucket does not exist",
+				HTTPStatusCode: http.StatusNotFound,
+			}
+		}
 		return err
 	}
 
@@ -695,6 +715,44 @@ func (r *BatchJobReplicateV1) Validate(ctx context.Context, o ObjectLayer) error
 		return err
 	}
 
+	u, err := url.Parse(r.Target.Endpoint)
+	if err != nil {
+		return err
+	}
+
+	cred := r.Target.Creds
+
+	c, err := miniogo.NewCore(u.Host, &miniogo.Options{
+		Creds:     credentials.NewStaticV4(cred.AccessKey, cred.SecretKey, cred.SessionToken),
+		Secure:    u.Scheme == "https",
+		Transport: getRemoteInstanceTransport,
+	})
+	if err != nil {
+		return err
+	}
+
+	vcfg, err := c.GetBucketVersioning(ctx, r.Target.Bucket)
+	if err != nil {
+		if miniogo.ToErrorResponse(err).Code == "NoSuchBucket" {
+			return batchReplicationJobError{
+				Code:           "NoSuchTargetBucket",
+				Description:    "The specified target bucket does not exist",
+				HTTPStatusCode: http.StatusNotFound,
+			}
+		}
+		return err
+	}
+
+	if info.Versioning && !vcfg.Enabled() {
+		return batchReplicationJobError{
+			Code: "InvalidBucketState",
+			Description: fmt.Sprintf("The source '%s' has versioning enabled, target '%s' must have versioning enabled",
+				r.Source.Bucket, r.Target.Bucket),
+			HTTPStatusCode: http.StatusBadRequest,
+		}
+	}
+
+	r.clnt = c
 	return nil
 }
 
diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go
index 738955fb8..409493240 100644
--- a/cmd/erasure-server-pool.go
+++ b/cmd/erasure-server-pool.go
@@ -1588,6 +1588,8 @@ func (z *erasureServerPools) GetBucketInfo(ctx context.Context, bucket string, o
 		meta, err := globalBucketMetadataSys.Get(bucket)
 		if err == nil {
 			bucketInfo.Created = meta.Created
+			bucketInfo.Versioning = meta.LockEnabled || globalBucketVersioningSys.Enabled(bucket)
+			bucketInfo.ObjectLocking = meta.LockEnabled
 		}
 		return bucketInfo, nil
 	}
@@ -1602,6 +1604,8 @@ func (z *erasureServerPools) GetBucketInfo(ctx context.Context, bucket string, o
 		meta, err := globalBucketMetadataSys.Get(bucket)
 		if err == nil {
 			bucketInfo.Created = meta.Created
+			bucketInfo.Versioning = meta.LockEnabled || globalBucketVersioningSys.Enabled(bucket)
+			bucketInfo.ObjectLocking = meta.LockEnabled
 		}
 		return bucketInfo, nil
 	}
diff --git a/cmd/erasure-single-drive.go b/cmd/erasure-single-drive.go
index 647b42acf..9f525dcf5 100644
--- a/cmd/erasure-single-drive.go
+++ b/cmd/erasure-single-drive.go
@@ -344,7 +344,14 @@ func (es *erasureSingle) GetBucketInfo(ctx context.Context, bucket string, opts
 		}
 		return bi, toObjectErr(err, bucket)
 	}
-	return BucketInfo{Name: volInfo.Name, Created: volInfo.Created}, nil
+	bi = BucketInfo{Name: volInfo.Name, Created: volInfo.Created}
+	meta, err := globalBucketMetadataSys.Get(bucket)
+	if err == nil {
+		bi.Created = meta.Created
+		bi.Versioning = meta.LockEnabled || globalBucketVersioningSys.Enabled(bucket)
+		bi.ObjectLocking = meta.LockEnabled
+	}
+	return bi, nil
 }
 
 // DeleteBucket - deletes a bucket.
diff --git a/cmd/object-api-datatypes.go b/cmd/object-api-datatypes.go
index 94b9731f0..e585c407e 100644
--- a/cmd/object-api-datatypes.go
+++ b/cmd/object-api-datatypes.go
@@ -79,6 +79,9 @@ type BucketInfo struct {
 	// Date and time when the bucket was created.
 	Created time.Time
 	Deleted time.Time
+
+	// Bucket features enabled
+	Versioning, ObjectLocking bool
 }
 
 // ObjectInfo - represents object metadata.