Bug 2228785: object: avoid creating same bucket for two different OBC

If bucket exists for Provision(), then check whether user in the OBC and owner of bucket are same. Signed-off-by: Jiffin Tony Thottan <[email protected]> (cherry picked from commit b39e813)
red-hat-storage · Sep 6, 2023 · 9229b96 · 9229b96
1 parent 236b032
commit 9229b96
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 7 deletions.
diff --git a/pkg/operator/ceph/object/bucket/provisioner.go b/pkg/operator/ceph/object/bucket/provisioner.go
@@ -104,7 +104,8 @@ func (p Provisioner) Provision(options *apibkt.BucketOptions) (*bktv1alpha1.Obje
 
 	// create the bucket
 	var bucketExists bool
-	bucketExists, err = p.bucketExists(p.bucketName)
+	var owner string
+	bucketExists, owner, err = p.bucketExists(p.bucketName)
 	if err != nil {
 		return nil, errors.Wrapf(err, "error creating bucket %q. failed to check if bucket already exists", p.bucketName)
 	}
@@ -116,6 +117,8 @@ func (p Provisioner) Provision(options *apibkt.BucketOptions) (*bktv1alpha1.Obje
 		if err != nil {
 			return nil, errors.Wrapf(err, "error creating bucket %q", p.bucketName)
 		}
+	} else if owner != options.UserID {
+		return nil, errors.Errorf("bucket %q already exists and is owned by %q for different OBC", p.bucketName, owner)
 	} else {
 		logger.Debugf("bucket %q already exists", p.bucketName)
 	}
@@ -149,7 +152,7 @@ func (p Provisioner) Grant(options *apibkt.BucketOptions) (*bktv1alpha1.ObjectBu
 
 	// check and make sure the bucket exists
 	logger.Infof("Checking for existing bucket %q", p.bucketName)
-	if exists, err := p.bucketExists(p.bucketName); !exists {
+	if exists, _, err := p.bucketExists(p.bucketName); !exists {
 		return nil, errors.Wrapf(err, "bucket %s does not exist", p.bucketName)
 	}
 

diff --git a/pkg/operator/ceph/object/bucket/rgw-handlers.go b/pkg/operator/ceph/object/bucket/rgw-handlers.go
@@ -5,15 +5,15 @@ import (
 	"github.com/pkg/errors"
 )
 
-func (p *Provisioner) bucketExists(name string) (bool, error) {
-	_, err := p.adminOpsClient.GetBucketInfo(p.clusterInfo.Context, admin.Bucket{Bucket: name})
+func (p *Provisioner) bucketExists(name string) (bool, string, error) {
+	bucket, err := p.adminOpsClient.GetBucketInfo(p.clusterInfo.Context, admin.Bucket{Bucket: name})
 	if err != nil {
 		if errors.Is(err, admin.ErrNoSuchBucket) {
-			return false, nil
+			return false, "", nil
 		}
-		return false, errors.Wrapf(err, "failed to get ceph bucket %q", name)
+		return false, "", errors.Wrapf(err, "failed to get ceph bucket %q", name)
 	}
-	return true, nil
+	return true, bucket.Owner, nil
 }
 
 // Create a Ceph user based on the passed-in name or a generated name. Return the