From 7ac36bcef6149a9280e057115b0d00c2881390ab Mon Sep 17 00:00:00 2001 From: Joshua Hoblitt Date: Mon, 21 Oct 2024 16:48:21 -0700 Subject: [PATCH 1/4] object: factor out bucket.setS3Agent() Add a s3Agent field to bucket.Provisioner struct as a step towards allowing the s3Agent / s3 client to be mocked in unit tests. Signed-off-by: Joshua Hoblitt (cherry picked from commit e5f79dba7f9bed5f004ddff7ca514f63dc94f84e) --- .../ceph/object/bucket/provisioner.go | 63 +++++++++---------- 1 file changed, 28 insertions(+), 35 deletions(-) diff --git a/pkg/operator/ceph/object/bucket/provisioner.go b/pkg/operator/ceph/object/bucket/provisioner.go index 074905625..4e3dec79f 100644 --- a/pkg/operator/ceph/object/bucket/provisioner.go +++ b/pkg/operator/ceph/object/bucket/provisioner.go @@ -54,6 +54,7 @@ type Provisioner struct { tlsCert []byte insecureTLS bool adminOpsClient *admin.API + s3Agent *object.S3Agent } type additionalConfigSpec struct { @@ -95,12 +96,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 +112,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) } @@ -172,29 +168,13 @@ 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 } // 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,7 +195,7 @@ 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 { @@ -280,19 +260,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 +295,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 +307,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 } @@ -754,3 +732,18 @@ func (p *Provisioner) setAdminOpsAPIClient() error { return nil } + +func (p *Provisioner) setS3Agent() error { + var err error + + if p.insecureTLS { + p.s3Agent, err = object.NewInsecureS3Agent(p.accessKeyID, p.secretAccessKey, p.getObjectStoreEndpoint(), logger.LevelAt(capnslog.DEBUG)) + } else { + p.s3Agent, err = object.NewS3Agent(p.accessKeyID, p.secretAccessKey, p.getObjectStoreEndpoint(), logger.LevelAt(capnslog.DEBUG), p.tlsCert) + } + if err != nil { + return err + } + + return nil +} From 40e5026b263e1fddd882ff2a6c0f707f6d6e7ea0 Mon Sep 17 00:00:00 2001 From: Joshua Hoblitt Date: Tue, 22 Oct 2024 15:09:04 -0700 Subject: [PATCH 2/4] object: add httpClient param to object.NewS3Agent() To allow the caller to pass in their own transport when testing. Signed-off-by: Joshua Hoblitt (cherry picked from commit 57b7eeec800bc008507b7e2ff5b47d50d4503ece) --- .../ceph/object/bucket/provisioner.go | 13 ++------ .../ceph/object/notification/provisioner.go | 5 +-- pkg/operator/ceph/object/s3-handlers.go | 25 +++++++-------- pkg/operator/ceph/object/s3-handlers_test.go | 32 ++++++++++++++++--- tests/framework/clients/bucket.go | 6 +--- .../ceph_bucket_notification_test.go | 14 +++----- tests/integration/ceph_object_test.go | 14 +++----- 7 files changed, 50 insertions(+), 59 deletions(-) diff --git a/pkg/operator/ceph/object/bucket/provisioner.go b/pkg/operator/ceph/object/bucket/provisioner.go index 4e3dec79f..eba551186 100644 --- a/pkg/operator/ceph/object/bucket/provisioner.go +++ b/pkg/operator/ceph/object/bucket/provisioner.go @@ -735,15 +735,6 @@ func (p *Provisioner) setAdminOpsAPIClient() error { func (p *Provisioner) setS3Agent() error { var err error - - if p.insecureTLS { - p.s3Agent, err = object.NewInsecureS3Agent(p.accessKeyID, p.secretAccessKey, p.getObjectStoreEndpoint(), logger.LevelAt(capnslog.DEBUG)) - } else { - p.s3Agent, err = object.NewS3Agent(p.accessKeyID, p.secretAccessKey, p.getObjectStoreEndpoint(), logger.LevelAt(capnslog.DEBUG), p.tlsCert) - } - if err != nil { - return err - } - - return nil + p.s3Agent, err = object.NewS3Agent(p.accessKeyID, p.secretAccessKey, p.getObjectStoreEndpoint(), logger.LevelAt(capnslog.DEBUG), p.tlsCert, p.insecureTLS, nil) + return err } diff --git a/pkg/operator/ceph/object/notification/provisioner.go b/pkg/operator/ceph/object/notification/provisioner.go index dc1579fa0..47d0d37e5 100644 --- a/pkg/operator/ceph/object/notification/provisioner.go +++ b/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 diff --git a/pkg/operator/ceph/object/s3-handlers.go b/pkg/operator/ceph/object/s3-handlers.go index 13eccb826..6bc510ef0 100644 --- a/pkg/operator/ceph/object/s3-handlers.go +++ b/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 { diff --git a/pkg/operator/ceph/object/s3-handlers_test.go b/pkg/operator/ceph/object/s3-handlers_test.go index f7f20fe0c..628dea1e5 100644 --- a/pkg/operator/ceph/object/s3-handlers_test.go +++ b/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) + }) } diff --git a/tests/framework/clients/bucket.go b/tests/framework/clients/bucket.go index 6b86aca37..fd0216d7e 100644 --- a/tests/framework/clients/bucket.go +++ b/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 diff --git a/tests/integration/ceph_bucket_notification_test.go b/tests/integration/ceph_bucket_notification_test.go index d8330ec99..8e0bdfdfa 100644 --- a/tests/integration/ceph_bucket_notification_test.go +++ b/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) diff --git a/tests/integration/ceph_object_test.go b/tests/integration/ceph_object_test.go index 3763a3c1e..8ef593ca4 100644 --- a/tests/integration/ceph_object_test.go +++ b/tests/integration/ceph_object_test.go @@ -218,11 +218,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 +331,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) }) From fa9da8f12a2fc62cc5347c7c4f748930d5efc3e8 Mon Sep 17 00:00:00 2001 From: Joshua Hoblitt Date: Mon, 21 Oct 2024 17:21:27 -0700 Subject: [PATCH 3/4] object: add obc bucketPolicy Signed-off-by: Joshua Hoblitt (cherry picked from commit 97b904c717f69cf6f918bc2334407589cb1537ce) --- PendingReleaseNotes.md | 1 + .../ceph/object/bucket/provisioner.go | 94 +++- .../ceph/object/bucket/provisioner_test.go | 400 +++++++++--------- pkg/operator/ceph/object/bucket/util.go | 5 + 4 files changed, 285 insertions(+), 215 deletions(-) diff --git a/PendingReleaseNotes.md b/PendingReleaseNotes.md index c850f99df..7683a1a58 100644 --- a/PendingReleaseNotes.md +++ b/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)). diff --git a/pkg/operator/ceph/object/bucket/provisioner.go b/pkg/operator/ceph/object/bucket/provisioner.go index eba551186..d2d7cb9cc 100644 --- a/pkg/operator/ceph/object/bucket/provisioner.go +++ b/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" @@ -62,6 +63,7 @@ type additionalConfigSpec struct { maxSize *int64 bucketMaxObjects *int64 bucketMaxSize *int64 + bucketPolicy *string } var _ apibkt.Provisioner = &Provisioner{} @@ -129,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) } @@ -173,6 +180,24 @@ func (p Provisioner) Grant(options *apibkt.BucketOptions) (*bktv1alpha1.ObjectBu 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 := p.s3Agent.GetBucketPolicy(p.bucketName) if err != nil { @@ -202,12 +227,6 @@ func (p Provisioner) Grant(options *apibkt.BucketOptions) (*bktv1alpha1.ObjectBu 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 } @@ -552,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") } @@ -568,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 } @@ -675,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 { diff --git a/pkg/operator/ceph/object/bucket/provisioner_test.go b/pkg/operator/ceph/object/bucket/provisioner_test.go index 467b0a54f..e5a6ea478 100644 --- a/pkg/operator/ceph/object/bucket/provisioner_test.go +++ b/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) diff --git a/pkg/operator/ceph/object/bucket/util.go b/pkg/operator/ceph/object/bucket/util.go index de1936d26..22a0e04db 100644 --- a/pkg/operator/ceph/object/bucket/util.go +++ b/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 } From 04624782ea1c9bcf5dfc4baf74f1bfaaaa8e21e1 Mon Sep 17 00:00:00 2001 From: Joshua Hoblitt Date: Mon, 4 Nov 2024 15:41:05 -0700 Subject: [PATCH 4/4] test: add obc bucketPolicy integration test Signed-off-by: Joshua Hoblitt (cherry picked from commit 8fd34e88bbaf105d4085dfa9c08213320e5d8896) --- tests/integration/ceph_object_test.go | 245 ++++++++++++++++++++++++++ 1 file changed, 245 insertions(+) diff --git a/tests/integration/ceph_object_test.go b/tests/integration/ceph_object_test.go index 8ef593ca4..d3faf0412 100644 --- a/tests/integration/ceph_object_test.go +++ b/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" @@ -420,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