diff --git a/deploy/charts/rook-ceph/templates/resources.yaml b/deploy/charts/rook-ceph/templates/resources.yaml
index 4a9ba1e33081..2d28ce4bfcc4 100644
--- a/deploy/charts/rook-ceph/templates/resources.yaml
+++ b/deploy/charts/rook-ceph/templates/resources.yaml
@@ -10053,6 +10053,34 @@ spec:
                   items:
                     type: string
                   type: array
+                auth:
+                  description: The authentication configuration
+                  nullable: true
+                  properties:
+                    keystone:
+                      description: KeystoneSpec represents the Keystone authentication configuration of a Ceph Object Store Gateway
+                      nullable: true
+                      properties:
+                        acceptedRoles:
+                          items:
+                            type: string
+                          type: array
+                        implicitTenants:
+                          type: string
+                        revocationInterval:
+                          type: integer
+                        serviceUserSecretName:
+                          type: string
+                        tokenCacheSize:
+                          type: integer
+                        url:
+                          type: string
+                      required:
+                        - acceptedRoles
+                        - serviceUserSecretName
+                        - url
+                      type: object
+                  type: object
                 dataPool:
                   description: The data pool settings
                   nullable: true
@@ -11294,6 +11322,31 @@ spec:
                 preservePoolsOnDelete:
                   description: Preserve pools on object store deletion
                   type: boolean
+                protocols:
+                  description: The protocol specification
+                  nullable: true
+                  properties:
+                    s3:
+                      description: The spec for S3
+                      nullable: true
+                      properties:
+                        authUseKeystone:
+                          type: boolean
+                        enabled:
+                          type: boolean
+                      type: object
+                    swift:
+                      description: The spec for S3
+                      nullable: true
+                      properties:
+                        accountInUrl:
+                          type: boolean
+                        urlPrefix:
+                          type: string
+                        versioningEnabled:
+                          type: boolean
+                      type: object
+                  type: object
                 security:
                   description: Security represents security settings
                   nullable: true
@@ -11619,6 +11672,19 @@ spec:
                 store:
                   description: The store the user will be created in
                   type: string
+                subUsers:
+                  items:
+                    properties:
+                      access:
+                        type: string
+                      name:
+                        type: string
+                    required:
+                      - access
+                      - name
+                    type: object
+                  nullable: true
+                  type: array
               type: object
             status:
               description: ObjectStoreUserStatus represents the status Ceph Object Store Gateway User
diff --git a/deploy/examples/crds.yaml b/deploy/examples/crds.yaml
index 5b3783a8490b..323cbb0d191d 100644
--- a/deploy/examples/crds.yaml
+++ b/deploy/examples/crds.yaml
@@ -10044,6 +10044,34 @@ spec:
                   items:
                     type: string
                   type: array
+                auth:
+                  description: The authentication configuration
+                  nullable: true
+                  properties:
+                    keystone:
+                      description: KeystoneSpec represents the Keystone authentication configuration of a Ceph Object Store Gateway
+                      nullable: true
+                      properties:
+                        acceptedRoles:
+                          items:
+                            type: string
+                          type: array
+                        implicitTenants:
+                          type: string
+                        revocationInterval:
+                          type: integer
+                        serviceUserSecretName:
+                          type: string
+                        tokenCacheSize:
+                          type: integer
+                        url:
+                          type: string
+                      required:
+                        - acceptedRoles
+                        - serviceUserSecretName
+                        - url
+                      type: object
+                  type: object
                 dataPool:
                   description: The data pool settings
                   nullable: true
@@ -11285,6 +11313,31 @@ spec:
                 preservePoolsOnDelete:
                   description: Preserve pools on object store deletion
                   type: boolean
+                protocols:
+                  description: The protocol specification
+                  nullable: true
+                  properties:
+                    s3:
+                      description: The spec for S3
+                      nullable: true
+                      properties:
+                        authUseKeystone:
+                          type: boolean
+                        enabled:
+                          type: boolean
+                      type: object
+                    swift:
+                      description: The spec for S3
+                      nullable: true
+                      properties:
+                        accountInUrl:
+                          type: boolean
+                        urlPrefix:
+                          type: string
+                        versioningEnabled:
+                          type: boolean
+                      type: object
+                  type: object
                 security:
                   description: Security represents security settings
                   nullable: true
@@ -11609,6 +11662,19 @@ spec:
                 store:
                   description: The store the user will be created in
                   type: string
+                subUsers:
+                  items:
+                    properties:
+                      access:
+                        type: string
+                      name:
+                        type: string
+                    required:
+                      - access
+                      - name
+                    type: object
+                  nullable: true
+                  type: array
               type: object
             status:
               description: ObjectStoreUserStatus represents the status Ceph Object Store Gateway User
diff --git a/design/ceph/object/swift-and-keystone-integration.md b/design/ceph/object/swift-and-keystone-integration.md
index 27717ade5093..3a27938b8ab7 100644
--- a/design/ceph/object/swift-and-keystone-integration.md
+++ b/design/ceph/object/swift-and-keystone-integration.md
@@ -80,11 +80,11 @@ Annotations:
   options](https://docs.ceph.com/en/octopus/radosgw/config-ref/#keystone-settings),
   the corresponding RGW option is formed by prefixing it with
   `rgw_keystone_` and replacing upper case letters by their lower case
-  letter followed by an underscore. E.g. `tokenCacheSize` maps to
+  letter preceded by an underscore. E.g. `tokenCacheSize` maps to
   `rgw_keystone_token_cache_size`.
 * `[2]` These settings are required in the `keystone` section if
   present.
-* `[1]` The name of the secret containing the credentials for the
+* `[3]` The name of the secret containing the credentials for the
   service user account used by RGW. It has to be in the same namespace
   as the object store resource.
 
@@ -173,12 +173,12 @@ Annotations:
   options](https://docs.ceph.com/en/octopus/radosgw/config-ref/#swift-settings),
   the corresponding RGW option is formed by prefixing it with
   `rgw_swift_` and replacing upper case letters by their lower case
-  letter followed by an underscore. E.g. `urlPrefix` maps to
+  letter preceded by an underscore. E.g. `urlPrefix` maps to
   `rgw_swift_url_prefix`. They are optional. If not given, the defaults
   of the corresponding RGW option apply.
 
-The access to the Swift API is granted by creating a subuser of an RGW
-user. While commonly the access is granted via projects
+Access to the Swift API is granted by creating a subuser of an RGW
+user. While commonly access is granted via projects
 mapped from Keystone, explicit creation of subusers is supported by
 extending the `cephobjectstoreuser` resource with a new optional section
 `spec.subUsers`:
diff --git a/pkg/apis/ceph.rook.io/v1/spec_test.go b/pkg/apis/ceph.rook.io/v1/spec_test.go
index 3bbbe6142989..2875bf5520a5 100644
--- a/pkg/apis/ceph.rook.io/v1/spec_test.go
+++ b/pkg/apis/ceph.rook.io/v1/spec_test.go
@@ -92,3 +92,104 @@ storage:
 
 	assert.Equal(t, expectedSpec, clusterSpec)
 }
+
+func newTrue() *bool {
+	t := true
+	return &t
+}
+
+func newFalse() *bool {
+	t := false
+	return &t
+}
+
+func newInt(val int) *int {
+	return &val
+}
+
+func newString(val string) *string {
+	return &val
+}
+
+func TestObjectStoreSpecMarhsalSwiftAndKeystone(t *testing.T) {
+	// Assert that the new ObjectStoreSpec fields specified in <design/ceph/object/swift-and-keystone-integration.md> are correctly parsed
+	specYaml := []byte(`
+auth:
+  keystone:
+    url: https://keystone:5000/
+    acceptedRoles: ["_member_", "service", "admin"]
+    implicitTenants: swift
+    tokenCacheSize: 1000
+    revocationInterval: 1200
+    serviceUserSecretName: rgw-service-user
+protocols:
+  swift:
+    accountInUrl: true
+    urlPrefix: /example
+    versioningEnabled: false
+  s3:
+    enabled: false
+    authUseKeystone: true
+`)
+	rawJSON, err := yaml.ToJSON(specYaml)
+	assert.Nil(t, err)
+	fmt.Printf("rawJSON: %s\n", string(rawJSON))
+
+	// unmarshal the JSON into a strongly typed storage spec object
+	var objectStoreSpec ObjectStoreSpec
+	err = json.Unmarshal(rawJSON, &objectStoreSpec)
+	assert.Nil(t, err)
+
+	// the unmarshalled storage spec should equal the expected spec below
+	expectedSpec := ObjectStoreSpec{
+		Auth: AuthSpec{
+			Keystone: &KeystoneSpec{
+				Url:                   "https://keystone:5000/",
+				AcceptedRoles:         []string{"_member_", "service", "admin"},
+				ImplicitTenants:       "swift",
+				TokenCacheSize:        newInt(1000),
+				RevocationInterval:    newInt(1200),
+				ServiceUserSecretName: "rgw-service-user",
+			},
+		},
+		Protocols: ProtocolSpec{
+			S3: &S3Spec{
+				Enabled:         newFalse(),
+				AuthUseKeystone: newTrue(),
+			},
+			Swift: &SwiftSpec{
+				AccountInUrl:      newTrue(),
+				UrlPrefix:         newString("/example"),
+				VersioningEnabled: newFalse(),
+			},
+		},
+	}
+
+	assert.Equal(t, expectedSpec, objectStoreSpec)
+}
+
+func TestObjectStoreUserSpecMarhsalSubuser(t *testing.T) {
+	// Assert that the new ObjectStoreUserSpec fields specified in <design/ceph/object/swift-and-keystone-integration.md> parse
+	specYaml := []byte(`
+subUsers:
+- name: swift
+  access: full
+`)
+	rawJSON, err := yaml.ToJSON(specYaml)
+	assert.Nil(t, err)
+	fmt.Printf("rawJSON: %s\n", string(rawJSON))
+
+	// unmarshal the JSON into a strongly typed storage spec object
+	var objectStoreUserSpec ObjectStoreUserSpec
+	err = json.Unmarshal(rawJSON, &objectStoreUserSpec)
+	assert.Nil(t, err)
+
+	// the unmarshalled storage spec should equal the expected spec below
+	expectedSpec := ObjectStoreUserSpec{
+		Subusers: []SubuserSpec{
+			{Name: "swift", Access: "full"},
+		},
+	}
+
+	assert.Equal(t, expectedSpec, objectStoreUserSpec)
+}
diff --git a/pkg/apis/ceph.rook.io/v1/types.go b/pkg/apis/ceph.rook.io/v1/types.go
index baf337effa92..0c3598c1f0c8 100755
--- a/pkg/apis/ceph.rook.io/v1/types.go
+++ b/pkg/apis/ceph.rook.io/v1/types.go
@@ -1379,6 +1379,16 @@ type ObjectStoreSpec struct {
 	// +nullable
 	Gateway GatewaySpec `json:"gateway"`
 
+	// The protocol specification
+	// +optional
+	// +nullable
+	Protocols ProtocolSpec `json:"protocols,omitempty"`
+
+	// The authentication configuration
+	// +optional
+	// +nullable
+	Auth AuthSpec `json:"auth,omitempty"`
+
 	// The multisite info
 	// +optional
 	// +nullable
@@ -1528,6 +1538,59 @@ type EndpointAddress struct {
 	Hostname string `json:"hostname,omitempty" protobuf:"bytes,3,opt,name=hostname"`
 }
 
+// ProtocolSpec represents a Ceph Object Store protocol specification
+type ProtocolSpec struct {
+	// The spec for S3
+	// +optional
+	// +nullable
+	S3 *S3Spec `json:"s3,omitempty"`
+
+	// The spec for S3
+	// +optional
+	// +nullable
+	Swift *SwiftSpec `json:"swift"`
+}
+
+// S3Spec represents Ceph Object Store specification for the S3 API
+type S3Spec struct {
+	Enabled         *bool `json:"enabled,omitempty"`
+	AuthUseKeystone *bool `json:"authUseKeystone,omitempty"`
+}
+
+// S3Spec represents Ceph Object Store specification for the Swift API
+type SwiftSpec struct {
+	AccountInUrl      *bool   `json:"accountInUrl,omitempty"`
+	UrlPrefix         *string `json:"urlPrefix,omitempty"`
+	VersioningEnabled *bool   `json:"versioningEnabled,omitempty"`
+}
+
+// AuthSpec represents the authentication protocol configuration of a Ceph Object Store Gateway
+type AuthSpec struct {
+	// +optional
+	// +nullable
+	Keystone *KeystoneSpec `json:"keystone,omitempty"`
+}
+
+// KeystoneSpec represents the Keystone authentication configuration of a Ceph Object Store Gateway
+type KeystoneSpec struct {
+	Url                   string                `json:"url"`
+	ServiceUserSecretName string                `json:"serviceUserSecretName"`
+	AcceptedRoles         []string              `json:"acceptedRoles"`
+	ImplicitTenants       ImplicitTenantSetting `json:"implicitTenants,omitempty"`
+	TokenCacheSize        *int                  `json:"tokenCacheSize,omitempty"`
+	RevocationInterval    *int                  `json:"revocationInterval,omitempty"`
+}
+
+type ImplicitTenantSetting string
+
+const (
+	ImplicitTenantSwift   ImplicitTenantSetting = "swift"
+	ImplicitTenantS3      ImplicitTenantSetting = "s3"
+	ImplicitTenantTrue    ImplicitTenantSetting = "true"
+	ImplicitTenantFalse   ImplicitTenantSetting = "false"
+	ImplicitTenantDefault ImplicitTenantSetting = ""
+)
+
 // ZoneSpec represents a Ceph Object Store Gateway Zone specification
 type ZoneSpec struct {
 	// RGW Zone the Object Store is in
@@ -1615,6 +1678,9 @@ type ObjectStoreUserSpec struct {
 	// The namespace where the parent CephCluster and CephObjectStore are found
 	// +optional
 	ClusterNamespace string `json:"clusterNamespace,omitempty"`
+	// +optional
+	// +nullable
+	Subusers []SubuserSpec `json:"subUsers,omitemtpy"`
 }
 
 // Additional admin-level capabilities for the Ceph object store user
@@ -1702,6 +1768,21 @@ type ObjectUserQuotaSpec struct {
 	MaxObjects *int64 `json:"maxObjects,omitempty"`
 }
 
+type SubuserSpec struct {
+	Name   string     `json:"name"`
+	Access AccessSpec `json:"access"`
+}
+
+type AccessSpec string
+
+const (
+	AccessSpecFull      AccessSpec = "full"
+	AccessSpecRead      AccessSpec = "read"
+	AccessSpecWrite     AccessSpec = "write"
+	AccessSpecReadWrite AccessSpec = "readwrite"
+)
+
+// CephObjectRealm represents a Ceph Object Store Gateway Realm
 // +genclient
 // +genclient:noStatus
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
diff --git a/pkg/operator/ceph/config/monstore.go b/pkg/operator/ceph/config/monstore.go
index 96dd1f7f77d1..e5533cb46620 100644
--- a/pkg/operator/ceph/config/monstore.go
+++ b/pkg/operator/ceph/config/monstore.go
@@ -58,7 +58,7 @@ type Option struct {
 
 // SetIfChanged sets a config in the centralized mon configuration database if the config has
 // changed value.
-// https://docs.ceph.com/docs/master/rados/configuration/ceph-conf/#monitor-configuration-database
+// https://docs.ceph.com/en/latest/rados/configuration/ceph-conf/#monitor-configuration-database
 //
 // There is a bug through at least Ceph v18 where `ceph config get global <option>` does not work.
 // As a workaround it is possible to use `ceph config get client <option>` as long as the config
@@ -87,7 +87,7 @@ func (m *MonStore) SetIfChanged(who, option, value string) (bool, error) {
 }
 
 // Set sets a config in the centralized mon configuration database.
-// https://docs.ceph.com/docs/master/rados/configuration/ceph-conf/#monitor-configuration-database
+// https://docs.ceph.com/en/latest/rados/configuration/ceph-conf/#monitor-configuration-database
 func (m *MonStore) Set(who, option, value string) error {
 	logger.Infof("setting %q=%q=%q option to the mon configuration database", who, option, value)
 	args := []string{"config", "set", who, normalizeKey(option), value}
@@ -118,7 +118,7 @@ func (m *MonStore) Delete(who, option string) error {
 }
 
 // Get retrieves a config in the centralized mon configuration database.
-// https://docs.ceph.com/docs/master/rados/configuration/ceph-conf/#monitor-configuration-database
+// https://docs.ceph.com/en/latest/rados/configuration/ceph-conf/#monitor-configuration-database
 func (m *MonStore) Get(who, option string) (string, error) {
 	args := []string{"config", "get", who, normalizeKey(option)}
 	cephCmd := client.NewCephCommand(m.context, m.clusterInfo, args)
diff --git a/pkg/operator/ceph/object/config.go b/pkg/operator/ceph/object/config.go
index 8ebe1237a995..7663237aa35d 100644
--- a/pkg/operator/ceph/object/config.go
+++ b/pkg/operator/ceph/object/config.go
@@ -110,7 +110,72 @@ func (c *clusterConfig) generateKeyring(rgwConfig *rgwConfig) (string, error) {
 	return keyring, s.CreateOrUpdate(rgwConfig.ResourceName, keyring)
 }
 
+func mapKeystoneSecretToConfig(cfg map[string]string, secret *v1.Secret) (map[string]string, error) {
+
+	data := make(map[string]string)
+	for key, value := range secret.Data {
+
+		logger.Debugf("keystone secret %s => %s", key, value)
+		data[key] = string(value[:])
+
+	}
+
+	authType, ok := data["OS_AUTH_TYPE"]
+	if ok {
+		if authType != "password" {
+			return nil, errors.New(fmt.Sprintf("OS_AUTHTYPE %s is not supported. Only OS_AUTH_TYPE password is supported!", authType))
+		}
+	}
+
+	apiVersion, ok := data["OS_IDENTITY_API_VERSION"]
+	if ok {
+		if apiVersion != "3" {
+			return nil, errors.New(fmt.Sprintf("OS_IDENTITY_API_VERSION %s is not supported! Only OS_IDENTITY_API_VERSION 3 is supported!", apiVersion))
+		}
+	}
+
+	projectDomain, ok := data["OS_PROJECT_DOMAIN_NAME"]
+	if !ok {
+		return nil, errors.New("Missing OS_PROJECT_DOMAIN_NAME")
+	}
+
+	userDomain, ok := data["OS_USER_DOMAIN_NAME"]
+	if !ok {
+		return nil, errors.New("Missing OS_USER_DOMAIN_NAME")
+	}
+
+	if projectDomain != userDomain {
+		return nil, errors.New("The user domain name does not match the project domain name.")
+	}
+
+	project, ok := data["OS_PROJECT_NAME"]
+	if !ok {
+		return nil, errors.New("No OS_PROJECT_NAME set.")
+	}
+
+	username, ok := data["OS_USERNAME"]
+	if !ok {
+		return nil, errors.New("No OS_USERNAME set.")
+	}
+
+	password, ok := data["OS_PASSWORD"]
+	if !ok {
+		return nil, errors.New("No OS_PASSWORD set.")
+	}
+
+	// keystone api version 3 should be used today
+	cfg["rgw_keystone_api_version"] = "3"
+	cfg["rgw_keystone_admin_domain"] = userDomain
+	cfg["rgw_keystone_admin_project"] = project
+	cfg["rgw_keystone_admin_user"] = username
+	cfg["rgw_keystone_admin_password"] = password
+
+	return cfg, nil
+}
+
 func (c *clusterConfig) setFlagsMonConfigStore(rgwConfig *rgwConfig) error {
+	var err error
+
 	monStore := cephconfig.GetMonStore(c.context, c.clusterInfo)
 	who := generateCephXUser(rgwConfig.ResourceName)
 	configOptions := make(map[string]string)
@@ -126,9 +191,72 @@ func (c *clusterConfig) setFlagsMonConfigStore(rgwConfig *rgwConfig) error {
 	configOptions["rgw_zone"] = rgwConfig.Zone
 	configOptions["rgw_zonegroup"] = rgwConfig.ZoneGroup
 
+	if ks := rgwConfig.Auth.Keystone; ks != nil {
+
+		logger.Info("Configuring Authentication with keystone")
+
+		configOptions["rgw_keystone_url"] = ks.Url
+		configOptions["rgw_keystone_accepted_roles"] = strings.Join(ks.AcceptedRoles, ",")
+		if ks.ImplicitTenants != "" {
+			// XXX: where do we validate this?
+			configOptions["rgw_keystone_implicit_tenants"] = string(ks.ImplicitTenants)
+		}
+		if ks.TokenCacheSize != nil {
+			configOptions["rgw_keystone_token_cache_size"] = fmt.Sprintf("%d", *ks.TokenCacheSize)
+		}
+		if rgwConfig.KeystoneSecret == nil {
+			return errors.New("Cannot find keystone secret")
+		}
+
+		configOptions, err = mapKeystoneSecretToConfig(configOptions, rgwConfig.KeystoneSecret)
+		if err != nil {
+			logger.Infof("error mapping keystone secret %s to config: %s", rgwConfig.KeystoneSecret.Name, err)
+			return err
+		}
+	} else {
+		logger.Info("Authentication with keystone disabled")
+	}
+
+	s3disabled := false
+	if s3 := rgwConfig.Protocols.S3; s3 != nil {
+		if s3.Enabled != nil && !*s3.Enabled {
+			s3disabled = true
+		}
+
+		if s3.AuthUseKeystone != nil {
+			configOptions["rgw_s3_auth_use_keystone"] = fmt.Sprintf("%t", *s3.AuthUseKeystone)
+		}
+	}
+
+	if swift := rgwConfig.Protocols.Swift; swift != nil {
+		if swift.AccountInUrl != nil {
+			configOptions["rgw_swift_account_in_url"] = fmt.Sprintf("%t", *swift.AccountInUrl)
+		}
+		if swift.UrlPrefix != nil {
+			configOptions["rgw_swift_url_prefix"] = *swift.UrlPrefix
+		}
+		if swift.VersioningEnabled != nil {
+			configOptions["rgw_swift_versioning_enabled"] = fmt.Sprintf("%t", *swift.VersioningEnabled)
+		}
+	}
+
+	if s3disabled {
+		// XXX: how to handle enabled APIs? We only configure s3 and
+		// swift in the resource, `admin` is required for the operator to
+		// work, `swift_auth` is required to access swift without keystone
+		// – not sure about the additional APIs
+
+		// Swift was enabled so far already by default, so perhaps better
+		// not change that if someon relies on it.
+
+		configOptions["rgw_enabled_apis"] = "s3website, swift, swift_auth, admin, sts, iam, notifications"
+	}
+
 	for flag, val := range configOptions {
+		logger.Debugf("Set %s to %q on %q", flag, val, who)
 		err := monStore.Set(who, flag, val)
 		if err != nil {
+			logger.Debugf("failed to setup config option: %s to %q on %q: %s", flag, val, who, err)
 			return errors.Wrapf(err, "failed to set %q to %q on %q", flag, val, who)
 		}
 	}
diff --git a/pkg/operator/ceph/object/controller.go b/pkg/operator/ceph/object/controller.go
index a5b8333f062b..8082910e7e94 100644
--- a/pkg/operator/ceph/object/controller.go
+++ b/pkg/operator/ceph/object/controller.go
@@ -300,7 +300,7 @@ func (r *ReconcileCephObjectStore) reconcile(request reconcile.Request) (reconci
 		)
 		if err != nil {
 			if strings.Contains(err.Error(), opcontroller.UninitializedCephConfigError) {
-				logger.Info(opcontroller.OperatorNotInitializedMessage)
+				logger.Infof("%s: %s", opcontroller.OperatorNotInitializedMessage, err.Error())
 				return opcontroller.WaitForRequeueIfOperatorNotInitialized, *cephObjectStore, nil
 			}
 			return reconcile.Result{}, *cephObjectStore, errors.Wrap(err, "failed to detect running and desired ceph version")
@@ -327,7 +327,7 @@ func (r *ReconcileCephObjectStore) reconcile(request reconcile.Request) (reconci
 	// CREATE/UPDATE
 	_, err = r.reconcileCreateObjectStore(cephObjectStore, request.NamespacedName, cephCluster.Spec)
 	if err != nil && kerrors.IsNotFound(err) {
-		logger.Info(opcontroller.OperatorNotInitializedMessage)
+		logger.Infof("%s %s", opcontroller.OperatorNotInitializedMessage, err.Error())
 		return opcontroller.WaitForRequeueIfOperatorNotInitialized, *cephObjectStore, nil
 	} else if err != nil {
 		result, err := r.setFailedStatus(k8sutil.ObservedGenerationNotAvailable, request.NamespacedName, "failed to create object store deployments", err)
@@ -458,8 +458,17 @@ func (r *ReconcileCephObjectStore) reconcileCreateObjectStore(cephObjectStore *c
 			return r.setFailedStatus(k8sutil.ObservedGenerationNotAvailable, namespacedName, "failed to configure multisite for object store", err)
 		}
 
-		// Create or Update Store
-		err = cfg.createOrUpdateStore(realmName, zoneGroupName, zoneName)
+		// Retrieve the keystone secret if specified
+		var keystoneSecret *corev1.Secret
+		if ks := cephObjectStore.Spec.Auth.Keystone; ks != nil {
+			keystoneSecret, err = objContext.Context.Clientset.CoreV1().Secrets(objContext.clusterInfo.Namespace).Get(objContext.clusterInfo.Context, ks.ServiceUserSecretName, metav1.GetOptions{})
+			if err != nil {
+				return reconcile.Result{}, errors.Wrapf(err, "failed to get the keystone credential secret")
+			}
+		}
+
+		// Create or Update store
+		err = cfg.createOrUpdateStore(realmName, zoneGroupName, zoneName, keystoneSecret)
 		if err != nil {
 			return reconcile.Result{}, errors.Wrapf(err, "failed to create object store %q", cephObjectStore.Name)
 		}
@@ -559,7 +568,7 @@ func (r *ReconcileCephObjectStore) reconcileCOSIUser(cephObjectStore *cephv1.Cep
 	}
 
 	// Create COSI user secret
-	return ReconcileCephUserSecret(r.opManagerContext, r.client, r.scheme, cephObjectStore, &user, objCtx.Endpoint, cephObjectStore.Namespace, cephObjectStore.Name, cephObjectStore.Spec.Gateway.SSLCertificateRef)
+	return ReconcileCephUserSecrets(r.opManagerContext, r.client, r.scheme, cephObjectStore, &user, objCtx.Endpoint, cephObjectStore.Namespace, cephObjectStore.Name, cephObjectStore.Spec.Gateway.SSLCertificateRef)
 }
 
 func generateCOSIUserConfig() *admin.User {
diff --git a/pkg/operator/ceph/object/rgw.go b/pkg/operator/ceph/object/rgw.go
index 5e2f3d637167..7e2b082af716 100644
--- a/pkg/operator/ceph/object/rgw.go
+++ b/pkg/operator/ceph/object/rgw.go
@@ -62,6 +62,10 @@ type rgwConfig struct {
 	Realm        string
 	ZoneGroup    string
 	Zone         string
+
+	Auth           cephv1.AuthSpec
+	KeystoneSecret *v1.Secret
+	Protocols      cephv1.ProtocolSpec
 }
 
 var updateDeploymentAndWait = mon.UpdateCephDeploymentAndWait
@@ -70,10 +74,10 @@ var (
 	insecureSkipVerify = "insecureSkipVerify"
 )
 
-func (c *clusterConfig) createOrUpdateStore(realmName, zoneGroupName, zoneName string) error {
+func (c *clusterConfig) createOrUpdateStore(realmName, zoneGroupName, zoneName string, keystoneSecret *v1.Secret) error {
 	logger.Infof("creating object store %q in namespace %q", c.store.Name, c.store.Namespace)
 
-	if err := c.startRGWPods(realmName, zoneGroupName, zoneName); err != nil {
+	if err := c.startRGWPods(realmName, zoneGroupName, zoneName, keystoneSecret); err != nil {
 		return errors.Wrap(err, "failed to start rgw pods")
 	}
 
@@ -95,7 +99,7 @@ func (c *clusterConfig) createOrUpdateStore(realmName, zoneGroupName, zoneName s
 	return nil
 }
 
-func (c *clusterConfig) startRGWPods(realmName, zoneGroupName, zoneName string) error {
+func (c *clusterConfig) startRGWPods(realmName, zoneGroupName, zoneName string, keystoneSecret *v1.Secret) error {
 	// backward compatibility, triggered during updates
 	if c.store.Spec.Gateway.Instances < 1 {
 		// Set the minimum of at least one instance
@@ -127,11 +131,13 @@ func (c *clusterConfig) startRGWPods(realmName, zoneGroupName, zoneName string)
 		resourceName := fmt.Sprintf("%s-%s-%s", AppName, c.store.Name, daemonLetterID)
 
 		rgwConfig := &rgwConfig{
-			ResourceName: resourceName,
-			DaemonID:     daemonName,
-			Realm:        realmName,
-			ZoneGroup:    zoneGroupName,
-			Zone:         zoneName,
+			ResourceName:   resourceName,
+			DaemonID:       daemonName,
+			Realm:          realmName,
+			ZoneGroup:      zoneGroupName,
+			Zone:           zoneName,
+			Auth:           c.store.Spec.Auth,
+			KeystoneSecret: keystoneSecret,
 		}
 
 		// We set the owner reference of the Secret to the Object controller instead of the replicaset
@@ -151,6 +157,9 @@ func (c *clusterConfig) startRGWPods(realmName, zoneGroupName, zoneName string)
 		if err != nil {
 			// Getting EPERM typically happens when the flag may not be modified at runtime
 			// This is fine to ignore
+			// No it is not: As errors in the mapKeystoneSecretToConfig function are returned
+			// directly and the situation may occur, that only part of the configuration is set
+			// TODO have another look at the error handling here (and in setFlagsMonConfigStore)
 			code, ok := exec.ExitStatus(err)
 			if ok && code != int(syscall.EPERM) {
 				return errors.Wrap(err, "failed to set default rgw config options")
diff --git a/pkg/operator/ceph/object/rgw_test.go b/pkg/operator/ceph/object/rgw_test.go
index 2723ca5a3a5b..264c15b01913 100644
--- a/pkg/operator/ceph/object/rgw_test.go
+++ b/pkg/operator/ceph/object/rgw_test.go
@@ -79,10 +79,12 @@ func TestStartRGW(t *testing.T) {
 	// start a basic cluster
 	ownerInfo := client.NewMinimumOwnerInfoWithOwnerRef()
 	c := &clusterConfig{context, info, store, version, &cephv1.ClusterSpec{}, ownerInfo, data, r.client}
+	err := c.startRGWPods(store.Name, store.Name, store.Name, nil)
+	assert.Nil(t, err)
 
 	t.Run("Deployment is created", func(t *testing.T) {
 		store.Spec.Gateway.Instances = 1
-		err := c.startRGWPods(store.Name, store.Name, store.Name)
+		err := c.startRGWPods(store.Name, store.Name, store.Name, nil)
 		assert.Nil(t, err)
 
 		validateStart(ctx, t, c, clientset)
@@ -95,7 +97,7 @@ func TestStartRGW(t *testing.T) {
 		// Purge store of configurations applied to gateways
 		appliedRgwConfigurations = make(map[string]string)
 
-		err := c.startRGWPods(store.Name, store.Name, store.Name)
+		err := c.startRGWPods(store.Name, store.Name, store.Name, nil)
 		assert.Nil(t, err)
 
 		assert.Contains(t, appliedRgwConfigurations, "rgw_run_sync_thread")
@@ -109,7 +111,7 @@ func TestStartRGW(t *testing.T) {
 		// Purge store of configurations applied to gateways
 		appliedRgwConfigurations = make(map[string]string)
 
-		err := c.startRGWPods(store.Name, store.Name, store.Name)
+		err := c.startRGWPods(store.Name, store.Name, store.Name, nil)
 		assert.Nil(t, err)
 
 		assert.Contains(t, appliedRgwConfigurations, "rgw_run_sync_thread")
@@ -157,7 +159,7 @@ func TestCreateObjectStore(t *testing.T) {
 	r := &ReconcileCephObjectStore{client: cl, scheme: s}
 	ownerInfo := client.NewMinimumOwnerInfoWithOwnerRef()
 	c := &clusterConfig{context, info, store, "1.2.3.4", &cephv1.ClusterSpec{}, ownerInfo, data, r.client}
-	err := c.createOrUpdateStore(store.Name, store.Name, store.Name)
+	err := c.createOrUpdateStore(store.Name, store.Name, store.Name, nil)
 	assert.Nil(t, err)
 }
 
diff --git a/pkg/operator/ceph/object/user.go b/pkg/operator/ceph/object/user.go
index 7cb2b670b682..3863f9bd9f70 100644
--- a/pkg/operator/ceph/object/user.go
+++ b/pkg/operator/ceph/object/user.go
@@ -269,7 +269,35 @@ func generateCephUserSecret(userConfig *admin.User, endpoint, namespace, storeNa
 	return secret
 }
 
-func ReconcileCephUserSecret(ctx context.Context, k8sclient client.Client, scheme *runtime.Scheme, ownerRef metav1.Object, userConfig *admin.User, endpoint, namespace, storeName, tlsSecretName string) (reconcile.Result, error) {
+func generateCephSubuserSecretName(store, username, subusername string) string {
+	return fmt.Sprintf("rook-ceph-object-subuser-%s-%s-%s", store, username, subusername)
+}
+
+func generateCephSubuserSecret(userConfig *admin.User, endpoint, namespace, storeName string, subuser *admin.SwiftKeySpec) *corev1.Secret {
+	secrets := map[string]string{
+		"SWIFT_USER":          subuser.User,
+		"SWIFT_SECRET_KEY":    subuser.SecretKey,
+		"SWIFT_AUTH_ENDPOINT": "TODO", // TODO
+	}
+	secret := &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      generateCephSubuserSecretName(storeName, userConfig.ID, subuser.User),
+			Namespace: namespace,
+			Labels: map[string]string{
+				"app":  AppName,
+				"user": userConfig.ID,
+				// XXX: should we add a subuser label?
+				"rook_cluster":      namespace,
+				"rook_object_store": storeName,
+			},
+		},
+		StringData: secrets,
+		Type:       k8sutil.RookType,
+	}
+	return secret
+}
+
+func ReconcileCephUserSecrets(ctx context.Context, k8sclient client.Client, scheme *runtime.Scheme, ownerRef metav1.Object, userConfig *admin.User, endpoint, namespace, storeName, tlsSecretName string) (reconcile.Result, error) {
 	// Generate Kubernetes Secret
 	secret := generateCephUserSecret(userConfig, endpoint, namespace, storeName, tlsSecretName)
 
@@ -284,5 +312,21 @@ func ReconcileCephUserSecret(ctx context.Context, k8sclient client.Client, schem
 	if err != nil {
 		return reconcile.Result{}, errors.Wrapf(err, "failed to create or update ceph object user %q secret", secret.Name)
 	}
+
+	for _, key := range userConfig.SwiftKeys {
+		secret = generateCephSubuserSecret(userConfig, endpoint, namespace, storeName, &key)
+
+		err = controllerutil.SetControllerReference(ownerRef, secret, scheme)
+		if err != nil {
+			return reconcile.Result{}, errors.Wrapf(err, "failed to set owner reference of ceph object subuser secret %q", secret.Name)
+		}
+
+		// Create Kubernetes Secret
+		err = opcontroller.CreateOrUpdateObject(ctx, k8sclient, secret)
+		if err != nil {
+			return reconcile.Result{}, errors.Wrapf(err, "failed to create or update ceph object subuser %q secret", secret.Name)
+		}
+	}
+
 	return reconcile.Result{}, nil
 }
diff --git a/pkg/operator/ceph/object/user/controller.go b/pkg/operator/ceph/object/user/controller.go
index 1c7624e6c7c4..5042fca3e4c2 100644
--- a/pkg/operator/ceph/object/user/controller.go
+++ b/pkg/operator/ceph/object/user/controller.go
@@ -21,6 +21,7 @@ import (
 	"context"
 	"fmt"
 	"reflect"
+	"sort"
 
 	"github.com/ceph/go-ceph/rgw/admin"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
@@ -263,7 +264,7 @@ func (r *ReconcileObjectStoreUser) reconcile(request reconcile.Request) (reconci
 	}
 
 	tlsSecretName := store.Spec.Gateway.SSLCertificateRef
-	reconcileResponse, err = object.ReconcileCephUserSecret(r.opManagerContext, r.client, r.scheme, cephObjectStoreUser, r.userConfig, r.objContext.Endpoint, cephObjectStoreUser.Namespace, cephObjectStoreUser.Spec.Store, tlsSecretName)
+	reconcileResponse, err = object.ReconcileCephUserSecrets(r.opManagerContext, r.client, r.scheme, cephObjectStoreUser, r.userConfig, r.objContext.Endpoint, cephObjectStoreUser.Namespace, cephObjectStoreUser.Spec.Store, tlsSecretName)
 	if err != nil {
 		r.updateStatus(k8sutil.ObservedGenerationNotAvailable, request.NamespacedName, k8sutil.ReconcileFailedStatus)
 		return reconcileResponse, *cephObjectStoreUser, err
@@ -362,17 +363,120 @@ func (r *ReconcileObjectStoreUser) createOrUpdateCephUser(u *cephv1.CephObjectSt
 		return errors.Wrapf(err, "failed to set quotas for user %q", u.Name)
 	}
 
+	// Reconcile subusers
+	user, err = r.reconcileSubusers(user, u)
+	if err != nil {
+		return err
+	}
+
 	// Set access and secret key
 	if r.userConfig.Keys == nil {
 		r.userConfig.Keys = make([]admin.UserKeySpec, 1)
 	}
 	r.userConfig.Keys[0].AccessKey = user.Keys[0].AccessKey
 	r.userConfig.Keys[0].SecretKey = user.Keys[0].SecretKey
+
+	// Copy the subuser keys – XXX: should this be a deep copy for safety?
+	r.userConfig.SwiftKeys = append([]admin.SwiftKeySpec{}, user.SwiftKeys...)
+
 	logger.Info(logCreateOrUpdate)
 
 	return nil
 }
 
+var accessLevelMap = map[admin.SubuserAccess]admin.SubuserAccess{
+	admin.SubuserAccessReplyNone:      admin.SubuserAccessNone,
+	admin.SubuserAccessReplyRead:      admin.SubuserAccessRead,
+	admin.SubuserAccessReplyWrite:     admin.SubuserAccessWrite,
+	admin.SubuserAccessReplyReadWrite: admin.SubuserAccessReadWrite,
+	admin.SubuserAccessReplyFull:      admin.SubuserAccessFull,
+}
+
+func mapAccessLevel(outputAccessLevel admin.SubuserAccess) (admin.SubuserAccess, error) {
+	level, ok := accessLevelMap[outputAccessLevel]
+	if !ok {
+		return admin.SubuserAccessNone, fmt.Errorf("invalid access level returned by RGW %s", outputAccessLevel)
+	}
+	return level, nil
+}
+
+func (r *ReconcileObjectStoreUser) reconcileSubusers(userIs admin.User, userSpec *cephv1.CephObjectStoreUser) (admin.User, error) {
+	var err error
+
+	mustRefreshUser := false
+
+	subusersSpec := append([]cephv1.SubuserSpec{}, userSpec.Spec.Subusers...)
+	subusersIs := append([]admin.SubuserSpec{}, userIs.Subusers...)
+
+	sort.Slice(subusersSpec, func(i, j int) bool {
+		return subusersSpec[i].Name < subusersSpec[j].Name
+	})
+	sort.Slice(subusersIs, func(i, j int) bool {
+		return subusersIs[i].Name < subusersIs[j].Name
+	})
+
+	toCreate := make([]cephv1.SubuserSpec, 0)
+	toDelete := make([]admin.SubuserSpec, 0)
+	for len(subusersSpec) > 0 && len(subusersIs) > 0 {
+		subuserFullNameSpec := fmt.Sprintf("%s:%s", userIs.ID, subusersSpec[0].Name)
+
+		if subuserFullNameSpec < subusersIs[0].Name {
+			toCreate = append(toCreate, subusersSpec[0])
+			subusersSpec = subusersSpec[1:]
+		} else if subuserFullNameSpec > subusersIs[0].Name {
+			toDelete = append(toDelete, subusersIs[0])
+			subusersIs = subusersIs[1:]
+		} else { // subusersSpec[0].Name == subusersIs[0].Name
+			accessLevelIs, err := mapAccessLevel(subusersIs[0].Access)
+			if err != nil {
+				return userIs, errors.Wrapf(err, "failed to reconcile subusers")
+			}
+			if string(subusersSpec[0].Access) != string(accessLevelIs) {
+				modifiedSubuser := admin.SubuserSpec{
+					Name:   subuserFullNameSpec,
+					Access: admin.SubuserAccess(subusersSpec[0].Access),
+				}
+				mustRefreshUser = true
+				err = r.objContext.AdminOpsClient.ModifySubuser(r.opManagerContext, userIs, modifiedSubuser)
+				if err != nil {
+					return userIs, errors.Wrapf(err, "failed to modify subuser %q", modifiedSubuser.Name)
+				}
+			}
+
+			subusersSpec = subusersSpec[1:]
+			subusersIs = subusersIs[1:]
+		}
+	}
+	toCreate = append(toCreate, subusersSpec...)
+	toDelete = append(toDelete, subusersIs...)
+
+	for _, subuserToCreate := range toCreate {
+		err = r.objContext.AdminOpsClient.CreateSubuser(r.opManagerContext, userIs, admin.SubuserSpec{
+			Name:   fmt.Sprintf("%s:%s", userIs.ID, subuserToCreate.Name),
+			Access: admin.SubuserAccess(subuserToCreate.Access),
+		})
+		mustRefreshUser = true
+		if err != nil {
+			return userIs, errors.Wrapf(err, "failed to create subuser %q", subuserToCreate.Name)
+		}
+	}
+
+	for _, subuserToDelete := range toDelete {
+		mustRefreshUser = true
+		err = r.objContext.AdminOpsClient.RemoveSubuser(r.opManagerContext, userIs, subuserToDelete)
+		if err != nil {
+			return userIs, errors.Wrapf(err, "failed to delete subuser %q", subuserToDelete.Name)
+		}
+	}
+
+	if mustRefreshUser {
+		// As the subuser commands don't give us the full user object, we need to get it here to have an accurate representation of the user state after the reconcile
+		userIs, err = r.objContext.AdminOpsClient.GetUser(r.opManagerContext, userIs)
+	}
+
+	return userIs, nil
+}
+
 func (r *ReconcileObjectStoreUser) initializeObjectStoreContext(u *cephv1.CephObjectStoreUser) error {
 	err := r.objectStoreInitialized(u)
 	if err != nil {
@@ -438,6 +542,7 @@ func generateUserConfig(user *cephv1.CephObjectStoreUser) admin.User {
 	userConfig := admin.User{
 		ID:          user.Name,
 		DisplayName: displayName,
+		Keys:        make([]admin.UserKeySpec, 0),
 	}
 
 	defaultMaxBuckets := 1000
diff --git a/pkg/operator/ceph/object/user/controller_test.go b/pkg/operator/ceph/object/user/controller_test.go
index 498e6a503e5a..f159e4532d31 100644
--- a/pkg/operator/ceph/object/user/controller_test.go
+++ b/pkg/operator/ceph/object/user/controller_test.go
@@ -20,14 +20,17 @@ package objectuser
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
+	"net/url"
 	"testing"
 	"time"
 
 	"github.com/ceph/go-ceph/rgw/admin"
 	"github.com/coreos/pkg/capnslog"
+	"github.com/pkg/errors"
 	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/operator/k8sutil"
@@ -290,9 +293,9 @@ 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&uid=my-user" && req.Method == http.MethodPost && req.URL.Path == "rook-ceph-rgw-my-store.mycluster.svc/admin/user") || // TODO: Leave this in?
 						(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") {
+						(req.URL.RawQuery == "format=json&uid=my-user" && (req.Method == http.MethodGet || req.Method == http.MethodPost) && req.URL.Path == "rook-ceph-rgw-my-store.mycluster.svc/admin/user") {
 						return &http.Response{
 							StatusCode: 200,
 							Body:       io.NopCloser(bytes.NewReader([]byte(userCreateJSON))),
@@ -435,15 +438,6 @@ func TestCreateOrUpdateCephUser(t *testing.T) {
 				return nil, fmt.Errorf("unexpected url path %q", req.URL.Path)
 			}
 
-			if req.Method == http.MethodGet {
-				if req.URL.RawQuery == "format=json&uid=my-user" {
-					return &http.Response{
-						StatusCode: 200,
-						Body:       io.NopCloser(bytes.NewReader([]byte(userCreateJSON))),
-					}, nil
-				}
-			}
-
 			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" ||
@@ -590,3 +584,275 @@ func TestCreateOrUpdateCephUser(t *testing.T) {
 		assert.NoError(t, err)
 	})
 }
+
+func TestCreateorUpdateSubusers(t *testing.T) {
+	// Set DEBUG logging
+	capnslog.SetGlobalLogLevel(capnslog.DEBUG)
+
+	objectUser := &cephv1.CephObjectStoreUser{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "",
+			Namespace: namespace,
+		},
+		Spec: cephv1.ObjectStoreUserSpec{
+			Store: store,
+		},
+		TypeMeta: metav1.TypeMeta{
+			Kind: "CephObjectStoreUser",
+		},
+	}
+
+	users := make(map[string]*admin.User)
+	mockClient := &cephobject.MockClient{
+		MockDo: func(req *http.Request) (*http.Response, error) {
+			fmt.Printf("REQUEST %s %s\n", req.Method, req.URL)
+
+			if req.URL.Path != "rook-ceph-rgw-my-store.mycluster.svc/admin/user" {
+				return nil, fmt.Errorf("unexpected url path %q", req.URL.Path)
+			}
+
+			values, err := url.ParseQuery(req.URL.RawQuery)
+			if err != nil {
+				return nil, errors.Wrapf(err, "invalid query")
+			}
+
+			if values.Get("format") != "json" {
+				return nil, fmt.Errorf("unexpected format %q", values.Get("format"))
+			}
+
+			if req.Method == http.MethodGet {
+				user, ok := users[values.Get("uid")]
+				if !ok {
+					return &http.Response{
+						StatusCode: 404,
+						Body:       io.NopCloser(bytes.NewReader([]byte(`{"Code":"NoSuchUser","RequestId":"tx0000000000000000005a9-00608957a2-10496-my-store","HostId":"10496-my-store-my-store"}`))),
+					}, nil
+				}
+				resp, err := json.Marshal(user)
+				fmt.Printf("GET RESPONSE: %s\n", resp)
+				if err != nil {
+					return nil, err
+				}
+
+				return &http.Response{
+					StatusCode: 201,
+					Body:       io.NopCloser(bytes.NewReader(resp)),
+				}, nil
+			}
+
+			if req.Method == http.MethodPost {
+				user, ok := users[values.Get("uid")]
+				if values.Has("subuser") {
+					if !ok {
+						return nil, fmt.Errorf("trying to modify a subuser for non-existant user %q", values.Get("uid"))
+					}
+
+					for i, subuser := range user.Subusers {
+						if subuser.Name == values.Get("subuser") {
+							user.Subusers[i].Access = admin.SubuserAccess(values.Get("access"))
+						}
+						return &http.Response{
+							StatusCode: 200,
+							Body:       io.NopCloser(bytes.NewReader([]byte("{}"))),
+						}, nil
+					}
+
+					return nil, fmt.Errorf("trying to modify non-existant subuser %q", values.Get("subuser"))
+				}
+
+				if !ok {
+					return nil, fmt.Errorf("trying to modify existing user %q", values.Get("uid"))
+				}
+
+				u := users[values.Get("uid")]
+
+				if values.Has("max-buckets") {
+					var maxBuckets int
+					fmt.Scanf(values.Get("max-buckets"), "%d", &maxBuckets)
+					u.MaxBuckets = &maxBuckets
+				}
+
+				if values.Has("display-name") {
+					u.DisplayName = values.Get("DisplayName")
+				}
+
+				resp, err := json.Marshal(u)
+				if err != nil {
+					return nil, err
+				}
+
+				return &http.Response{
+					StatusCode: 201,
+					Body:       io.NopCloser(bytes.NewReader(resp)),
+				}, nil
+			}
+
+			if req.Method == http.MethodDelete {
+				user, ok := users[values.Get("uid")]
+				if values.Has("subuser") {
+					if !ok {
+						return nil, fmt.Errorf("trying to create a subuser for non-existant user %q", values.Get("uid"))
+					}
+
+					newSubusers := make([]admin.SubuserSpec, 0)
+					deleted := false
+					for _, subuser := range user.Subusers {
+						if subuser.Name == values.Get("subuser") {
+							deleted = true
+						} else {
+							newSubusers = append(newSubusers, subuser)
+						}
+					}
+
+					if !deleted {
+						return nil, fmt.Errorf("trying to delete non-existant subuser %q", values.Get("subuser"))
+					}
+
+					user.Subusers = newSubusers
+
+					return &http.Response{
+						StatusCode: 201,
+						Body:       io.NopCloser(bytes.NewReader([]byte("{}"))),
+					}, nil
+				}
+
+				if !ok {
+					return nil, fmt.Errorf("trying to delete non-existant user %q", values.Get("uid"))
+				}
+				delete(users, values.Get("uid"))
+				return &http.Response{
+					StatusCode: 200,
+					Body:       io.NopCloser(bytes.NewReader([]byte("{}"))),
+				}, nil
+			}
+			if req.Method == http.MethodPut {
+				user, ok := users[values.Get("uid")]
+				if values.Has("subuser") {
+					if !ok {
+						return nil, fmt.Errorf("trying to create a subuser for non-existant user %q", values.Get("uid"))
+					}
+
+					for _, subuser := range user.Subusers {
+						if subuser.Name == values.Get("subuser") {
+							return nil, fmt.Errorf("trying to create existing subuser %q", values.Get("subuser"))
+						}
+					}
+
+					user.Subusers = append(user.Subusers, admin.SubuserSpec{
+						Name:   values.Get("subuser"),
+						Access: admin.SubuserAccess(values.Get("access")), // TODO: map this to the reply values
+					})
+
+					return &http.Response{
+						StatusCode: 201,
+						Body:       io.NopCloser(bytes.NewReader([]byte("{}"))),
+					}, nil
+				}
+
+				if values.Has("quota") {
+					return &http.Response{
+						StatusCode: 200,
+						Body:       io.NopCloser(bytes.NewReader([]byte("{}"))),
+					}, nil
+				}
+
+				if ok {
+					return nil, fmt.Errorf("trying to create existing user %q", values.Get("uid"))
+				}
+
+				maxBuckets := -1
+				if values.Has("max-buckets") {
+					fmt.Scanf(values.Get("max-buckets"), "%d", &maxBuckets)
+				}
+
+				u := &admin.User{
+					ID: values.Get("uid"),
+					Keys: []admin.UserKeySpec{
+						{User: values.Get("uid"), AccessKey: "access_key", SecretKey: "secret_key"},
+					},
+					MaxBuckets:  &maxBuckets,
+					DisplayName: values.Get("display-name"),
+				}
+				users[values.Get("uid")] = u
+
+				resp, err := json.Marshal(u)
+				if err != nil {
+					return nil, err
+				}
+
+				return &http.Response{
+					StatusCode: 201,
+					Body:       io.NopCloser(bytes.NewReader(resp)),
+				}, nil
+			}
+			return nil, fmt.Errorf("unexpected request: %q. method %q. path %q", req.URL.RawQuery, req.Method, req.URL.Path)
+		},
+	}
+
+	adminClient, err := admin.New("rook-ceph-rgw-my-store.mycluster.svc", "53S6B9S809NUP19IJ2K3", "1bXPegzsGClvoGAiJdHQD1uOW2sQBLAZM9j9VtXR", mockClient)
+	assert.NoError(t, err)
+	userConfig := generateUserConfig(objectUser)
+
+	r := &ReconcileObjectStoreUser{
+		objContext: &cephobject.AdminOpsContext{
+			AdminOpsClient: adminClient,
+		},
+		userConfig:       &userConfig,
+		opManagerContext: context.TODO(),
+	}
+
+	t.Run("user without any Quotas or Capabilities", func(t *testing.T) {
+		objectUser.Name = name
+		userConfig = generateUserConfig(objectUser)
+		r.userConfig = &userConfig
+		err = r.createOrUpdateCephUser(objectUser)
+		assert.NoError(t, err)
+	})
+
+	t.Run("add a subuser", func(t *testing.T) {
+		objectUser.Spec.Subusers = []cephv1.SubuserSpec{
+			{
+				Name:   "swift",
+				Access: cephv1.AccessSpecRead,
+			},
+		}
+		userConfig = generateUserConfig(objectUser)
+		r.userConfig = &userConfig
+		err = r.createOrUpdateCephUser(objectUser)
+		assert.NoError(t, err)
+	})
+
+	t.Run("modify a subuser", func(t *testing.T) {
+		objectUser.Spec.Subusers = []cephv1.SubuserSpec{
+			{
+				Name:   "swift",
+				Access: cephv1.AccessSpecWrite,
+			},
+		}
+		userConfig = generateUserConfig(objectUser)
+		r.userConfig = &userConfig
+		err = r.createOrUpdateCephUser(objectUser)
+		assert.NoError(t, err)
+	})
+
+	t.Run("replace the subuser", func(t *testing.T) {
+		objectUser.Spec.Subusers = []cephv1.SubuserSpec{
+			{
+				Name:   "swift-wo",
+				Access: cephv1.AccessSpecWrite,
+			},
+		}
+		userConfig = generateUserConfig(objectUser)
+		r.userConfig = &userConfig
+		err = r.createOrUpdateCephUser(objectUser)
+		assert.NoError(t, err)
+	})
+
+	t.Run("remove all subusers", func(t *testing.T) {
+		objectUser.Spec.Subusers = []cephv1.SubuserSpec{}
+		userConfig = generateUserConfig(objectUser)
+		r.userConfig = &userConfig
+		err = r.createOrUpdateCephUser(objectUser)
+		assert.NoError(t, err)
+	})
+}
diff --git a/tests/framework/clients/object.go b/tests/framework/clients/object.go
index a5df6584702d..02a40622d292 100644
--- a/tests/framework/clients/object.go
+++ b/tests/framework/clients/object.go
@@ -40,10 +40,10 @@ func CreateObjectOperation(k8sh *utils.K8sHelper, manifests installer.CephManife
 }
 
 // ObjectCreate Function to create a object store in rook
-func (o *ObjectOperation) Create(namespace, storeName string, replicaCount int32, tlsEnable bool) error {
+func (o *ObjectOperation) Create(namespace, storeName string, replicaCount int32, tlsEnable bool, swiftAndKeystone bool) error {
 
 	logger.Info("creating the object store via CRD")
-	if err := o.k8sh.ResourceOperation("apply", o.manifests.GetObjectStore(storeName, int(replicaCount), rgwPort, tlsEnable)); err != nil {
+	if err := o.k8sh.ResourceOperation("apply", o.manifests.GetObjectStore(storeName, int(replicaCount), rgwPort, tlsEnable, swiftAndKeystone)); err != nil {
 		return err
 	}
 
diff --git a/tests/framework/installer/ceph_helm_installer.go b/tests/framework/installer/ceph_helm_installer.go
index ed2f0df921a9..f7becceb6b4b 100644
--- a/tests/framework/installer/ceph_helm_installer.go
+++ b/tests/framework/installer/ceph_helm_installer.go
@@ -281,7 +281,7 @@ func (h *CephInstaller) CreateFileSystemConfiguration(values map[string]interfac
 
 // CreateObjectStoreConfiguration creates an object store configuration
 func (h *CephInstaller) CreateObjectStoreConfiguration(values map[string]interface{}, name, scName string) error {
-	testObjectStoreBytes := []byte(h.Manifests.GetObjectStore(name, 2, 8080, false))
+	testObjectStoreBytes := []byte(h.Manifests.GetObjectStore(name, 2, 8080, false, false))
 	var testObjectStoreCRD map[string]interface{}
 	if err := yaml.Unmarshal(testObjectStoreBytes, &testObjectStoreCRD); err != nil {
 		return err
diff --git a/tests/framework/installer/ceph_manifests.go b/tests/framework/installer/ceph_manifests.go
index d0bc1e3134d6..5cba689e20db 100644
--- a/tests/framework/installer/ceph_manifests.go
+++ b/tests/framework/installer/ceph_manifests.go
@@ -47,7 +47,7 @@ type CephManifests interface {
 	GetNFS(name string, daemonCount int) string
 	GetNFSPool() string
 	GetRBDMirror(name string, daemonCount int) string
-	GetObjectStore(name string, replicaCount, port int, tlsEnable bool) string
+	GetObjectStore(name string, replicaCount, port int, tlsEnable bool, swiftAndKeystone bool) string
 	GetObjectStoreUser(name, displayName, store, usercaps, maxsize string, maxbuckets, maxobjects int) string
 	GetBucketStorageClass(storeName, storageClassName, reclaimPolicy string) string
 	GetOBC(obcName, storageClassName, bucketName string, maxObject string, createBucket bool) string
@@ -456,35 +456,30 @@ spec:
     requireSafeReplicaSize: false`
 }
 
-func (m *CephManifestsMaster) GetObjectStore(name string, replicaCount, port int, tlsEnable bool) string {
-	if tlsEnable {
-		return `apiVersion: ceph.rook.io/v1
-kind: CephObjectStore
-metadata:
-  name: ` + name + `
-  namespace: ` + m.settings.Namespace + `
-spec:
-  metadataPool:
-    replicated:
-      size: 1
-      requireSafeReplicaSize: false
-    compressionMode: passive
-  dataPool:
-    replicated:
-      size: 1
-      requireSafeReplicaSize: false
-  gateway:
-    resources: null
-    securePort: ` + strconv.Itoa(port) + `
-    instances: ` + strconv.Itoa(replicaCount) + `
-    sslCertificateRef: ` + name + `
-`
+func (m *CephManifestsMaster) GetObjectStore(name string, replicaCount, port int, tlsEnable bool, swiftAndKeystone bool) string {
+	type Spec struct {
+		Name             string
+		TLS              bool
+		Port             int
+		ReplicaCount     int
+		SwiftAndKeystone bool
+		Manifests        *CephManifestsMaster
 	}
-	return `apiVersion: ceph.rook.io/v1
+
+	spec := Spec{
+		Name:             name,
+		TLS:              tlsEnable,
+		ReplicaCount:     replicaCount,
+		Port:             port,
+		SwiftAndKeystone: swiftAndKeystone,
+		Manifests:        m,
+	}
+
+	tmpl := `apiVersion: ceph.rook.io/v1
 kind: CephObjectStore
 metadata:
-  name: ` + name + `
-  namespace: ` + m.settings.Namespace + `
+  name: {{ .Name }}
+  namespace: {{ .Manifests.Settings.Namespace }}
 spec:
   metadataPool:
     replicated:
@@ -495,11 +490,24 @@ spec:
     replicated:
       size: 1
       requireSafeReplicaSize: false
+  {{ if .SwiftAndKeystone }}
+  protocols:
+    swift:
+      accountInUrl: true
+      urlPrefix: /swift
+  {{ end }}
   gateway:
     resources: null
-    port: ` + strconv.Itoa(port) + `
-    instances: ` + strconv.Itoa(replicaCount) + `
-`
+    type: s3
+    {{ if .TLS }}securePort: {{ .Port }}{{ else }}port: {{ .Port }}{{ end }}
+    instances: {{ .ReplicaCount }}
+    {{ if .TLS }}sslCertificateRef: {{ .Name }}{{ end }}
+  healthCheck:
+    bucket:
+      disabled: false
+      interval: {{ if .TLS }}10s{{ else }}5s{{ end }}`
+
+	return renderTemplate(tmpl, spec)
 }
 
 func (m *CephManifestsMaster) GetObjectStoreUser(name, displayName, store, usercaps, maxsize string, maxbuckets, maxobjects int) string {
diff --git a/tests/framework/installer/ceph_manifests_previous.go b/tests/framework/installer/ceph_manifests_previous.go
index ff41757d7a47..f2eb759eebc3 100644
--- a/tests/framework/installer/ceph_manifests_previous.go
+++ b/tests/framework/installer/ceph_manifests_previous.go
@@ -131,8 +131,11 @@ func (m *CephManifestsPreviousVersion) GetNFSPool() string {
 	return m.latest.GetNFSPool()
 }
 
-func (m *CephManifestsPreviousVersion) GetObjectStore(name string, replicaCount, port int, tlsEnable bool) string {
-	return m.latest.GetObjectStore(name, replicaCount, port, tlsEnable)
+func (m *CephManifestsPreviousVersion) GetObjectStore(name string, replicaCount, port int, tlsEnable bool, swiftAndKeystone bool) string {
+	if swiftAndKeystone {
+		panic("Previous version does not support swift or keystone")
+	}
+	return m.latest.GetObjectStore(name, replicaCount, port, tlsEnable, false)
 }
 
 func (m *CephManifestsPreviousVersion) GetObjectStoreUser(name, displayName, store, usercaps, maxsize string, maxbuckets, maxobjects int) string {
diff --git a/tests/framework/installer/installer.go b/tests/framework/installer/installer.go
index c44602b9516e..33aba976e4a6 100644
--- a/tests/framework/installer/installer.go
+++ b/tests/framework/installer/installer.go
@@ -18,7 +18,9 @@ package installer
 
 import (
 	"fmt"
+	"strings"
 	"testing"
+	"text/template"
 
 	"github.com/coreos/pkg/capnslog"
 	"github.com/rook/rook/tests/framework/utils"
@@ -59,3 +61,16 @@ func checkError(t *testing.T, err error, message string) {
 	}
 	assert.NoError(t, err, "%s. %+v", message, err)
 }
+
+func renderTemplate(tmplSrc string, data any) string {
+	tmpl, err := template.New("template").Parse(tmplSrc)
+	if err != nil {
+		panic(fmt.Errorf("syntax error in template: %s", err))
+	}
+	var b strings.Builder
+	err = tmpl.Execute(&b, data)
+	if err != nil {
+		panic(fmt.Errorf("error while rendering the template: %s", err))
+	}
+	return b.String()
+}
diff --git a/tests/integration/ceph_auth_keystone_test.go b/tests/integration/ceph_auth_keystone_test.go
new file mode 100644
index 000000000000..76ab1b7282d8
--- /dev/null
+++ b/tests/integration/ceph_auth_keystone_test.go
@@ -0,0 +1 @@
+package integration
diff --git a/tests/integration/ceph_base_object_test.go b/tests/integration/ceph_base_object_test.go
index 5f8779a2b2ef..8e3cb83fa49f 100644
--- a/tests/integration/ceph_base_object_test.go
+++ b/tests/integration/ceph_base_object_test.go
@@ -73,7 +73,7 @@ func runObjectE2ETestLite(t *testing.T, helper *clients.TestClient, k8sh *utils.
 	}
 	logger.Infof("test creating %s object store %q in namespace %q", andDeleting, storeName, namespace)
 
-	createCephObjectStore(t, helper, k8sh, installer, namespace, storeName, replicaSize, enableTLS)
+	createCephObjectStore(t, helper, k8sh, installer, namespace, storeName, replicaSize, enableTLS, false)
 
 	if deleteStore {
 		t.Run("delete object store", func(t *testing.T) {
@@ -84,7 +84,7 @@ func runObjectE2ETestLite(t *testing.T, helper *clients.TestClient, k8sh *utils.
 }
 
 // create a CephObjectStore and wait for it to report ready status
-func createCephObjectStore(t *testing.T, helper *clients.TestClient, k8sh *utils.K8sHelper, installer *installer.CephInstaller, namespace, storeName string, replicaSize int, tlsEnable bool) {
+func createCephObjectStore(t *testing.T, helper *clients.TestClient, k8sh *utils.K8sHelper, installer *installer.CephInstaller, namespace, storeName string, replicaSize int, tlsEnable bool, swiftAndKeystone bool) {
 	logger.Infof("Create Object Store %q with replica count %d", storeName, replicaSize)
 	rgwServiceName := "rook-ceph-rgw-" + storeName
 	if tlsEnable {
@@ -93,7 +93,7 @@ func createCephObjectStore(t *testing.T, helper *clients.TestClient, k8sh *utils
 		})
 	}
 	t.Run("create CephObjectStore", func(t *testing.T) {
-		err := helper.ObjectClient.Create(namespace, storeName, int32(replicaSize), tlsEnable)
+		err := helper.ObjectClient.Create(namespace, storeName, int32(replicaSize), tlsEnable, swiftAndKeystone)
 		assert.Nil(t, err)
 	})
 
diff --git a/tests/integration/ceph_cosi_test.go b/tests/integration/ceph_cosi_test.go
index 732771aef395..a4cfa8050941 100644
--- a/tests/integration/ceph_cosi_test.go
+++ b/tests/integration/ceph_cosi_test.go
@@ -38,7 +38,7 @@ func testCOSIDriver(s *suite.Suite, helper *clients.TestClient, k8sh *utils.K8sH
 		assert.NoError(t, err, "failed to create COSI controller")
 	})
 
-	createCephObjectStore(s.T(), helper, k8sh, cephinstaller, namespace, objectStoreCOSI, 1, false)
+	createCephObjectStore(s.T(), helper, k8sh, cephinstaller, namespace, objectStoreCOSI, 1, false, false)
 
 	t.Run("Creating CephCOSIDriver CRD", func(t *testing.T) {
 		err := helper.COSIClient.CreateCOSI()
diff --git a/tests/integration/ceph_object_test.go b/tests/integration/ceph_object_test.go
index 63aded047983..7f903f0d6a49 100644
--- a/tests/integration/ceph_object_test.go
+++ b/tests/integration/ceph_object_test.go
@@ -96,15 +96,10 @@ func (s *ObjectSuite) TestWithTLS() {
 	}
 
 	tls := true
+	swiftAndKeystone := false
 	objectStoreServicePrefix = objectStoreServicePrefixUniq
-	runObjectE2ETest(s.helper, s.k8sh, s.installer, &s.Suite, s.settings.Namespace, tls)
-	err := s.k8sh.Clientset.CoreV1().Secrets(s.settings.Namespace).Delete(context.TODO(), objectTLSSecretName, metav1.DeleteOptions{})
-	if err != nil {
-		if !errors.IsNotFound(err) {
-			logger.Fatal("failed to deleted store TLS secret")
-		}
-	}
-	logger.Info("successfully deleted store TLS secret")
+	runObjectE2ETest(s.helper, s.k8sh, s.installer, &s.Suite, s.settings.Namespace, tls, swiftAndKeystone)
+	cleanUpTLS(s)
 }
 
 func (s *ObjectSuite) TestWithoutTLS() {
@@ -113,22 +108,41 @@ func (s *ObjectSuite) TestWithoutTLS() {
 	}
 
 	tls := false
+	swiftAndKeystone := false
+	objectStoreServicePrefix = objectStoreServicePrefixUniq
+	runObjectE2ETest(s.helper, s.k8sh, s.installer, &s.Suite, s.settings.Namespace, tls, swiftAndKeystone)
+}
+
+func (s *ObjectSuite) TestWithSwiftAndKeystone() {
+	tls := true
+	swiftAndKeystone := true
 	objectStoreServicePrefix = objectStoreServicePrefixUniq
-	runObjectE2ETest(s.helper, s.k8sh, s.installer, &s.Suite, s.settings.Namespace, tls)
+	runObjectE2ETest(s.helper, s.k8sh, s.installer, &s.Suite, s.settings.Namespace, tls, swiftAndKeystone)
+	cleanUpTLS(s)
+}
+
+func cleanUpTLS(s *ObjectSuite) {
+	err := s.k8sh.Clientset.CoreV1().Secrets(s.settings.Namespace).Delete(context.TODO(), objectTLSSecretName, metav1.DeleteOptions{})
+	if err != nil {
+		if !errors.IsNotFound(err) {
+			logger.Fatal("failed to deleted store TLS secret")
+		}
+	}
+	logger.Info("successfully deleted store TLS secret")
 }
 
 // Smoke Test for ObjectStore - Test check the following operations on ObjectStore in order
 // Create object store, Create User, Connect to Object Store, Create Bucket, Read/Write/Delete to bucket,
 // Check issues in MGRs, Delete Bucket and Delete user
 // Test for ObjectStore with and without TLS enabled
-func runObjectE2ETest(helper *clients.TestClient, k8sh *utils.K8sHelper, installer *installer.CephInstaller, s *suite.Suite, namespace string, tlsEnable bool) {
+func runObjectE2ETest(helper *clients.TestClient, k8sh *utils.K8sHelper, installer *installer.CephInstaller, s *suite.Suite, namespace string, tlsEnable bool, swiftAndKeystone bool) {
 	storeName := "test-store"
 	if tlsEnable {
 		storeName = objectStoreTLSName
 	}
 
 	logger.Infof("Running on Rook Cluster %s", namespace)
-	createCephObjectStore(s.T(), helper, k8sh, installer, namespace, storeName, 3, tlsEnable)
+	createCephObjectStore(s.T(), helper, k8sh, installer, namespace, storeName, 3, tlsEnable, swiftAndKeystone)
 
 	// test that a second object store can be created (and deleted) while the first exists
 	s.T().Run("run a second object store", func(t *testing.T) {
@@ -143,7 +157,7 @@ func runObjectE2ETest(helper *clients.TestClient, k8sh *utils.K8sHelper, install
 	testObjectStoreOperations(s, helper, k8sh, namespace, storeName)
 
 	bucketNotificationTestStoreName := "bucket-notification-" + storeName
-	createCephObjectStore(s.T(), helper, k8sh, installer, namespace, bucketNotificationTestStoreName, 1, tlsEnable)
+	createCephObjectStore(s.T(), helper, k8sh, installer, namespace, bucketNotificationTestStoreName, 1, tlsEnable, false)
 	testBucketNotifications(s, helper, k8sh, namespace, bucketNotificationTestStoreName)
 	if !tlsEnable {
 		// TODO : need to fix COSI driver to support TLS
diff --git a/tests/scripts/create-dev-cluster.sh b/tests/scripts/create-dev-cluster.sh
index fff7195b1443..193e2fd7421a 100755
--- a/tests/scripts/create-dev-cluster.sh
+++ b/tests/scripts/create-dev-cluster.sh
@@ -81,7 +81,7 @@ setup_minikube_env() {
     minikube_driver="$(get_minikube_driver)"
     echo "Setting up minikube env for profile '$ROOK_PROFILE_NAME' (using $minikube_driver driver)"
     $MINIKUBE delete
-    $MINIKUBE start --disk-size=40g --extra-disks=3 --driver "$minikube_driver"
+    $MINIKUBE start --disk-size=40g --extra-disks=3 --driver "$minikube_driver" $ROOK_MINIKUBE_EXTRA_ARGS
     eval "$($MINIKUBE docker-env)"
 }