From 0e25b2089f5dd4c7a2b9294b51afcf3bbdc3dabc Mon Sep 17 00:00:00 2001 From: Joshua Hoblitt Date: Tue, 13 Jan 2026 13:59:45 -0700 Subject: [PATCH] object: add ObjectStoreUserSpec.OpMask field Signed-off-by: Joshua Hoblitt (cherry picked from commit 3ef0c90ef8f06a5478cffd4beac7c670cc183b60) --- .../ceph-object-store-user-crd.md | 8 + Documentation/CRDs/specification.md | 33 ++ .../charts/rook-ceph/templates/resources.yaml | 13 + deploy/examples/crds.yaml | 13 + pkg/apis/ceph.rook.io/v1/types.go | 11 + .../ceph.rook.io/v1/zz_generated.deepcopy.go | 9 + pkg/operator/ceph/object/user/controller.go | 119 ++++++-- .../ceph/object/user/controller_test.go | 93 +++++- .../ceph/object/user/opmask/opmask.go | 78 +++++ .../ceph/object/user/opmask/opmask_test.go | 73 +++++ tests/integration/ceph_object_test.go | 2 + .../integration/object/user/opmask/opmask.go | 285 ++++++++++++++++++ 12 files changed, 694 insertions(+), 43 deletions(-) create mode 100644 pkg/operator/ceph/object/user/opmask/opmask.go create mode 100644 pkg/operator/ceph/object/user/opmask/opmask_test.go create mode 100644 tests/integration/object/user/opmask/opmask.go diff --git a/Documentation/CRDs/Object-Storage/ceph-object-store-user-crd.md b/Documentation/CRDs/Object-Storage/ceph-object-store-user-crd.md index af84691b2..c22dc37c8 100644 --- a/Documentation/CRDs/Object-Storage/ceph-object-store-user-crd.md +++ b/Documentation/CRDs/Object-Storage/ceph-object-store-user-crd.md @@ -23,6 +23,10 @@ spec: capabilities: user: "*" bucket: "*" + opMask: + - read + - write + - delete ``` ## Object Store User Settings @@ -60,3 +64,7 @@ spec: * `user-policy` * `odic-provider` * `ratelimit` +* `opMask`: Internally, RGW labels "operations" on persistent state as `RGW_OP_TYPE_READ` (`read`), `RGW_OP_TYPE_WRITE` (`write`), or `RGW_OP_TYPE_DELETE` (`delete`). All RGW users have an "operation mask", which does not function as mask or filter as is typically implied by the word "mask", but as a set of allowed or permissible "operation" types the user is able to perform. The "operation mask" is applied regardless of the bucket or IAM policy. For example, in order for an RGW user to be able to read an object from a bucket, that user must have **both** the `read` "op mask" bit and an IAM/bucket policy that allows `s3:GetObject`. The default operations allowed are `read`, `write`, and `delete`. Setting the value to `[]` (an empty YAML sequence) causes all "operations" in the mask to be removed, meaning that the user will not be able to perform any operations. These operation masks are supported: + * `read` + * `write` + * `delete` diff --git a/Documentation/CRDs/specification.md b/Documentation/CRDs/specification.md index f52965fee..d22daa969 100644 --- a/Documentation/CRDs/specification.md +++ b/Documentation/CRDs/specification.md @@ -2204,6 +2204,20 @@ string

The namespace where the parent CephCluster and CephObjectStore are found

+ + +opMask
+ + +[]github.com/rook/rook/pkg/apis/ceph.rook.io/v1.ObjectUserOpMask + + + + +(Optional) +

The op-mask of the user.

+ + @@ -12131,6 +12145,20 @@ string

The namespace where the parent CephCluster and CephObjectStore are found

+ + +opMask
+ + +[]github.com/rook/rook/pkg/apis/ceph.rook.io/v1.ObjectUserOpMask + + + + +(Optional) +

The op-mask of the user.

+ +

ObjectStoreUserStatus @@ -12453,6 +12481,11 @@ Kubernetes core/v1.SecretKeySelector +

ObjectUserOpMask +(string alias)

+
+

Internally, RGW labels “operations” on persistent state as RGW_OP_TYPE_READ (read), RGW_OP_TYPE_WRITE (write), or RGW_OP_TYPE_DELETE (delete). All RGW users have an “operation mask”, which does not function as mask or filter as is typically implied by the word “mask”, but as a set of allowed or permissible “operation” types the user is able to perform. The “operation mask” is applied regardless of the bucket or IAM policy. For example, in order for an RGW user to be able to read an object from a bucket, that user must have both the read “op mask” bit and an IAM/bucket policy that allows s3:GetObject. The default operations allowed are read, write, and delete. Setting the value to [] (an empty YAML sequence) causes all “operations” in the mask to be removed, meaning that the user will not be able to perform any operations. These operation masks are supported: read, write, delete

+

ObjectUserQuotaSpec

diff --git a/deploy/charts/rook-ceph/templates/resources.yaml b/deploy/charts/rook-ceph/templates/resources.yaml index 65b7e61ee..aac9a8ee8 100644 --- a/deploy/charts/rook-ceph/templates/resources.yaml +++ b/deploy/charts/rook-ceph/templates/resources.yaml @@ -15042,6 +15042,19 @@ spec: x-kubernetes-map-type: atomic type: object type: array + opMask: + description: The op-mask of the user. + items: + description: 'Internally, RGW labels "operations" on persistent state as `RGW_OP_TYPE_READ` (`read`), `RGW_OP_TYPE_WRITE` (`write`), or `RGW_OP_TYPE_DELETE` (`delete`). All RGW users have an "operation mask", which does not function as mask or filter as is typically implied by the word "mask", but as a set of allowed or permissible "operation" types the user is able to perform. The "operation mask" is applied regardless of the bucket or IAM policy. For example, in order for an RGW user to be able to read an object from a bucket, that user must have **both** the `read` "op mask" bit and an IAM/bucket policy that allows `s3:GetObject`. The default operations allowed are `read`, `write`, and `delete`. Setting the value to `[]` (an empty YAML sequence) causes all "operations" in the mask to be removed, meaning that the user will not be able to perform any operations. These operation masks are supported: `read`, `write`, `delete`' + enum: + - read + - write + - delete + type: string + maxItems: 3 + minItems: 0 + type: array + x-kubernetes-list-type: set quotas: description: ObjectUserQuotaSpec can be used to set quotas for the object store user to limit their usage. See the [Ceph docs](https://docs.ceph.com/en/latest/radosgw/admin/?#quota-management) for more nullable: true diff --git a/deploy/examples/crds.yaml b/deploy/examples/crds.yaml index 87c9f5d76..42f56a6bd 100644 --- a/deploy/examples/crds.yaml +++ b/deploy/examples/crds.yaml @@ -15031,6 +15031,19 @@ spec: x-kubernetes-map-type: atomic type: object type: array + opMask: + description: The op-mask of the user. + items: + description: 'Internally, RGW labels "operations" on persistent state as `RGW_OP_TYPE_READ` (`read`), `RGW_OP_TYPE_WRITE` (`write`), or `RGW_OP_TYPE_DELETE` (`delete`). All RGW users have an "operation mask", which does not function as mask or filter as is typically implied by the word "mask", but as a set of allowed or permissible "operation" types the user is able to perform. The "operation mask" is applied regardless of the bucket or IAM policy. For example, in order for an RGW user to be able to read an object from a bucket, that user must have **both** the `read` "op mask" bit and an IAM/bucket policy that allows `s3:GetObject`. The default operations allowed are `read`, `write`, and `delete`. Setting the value to `[]` (an empty YAML sequence) causes all "operations" in the mask to be removed, meaning that the user will not be able to perform any operations. These operation masks are supported: `read`, `write`, `delete`' + enum: + - read + - write + - delete + type: string + maxItems: 3 + minItems: 0 + type: array + x-kubernetes-list-type: set quotas: description: ObjectUserQuotaSpec can be used to set quotas for the object store user to limit their usage. See the [Ceph docs](https://docs.ceph.com/en/latest/radosgw/admin/?#quota-management) for more nullable: true diff --git a/pkg/apis/ceph.rook.io/v1/types.go b/pkg/apis/ceph.rook.io/v1/types.go index 8c12dcbe6..affb98aa9 100755 --- a/pkg/apis/ceph.rook.io/v1/types.go +++ b/pkg/apis/ceph.rook.io/v1/types.go @@ -2268,8 +2268,19 @@ type ObjectStoreUserSpec struct { // The namespace where the parent CephCluster and CephObjectStore are found // +optional ClusterNamespace string `json:"clusterNamespace,omitempty"` + // The op-mask of the user. + // +optional + // +kubebuilder:validation:MinItems=0 + // +kubebuilder:validation:MaxItems=3 + // +listType=set + OpMask *[]ObjectUserOpMask `json:"opMask,omitempty"` } +// Internally, RGW labels "operations" on persistent state as `RGW_OP_TYPE_READ` (`read`), `RGW_OP_TYPE_WRITE` (`write`), or `RGW_OP_TYPE_DELETE` (`delete`). All RGW users have an "operation mask", which does not function as mask or filter as is typically implied by the word "mask", but as a set of allowed or permissible "operation" types the user is able to perform. The "operation mask" is applied regardless of the bucket or IAM policy. For example, in order for an RGW user to be able to read an object from a bucket, that user must have **both** the `read` "op mask" bit and an IAM/bucket policy that allows `s3:GetObject`. The default operations allowed are `read`, `write`, and `delete`. Setting the value to `[]` (an empty YAML sequence) causes all "operations" in the mask to be removed, meaning that the user will not be able to perform any operations. These operation masks are supported: `read`, `write`, `delete` +// +enum +// +kubebuilder:validation:Enum=read;write;delete +type ObjectUserOpMask string + // Additional admin-level capabilities for the Ceph object store user type ObjectUserCapSpec struct { // +optional diff --git a/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go b/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go index c480a40ab..e615f19a1 100644 --- a/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go +++ b/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go @@ -4387,6 +4387,15 @@ func (in *ObjectStoreUserSpec) DeepCopyInto(out *ObjectStoreUserSpec) { (*in)[i].DeepCopyInto(&(*out)[i]) } } + if in.OpMask != nil { + in, out := &in.OpMask, &out.OpMask + *out = new([]ObjectUserOpMask) + if **in != nil { + in, out := *in, *out + *out = make([]ObjectUserOpMask, len(*in)) + copy(*out, *in) + } + } return } diff --git a/pkg/operator/ceph/object/user/controller.go b/pkg/operator/ceph/object/user/controller.go index b24f1dfde..2981ef09e 100644 --- a/pkg/operator/ceph/object/user/controller.go +++ b/pkg/operator/ceph/object/user/controller.go @@ -49,6 +49,7 @@ import ( cephclient "github.com/rook/rook/pkg/daemon/ceph/client" opcontroller "github.com/rook/rook/pkg/operator/ceph/controller" "github.com/rook/rook/pkg/operator/ceph/object" + opmask "github.com/rook/rook/pkg/operator/ceph/object/user/opmask" "github.com/rook/rook/pkg/operator/ceph/reporting" "github.com/rook/rook/pkg/operator/k8sutil" "github.com/rook/rook/pkg/util/log" @@ -334,7 +335,10 @@ func (r *ReconcileObjectStoreUser) reconcile(request reconcile.Request) (reconci // CR is not deleted, continue reconciling // Generate user config - userConfig := generateUserConfig(cephObjectStoreUser) + userConfig, err := generateUserConfig(cephObjectStoreUser) + if err != nil { + return reconcile.Result{}, *cephObjectStoreUser, errors.Wrapf(err, "failed to generate user config for %q", cephObjectStoreUser.Name) + } referencedSecrets := &map[types.UID]*corev1.Secret{} // Set any provided key pair(s) @@ -395,20 +399,18 @@ func (r *ReconcileObjectStoreUser) reconcileCephUser(cephObjectStoreUser *cephv1 return reconcile.Result{}, nil } -func (r *ReconcileObjectStoreUser) createOrUpdateCephUser(u *cephv1.CephObjectStoreUser, userConfig *admin.User) error { +func (r *ReconcileObjectStoreUser) createOrUpdateCephUser(u *cephv1.CephObjectStoreUser, targetUser *admin.User) error { nsName := opcontroller.NsName(u.Namespace, u.Name) log.NamedInfo(nsName, logger, "creating ceph object user") logCreateOrUpdate := fmt.Sprintf("retrieved existing ceph object user %q", u.Name) - var user admin.User - var err error // lookup user by name only and not by access key - user, err = r.objContext.AdminOpsClient.GetUser(r.opManagerContext, admin.User{ID: u.Name}) + liveUser, err := r.objContext.AdminOpsClient.GetUser(r.opManagerContext, admin.User{ID: u.Name}) if err != nil { if errors.Is(err, admin.ErrNoSuchUser) { - user, err = r.objContext.AdminOpsClient.CreateUser(r.opManagerContext, *userConfig) + liveUser, err = r.objContext.AdminOpsClient.CreateUser(r.opManagerContext, *targetUser) if err != nil { - return errors.Wrapf(err, "failed to create ceph object user %v", &userConfig.ID) + return errors.Wrapf(err, "failed to create ceph object user %v", &targetUser.ID) } logCreateOrUpdate = fmt.Sprintf("created ceph object user %q", u.Name) } else { @@ -416,33 +418,32 @@ func (r *ReconcileObjectStoreUser) createOrUpdateCephUser(u *cephv1.CephObjectSt } } - // Update max bucket if necessary - log.NamedTrace(nsName, logger, "user capabilities(id: %s, caps: %#v, user caps: %s, op mask: %s)", - user.ID, user.Caps, user.UserCaps, user.OpMask) - if *user.MaxBuckets != *userConfig.MaxBuckets { - user, err = r.objContext.AdminOpsClient.ModifyUser(r.opManagerContext, *userConfig) + // Update simple scalar fields supported by admin.ModifyUser() excluding keys, as this method is unable to handle multiple keys. + if !isUserSync(targetUser, &liveUser) { + liveUser, err = r.objContext.AdminOpsClient.ModifyUser(r.opManagerContext, *targetUser) if err != nil { - return errors.Wrapf(err, "failed to update ceph object user %q max buckets", userConfig.ID) + return errors.Wrapf(err, "failed to update ceph object user %q", targetUser.ID) } logCreateOrUpdate = fmt.Sprintf("updated ceph object user %q", u.Name) } - // Update caps if necessary - user.UserCaps = generateUserCaps(user) - if user.UserCaps != userConfig.UserCaps { + // Update caps, if necessary + log.NamedTrace(nsName, logger, "user capabilities(id: %s, caps: %#v, user caps: %s, op mask: %s)", + liveUser.ID, liveUser.Caps, liveUser.UserCaps, liveUser.OpMask) + if targetUser.UserCaps != liveUser.UserCaps { // If they are no caps to be removed, the API will return an error "missing user capabilities" - if user.UserCaps != "" { - log.NamedTrace(nsName, logger, "remove capabilities %s from user %s", user.UserCaps, userConfig.ID) - _, err = r.objContext.AdminOpsClient.RemoveUserCap(r.opManagerContext, userConfig.ID, user.UserCaps) + if liveUser.UserCaps != "" { + log.NamedTrace(nsName, logger, "remove capabilities %s from user %s", liveUser.UserCaps, targetUser.ID) + _, err = r.objContext.AdminOpsClient.RemoveUserCap(r.opManagerContext, targetUser.ID, liveUser.UserCaps) if err != nil { - return errors.Wrapf(err, "failed to remove current ceph object user %q capabilities", userConfig.ID) + return errors.Wrapf(err, "failed to remove current ceph object user %q capabilities", targetUser.ID) } } - if userConfig.UserCaps != "" { - log.NamedTrace(nsName, logger, "set capabilities %s for user %s", userConfig.UserCaps, userConfig.ID) - _, err = r.objContext.AdminOpsClient.AddUserCap(r.opManagerContext, userConfig.ID, userConfig.UserCaps) + if targetUser.UserCaps != "" { + log.NamedTrace(nsName, logger, "set capabilities %s for user %s", targetUser.UserCaps, targetUser.ID) + _, err = r.objContext.AdminOpsClient.AddUserCap(r.opManagerContext, targetUser.ID, targetUser.UserCaps) if err != nil { - return errors.Wrapf(err, "failed to update ceph object user %q capabilities", userConfig.ID) + return errors.Wrapf(err, "failed to update ceph object user %q capabilities", targetUser.ID) } } logCreateOrUpdate = fmt.Sprintf("updated ceph object user %q", u.Name) @@ -472,18 +473,18 @@ func (r *ReconcileObjectStoreUser) createOrUpdateCephUser(u *cephv1.CephObjectSt return errors.Wrapf(err, "failed to set quotas for user %q", u.Name) } - if len(userConfig.Keys) == 0 { + if len(targetUser.Keys) == 0 { // use the keys already set on the user & remove all but one key - if len(user.Keys) == 0 { + if len(liveUser.Keys) == 0 { // something is wrong, there should be at least one key return errors.Wrapf(err, "no keys set for user %q", u.Name) } - userConfig.Keys = []admin.UserKeySpec{user.Keys[0]} - log.NamedDebug(nsName, logger, "reducing user %q keypairs to %v", u.Name, userConfig.Keys) + targetUser.Keys = []admin.UserKeySpec{liveUser.Keys[0]} + log.NamedDebug(nsName, logger, "reducing user %q keypairs to %v", u.Name, targetUser.Keys) } - if err := r.reconcileUserKeys(nsName, userConfig.Keys); err != nil { + if err := r.reconcileUserKeys(nsName, targetUser.Keys); err != nil { return errors.Wrapf(err, "failed to reconcile keys for user %q", u.Name) } log.NamedInfo(nsName, logger, "%s", logCreateOrUpdate) @@ -543,7 +544,7 @@ func userInNamespaceAllowed(requestedNamespace string, allowedNamespaces []strin return false } -func generateUserCaps(user admin.User) string { +func generateUserCaps(user *admin.User) string { var caps string for _, c := range user.Caps { caps += fmt.Sprintf("%s=%s;", c.Type, c.Perm) @@ -551,7 +552,7 @@ func generateUserCaps(user admin.User) string { return caps } -func generateUserConfig(user *cephv1.CephObjectStoreUser) *admin.User { +func generateUserConfig(user *cephv1.CephObjectStoreUser) (*admin.User, error) { // Set DisplayName to match Name if DisplayName is not set displayName := user.Spec.DisplayName if len(displayName) == 0 { @@ -622,7 +623,32 @@ func generateUserConfig(user *cephv1.CephObjectStoreUser) *admin.User { } } - return userConfig + nsName := opcontroller.NsName(user.Namespace, user.Name) + + var opMask string + if user.Spec.OpMask != nil && *user.Spec.OpMask != nil { + mask, err := opmask.FromSlice(*user.Spec.OpMask) + if err != nil { + return nil, errors.Wrapf(err, "invalid op-mask specified for user %q", user.Name) + } + opMask = mask.String() + + if opMask == "" { + log.NamedTrace(nsName, logger, "removing all operations from op-mask for user %q", user.Name) + } else { + log.NamedTrace(nsName, logger, "setting op-mask for user %q: %v", user.Name, user.Spec.OpMask) + } + } else { + // If no explicit opmask is set, default to read/write/delete permissions, which is the default that the RGW applies when creating a user via the admin ops API. + opMask = "read, write, delete" + + log.NamedTrace(nsName, logger, "setting default op-mask for user %q: %v", user.Name, opMask) + } + + userConfig.OpMask = opMask + userConfig.UserCaps = generateUserCaps(userConfig) + + return userConfig, nil } func generateCephUserSecretName(u *cephv1.CephObjectStoreUser) string { @@ -987,3 +1013,32 @@ func (r *ReconcileObjectStoreUser) generateUserKeySpec(user *cephv1.CephObjectSt return &keys, &referencedSecrets, nil } + +// Determine if the simple scalar fields, supported by CephObjectStoreUser, of two `admin.User`s are in sync. +// This is intended to determine if an update is needed/possible via AdminOpsClient.ModifyUser(). +// AdminOpsClient.ModifyUser() does not currently support updating all types of user configuration and only supports: +// +// []string{"uid", "display-name", "default-placement", "email", "generate-key", "access-key", "secret-key", "key-type", "max-buckets", "suspended", "op-mask"})) +// +// While this method does have support for updating keys, it is unable to handle multiple keys, so key reconciliation needs to be handled separately. +// There is also no support for updating capabilities via ModifyUser(). +func isUserSync(targetUser, liveUser *admin.User) bool { + if targetUser.DisplayName != liveUser.DisplayName { + return false + } + + if targetUser.MaxBuckets == nil && liveUser.MaxBuckets != nil || + targetUser.MaxBuckets != nil && liveUser.MaxBuckets == nil { + return false + } + + if targetUser.MaxBuckets != nil && liveUser.MaxBuckets != nil && *targetUser.MaxBuckets != *liveUser.MaxBuckets { + return false + } + + if targetUser.OpMask != liveUser.OpMask { + return false + } + + return true +} diff --git a/pkg/operator/ceph/object/user/controller_test.go b/pkg/operator/ceph/object/user/controller_test.go index ca28e03c9..2a9fec4ae 100644 --- a/pkg/operator/ceph/object/user/controller_test.go +++ b/pkg/operator/ceph/object/user/controller_test.go @@ -294,7 +294,7 @@ func TestCephObjectStoreUserController(t *testing.T) { newMultisiteAdminOpsCtxFunc = func(objContext *cephobject.Context, spec *cephv1.ObjectStoreSpec) (*cephobject.AdminOpsContext, error) { mockClient := &cephobject.MockClient{ MockDo: func(req *http.Request) (*http.Response, error) { - if (req.URL.RawQuery == "display-name=my-user&format=json&max-buckets=1000&uid=my-user" && req.Method == http.MethodPost && req.URL.Path == "rook-ceph-rgw-my-store.mycluster.svc/admin/user") || + if (req.URL.RawQuery == "display-name=my-user&format=json&max-buckets=1000&op-mask=read%2C%20write%2C%20delete&uid=my-user" && req.Method == http.MethodPost && req.URL.Path == "rook-ceph-rgw-my-store.mycluster.svc/admin/user") || (req.URL.RawQuery == "enabled=false&format=json&max-objects=-1&max-size=-1"a="a-type=user&uid=my-user" && req.Method == http.MethodPut && req.URL.Path == "rook-ceph-rgw-my-store.mycluster.svc/admin/user") || (req.URL.RawQuery == "format=json&uid=my-user" && req.Method == http.MethodGet && req.URL.Path == "rook-ceph-rgw-my-store.mycluster.svc/admin/user") { return &http.Response{ @@ -451,7 +451,7 @@ func TestCreateOrUpdateCephUser(t *testing.T) { if req.Method == http.MethodPost { if req.URL.RawQuery == "display-name=my-user&format=json&max-buckets=1000&uid=my-user" || - req.URL.RawQuery == "display-name=my-user&format=json&max-buckets=200&uid=my-user" || + req.URL.RawQuery == "display-name=my-user&format=json&max-buckets=200&op-mask=read%2C%20write%2C%20delete&uid=my-user" || req.URL.RawQuery == "display-name=my-user&format=json&max-buckets=1000&uid=my-user&user-caps=users%3Dread%3Bbuckets%3Dread%3B" || req.URL.RawQuery == "display-name=my-user&format=json&max-buckets=200&uid=my-user&user-caps=users%3Dread%3Bbuckets%3Dread%3B" { return &http.Response{ @@ -515,14 +515,16 @@ func TestCreateOrUpdateCephUser(t *testing.T) { t.Run("user without any Quotas or Capabilities", func(t *testing.T) { objectUser.Name = name - userConfig := generateUserConfig(objectUser) + userConfig, err := generateUserConfig(objectUser) + require.NoError(t, err) err = r.createOrUpdateCephUser(objectUser, userConfig) assert.NoError(t, err) }) t.Run("setting MaxBuckets for the user", func(t *testing.T) { objectUser.Spec.Quotas = &cephv1.ObjectUserQuotaSpec{MaxBuckets: &maxbucket} - userConfig := generateUserConfig(objectUser) + userConfig, err := generateUserConfig(objectUser) + require.NoError(t, err) err = r.createOrUpdateCephUser(objectUser, userConfig) assert.NoError(t, err) }) @@ -535,7 +537,8 @@ func TestCreateOrUpdateCephUser(t *testing.T) { Roles: "*", Info: "read, write", } - userConfig := generateUserConfig(objectUser) + userConfig, err := generateUserConfig(objectUser) + require.NoError(t, err) err = r.createOrUpdateCephUser(objectUser, userConfig) assert.NoError(t, err) }) @@ -544,35 +547,35 @@ func TestCreateOrUpdateCephUser(t *testing.T) { t.Run("setting MaxObjects for the user", func(t *testing.T) { objectUser.Spec.Capabilities = nil objectUser.Spec.Quotas = &cephv1.ObjectUserQuotaSpec{MaxObjects: &maxobject} - userConfig := generateUserConfig(objectUser) + userConfig, err := generateUserConfig(objectUser) require.NoError(t, err) err = r.createOrUpdateCephUser(objectUser, userConfig) assert.NoError(t, err) }) t.Run("setting MaxSize for the user", func(t *testing.T) { objectUser.Spec.Quotas = &cephv1.ObjectUserQuotaSpec{MaxSize: &maxsize} - userConfig := generateUserConfig(objectUser) + userConfig, err := generateUserConfig(objectUser) require.NoError(t, err) err = r.createOrUpdateCephUser(objectUser, userConfig) assert.NoError(t, err) }) t.Run("resetting MaxSize and MaxObjects for the user", func(t *testing.T) { objectUser.Spec.Quotas = nil - userConfig := generateUserConfig(objectUser) + userConfig, err := generateUserConfig(objectUser) require.NoError(t, err) err = r.createOrUpdateCephUser(objectUser, userConfig) assert.NoError(t, err) }) t.Run("setting both MaxSize and MaxObjects for the user", func(t *testing.T) { objectUser.Spec.Quotas = &cephv1.ObjectUserQuotaSpec{MaxObjects: &maxobject, MaxSize: &maxsize} - userConfig := generateUserConfig(objectUser) + userConfig, err := generateUserConfig(objectUser) require.NoError(t, err) err = r.createOrUpdateCephUser(objectUser, userConfig) assert.NoError(t, err) }) t.Run("resetting MaxSize and MaxObjects again for the user", func(t *testing.T) { objectUser.Spec.Quotas = nil - userConfig := generateUserConfig(objectUser) + userConfig, err := generateUserConfig(objectUser) require.NoError(t, err) err = r.createOrUpdateCephUser(objectUser, userConfig) assert.NoError(t, err) @@ -586,9 +589,77 @@ func TestCreateOrUpdateCephUser(t *testing.T) { Info: "read, write", } objectUser.Spec.Quotas = &cephv1.ObjectUserQuotaSpec{MaxBuckets: &maxbucket, MaxObjects: &maxobject, MaxSize: &maxsize} - userConfig := generateUserConfig(objectUser) + userConfig, err := generateUserConfig(objectUser) require.NoError(t, err) err = r.createOrUpdateCephUser(objectUser, userConfig) assert.NoError(t, err) }) } + +func TestIsUserSync(t *testing.T) { + t.Run("DisplayName same", func(t *testing.T) { + a := admin.User{ + DisplayName: "Alice", + } + b := admin.User{ + DisplayName: "Alice", + } + + assert.True(t, isUserSync(&a, &b)) + }) + + t.Run("DisplayName different", func(t *testing.T) { + a := admin.User{ + DisplayName: "Apollo", + } + b := admin.User{ + DisplayName: "Zeus", + } + + assert.False(t, isUserSync(&a, &b)) + }) + + t.Run("MaxBuckets same", func(t *testing.T) { + a := admin.User{ + MaxBuckets: &[]int{500}[0], + } + b := admin.User{ + MaxBuckets: &[]int{500}[0], + } + + assert.True(t, isUserSync(&a, &b)) + }) + + t.Run("MaxBuckets different", func(t *testing.T) { + a := admin.User{ + MaxBuckets: &[]int{400}[0], + } + b := admin.User{ + MaxBuckets: &[]int{600}[0], + } + + assert.False(t, isUserSync(&a, &b)) + }) + + t.Run("OpMask same", func(t *testing.T) { + a := admin.User{ + OpMask: "read, write, delete", + } + b := admin.User{ + OpMask: "read, write, delete", + } + + assert.True(t, isUserSync(&a, &b)) + }) + + t.Run("OpMask different", func(t *testing.T) { + a := admin.User{ + OpMask: "read, write, delete", + } + b := admin.User{ + OpMask: "read", + } + + assert.False(t, isUserSync(&a, &b)) + }) +} diff --git a/pkg/operator/ceph/object/user/opmask/opmask.go b/pkg/operator/ceph/object/user/opmask/opmask.go new file mode 100644 index 000000000..c10c2eb3e --- /dev/null +++ b/pkg/operator/ceph/object/user/opmask/opmask.go @@ -0,0 +1,78 @@ +/* +Copyright 2026 The Rook Authors. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package opmask + +import ( + "slices" + "strings" + + "github.com/pkg/errors" + + cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" +) + +type OpMask struct { + read bool + write bool + delete bool +} + +// Convert a slice of strings of operation names into an OpMask. +func FromSlice(ops []cephv1.ObjectUserOpMask) (*OpMask, error) { + mask := &OpMask{} + + if ops == nil { + return nil, errors.New("nil slice provided") + } + + if len(ops) == 0 { + return mask, nil + } + + if slices.Contains(ops, "read") { + mask.read = true + } + if slices.Contains(ops, "write") { + mask.write = true + } + if slices.Contains(ops, "delete") { + mask.delete = true + } + + return mask, nil +} + +// Format the mask as a string with the operators normalized into the same order as is returned by the admin api and radosgw-admin. +func (mask *OpMask) String() string { + // When no operations are set, the admin api and radosgw-admin return "". + if !mask.read && !mask.write && !mask.delete { + return "" + } + + ops := []string{} + if mask.read { + ops = append(ops, "read") + } + if mask.write { + ops = append(ops, "write") + } + if mask.delete { + ops = append(ops, "delete") + } + + return strings.Join(ops, ", ") +} diff --git a/pkg/operator/ceph/object/user/opmask/opmask_test.go b/pkg/operator/ceph/object/user/opmask/opmask_test.go new file mode 100644 index 000000000..fbd09335a --- /dev/null +++ b/pkg/operator/ceph/object/user/opmask/opmask_test.go @@ -0,0 +1,73 @@ +/* +Copyright 2026 The Rook Authors. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package opmask + +import ( + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + + cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" +) + +func TestFromSlice(t *testing.T) { + examples := []struct { + enum []cephv1.ObjectUserOpMask + mask *OpMask + }{ + {[]cephv1.ObjectUserOpMask{"read", "write", "delete"}, &OpMask{read: true, write: true, delete: true}}, + {[]cephv1.ObjectUserOpMask{"delete", "read", "write"}, &OpMask{read: true, write: true, delete: true}}, + {[]cephv1.ObjectUserOpMask{"read"}, &OpMask{read: true}}, + {[]cephv1.ObjectUserOpMask{"write"}, &OpMask{write: true}}, + {[]cephv1.ObjectUserOpMask{"delete"}, &OpMask{delete: true}}, + {[]cephv1.ObjectUserOpMask{"read", "write"}, &OpMask{read: true, write: true}}, + {[]cephv1.ObjectUserOpMask{"read", "delete"}, &OpMask{read: true, delete: true}}, + {[]cephv1.ObjectUserOpMask{"write", "delete"}, &OpMask{write: true, delete: true}}, + {[]cephv1.ObjectUserOpMask{}, &OpMask{}}, + } + + for _, ex := range examples { + t.Run(fmt.Sprintf("parse slice %v", ex.enum), func(t *testing.T) { + mask, err := FromSlice(ex.enum) + assert.NoError(t, err) + assert.Equal(t, ex.mask, mask) + }) + } +} + +func TestString(t *testing.T) { + examples := []struct { + mask *OpMask + str string + }{ + {&OpMask{read: true, write: true, delete: true}, "read, write, delete"}, + {&OpMask{read: true}, "read"}, + {&OpMask{write: true}, "write"}, + {&OpMask{delete: true}, "delete"}, + {&OpMask{read: true, write: true}, "read, write"}, + {&OpMask{read: true, delete: true}, "read, delete"}, + {&OpMask{write: true, delete: true}, "write, delete"}, + {&OpMask{}, ""}, + } + + for _, ex := range examples { + t.Run(fmt.Sprintf("stringify %v", ex.mask), func(t *testing.T) { + assert.Equal(t, ex.mask.String(), ex.str) + }) + } +} diff --git a/tests/integration/ceph_object_test.go b/tests/integration/ceph_object_test.go index f20599531..cfa42caa7 100644 --- a/tests/integration/ceph_object_test.go +++ b/tests/integration/ceph_object_test.go @@ -42,6 +42,7 @@ import ( "github.com/rook/rook/tests/framework/utils" "github.com/rook/rook/tests/integration/object/bucketowner" topickafka "github.com/rook/rook/tests/integration/object/topic/kafka" + "github.com/rook/rook/tests/integration/object/user/opmask" "github.com/rook/rook/tests/integration/object/user/userkeys" ) @@ -159,6 +160,7 @@ func runObjectE2ETest(helper *clients.TestClient, k8sh *utils.K8sHelper, install bucketowner.TestObjectBucketClaimBucketOwner(s.T(), k8sh, installer, logger, tlsEnable) userkeys.TestObjectStoreUserKeys(s.T(), k8sh, installer, logger, tlsEnable) topickafka.TestBucketTopicKafka(s.T(), k8sh, installer, logger, tlsEnable) + opmask.TestObjectStoreUserOpMask(s.T(), k8sh, installer, logger, tlsEnable) bucketNotificationTestStoreName := "bucket-notification-" + storeName createCephObjectStore(s.T(), helper, k8sh, installer, namespace, bucketNotificationTestStoreName, 1, tlsEnable, swiftAndKeystone) diff --git a/tests/integration/object/user/opmask/opmask.go b/tests/integration/object/user/opmask/opmask.go new file mode 100644 index 000000000..157ba7df8 --- /dev/null +++ b/tests/integration/object/user/opmask/opmask.go @@ -0,0 +1,285 @@ +/* +Copyright 2026 The Rook Authors. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package opmask + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/ceph/go-ceph/rgw/admin" + "github.com/coreos/pkg/capnslog" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" + + cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" + "github.com/rook/rook/tests/framework/installer" + "github.com/rook/rook/tests/framework/utils" + utiladmin "github.com/rook/rook/tests/integration/object/util/admin" +) + +var ( + defaultName = "test-cosu-opmask" + + ns = &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + Name: defaultName, + }, + } + + objectStore = &cephv1.CephObjectStore{ + ObjectMeta: metav1.ObjectMeta{ + Name: defaultName, + // the CephObjectstore must be in the same ns as the CephCluster + Namespace: "object-ns", + }, + Spec: cephv1.ObjectStoreSpec{ + MetadataPool: cephv1.PoolSpec{ + Replicated: cephv1.ReplicatedSpec{ + Size: 1, + RequireSafeReplicaSize: false, + }, + }, + DataPool: cephv1.PoolSpec{ + Replicated: cephv1.ReplicatedSpec{ + Size: 1, + RequireSafeReplicaSize: false, + }, + }, + Gateway: cephv1.GatewaySpec{ + Port: 80, + Instances: 1, + }, + AllowUsersInNamespaces: []string{ns.Name}, + }, + } + + objectStoreSvc = &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: objectStore.Name, + Namespace: objectStore.Namespace, + }, + Spec: corev1.ServiceSpec{ + Selector: map[string]string{ + "app": "rook-ceph-rgw", + "rook_cluster": objectStore.Namespace, + "rook_object_store": objectStore.Name, + }, + Ports: []corev1.ServicePort{ + { + Name: "http", + Port: 80, + Protocol: corev1.ProtocolTCP, + TargetPort: intstr.FromInt(8080), + }, + }, + SessionAffinity: corev1.ServiceAffinityNone, + Type: corev1.ServiceTypeNodePort, + }, + } + + osu1 = cephv1.CephObjectStoreUser{ + ObjectMeta: metav1.ObjectMeta{ + Name: defaultName + "-user1", + Namespace: ns.Name, + }, + Spec: cephv1.ObjectStoreUserSpec{ + Store: objectStore.Name, + ClusterNamespace: objectStore.Namespace, + }, + } +) + +func TestObjectStoreUserOpMask(t *testing.T, k8sh *utils.K8sHelper, installer *installer.CephInstaller, logger *capnslog.PackageLogger, tlsEnable bool) { + t.Run("ObjectStoreUser keys", func(t *testing.T) { + if tlsEnable { + // Skip testing with and without TLS to reduce test time + t.Skip("skipping test for TLS enabled clusters") + } + + var adminClient *admin.API + ctx := context.TODO() + + t.Run(fmt.Sprintf("create ns %q", ns.Name), func(t *testing.T) { + _, err := k8sh.Clientset.CoreV1().Namespaces().Create(ctx, ns, metav1.CreateOptions{}) + require.NoError(t, err) + }) + + t.Run(fmt.Sprintf("create CephObjectStore %q", objectStore.Name), func(t *testing.T) { + objectStore, err := k8sh.RookClientset.CephV1().CephObjectStores(objectStore.Namespace).Create(ctx, objectStore, metav1.CreateOptions{}) + require.NoError(t, err) + + osReady := utils.Retry(180, time.Second, "CephObjectStore is Ready", func() bool { + liveOs, err := k8sh.RookClientset.CephV1().CephObjectStores(objectStore.Namespace).Get(ctx, objectStore.Name, metav1.GetOptions{}) + if err != nil { + return false + } + + if liveOs.Status == nil { + return false + } + + return liveOs.Status.Phase == cephv1.ConditionReady + }) + require.True(t, osReady) + }) + + t.Run(fmt.Sprintf("create svc %q", objectStoreSvc.Name), func(t *testing.T) { + _, err := k8sh.Clientset.CoreV1().Services(objectStore.Namespace).Create(ctx, objectStoreSvc, metav1.CreateOptions{}) + require.NoError(t, err) + }) + + t.Run(fmt.Sprintf("create CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { + _, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Create(ctx, &osu1, metav1.CreateOptions{}) + require.NoError(t, err) + + // user creation may be slow right after rgw start up + osuReady := utils.Retry(120, time.Second, "CephObjectStoreUser is Ready", func() bool { + liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) + if err != nil { + return false + } + + if liveOsu.Status == nil { + return false + } + + return liveOsu.Status.Phase == string(cephv1.ConditionReady) + }) + require.True(t, osuReady) + }) + + // wait for cosu user to be ready so we know rgw admin api is ready + t.Run("setup rgw admin api client", func(t *testing.T) { + var err error + adminClient, err = utiladmin.NewAdminClient(objectStore, installer, k8sh, tlsEnable) + require.NoError(t, err) + }) + + t.Run(fmt.Sprintf("verify op_mask set on CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { + liveUser, err := adminClient.GetUser(ctx, admin.User{ID: osu1.Name}) + require.NoError(t, err) + assert.Equal(t, "read, write, delete", liveUser.OpMask) + }) + + t.Run(fmt.Sprintf("update op-mask on CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { + liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) + require.NoError(t, err) + + liveOsu.Spec.OpMask = &[]cephv1.ObjectUserOpMask{"read"} + + _, err = k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Update(ctx, liveOsu, metav1.UpdateOptions{}) + require.NoError(t, err) + }) + + t.Run(fmt.Sprintf("verify op_mask set on CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { + converged := utils.Retry(30, time.Second, "OpMask updated", func() bool { + liveUser, err := adminClient.GetUser(ctx, admin.User{ID: osu1.Name}) + if err != nil { + return false + } + + return liveUser.OpMask == "read" + }) + + assert.True(t, converged) + }) + + t.Run(fmt.Sprintf("remove op-mask on CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { + liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) + require.NoError(t, err) + + liveOsu.Spec.OpMask = nil + + _, err = k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Update(ctx, liveOsu, metav1.UpdateOptions{}) + require.NoError(t, err) + }) + + t.Run(fmt.Sprintf("verify op_mask set on CephObjectStoreUser %q returns to default", osu1.Name), func(t *testing.T) { + converged := utils.Retry(30, time.Second, "OpMask updated", func() bool { + liveUser, err := adminClient.GetUser(ctx, admin.User{ID: osu1.Name}) + if err != nil { + return false + } + + return liveUser.OpMask == "read, write, delete" + }) + + assert.True(t, converged) + }) + + t.Run(fmt.Sprintf("set empty op-mask (remove all ops) on CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { + liveOsu, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) + require.NoError(t, err) + + liveOsu.Spec.OpMask = &[]cephv1.ObjectUserOpMask{} + + _, err = k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Update(ctx, liveOsu, metav1.UpdateOptions{}) + require.NoError(t, err) + }) + + t.Run(fmt.Sprintf("verify op_mask set on CephObjectStoreUser %q is empty", osu1.Name), func(t *testing.T) { + converged := utils.Retry(30, time.Second, "OpMask updated", func() bool { + liveUser, err := adminClient.GetUser(ctx, admin.User{ID: osu1.Name}) + if err != nil { + return false + } + + return liveUser.OpMask == "" + }) + + assert.True(t, converged) + }) + + t.Run(fmt.Sprintf("delete CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { + err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Delete(ctx, osu1.Name, metav1.DeleteOptions{}) + require.NoError(t, err) + + absent := utils.Retry(40, time.Second, "CephObjectStoreUser is absent", func() bool { + _, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).Get(ctx, osu1.Name, metav1.GetOptions{}) + return err != nil + }) + assert.True(t, absent) + }) + + t.Run(fmt.Sprintf("no CephObjectStoreUsers in ns %q", ns.Name), func(t *testing.T) { + osus, err := k8sh.RookClientset.CephV1().CephObjectStoreUsers(ns.Name).List(ctx, metav1.ListOptions{}) + require.NoError(t, err) + + assert.Len(t, osus.Items, 0) + }) + + t.Run(fmt.Sprintf("delete svc %q", objectStoreSvc.Name), func(t *testing.T) { + err := k8sh.Clientset.CoreV1().Services(objectStore.Namespace).Delete(ctx, objectStoreSvc.Name, metav1.DeleteOptions{}) + require.NoError(t, err) + }) + + t.Run(fmt.Sprintf("delete CephObjectStore %q", objectStore.Name), func(t *testing.T) { + err := k8sh.RookClientset.CephV1().CephObjectStores(objectStore.Namespace).Delete(ctx, objectStore.Name, metav1.DeleteOptions{}) + require.NoError(t, err) + }) + + t.Run(fmt.Sprintf("delete ns %q", ns.Name), func(t *testing.T) { + err := k8sh.Clientset.CoreV1().Namespaces().Delete(ctx, ns.Name, metav1.DeleteOptions{}) + require.NoError(t, err) + }) + }) +}