object: add ObjectStoreUserSpec.OpMask field

Signed-off-by: Joshua Hoblitt <josh@hoblitt.com>
(cherry picked from commit 3ef0c90ef8)
This commit is contained in:
Joshua Hoblitt authored and Mergify committed 2026-02-11 17:11:07 +00:00
1 parent 7be1e02dca
commit 0e25b2089f
12 files changed
+694 -43

No files matched your search

@@ -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`
+33
View File
@@ -2204,6 +2204,20 @@ string
<p>The namespace where the parent CephCluster and CephObjectStore are found</p>
</td>
</tr>
<tr>
<td>
<code>opMask</code><br/>
<em>
<a href="#ceph.rook.io/v1.[]github.com/rook/rook/pkg/apis/ceph.rook.io/v1.ObjectUserOpMask">
[]github.com/rook/rook/pkg/apis/ceph.rook.io/v1.ObjectUserOpMask
</a>
</em>
</td>
<td>
<em>(Optional)</em>
<p>The op-mask of the user.</p>
</td>
</tr>
</table>
</td>
</tr>
@@ -12131,6 +12145,20 @@ string
<p>The namespace where the parent CephCluster and CephObjectStore are found</p>
</td>
</tr>
<tr>
<td>
<code>opMask</code><br/>
<em>
<a href="#ceph.rook.io/v1.[]github.com/rook/rook/pkg/apis/ceph.rook.io/v1.ObjectUserOpMask">
[]github.com/rook/rook/pkg/apis/ceph.rook.io/v1.ObjectUserOpMask
</a>
</em>
</td>
<td>
<em>(Optional)</em>
<p>The op-mask of the user.</p>
</td>
</tr>
</tbody>
</table>
<h3 id="ceph.rook.io/v1.ObjectStoreUserStatus">ObjectStoreUserStatus
@@ -12453,6 +12481,11 @@ Kubernetes core/v1.SecretKeySelector
</tr>
</tbody>
</table>
<h3 id="ceph.rook.io/v1.ObjectUserOpMask">ObjectUserOpMask
(<code>string</code> alias)</h3>
<div>
<p>Internally, RGW labels &ldquo;operations&rdquo; on persistent state as <code>RGW_OP_TYPE_READ</code> (<code>read</code>), <code>RGW_OP_TYPE_WRITE</code> (<code>write</code>), or <code>RGW_OP_TYPE_DELETE</code> (<code>delete</code>). All RGW users have an &ldquo;operation mask&rdquo;, which does not function as mask or filter as is typically implied by the word &ldquo;mask&rdquo;, but as a set of allowed or permissible &ldquo;operation&rdquo; types the user is able to perform. The &ldquo;operation mask&rdquo; 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 <strong>both</strong> the <code>read</code> &ldquo;op mask&rdquo; bit and an IAM/bucket policy that allows <code>s3:GetObject</code>. The default operations allowed are <code>read</code>, <code>write</code>, and <code>delete</code>. Setting the value to <code>[]</code> (an empty YAML sequence) causes all &ldquo;operations&rdquo; in the mask to be removed, meaning that the user will not be able to perform any operations. These operation masks are supported: <code>read</code>, <code>write</code>, <code>delete</code></p>
</div>
<h3 id="ceph.rook.io/v1.ObjectUserQuotaSpec">ObjectUserQuotaSpec
</h3>
<p>
@@ -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
+13
View File
@@ -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
+11
View File
@@ -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
@@ -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
}
+87 -32
View File
@@ -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 == "<none>" {
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
}
@@ -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&quota=&quota-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))
})
}
@@ -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 "<none>".
if !mask.read && !mask.write && !mask.delete {
return "<none>"
}
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, ", ")
}
@@ -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{}, "<none>"},
}
for _, ex := range examples {
t.Run(fmt.Sprintf("stringify %v", ex.mask), func(t *testing.T) {
assert.Equal(t, ex.mask.String(), ex.str)
})
}
}
+2
View File
@@ -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)
@@ -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 == "<none>"
})
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)
})
})
}