Merge pull request #15159 from rook/mergify/bp/release-1.16/pr-15138

object: add bucketPolicy to obc (backport #15138)

PendingReleaseNotes.md

@@ -12,3 +12,4 @@
 - Enable periodic monitoring for CephBlockPoolRadosNamespaces mirroring (see [#14896](https://github.com/rook/rook/pull/14896)).
 - Allow migration of PVC based OSDs to enable or disable encryption (see [#14776](https://github.com/rook/rook/pull/14776)).
 - Support `rgw_enable_apis` option for CephObjectStore (see [#15064](https://github.com/rook/rook/pull/15064)).
+- ObjectBucketClaim management of s3 bucket policy via the `bucketPolicy` field (see [#15138](https://github.com/rook/rook/pull/15138)).

pkg/operator/ceph/object/bucket/provisioner.go

@@ -22,6 +22,7 @@ import (
 	"strings"
 
 	"github.com/aws/aws-sdk-go/aws/awserr"
+	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/ceph/go-ceph/rgw/admin"
 	"github.com/coreos/pkg/capnslog"
 	"github.com/google/go-cmp/cmp"
@@ -54,6 +55,7 @@ type Provisioner struct {
 	tlsCert              []byte
 	insecureTLS          bool
 	adminOpsClient       *admin.API
+	s3Agent              *object.S3Agent
 }
 
 type additionalConfigSpec struct {
@@ -61,6 +63,7 @@ type additionalConfigSpec struct {
 	maxSize          *int64
 	bucketMaxObjects *int64
 	bucketMaxSize    *int64
+	bucketPolicy     *string
 }
 
 var _ apibkt.Provisioner = &Provisioner{}
@@ -95,12 +98,7 @@ func (p Provisioner) Provision(options *apibkt.BucketOptions) (*bktv1alpha1.Obje
 		return nil, errors.Wrap(err, "Provision: can't create ceph user")
 	}
 
-	var s3svc *object.S3Agent
-	if p.insecureTLS {
-		s3svc, err = object.NewInsecureS3Agent(p.accessKeyID, p.secretAccessKey, p.getObjectStoreEndpoint(), logger.LevelAt(capnslog.DEBUG))
-	} else {
-		s3svc, err = object.NewS3Agent(p.accessKeyID, p.secretAccessKey, p.getObjectStoreEndpoint(), logger.LevelAt(capnslog.DEBUG), p.tlsCert)
-	}
+	err = p.setS3Agent()
 	if err != nil {
 		return nil, err
 	}
@@ -116,7 +114,7 @@ func (p Provisioner) Provision(options *apibkt.BucketOptions) (*bktv1alpha1.Obje
 		// if bucket already exists, this returns error: TooManyBuckets because we set the quota
 		// below. If it already exists, assume we are good to go
 		logger.Debugf("creating bucket %q", p.bucketName)
-		err = s3svc.CreateBucket(p.bucketName)
+		err = p.s3Agent.CreateBucket(p.bucketName)
 		if err != nil {
 			return nil, errors.Wrapf(err, "error creating bucket %q", p.bucketName)
 		}
@@ -133,7 +131,12 @@ func (p Provisioner) Provision(options *apibkt.BucketOptions) (*bktv1alpha1.Obje
 	}
 	logger.Infof("set user %q bucket max to %d", p.cephUserName, singleBucketQuota)
 
-	err = p.setAdditionalSettings(options)
+	additionalConfig, err := additionalConfigSpecFromMap(options.ObjectBucketClaim.Spec.AdditionalConfig)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to process additionalConfig")
+	}
+
+	err = p.setAdditionalSettings(additionalConfig)
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to set additional settings for OBC %q in NS %q associated with CephObjectStore %q in NS %q", options.ObjectBucketClaim.Name, options.ObjectBucketClaim.Namespace, p.objectStoreName, p.clusterInfo.Namespace)
 	}
@@ -172,29 +175,31 @@ func (p Provisioner) Grant(options *apibkt.BucketOptions) (*bktv1alpha1.ObjectBu
 		return nil, err
 	}
 
-	// get the bucket's owner via the bucket metadata
-	stats, err := p.adminOpsClient.GetBucketInfo(p.clusterInfo.Context, admin.Bucket{Bucket: p.bucketName})
-	if err != nil {
-		return nil, errors.Wrapf(err, "failed to get bucket %q stats", p.bucketName)
-	}
-
-	objectUser, err := p.adminOpsClient.GetUser(p.clusterInfo.Context, admin.User{ID: stats.Owner})
-	if err != nil {
-		return nil, errors.Wrapf(err, "failed to get user %q", stats.Owner)
-	}
-
-	var s3svc *object.S3Agent
-	if p.insecureTLS {
-		s3svc, err = object.NewInsecureS3Agent(objectUser.Keys[0].AccessKey, objectUser.Keys[0].SecretKey, p.getObjectStoreEndpoint(), logger.LevelAt(capnslog.DEBUG))
-	} else {
-		s3svc, err = object.NewS3Agent(objectUser.Keys[0].AccessKey, objectUser.Keys[0].SecretKey, p.getObjectStoreEndpoint(), logger.LevelAt(capnslog.DEBUG), p.tlsCert)
-	}
+	err = p.setS3Agent()
 	if err != nil {
 		return nil, err
 	}
 
+	additionalConfig, err := additionalConfigSpecFromMap(options.ObjectBucketClaim.Spec.AdditionalConfig)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to process additionalConfig")
+	}
+
+	// setting quota limit if it is enabled
+	err = p.setAdditionalSettings(additionalConfig)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to set additional settings for OBC %q in NS %q associated with CephObjectStore %q in NS %q", options.ObjectBucketClaim.Name, options.ObjectBucketClaim.Namespace, p.objectStoreName, p.clusterInfo.Namespace)
+	}
+
+	if additionalConfig.bucketPolicy != nil {
+		// if the user is managing the bucket policy, there's nothing else to do
+		return p.composeObjectBucket(), nil
+	}
+
+	// generate the bucket policy if it isn't managed by the user
+
 	// if the policy does not exist, we'll create a new and append the statement to it
-	policy, err := s3svc.GetBucketPolicy(p.bucketName)
+	policy, err := p.s3Agent.GetBucketPolicy(p.bucketName)
 	if err != nil {
 		if aerr, ok := err.(awserr.Error); ok {
 			if aerr.Code() != "NoSuchBucketPolicy" {
@@ -215,19 +220,13 @@ func (p Provisioner) Grant(options *apibkt.BucketOptions) (*bktv1alpha1.ObjectBu
 	} else {
 		policy = policy.ModifyBucketPolicy(*statement)
 	}
-	out, err := s3svc.PutBucketPolicy(p.bucketName, *policy)
+	out, err := p.s3Agent.PutBucketPolicy(p.bucketName, *policy)
 
 	logger.Infof("PutBucketPolicy output: %v", out)
 	if err != nil {
 		return nil, err
 	}
 
-	// setting quota limit if it is enabled
-	err = p.setAdditionalSettings(options)
-	if err != nil {
-		return nil, errors.Wrapf(err, "failed to set additional settings for OBC %q in NS %q associated with CephObjectStore %q in NS %q", options.ObjectBucketClaim.Name, options.ObjectBucketClaim.Namespace, p.objectStoreName, p.clusterInfo.Namespace)
-	}
-
 	// returned ob with connection info
 	return p.composeObjectBucket(), nil
 }
@@ -280,19 +279,17 @@ func (p Provisioner) Revoke(ob *bktv1alpha1.ObjectBucket) error {
 			return err
 		}
 
-		var s3svc *object.S3Agent
-		if p.insecureTLS {
-			s3svc, err = object.NewInsecureS3Agent(user.Keys[0].AccessKey, user.Keys[0].SecretKey, p.getObjectStoreEndpoint(), logger.LevelAt(capnslog.DEBUG))
-		} else {
-			s3svc, err = object.NewS3Agent(user.Keys[0].AccessKey, user.Keys[0].SecretKey, p.getObjectStoreEndpoint(), logger.LevelAt(capnslog.DEBUG), p.tlsCert)
-		}
+		p.accessKeyID = user.Keys[0].AccessKey
+		p.secretAccessKey = user.Keys[0].SecretKey
+
+		err = p.setS3Agent()
 		if err != nil {
 			return err
 		}
 
 		// Ignore cases where there is no bucket policy. This may have occurred if an error ended a Grant()
 		// call before the policy was attached to the bucket
-		policy, err := s3svc.GetBucketPolicy(p.bucketName)
+		policy, err := p.s3Agent.GetBucketPolicy(p.bucketName)
 		if err != nil {
 			if aerr, ok := err.(awserr.Error); ok && aerr.Code() == "NoSuchBucketPolicy" {
 				policy = nil
@@ -317,7 +314,7 @@ func (p Provisioner) Revoke(ob *bktv1alpha1.ObjectBucket) error {
 			} else {
 				policy = policy.ModifyBucketPolicy(*statement)
 			}
-			out, err := s3svc.PutBucketPolicy(p.bucketName, *policy)
+			out, err := p.s3Agent.PutBucketPolicy(p.bucketName, *policy)
 			logger.Infof("PutBucketPolicy output: %v", out)
 			if err != nil {
 				return errors.Wrap(err, "failed to update policy")
@@ -329,7 +326,7 @@ func (p Provisioner) Revoke(ob *bktv1alpha1.ObjectBucket) error {
 		// drop policy if present
 		if policy != nil {
 			policy = policy.DropPolicyStatements(p.cephUserName)
-			_, err := s3svc.PutBucketPolicy(p.bucketName, *policy)
+			_, err := p.s3Agent.PutBucketPolicy(p.bucketName, *policy)
 			if err != nil {
 				return err
 			}
@@ -574,13 +571,8 @@ func (p *Provisioner) populateDomainAndPort(sc *storagev1.StorageClass) error {
 }
 
 // Check for additional options mentioned in OBC and set them accordingly
-func (p *Provisioner) setAdditionalSettings(options *apibkt.BucketOptions) error {
-	additionalConfig, err := additionalConfigSpecFromMap(options.ObjectBucketClaim.Spec.AdditionalConfig)
-	if err != nil {
-		return errors.Wrap(err, "failed to process additionalConfig")
-	}
-
-	err = p.setUserQuota(additionalConfig)
+func (p *Provisioner) setAdditionalSettings(additionalConfig *additionalConfigSpec) error {
+	err := p.setUserQuota(additionalConfig)
 	if err != nil {
 		return errors.Wrap(err, "failed to set user quota")
 	}
@@ -590,6 +582,11 @@ func (p *Provisioner) setAdditionalSettings(options *apibkt.BucketOptions) error
 		return errors.Wrap(err, "failed to set bucket quota")
 	}
 
+	err = p.setBucketPolicy(additionalConfig)
+	if err != nil {
+		return errors.Wrap(err, "failed to set bucket policy")
+	}
+
 	return nil
 }
 
@@ -697,6 +694,53 @@ func (p *Provisioner) setBucketQuota(additionalConfig *additionalConfigSpec) err
 	return nil
 }
 
+func (p *Provisioner) setBucketPolicy(additionalConfig *additionalConfigSpec) error {
+	svc := p.s3Agent.Client
+	var livePolicy *string
+
+	policyResp, err := svc.GetBucketPolicy(&s3.GetBucketPolicyInput{
+		Bucket: &p.bucketName,
+	})
+	if err != nil {
+		// when no policy is set, an err with code "NoSuchBucketPolicy" is returned
+		if aerr, ok := err.(awserr.Error); ok {
+			if aerr.Code() != "NoSuchBucketPolicy" {
+				return errors.Wrapf(err, "failed to fetch policy for bucket %q", p.bucketName)
+			}
+		}
+	} else {
+		livePolicy = policyResp.Policy
+	}
+
+	diff := cmp.Diff(&livePolicy, additionalConfig.bucketPolicy)
+	if diff == "" {
+		// policy is in sync
+		return nil
+	}
+
+	logger.Debugf("Policy for bucket %q has changed. diff:%s", p.bucketName, diff)
+	if additionalConfig.bucketPolicy == nil {
+		// if policy is out of sync and the new policy is nil, we should delete the live policy
+		_, err = svc.DeleteBucketPolicy(&s3.DeleteBucketPolicyInput{
+			Bucket: &p.bucketName,
+		})
+		if err != nil {
+			return errors.Wrapf(err, "failed to delete policy for bucket %q", p.bucketName)
+		}
+	} else {
+		// set the new policy
+		_, err = svc.PutBucketPolicy(&s3.PutBucketPolicyInput{
+			Bucket: &p.bucketName,
+			Policy: additionalConfig.bucketPolicy,
+		})
+		if err != nil {
+			return errors.Wrapf(err, "failed to set policy for bucket %q", p.bucketName)
+		}
+	}
+
+	return nil
+}
+
 func (p *Provisioner) setTlsCaCert() error {
 	objStore, err := p.getObjectStore()
 	if err != nil {
@@ -754,3 +798,9 @@ func (p *Provisioner) setAdminOpsAPIClient() error {
 
 	return nil
 }
+
+func (p *Provisioner) setS3Agent() error {
+	var err error
+	p.s3Agent, err = object.NewS3Agent(p.accessKeyID, p.secretAccessKey, p.getObjectStoreEndpoint(), logger.LevelAt(capnslog.DEBUG), p.tlsCert, p.insecureTLS, nil)
+	return err
+}

pkg/operator/ceph/object/bucket/provisioner_test.go

@@ -25,9 +25,8 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/aws/aws-sdk-go/aws"
 	"github.com/ceph/go-ceph/rgw/admin"
-	"github.com/kube-object-storage/lib-bucket-provisioner/pkg/apis/objectbucket.io/v1alpha1"
-	apibkt "github.com/kube-object-storage/lib-bucket-provisioner/pkg/provisioner/api"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	rookclient "github.com/rook/rook/pkg/client/clientset/versioned/fake"
 	"github.com/rook/rook/pkg/clusterd"
@@ -146,7 +145,7 @@ func TestQuanityToInt64(t *testing.T) {
 	}
 }
 
-func TestProvisioner_setAdditionalSettings(t *testing.T) {
+func TestProvisioner_setUserQuota(t *testing.T) {
 	newProvisioner := func(t *testing.T, getResult *map[string]string, getSeen, putSeen *map[string][]string) *Provisioner {
 		mockClient := &object.MockClient{
 			MockDo: func(req *http.Request) (*http.Response, error) {
@@ -196,9 +195,194 @@ func TestProvisioner_setAdditionalSettings(t *testing.T) {
 		return p
 	}
 
-	t.Run("user and bucket quota should remain disabled", func(t *testing.T) {
+	t.Run("user quota should remain disabled", func(t *testing.T) {
+		getResult := map[string]string{
+			userPath: `{"enabled":false,"check_on_raw":false,"max_size":-1024,"max_size_kb":0,"max_objects":-1}`,
+		}
+		getSeen := map[string][]string{}
+		putSeen := map[string][]string{}
+
+		p := newProvisioner(t, &getResult, &getSeen, &putSeen)
+		p.setBucketName("bob")
+
+		err := p.setUserQuota(&additionalConfigSpec{})
+		assert.NoError(t, err)
+
+		assert.Len(t, getSeen[userPath], 1)
+		assert.Equal(t, numberOfCallsWithValue("uid=bob", getSeen[userPath]), 1)
+
+		assert.Len(t, putSeen[userPath], 0) // user quota should not be touched
+	})
+
+	t.Run("user quota should be disabled", func(t *testing.T) {
+		getResult := map[string]string{
+			userPath: `{"enabled":true,"check_on_raw":false,"max_size":-1024,"max_size_kb":0,"max_objects":2}`,
+		}
+		getSeen := map[string][]string{}
+		putSeen := map[string][]string{}
+
+		p := newProvisioner(t, &getResult, &getSeen, &putSeen)
+		p.setBucketName("bob")
+
+		err := p.setUserQuota(&additionalConfigSpec{})
+		assert.NoError(t, err)
+
+		assert.Len(t, getSeen[userPath], 1)
+		assert.Equal(t, numberOfCallsWithValue("uid=bob", getSeen[userPath]), 1)
+
+		assert.Len(t, putSeen[userPath], 1)
+		assert.Equal(t, 1, numberOfCallsWithValue("enabled=false", putSeen[userPath]))
+	})
+
+	t.Run("user maxSize quota should be enabled", func(t *testing.T) {
+		getResult := map[string]string{
+			userPath: `{"enabled":false,"check_on_raw":false,"max_size":-1024,"max_size_kb":0,"max_objects":-1}`,
+		}
+		getSeen := map[string][]string{}
+		putSeen := map[string][]string{}
+
+		p := newProvisioner(t, &getResult, &getSeen, &putSeen)
+		p.setBucketName("bob")
+
+		err := p.setUserQuota(&additionalConfigSpec{
+			maxSize: aws.Int64(2),
+		})
+		assert.NoError(t, err)
+
+		assert.Len(t, getSeen[userPath], 1)
+		assert.Equal(t, numberOfCallsWithValue("uid=bob", getSeen[userPath]), 1)
+
+		assert.Len(t, putSeen[userPath], 1)
+		assert.Equal(t, numberOfCallsWithValue("enabled=true", putSeen[userPath]), 1)
+		assert.Equal(t, numberOfCallsWithValue("max-size=2", putSeen[userPath]), 1)
+	})
+
+	t.Run("user maxObjects quota should be enabled", func(t *testing.T) {
+		getResult := map[string]string{
+			userPath: `{"enabled":false,"check_on_raw":false,"max_size":-1024,"max_size_kb":0,"max_objects":-1}`,
+		}
+		getSeen := map[string][]string{}
+		putSeen := map[string][]string{}
+
+		p := newProvisioner(t, &getResult, &getSeen, &putSeen)
+		p.setBucketName("bob")
+
+		err := p.setUserQuota(&additionalConfigSpec{
+			maxObjects: aws.Int64(2),
+		})
+		assert.NoError(t, err)
+
+		assert.Len(t, getSeen[userPath], 1)
+		assert.Equal(t, numberOfCallsWithValue("uid=bob", getSeen[userPath]), 1)
+
+		assert.Len(t, putSeen[userPath], 1)
+		assert.Equal(t, numberOfCallsWithValue("enabled=true", putSeen[userPath]), 1)
+		assert.Equal(t, numberOfCallsWithValue("max-objects=2", putSeen[userPath]), 1)
+	})
+
+	t.Run("user maxObjects and maxSize quotas should be enabled", func(t *testing.T) {
+		getResult := map[string]string{
+			userPath: `{"enabled":false,"check_on_raw":false,"max_size":-1024,"max_size_kb":0,"max_objects":-1}`,
+		}
+		getSeen := map[string][]string{}
+		putSeen := map[string][]string{}
+
+		p := newProvisioner(t, &getResult, &getSeen, &putSeen)
+		p.setBucketName("bob")
+
+		err := p.setUserQuota(&additionalConfigSpec{
+			maxObjects: aws.Int64(2),
+			maxSize:    aws.Int64(3),
+		})
+		assert.NoError(t, err)
+
+		assert.Len(t, getSeen[userPath], 1)
+		assert.Equal(t, numberOfCallsWithValue("uid=bob", getSeen[userPath]), 1)
+
+		assert.Len(t, putSeen[userPath], 1)
+		assert.Equal(t, numberOfCallsWithValue("enabled=true", putSeen[userPath]), 1)
+		assert.Equal(t, numberOfCallsWithValue("max-objects=2", putSeen[userPath]), 1)
+		assert.Equal(t, numberOfCallsWithValue("max-size=3", putSeen[userPath]), 1)
+	})
+
+	t.Run("user quotas are enabled and need updated enabled", func(t *testing.T) {
+		getResult := map[string]string{
+			userPath: `{"enabled":true,"check_on_raw":false,"max_size":1,"max_size_kb":0,"max_objects":1}`,
+		}
+		getSeen := map[string][]string{}
+		putSeen := map[string][]string{}
+
+		p := newProvisioner(t, &getResult, &getSeen, &putSeen)
+		p.setBucketName("bob")
+
+		err := p.setUserQuota(&additionalConfigSpec{
+			maxObjects: aws.Int64(12),
+			maxSize:    aws.Int64(13),
+		})
+		assert.NoError(t, err)
+
+		assert.Len(t, getSeen[userPath], 1)
+		assert.Equal(t, numberOfCallsWithValue("uid=bob", getSeen[userPath]), 1)
+
+		assert.Len(t, putSeen[userPath], 1)
+		assert.Equal(t, numberOfCallsWithValue("enabled=true", putSeen[userPath]), 1)
+		assert.Equal(t, numberOfCallsWithValue("max-objects=12", putSeen[userPath]), 1)
+		assert.Equal(t, numberOfCallsWithValue("max-size=13", putSeen[userPath]), 1)
+	})
+}
+
+func TestProvisioner_setBucketQuota(t *testing.T) {
+	newProvisioner := func(t *testing.T, getResult *map[string]string, getSeen, putSeen *map[string][]string) *Provisioner {
+		mockClient := &object.MockClient{
+			MockDo: func(req *http.Request) (*http.Response, error) {
+				path := req.URL.Path
+
+				// t.Logf("HTTP req: %#v", req)
+				t.Logf("HTTP %s: %s %s", req.Method, path, req.URL.RawQuery)
+
+				statusCode := 200
+				if _, ok := (*getResult)[path]; !ok {
+					// path not configured
+					statusCode = 500
+				}
+				responseBody := []byte(`[]`)
+
+				switch method := req.Method; method {
+				case http.MethodGet:
+					(*getSeen)[path] = append((*getSeen)[path], req.URL.RawQuery)
+					responseBody = []byte((*getResult)[path])
+				case http.MethodPut:
+					if (*putSeen)[path] == nil {
+						(*putSeen)[path] = []string{}
+					}
+					(*putSeen)[path] = append((*putSeen)[path], req.URL.RawQuery)
+				default:
+					panic(fmt.Sprintf("unexpected request: %q. method %q. path %q", req.URL.RawQuery, req.Method, path))
+				}
+
+				return &http.Response{
+					StatusCode: statusCode,
+					Body:       io.NopCloser(bytes.NewReader(responseBody)),
+				}, nil
+			},
+		}
+
+		adminClient, err := admin.New("rgw.test", "accesskey", "secretkey", mockClient)
+		assert.NoError(t, err)
+
+		p := &Provisioner{
+			clusterInfo: &client.ClusterInfo{
+				Context: context.Background(),
+			},
+			cephUserName:   "bob",
+			adminOpsClient: adminClient,
+		}
+
+		return p
+	}
+
+	t.Run("bucket quota should remain disabled", func(t *testing.T) {
 		getResult := map[string]string{
-			userPath:   `{"enabled":false,"check_on_raw":false,"max_size":-1024,"max_size_kb":0,"max_objects":-1}`,
 			bucketPath: `{"bucket_quota":{"enabled":false,"check_on_raw":false,"max_size":-1024,"max_size_kb":0,"max_objects":-1}}`,
 		}
 		getSeen := map[string][]string{}
@@ -207,27 +391,17 @@ func TestProvisioner_setAdditionalSettings(t *testing.T) {
 		p := newProvisioner(t, &getResult, &getSeen, &putSeen)
 		p.setBucketName("bob")
 
-		err := p.setAdditionalSettings(&apibkt.BucketOptions{
-			ObjectBucketClaim: &v1alpha1.ObjectBucketClaim{
-				Spec: v1alpha1.ObjectBucketClaimSpec{
-					AdditionalConfig: map[string]string{},
-				},
-			},
-		})
+		err := p.setBucketQuota(&additionalConfigSpec{})
 		assert.NoError(t, err)
 
-		assert.Len(t, getSeen[userPath], 1)
-		assert.Equal(t, numberOfCallsWithValue("uid=bob", getSeen[userPath]), 1)
 		assert.Len(t, getSeen[bucketPath], 1)
 		assert.Equal(t, numberOfCallsWithValue("bucket=bob", getSeen[bucketPath]), 1)
 
-		assert.Len(t, putSeen[userPath], 0)   // user quota should not be touched
 		assert.Len(t, putSeen[bucketPath], 0) // bucket quota should not be touched
 	})
 
-	t.Run("user and bucket quota should be disabled", func(t *testing.T) {
+	t.Run("bucket quota should be disabled", func(t *testing.T) {
 		getResult := map[string]string{
-			userPath:   `{"enabled":true,"check_on_raw":false,"max_size":-1024,"max_size_kb":0,"max_objects":2}`,
 			bucketPath: `{"owner": "bob","bucket_quota":{"enabled":true,"check_on_raw":false,"max_size":-1024,"max_size_kb":0,"max_objects":3}}`,
 		}
 		getSeen := map[string][]string{}
@@ -236,165 +410,18 @@ func TestProvisioner_setAdditionalSettings(t *testing.T) {
 		p := newProvisioner(t, &getResult, &getSeen, &putSeen)
 		p.setBucketName("bob")
 
-		err := p.setAdditionalSettings(&apibkt.BucketOptions{
-			ObjectBucketClaim: &v1alpha1.ObjectBucketClaim{
-				Spec: v1alpha1.ObjectBucketClaimSpec{
-					AdditionalConfig: map[string]string{},
-				},
-			},
-		})
+		err := p.setBucketQuota(&additionalConfigSpec{})
 		assert.NoError(t, err)
 
-		assert.Len(t, getSeen[userPath], 1)
-		assert.Equal(t, numberOfCallsWithValue("uid=bob", getSeen[userPath]), 1)
 		assert.Len(t, getSeen[bucketPath], 1)
 		assert.Equal(t, numberOfCallsWithValue("bucket=bob", getSeen[bucketPath]), 1)
 
-		assert.Len(t, putSeen[userPath], 1)
-		assert.Equal(t, 1, numberOfCallsWithValue("enabled=false", putSeen[userPath]))
 		assert.Len(t, putSeen[bucketPath], 1)
 		assert.Equal(t, 1, numberOfCallsWithValue("enabled=false", putSeen[bucketPath]))
 	})
 
-	t.Run("user maxSize quota should be enabled", func(t *testing.T) {
-		getResult := map[string]string{
-			userPath:   `{"enabled":false,"check_on_raw":false,"max_size":-1024,"max_size_kb":0,"max_objects":-1}`,
-			bucketPath: `{"bucket_quota":{"enabled":false,"check_on_raw":false,"max_size":-1,"max_size_kb":0,"max_objects":-1}}`,
-		}
-		getSeen := map[string][]string{}
-		putSeen := map[string][]string{}
-
-		p := newProvisioner(t, &getResult, &getSeen, &putSeen)
-		p.setBucketName("bob")
-
-		err := p.setAdditionalSettings(&apibkt.BucketOptions{
-			ObjectBucketClaim: &v1alpha1.ObjectBucketClaim{
-				Spec: v1alpha1.ObjectBucketClaimSpec{
-					AdditionalConfig: map[string]string{
-						"maxSize": "2",
-					},
-				},
-			},
-		})
-		assert.NoError(t, err)
-
-		assert.Len(t, getSeen[userPath], 1)
-		assert.Equal(t, numberOfCallsWithValue("uid=bob", getSeen[userPath]), 1)
-		assert.Len(t, getSeen[bucketPath], 1)
-		assert.Equal(t, numberOfCallsWithValue("bucket=bob", getSeen[bucketPath]), 1)
-
-		assert.Len(t, putSeen[userPath], 1)
-		assert.Equal(t, numberOfCallsWithValue("enabled=true", putSeen[userPath]), 1)
-		assert.Equal(t, numberOfCallsWithValue("max-size=2", putSeen[userPath]), 1)
-		assert.Len(t, putSeen[bucketPath], 0) // bucket quota should not be touched
-	})
-
-	t.Run("user maxObjects quota should be enabled", func(t *testing.T) {
-		getResult := map[string]string{
-			userPath:   `{"enabled":false,"check_on_raw":false,"max_size":-1024,"max_size_kb":0,"max_objects":-1}`,
-			bucketPath: `{"bucket_quota":{"enabled":false,"check_on_raw":false,"max_size":-1,"max_size_kb":0,"max_objects":-1}}`,
-		}
-		getSeen := map[string][]string{}
-		putSeen := map[string][]string{}
-
-		p := newProvisioner(t, &getResult, &getSeen, &putSeen)
-		p.setBucketName("bob")
-
-		err := p.setAdditionalSettings(&apibkt.BucketOptions{
-			ObjectBucketClaim: &v1alpha1.ObjectBucketClaim{
-				Spec: v1alpha1.ObjectBucketClaimSpec{
-					AdditionalConfig: map[string]string{
-						"maxObjects": "2",
-					},
-				},
-			},
-		})
-		assert.NoError(t, err)
-
-		assert.Len(t, getSeen[userPath], 1)
-		assert.Equal(t, numberOfCallsWithValue("uid=bob", getSeen[userPath]), 1)
-		assert.Len(t, getSeen[bucketPath], 1)
-		assert.Equal(t, numberOfCallsWithValue("bucket=bob", getSeen[bucketPath]), 1)
-
-		assert.Len(t, putSeen[userPath], 1)
-		assert.Equal(t, numberOfCallsWithValue("enabled=true", putSeen[userPath]), 1)
-		assert.Equal(t, numberOfCallsWithValue("max-objects=2", putSeen[userPath]), 1)
-		assert.Len(t, putSeen[bucketPath], 0) // bucket quota should not be touched
-	})
-
-	t.Run("user maxObjects and maxSize quotas should be enabled", func(t *testing.T) {
-		getResult := map[string]string{
-			userPath:   `{"enabled":false,"check_on_raw":false,"max_size":-1024,"max_size_kb":0,"max_objects":-1}`,
-			bucketPath: `{"bucket_quota":{"enabled":false,"check_on_raw":false,"max_size":-1,"max_size_kb":0,"max_objects":-1}}`,
-		}
-		getSeen := map[string][]string{}
-		putSeen := map[string][]string{}
-
-		p := newProvisioner(t, &getResult, &getSeen, &putSeen)
-		p.setBucketName("bob")
-
-		err := p.setAdditionalSettings(&apibkt.BucketOptions{
-			ObjectBucketClaim: &v1alpha1.ObjectBucketClaim{
-				Spec: v1alpha1.ObjectBucketClaimSpec{
-					AdditionalConfig: map[string]string{
-						"maxObjects": "2",
-						"maxSize":    "3",
-					},
-				},
-			},
-		})
-		assert.NoError(t, err)
-
-		assert.Len(t, getSeen[userPath], 1)
-		assert.Equal(t, numberOfCallsWithValue("uid=bob", getSeen[userPath]), 1)
-		assert.Len(t, getSeen[bucketPath], 1)
-		assert.Equal(t, numberOfCallsWithValue("bucket=bob", getSeen[bucketPath]), 1)
-
-		assert.Len(t, putSeen[userPath], 1)
-		assert.Equal(t, numberOfCallsWithValue("enabled=true", putSeen[userPath]), 1)
-		assert.Equal(t, numberOfCallsWithValue("max-objects=2", putSeen[userPath]), 1)
-		assert.Equal(t, numberOfCallsWithValue("max-size=3", putSeen[userPath]), 1)
-		assert.Len(t, putSeen[bucketPath], 0) // bucket quota should not be touched
-	})
-
-	t.Run("user quotas are enabled and need updated enabled", func(t *testing.T) {
-		getResult := map[string]string{
-			userPath:   `{"enabled":true,"check_on_raw":false,"max_size":1,"max_size_kb":0,"max_objects":1}`,
-			bucketPath: `{"bucket_quota":{"enabled":false,"check_on_raw":false,"max_size":-1,"max_size_kb":0,"max_objects":-1}}`,
-		}
-		getSeen := map[string][]string{}
-		putSeen := map[string][]string{}
-
-		p := newProvisioner(t, &getResult, &getSeen, &putSeen)
-		p.setBucketName("bob")
-
-		err := p.setAdditionalSettings(&apibkt.BucketOptions{
-			ObjectBucketClaim: &v1alpha1.ObjectBucketClaim{
-				Spec: v1alpha1.ObjectBucketClaimSpec{
-					AdditionalConfig: map[string]string{
-						"maxObjects": "12",
-						"maxSize":    "13",
-					},
-				},
-			},
-		})
-		assert.NoError(t, err)
-
-		assert.Len(t, getSeen[userPath], 1)
-		assert.Equal(t, numberOfCallsWithValue("uid=bob", getSeen[userPath]), 1)
-		assert.Len(t, getSeen[bucketPath], 1)
-		assert.Equal(t, numberOfCallsWithValue("bucket=bob", getSeen[bucketPath]), 1)
-
-		assert.Len(t, putSeen[userPath], 1)
-		assert.Equal(t, numberOfCallsWithValue("enabled=true", putSeen[userPath]), 1)
-		assert.Equal(t, numberOfCallsWithValue("max-objects=12", putSeen[userPath]), 1)
-		assert.Equal(t, numberOfCallsWithValue("max-size=13", putSeen[userPath]), 1)
-		assert.Len(t, putSeen[bucketPath], 0) // bucket quota should not be touched
-	})
-
 	t.Run("bucket maxObjects quota should be enabled", func(t *testing.T) {
 		getResult := map[string]string{
-			userPath:   `{"enabled":false,"check_on_raw":false,"max_size":-1024,"max_size_kb":0,"max_objects":-1}`,
 			bucketPath: `{"bucket_quota":{"enabled":false,"check_on_raw":false,"max_size":-1,"max_size_kb":0,"max_objects":-1}}`,
 		}
 		getSeen := map[string][]string{}
@@ -403,23 +430,14 @@ func TestProvisioner_setAdditionalSettings(t *testing.T) {
 		p := newProvisioner(t, &getResult, &getSeen, &putSeen)
 		p.setBucketName("bob")
 
-		err := p.setAdditionalSettings(&apibkt.BucketOptions{
-			ObjectBucketClaim: &v1alpha1.ObjectBucketClaim{
-				Spec: v1alpha1.ObjectBucketClaimSpec{
-					AdditionalConfig: map[string]string{
-						"bucketMaxObjects": "4",
-					},
-				},
-			},
+		err := p.setBucketQuota(&additionalConfigSpec{
+			bucketMaxObjects: aws.Int64(4),
 		})
 		assert.NoError(t, err)
 
-		assert.Len(t, getSeen[userPath], 1)
-		assert.Equal(t, numberOfCallsWithValue("uid=bob", getSeen[userPath]), 1)
 		assert.Len(t, getSeen[bucketPath], 1)
 		assert.Equal(t, numberOfCallsWithValue("bucket=bob", getSeen[bucketPath]), 1)
 
-		assert.Len(t, putSeen[userPath], 0) // user quota should not be touched
 		assert.Len(t, putSeen[bucketPath], 1)
 		assert.Equal(t, numberOfCallsWithValue("enabled=true", putSeen[bucketPath]), 1)
 		assert.Equal(t, numberOfCallsWithValue("max-objects=4", putSeen[bucketPath]), 1)
@@ -427,7 +445,6 @@ func TestProvisioner_setAdditionalSettings(t *testing.T) {
 
 	t.Run("bucket maxSize quota should be enabled", func(t *testing.T) {
 		getResult := map[string]string{
-			userPath:   `{"enabled":false,"check_on_raw":false,"max_size":-1024,"max_size_kb":0,"max_objects":-1}`,
 			bucketPath: `{"bucket_quota":{"enabled":false,"check_on_raw":false,"max_size":-1024,"max_size_kb":0,"max_objects":-1}}`,
 		}
 		getSeen := map[string][]string{}
@@ -436,23 +453,14 @@ func TestProvisioner_setAdditionalSettings(t *testing.T) {
 		p := newProvisioner(t, &getResult, &getSeen, &putSeen)
 		p.setBucketName("bob")
 
-		err := p.setAdditionalSettings(&apibkt.BucketOptions{
-			ObjectBucketClaim: &v1alpha1.ObjectBucketClaim{
-				Spec: v1alpha1.ObjectBucketClaimSpec{
-					AdditionalConfig: map[string]string{
-						"bucketMaxSize": "5",
-					},
-				},
-			},
+		err := p.setBucketQuota(&additionalConfigSpec{
+			bucketMaxSize: aws.Int64(5),
 		})
 		assert.NoError(t, err)
 
-		assert.Len(t, getSeen[userPath], 1)
-		assert.Equal(t, numberOfCallsWithValue("uid=bob", getSeen[userPath]), 1)
 		assert.Len(t, getSeen[bucketPath], 1)
 		assert.Equal(t, numberOfCallsWithValue("bucket=bob", getSeen[bucketPath]), 1)
 
-		assert.Len(t, putSeen[userPath], 0) // user quota should not be touched
 		assert.Len(t, putSeen[bucketPath], 1)
 		assert.Equal(t, numberOfCallsWithValue("enabled=true", putSeen[bucketPath]), 1)
 		assert.Equal(t, numberOfCallsWithValue("max-size=5", putSeen[bucketPath]), 1)
@@ -460,7 +468,6 @@ func TestProvisioner_setAdditionalSettings(t *testing.T) {
 
 	t.Run("bucket quotas are enabled and need updated enabled", func(t *testing.T) {
 		getResult := map[string]string{
-			userPath:   `{"enabled":false,"check_on_raw":false,"max_size":-1024,"max_size_kb":0,"max_objects":-1}`,
 			bucketPath: `{"bucket_quota":{"enabled":true,"check_on_raw":false,"max_size":5,"max_size_kb":0,"max_objects":4}}`,
 		}
 		getSeen := map[string][]string{}
@@ -469,24 +476,15 @@ func TestProvisioner_setAdditionalSettings(t *testing.T) {
 		p := newProvisioner(t, &getResult, &getSeen, &putSeen)
 		p.setBucketName("bob")
 
-		err := p.setAdditionalSettings(&apibkt.BucketOptions{
-			ObjectBucketClaim: &v1alpha1.ObjectBucketClaim{
-				Spec: v1alpha1.ObjectBucketClaimSpec{
-					AdditionalConfig: map[string]string{
-						"bucketMaxObjects": "14",
-						"bucketMaxSize":    "15",
-					},
-				},
-			},
+		err := p.setBucketQuota(&additionalConfigSpec{
+			bucketMaxObjects: aws.Int64(14),
+			bucketMaxSize:    aws.Int64(15),
 		})
 		assert.NoError(t, err)
 
-		assert.Len(t, getSeen[userPath], 1)
-		assert.Equal(t, numberOfCallsWithValue("uid=bob", getSeen[userPath]), 1)
 		assert.Len(t, getSeen[bucketPath], 1)
 		assert.Equal(t, numberOfCallsWithValue("bucket=bob", getSeen[bucketPath]), 1)
 
-		assert.Len(t, putSeen[userPath], 0) // user quota should not be touched
 		assert.Len(t, putSeen[bucketPath], 1)
 		assert.Equal(t, numberOfCallsWithValue("enabled=true", putSeen[bucketPath]), 1)
 		assert.Equal(t, numberOfCallsWithValue("max-objects=14", putSeen[bucketPath]), 1)

pkg/operator/ceph/object/bucket/util.go

@@ -120,6 +120,11 @@ func additionalConfigSpecFromMap(config map[string]string) (*additionalConfigSpe
 		}
 	}
 
+	if _, ok := config["bucketPolicy"]; ok {
+		policy := config["bucketPolicy"]
+		spec.bucketPolicy = &policy
+	}
+
 	return &spec, nil
 }
 

pkg/operator/ceph/object/notification/provisioner.go

@@ -91,10 +91,7 @@ func newS3Agent(p provisioner) (*object.S3Agent, error) {
 		}
 	}
 
-	if insecureTLS {
-		return object.NewInsecureS3Agent(accessKey, secretKey, objContext.Endpoint, logger.LevelAt(capnslog.DEBUG))
-	}
-	return object.NewS3Agent(accessKey, secretKey, objContext.Endpoint, logger.LevelAt(capnslog.DEBUG), tlsCert)
+	return object.NewS3Agent(accessKey, secretKey, objContext.Endpoint, logger.LevelAt(capnslog.DEBUG), tlsCert, insecureTLS, nil)
 }
 
 // TODO: convert all rules without restrictions once the AWS SDK supports that

pkg/operator/ceph/object/s3-handlers.go

@@ -39,27 +39,24 @@ type S3Agent struct {
 	Client *s3.S3
 }
 
-func NewS3Agent(accessKey, secretKey, endpoint string, debug bool, tlsCert []byte) (*S3Agent, error) {
-	return newS3Agent(accessKey, secretKey, endpoint, debug, tlsCert, false)
-}
-
-func NewInsecureS3Agent(accessKey, secretKey, endpoint string, debug bool) (*S3Agent, error) {
-	return newS3Agent(accessKey, secretKey, endpoint, debug, nil, true)
-}
-
-func newS3Agent(accessKey, secretKey, endpoint string, debug bool, tlsCert []byte, insecure bool) (*S3Agent, error) {
+func NewS3Agent(accessKey, secretKey, endpoint string, debug bool, tlsCert []byte, insecure bool, httpClient *http.Client) (*S3Agent, error) {
 	logLevel := aws.LogOff
 	if debug {
 		logLevel = aws.LogDebug
 	}
-	client := http.Client{
-		Timeout: HttpTimeOut,
-	}
 	tlsEnabled := false
 	if len(tlsCert) > 0 || insecure {
 		tlsEnabled = true
-		client.Transport = BuildTransportTLS(tlsCert, insecure)
 	}
+	if httpClient == nil {
+		httpClient = &http.Client{
+			Timeout: HttpTimeOut,
+		}
+		if tlsEnabled {
+			httpClient.Transport = BuildTransportTLS(tlsCert, insecure)
+		}
+	}
+
 	session, err := awssession.NewSession(
 		aws.NewConfig().
 			WithRegion(CephRegion).
@@ -68,7 +65,7 @@ func newS3Agent(accessKey, secretKey, endpoint string, debug bool, tlsCert []byt
 			WithS3ForcePathStyle(true).
 			WithMaxRetries(5).
 			WithDisableSSL(!tlsEnabled).
-			WithHTTPClient(&client).
+			WithHTTPClient(httpClient).
 			WithLogLevel(logLevel),
 	)
 	if err != nil {

pkg/operator/ceph/object/s3-handlers_test.go

@@ -32,7 +32,7 @@ func TestNewS3Agent(t *testing.T) {
 	t.Run("test without tls/debug", func(t *testing.T) {
 		debug := false
 		insecure := false
-		s3Agent, err := newS3Agent(accessKey, secretKey, endpoint, debug, nil, insecure)
+		s3Agent, err := NewS3Agent(accessKey, secretKey, endpoint, debug, nil, insecure, nil)
 		assert.NoError(t, err)
 		assert.NotEqual(t, aws.LogDebug, s3Agent.Client.Config.LogLevel)
 		assert.Equal(t, nil, s3Agent.Client.Config.HTTPClient.Transport)
@@ -42,7 +42,7 @@ func TestNewS3Agent(t *testing.T) {
 		debug := true
 		logLevel := aws.LogDebug
 		insecure := false
-		s3Agent, err := newS3Agent(accessKey, secretKey, endpoint, debug, nil, insecure)
+		s3Agent, err := NewS3Agent(accessKey, secretKey, endpoint, debug, nil, insecure, nil)
 		assert.NoError(t, err)
 		assert.Equal(t, &logLevel, s3Agent.Client.Config.LogLevel)
 		assert.Nil(t, s3Agent.Client.Config.HTTPClient.Transport)
@@ -51,7 +51,7 @@ func TestNewS3Agent(t *testing.T) {
 	t.Run("test without tls client cert but insecure tls", func(t *testing.T) {
 		debug := true
 		insecure := true
-		s3Agent, err := newS3Agent(accessKey, secretKey, endpoint, debug, nil, insecure)
+		s3Agent, err := NewS3Agent(accessKey, secretKey, endpoint, debug, nil, insecure, nil)
 		assert.NoError(t, err)
 		assert.NotNil(t, s3Agent.Client.Config.HTTPClient.Transport.(*http.Transport).TLSClientConfig.RootCAs) // still includes sys certs
 		assert.True(t, s3Agent.Client.Config.HTTPClient.Transport.(*http.Transport).TLSClientConfig.InsecureSkipVerify)
@@ -61,7 +61,7 @@ func TestNewS3Agent(t *testing.T) {
 		debug := true
 		insecure := false
 		tlsCert := []byte("tlsCert")
-		s3Agent, err := newS3Agent(accessKey, secretKey, endpoint, debug, tlsCert, insecure)
+		s3Agent, err := NewS3Agent(accessKey, secretKey, endpoint, debug, tlsCert, insecure, nil)
 		assert.NoError(t, err)
 		assert.NotNil(t, s3Agent.Client.Config.HTTPClient.Transport.(*http.Transport).TLSClientConfig.RootCAs)
 		assert.False(t, *s3Agent.Client.Config.DisableSSL)
@@ -70,10 +70,32 @@ func TestNewS3Agent(t *testing.T) {
 		debug := true
 		insecure := true
 		tlsCert := []byte("tlsCert")
-		s3Agent, err := newS3Agent(accessKey, secretKey, endpoint, debug, tlsCert, insecure)
+		s3Agent, err := NewS3Agent(accessKey, secretKey, endpoint, debug, tlsCert, insecure, nil)
 		assert.NoError(t, err)
 		assert.NotNil(t, s3Agent.Client.Config.HTTPClient.Transport)
 		assert.True(t, s3Agent.Client.Config.HTTPClient.Transport.(*http.Transport).TLSClientConfig.InsecureSkipVerify)
 		assert.False(t, *s3Agent.Client.Config.DisableSSL)
 	})
+	t.Run("test with custom http.Client", func(t *testing.T) {
+		debug := true
+		logLevel := aws.LogDebug
+		insecure := false
+		httpClient := &http.Client{
+			Transport: &http.Transport{
+				// set some values to check if they are passed through
+				MaxIdleConns:        7,
+				MaxIdleConnsPerHost: 13,
+				MaxConnsPerHost:     17,
+			},
+		}
+		s3Agent, err := NewS3Agent(accessKey, secretKey, endpoint, debug, nil, insecure, httpClient)
+		assert.NoError(t, err)
+		assert.Equal(t, &logLevel, s3Agent.Client.Config.LogLevel)
+		assert.NotNil(t, s3Agent.Client.Config.HTTPClient.Transport)
+		assert.True(t, *s3Agent.Client.Config.DisableSSL)
+		transport := s3Agent.Client.Config.HTTPClient.Transport
+		assert.Equal(t, 7, transport.(*http.Transport).MaxIdleConns)
+		assert.Equal(t, 13, transport.(*http.Transport).MaxIdleConnsPerHost)
+		assert.Equal(t, 17, transport.(*http.Transport).MaxConnsPerHost)
+	})
 }

tests/framework/clients/bucket.go

@@ -157,11 +157,7 @@ func (b *BucketOperation) CheckBucketNotificationSetonRGW(namespace, storeName,
 	s3endpoint, _ := helper.ObjectClient.GetEndPointUrl(namespace, storeName)
 	s3AccessKey, _ := helper.BucketClient.GetAccessKey(obcName)
 	s3SecretKey, _ := helper.BucketClient.GetSecretKey(obcName)
-	if tlsEnabled {
-		s3client, err = rgw.NewInsecureS3Agent(s3AccessKey, s3SecretKey, s3endpoint, true)
-	} else {
-		s3client, err = rgw.NewS3Agent(s3AccessKey, s3SecretKey, s3endpoint, true, nil)
-	}
+	s3client, err = rgw.NewS3Agent(s3AccessKey, s3SecretKey, s3endpoint, true, nil, tlsEnabled, nil)
 	if err != nil {
 		logger.Infof("failed to s3client due to %v", err)
 		return false

tests/integration/ceph_bucket_notification_test.go

@@ -123,11 +123,8 @@ func testBucketNotifications(s *suite.Suite, helper *clients.TestClient, k8sh *u
 		s3endpoint, _ := helper.ObjectClient.GetEndPointUrl(namespace, storeName)
 		s3AccessKey, _ := helper.BucketClient.GetAccessKey(obcName)
 		s3SecretKey, _ := helper.BucketClient.GetSecretKey(obcName)
-		if objectStore.Spec.IsTLSEnabled() {
-			s3client, err = rgw.NewInsecureS3Agent(s3AccessKey, s3SecretKey, s3endpoint, true)
-		} else {
-			s3client, err = rgw.NewS3Agent(s3AccessKey, s3SecretKey, s3endpoint, true, nil)
-		}
+		insecure := objectStore.Spec.IsTLSEnabled()
+		s3client, err = rgw.NewS3Agent(s3AccessKey, s3SecretKey, s3endpoint, true, nil, insecure, nil)
 
 		assert.Nil(t, err)
 		logger.Infof("endpoint (%s) Accesskey (%s) secret (%s)", s3endpoint, s3AccessKey, s3SecretKey)
@@ -227,11 +224,8 @@ func testBucketNotifications(s *suite.Suite, helper *clients.TestClient, k8sh *u
 		s3endpoint, _ := helper.ObjectClient.GetEndPointUrl(namespace, storeName)
 		s3AccessKey, _ := helper.BucketClient.GetAccessKey(obcName)
 		s3SecretKey, _ := helper.BucketClient.GetSecretKey(obcName)
-		if objectStore.Spec.IsTLSEnabled() {
-			s3client, err = rgw.NewInsecureS3Agent(s3AccessKey, s3SecretKey, s3endpoint, true)
-		} else {
-			s3client, err = rgw.NewS3Agent(s3AccessKey, s3SecretKey, s3endpoint, true, nil)
-		}
+		insecure := objectStore.Spec.IsTLSEnabled()
+		s3client, err = rgw.NewS3Agent(s3AccessKey, s3SecretKey, s3endpoint, true, nil, insecure, nil)
 
 		assert.Nil(t, err)
 		logger.Infof("endpoint (%s) Accesskey (%s) secret (%s)", s3endpoint, s3AccessKey, s3SecretKey)

tests/integration/ceph_object_test.go

@@ -23,6 +23,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/aws/aws-sdk-go/aws/awserr"
+	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/kube-object-storage/lib-bucket-provisioner/pkg/apis/objectbucket.io/v1alpha1"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
@@ -31,6 +33,7 @@ import (
 	"github.com/rook/rook/tests/framework/installer"
 	"github.com/rook/rook/tests/framework/utils"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -218,11 +221,8 @@ func testObjectStoreOperations(s *suite.Suite, helper *clients.TestClient, k8sh
 		s3endpoint, _ := helper.ObjectClient.GetEndPointUrl(namespace, storeName)
 		s3AccessKey, _ := helper.BucketClient.GetAccessKey(obcName)
 		s3SecretKey, _ := helper.BucketClient.GetSecretKey(obcName)
-		if objectStore.Spec.IsTLSEnabled() {
-			s3client, err = rgw.NewInsecureS3Agent(s3AccessKey, s3SecretKey, s3endpoint, true)
-		} else {
-			s3client, err = rgw.NewS3Agent(s3AccessKey, s3SecretKey, s3endpoint, true, nil)
-		}
+		insecure := objectStore.Spec.IsTLSEnabled()
+		s3client, err = rgw.NewS3Agent(s3AccessKey, s3SecretKey, s3endpoint, true, nil, insecure, nil)
 
 		assert.Nil(t, err)
 		logger.Infof("endpoint (%s) Accesskey (%s) secret (%s)", s3endpoint, s3AccessKey, s3SecretKey)
@@ -334,11 +334,8 @@ func testObjectStoreOperations(s *suite.Suite, helper *clients.TestClient, k8sh
 			s3AccessKey := string(secret.Data["AWS_ACCESS_KEY_ID"])
 			s3SecretKey := string(secret.Data["AWS_SECRET_ACCESS_KEY"])
 
-			if objectStore.Spec.IsTLSEnabled() {
-				s3client, err = rgw.NewInsecureS3Agent(s3AccessKey, s3SecretKey, s3endpoint, true)
-			} else {
-				s3client, err = rgw.NewS3Agent(s3AccessKey, s3SecretKey, s3endpoint, true, nil)
-			}
+			insecure := objectStore.Spec.IsTLSEnabled()
+			s3client, err = rgw.NewS3Agent(s3AccessKey, s3SecretKey, s3endpoint, true, nil, insecure, nil)
 			assert.Nil(t, err)
 			logger.Infof("endpoint (%s) Accesskey (%s) secret (%s)", s3endpoint, s3AccessKey, s3SecretKey)
 		})
@@ -426,6 +423,248 @@ func testObjectStoreOperations(s *suite.Suite, helper *clients.TestClient, k8sh
 		})
 	})
 
+	t.Run("OBC bucket policy management", func(t *testing.T) {
+		bucketName := "bucket-policy-test"
+		var obName string
+		var s3client *rgw.S3Agent
+		bucketPolicy1 := `
+{
+    "Version": "2012-10-17",
+    "Statement": [
+        {
+            "Effect": "Allow",
+            "Principal": {
+                "AWS": "arn:aws:iam:::user/foo"
+            },
+            "Action": [
+                "s3:GetObject",
+                "s3:PutObject",
+                "s3:DeleteObject",
+                "s3:ListBucket",
+                "s3:GetBucketLocation"
+            ],
+            "Resource": [
+                "arn:aws:s3:::bucket-policy-test/*"
+            ]
+        }
+    ]
+}
+		`
+
+		bucketPolicy2 := `
+{
+    "Version": "2012-10-17",
+    "Statement": [
+        {
+            "Effect": "Allow",
+            "Principal": {
+                "AWS": "arn:aws:iam:::user/bar"
+            },
+            "Action": [
+                "s3:GetObject",
+                "s3:ListBucket",
+                "s3:GetBucketLocation"
+            ],
+            "Resource": [
+                "arn:aws:s3:::bucket-policy-test/*"
+            ]
+        }
+    ]
+}
+		`
+
+		t.Run("create obc with bucketPolicy", func(t *testing.T) {
+			newObc := v1alpha1.ObjectBucketClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      bucketName,
+					Namespace: namespace,
+				},
+				Spec: v1alpha1.ObjectBucketClaimSpec{
+					BucketName:       bucketName,
+					StorageClassName: bucketStorageClassName,
+					AdditionalConfig: map[string]string{
+						"bucketPolicy": bucketPolicy1,
+					},
+				},
+			}
+			_, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(namespace).Create(ctx, &newObc, metav1.CreateOptions{})
+			assert.Nil(t, err)
+			obcBound := utils.Retry(20, 2*time.Second, "OBC is Bound", func() bool {
+				liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(namespace).Get(ctx, bucketName, metav1.GetOptions{})
+				if err != nil {
+					return false
+				}
+
+				return liveObc.Status.Phase == v1alpha1.ObjectBucketClaimStatusPhaseBound
+			})
+			assert.True(t, obcBound)
+
+			// wait until obc is Bound to lookup the ob name
+			obc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(namespace).Get(ctx, bucketName, metav1.GetOptions{})
+			assert.Nil(t, err)
+			obName = obc.Spec.ObjectBucketName
+
+			obBound := utils.Retry(20, 2*time.Second, "OB is Bound", func() bool {
+				liveOb, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets().Get(ctx, obName, metav1.GetOptions{})
+				if err != nil {
+					return false
+				}
+
+				return liveOb.Status.Phase == v1alpha1.ObjectBucketStatusPhaseBound
+			})
+			assert.True(t, obBound)
+
+			// verify that bucketPolicy is set
+			assert.Equal(t, bucketPolicy1, obc.Spec.AdditionalConfig["bucketPolicy"])
+
+			// as the tests are running external to k8s, the internal svc can't be used
+			labelSelector := "rgw=" + storeName
+			services, err := k8sh.Clientset.CoreV1().Services(namespace).List(ctx, metav1.ListOptions{LabelSelector: labelSelector})
+			assert.Nil(t, err)
+			assert.Equal(t, 1, len(services.Items))
+			s3endpoint := services.Items[0].Spec.ClusterIP + ":80"
+
+			secret, err := k8sh.Clientset.CoreV1().Secrets(namespace).Get(ctx, bucketName, metav1.GetOptions{})
+			assert.Nil(t, err)
+			s3AccessKey := string(secret.Data["AWS_ACCESS_KEY_ID"])
+			s3SecretKey := string(secret.Data["AWS_SECRET_ACCESS_KEY"])
+
+			insecure := objectStore.Spec.IsTLSEnabled()
+			s3client, err = rgw.NewS3Agent(s3AccessKey, s3SecretKey, s3endpoint, true, nil, insecure, nil)
+			assert.Nil(t, err)
+			logger.Infof("endpoint (%s) Accesskey (%s) secret (%s)", s3endpoint, s3AccessKey, s3SecretKey)
+		})
+
+		t.Run("policy was applied verbatim to bucket", func(t *testing.T) {
+			policyResp, err := s3client.Client.GetBucketPolicy(&s3.GetBucketPolicyInput{
+				Bucket: &bucketName,
+			})
+			require.NoError(t, err)
+			assert.Equal(t, bucketPolicy1, *policyResp.Policy)
+		})
+
+		t.Run("update obc bucketPolicy", func(t *testing.T) {
+			obc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(namespace).Get(ctx, bucketName, metav1.GetOptions{})
+			assert.Nil(t, err)
+
+			obc.Spec.AdditionalConfig["bucketPolicy"] = bucketPolicy2
+
+			_, err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(namespace).Update(ctx, obc, metav1.UpdateOptions{})
+			assert.Nil(t, err)
+			obcBound := utils.Retry(20, time.Second, "OBC is Bound", func() bool {
+				liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(namespace).Get(ctx, bucketName, metav1.GetOptions{})
+				if err != nil {
+					return false
+				}
+
+				return liveObc.Status.Phase == v1alpha1.ObjectBucketClaimStatusPhaseBound
+			})
+			assert.True(t, obcBound)
+
+			// wait until obc is Bound to lookup the ob name
+			obc, err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(namespace).Get(ctx, bucketName, metav1.GetOptions{})
+			assert.Nil(t, err)
+			obName = obc.Spec.ObjectBucketName
+
+			obBound := utils.Retry(20, time.Second, "OB is Bound", func() bool {
+				liveOb, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets().Get(ctx, obName, metav1.GetOptions{})
+				if err != nil {
+					return false
+				}
+
+				return liveOb.Status.Phase == v1alpha1.ObjectBucketStatusPhaseBound
+			})
+			assert.True(t, obBound)
+
+			// verify that updated bucketPolicy is set
+			assert.Equal(t, bucketPolicy2, obc.Spec.AdditionalConfig["bucketPolicy"])
+		})
+
+		t.Run("policy update applied verbatim to bucket", func(t *testing.T) {
+			var livePolicy string
+			utils.Retry(20, time.Second, "policy changed", func() bool {
+				policyResp, err := s3client.Client.GetBucketPolicy(&s3.GetBucketPolicyInput{
+					Bucket: &bucketName,
+				})
+				if err != nil {
+					return false
+				}
+
+				livePolicy = *policyResp.Policy
+				return bucketPolicy2 == livePolicy
+			})
+			assert.Equal(t, bucketPolicy2, livePolicy)
+		})
+
+		t.Run("remove obc bucketPolicy", func(t *testing.T) {
+			obc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(namespace).Get(ctx, bucketName, metav1.GetOptions{})
+			assert.Nil(t, err)
+
+			obc.Spec.AdditionalConfig = map[string]string{}
+
+			_, err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(namespace).Update(ctx, obc, metav1.UpdateOptions{})
+			assert.Nil(t, err)
+			obcBound := utils.Retry(20, time.Second, "OBC is Bound", func() bool {
+				liveObc, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(namespace).Get(ctx, bucketName, metav1.GetOptions{})
+				if err != nil {
+					return false
+				}
+
+				return liveObc.Status.Phase == v1alpha1.ObjectBucketClaimStatusPhaseBound
+			})
+			assert.True(t, obcBound)
+
+			// wait until obc is Bound to lookup the ob name
+			obc, err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(namespace).Get(ctx, bucketName, metav1.GetOptions{})
+			assert.Nil(t, err)
+			obName = obc.Spec.ObjectBucketName
+
+			obBound := utils.Retry(20, time.Second, "OB is Bound", func() bool {
+				liveOb, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets().Get(ctx, obName, metav1.GetOptions{})
+				if err != nil {
+					return false
+				}
+
+				return liveOb.Status.Phase == v1alpha1.ObjectBucketStatusPhaseBound
+			})
+			assert.True(t, obBound)
+
+			// verify that bucketPolicy is unset
+			assert.NotContains(t, obc.Spec.AdditionalConfig, "bucketPolicy")
+		})
+
+		t.Run("policy was removed from bucket", func(t *testing.T) {
+			var err error
+			utils.Retry(20, time.Second, "policy is gone", func() bool {
+				_, err = s3client.Client.GetBucketPolicy(&s3.GetBucketPolicyInput{
+					Bucket: &bucketName,
+				})
+				return err != nil
+			})
+			require.Error(t, err)
+			require.Implements(t, (*awserr.Error)(nil), err)
+			aerr, _ := err.(awserr.Error)
+			assert.Equal(t, aerr.Code(), "NoSuchBucketPolicy")
+		})
+
+		t.Run("delete bucket", func(t *testing.T) {
+			err = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(namespace).Delete(ctx, bucketName, metav1.DeleteOptions{})
+			assert.Nil(t, err)
+
+			absent := utils.Retry(20, time.Second, "OBC is absent", func() bool {
+				_, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(namespace).Get(ctx, bucketName, metav1.GetOptions{})
+				return err != nil
+			})
+			assert.True(t, absent)
+
+			absent = utils.Retry(20, time.Second, "OB is absent", func() bool {
+				_, err := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets().Get(ctx, obName, metav1.GetOptions{})
+				return err != nil
+			})
+			assert.True(t, absent)
+		})
+	})
+
 	t.Run("Regression check: OBC does not revert to Pending phase", func(t *testing.T) {
 		// A bug exists in older versions of lib-bucket-provisioner that will revert a bucket and claim
 		// back to "Pending" phase after being created and initially "Bound" by looping infinitely in