object: adding support for sse s3 for RGW

The RGW support server side encryption with help of s3 protocol, till
now the `sse:kms` was support in which keys will be provided by the user
and but it will be saved in external management service like vault. Now
the support for `sse:s3` is added so the entire encryption key
management is performed by RGW itsels.

Signed-off-by: Jiffin Tony Thottan <thottanjiffin@gmail.com>

Documentation/CRDs/Object-Storage/ceph-object-store-crd.md

@@ -201,9 +201,9 @@ Rook-Ceph always keeps the bucket and the user for the health check, it just doe
 
 ## Security settings
 
-Ceph RGW supports encryption via Key Management System (KMS) using HashiCorp Vault. Refer to the [vault kms section](../Cluster/ceph-cluster-crd.md#vault-kms) for detailed explanation.
-If these settings are defined, then RGW establish a connection between Vault and whenever S3 client sends a request with Server Side Encryption,
-it encrypts that using the key specified by the client. For more details w.r.t RGW, please refer [Ceph Vault documentation](https://docs.ceph.com/en/latest/radosgw/vault/)
+Ceph RGW supports Server Side Encryption as defined in [AWS S3 protocol](https://docs.aws.amazon.com/AmazonS3/latest/userguide/serv-side-encryption.html) with three different modes: AWS-SSE:C, AWS-SSE:KMS and AWS-SSE:S3. The last two modes require a Key Management System (KMS) like HashiCorp Vault. Currently, Vault is the only supported KMS backend for CephObjectStore.
+
+Refer to the [Vault KMS section](../../Storage-Configuration/Advanced/key-management-system.md#vault) for details about Vault. If these settings are defined, then RGW will establish a connection between Vault and whenever S3 client sends request with Server Side Encryption. [Ceph's Vault documentation](https://docs.ceph.com/en/latest/radosgw/vault/) has more details.
 
 The `security` section contains settings related to KMS encryption of the RGW.
 
@@ -217,13 +217,21 @@ security:
       VAULT_SECRET_ENGINE: kv
       VAULT_BACKEND: v2
     # name of the k8s secret containing the kms authentication token
-    tokenSecretName: rgw-vault-token
+    tokenSecretName: rgw-vault-kms-token
+  s3:
+    connectionDetails:
+      KMS_PROVIDER: vault
+      VAULT_ADDR: http://vault.default.svc.cluster.local:8200
+      VAULT_BACKEND_PATH: rgw
+      VAULT_SECRET_ENGINE: transit
+    # name of the k8s secret containing the kms authentication token
+    tokenSecretName: rgw-vault-s3-token
 ```
 
 For RGW, please note the following:
 
-* `VAULT_SECRET_ENGINE` option is specifically for RGW to mention about the secret engine which can be used, currently supports two: [kv](https://www.vaultproject.io/docs/secrets/kv) and [transit](https://www.vaultproject.io/docs/secrets/transit). And for kv engine only version 2 is supported.
-* The Storage administrator needs to create a secret in the Vault server so that S3 clients use that key for encryption
+* `VAULT_SECRET_ENGINE`: the secret engine which Vault should use. Currently supports [kv](https://www.vaultproject.io/docs/secrets/kv) and [transit](https://www.vaultproject.io/docs/secrets/transit). AWS-SSE:KMS supports `transit` engine and `kv` engine version 2. AWS-SSE:S3 only supports `transit` engine.
+* The Storage administrator needs to create a secret in the Vault server so that S3 clients use that key for encryption for AWS-SSE:KMS
 
 ```console
 vault kv put rook/<mybucketkey> key=$(openssl rand -base64 32) # kv engine
@@ -232,6 +240,10 @@ vault write -f transit/keys/<mybucketkey> exportable=true # transit engine
 
 * TLS authentication with custom certificates between Vault and CephObjectStore RGWs are supported from ceph v16.2.6 onwards
 
+* `tokenSecretName` can be (and often will be) the same for both kms and s3 configurations.
+
+* `AWS-SSE:S3` requires Ceph Quincy (v17.2.3) and later.
+
 ## Deleting a CephObjectStore
 
 During deletion of a CephObjectStore resource, Rook protects against accidental or premature

PendingReleaseNotes.md

@@ -7,3 +7,4 @@
 ## Features
 
 - The toolbox pod now uses the Ceph image directly instead of the Rook image
+- Add support for AWS Server Side Encryption with (AWS-SSE:S3)[https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingServerSideEncryption.html] for RGW.

cmd/rook/secret.go

@@ -73,7 +73,7 @@ func startSecret() *kms.Config {
 	}
 
 	// Validate connection details
-	err = kms.ValidateConnectionDetails(ctx, context, &cephCluster.Spec.Security, namespace)
+	err = kms.ValidateConnectionDetails(ctx, context, &cephCluster.Spec.Security.KeyManagementService, namespace)
 	if err != nil {
 		rook.TerminateFatal(errors.Wrap(err, "failed to validate kms connection details"))
 	}

deploy/charts/rook-ceph/templates/resources.yaml

@@ -8759,6 +8759,21 @@ spec:
                           description: TokenSecretName is the kubernetes secret containing the KMS token
                           type: string
                       type: object
+                    s3:
+                      description: The settings for supporting AWS-SSE:S3 with RGW
+                      nullable: true
+                      properties:
+                        connectionDetails:
+                          additionalProperties:
+                            type: string
+                          description: ConnectionDetails contains the KMS connection details (address, port etc)
+                          nullable: true
+                          type: object
+                          x-kubernetes-preserve-unknown-fields: true
+                        tokenSecretName:
+                          description: TokenSecretName is the kubernetes secret containing the KMS token
+                          type: string
+                      type: object
                   type: object
                 zone:
                   description: The multisite info

deploy/examples/crds.yaml

@@ -8751,6 +8751,21 @@ spec:
                           description: TokenSecretName is the kubernetes secret containing the KMS token
                           type: string
                       type: object
+                    s3:
+                      description: The settings for supporting AWS-SSE:S3 with RGW
+                      nullable: true
+                      properties:
+                        connectionDetails:
+                          additionalProperties:
+                            type: string
+                          description: ConnectionDetails contains the KMS connection details (address, port etc)
+                          nullable: true
+                          type: object
+                          x-kubernetes-preserve-unknown-fields: true
+                        tokenSecretName:
+                          description: TokenSecretName is the kubernetes secret containing the KMS token
+                          type: string
+                      type: object
                   type: object
                 zone:
                   description: The multisite info

deploy/examples/object-openshift.yaml

@@ -109,7 +109,7 @@ spec:
   # security oriented settings
   # security:
   # To enable the KMS configuration properly don't forget to uncomment the Secret at the end of the file
-  #   kms:
+  #   kms: # configures RGW with AWS-SSE:KMS settings
   #     # name of the config map containing all the kms connection details
   #     connectionDetails:
   #        KMS_PROVIDER: "vault"
@@ -119,6 +119,15 @@ spec:
   #        VAULT_BACKEND: v2
   #     # name of the secret containing the kms authentication token
   #     tokenSecretName: rook-vault-token
+  #   s3: # configures RGW with AWS-SSE:S3 settings
+  #     # name of the config map containing all the kms connection details
+  #     connectionDetails:
+  #        KMS_PROVIDER: "vault"
+  #        VAULT_ADDR: VAULT_ADDR_CHANGE_ME # e,g: http://vault.my-domain.com:8200
+  #        VAULT_BACKEND_PATH: "rook"
+  #        VAULT_SECRET_ENGINE: "transit"
+  #     # name of the secret containing the kms authentication token
+  #     tokenSecretName: rook-vault-token
 # # UNCOMMENT THIS TO ENABLE A KMS CONNECTION
 # # Also, do not forget to replace both:
 # #  * ROOK_TOKEN_CHANGE_ME: with a base64 encoded value of the token to use

deploy/examples/object.yaml

@@ -116,8 +116,8 @@ spec:
       disabled: false
   # security oriented settings
   # security:
-  # To enable the KMS configuration properly don't forget to uncomment the Secret at the end of the file
-  #   kms:
+  # To enable the Server Side Encryption configuration properly don't forget to uncomment the Secret at the end of the file
+  #   kms: # configures RGW with AWS-SSE:KMS
   #     # name of the config map containing all the kms connection details
   #     connectionDetails:
   #        KMS_PROVIDER: "vault"
@@ -127,6 +127,15 @@ spec:
   #        VAULT_BACKEND: v2
   #     # name of the secret containing the kms authentication token
   #     tokenSecretName: rook-vault-token
+  #   s3: # configures RGW with AWS-SSE:S3
+  #     # name of the config map containing all the kms connection details
+  #     connectionDetails:
+  #        KMS_PROVIDER: "vault"
+  #        VAULT_ADDR: VAULT_ADDR_CHANGE_ME # e,g: http://vault.my-domain.com:8200
+  #        VAULT_BACKEND_PATH: "rook"
+  #        VAULT_SECRET_ENGINE: "transit"
+  #     # name of the secret containing the kms authentication token
+  #     tokenSecretName: rook-vault-token
 # # UNCOMMENT THIS TO ENABLE A KMS CONNECTION
 # # Also, do not forget to replace both:
 # #  * ROOK_TOKEN_CHANGE_ME: with a base64 encoded value of the token to use

pkg/apis/ceph.rook.io/v1/types.go

@@ -240,6 +240,18 @@ type SecuritySpec struct {
 	KeyManagementService KeyManagementServiceSpec `json:"kms,omitempty"`
 }
 
+// ObjectStoreSecuritySpec is spec to define security features like encryption
+type ObjectStoreSecuritySpec struct {
+	// +optional
+	// +nullable
+	SecuritySpec `json:""`
+
+	// The settings for supporting AWS-SSE:S3 with RGW
+	// +optional
+	// +nullable
+	ServerSideEncryptionS3 KeyManagementServiceSpec `json:"s3,omitempty"`
+}
+
 // KeyManagementServiceSpec represent various details of the KMS server
 type KeyManagementServiceSpec struct {
 	// ConnectionDetails contains the KMS connection details (address, port etc)
@@ -1320,7 +1332,7 @@ type ObjectStoreSpec struct {
 	// Security represents security settings
 	// +optional
 	// +nullable
-	Security *SecuritySpec `json:"security,omitempty"`
+	Security *ObjectStoreSecuritySpec `json:"security,omitempty"`
 
 	// Whether host networking is enabled for the rgw daemon. If not set, the network settings from the cluster CR will be applied.
 	// +kubebuilder:pruning:PreserveUnknownFields

pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go

@@ -2975,6 +2975,24 @@ func (in *ObjectRealmSpec) DeepCopy() *ObjectRealmSpec {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ObjectStoreSecuritySpec) DeepCopyInto(out *ObjectStoreSecuritySpec) {
+	*out = *in
+	in.SecuritySpec.DeepCopyInto(&out.SecuritySpec)
+	in.ServerSideEncryptionS3.DeepCopyInto(&out.ServerSideEncryptionS3)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ObjectStoreSecuritySpec.
+func (in *ObjectStoreSecuritySpec) DeepCopy() *ObjectStoreSecuritySpec {
+	if in == nil {
+		return nil
+	}
+	out := new(ObjectStoreSecuritySpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ObjectStoreSpec) DeepCopyInto(out *ObjectStoreSpec) {
 	*out = *in
@@ -2985,7 +3003,7 @@ func (in *ObjectStoreSpec) DeepCopyInto(out *ObjectStoreSpec) {
 	in.HealthCheck.DeepCopyInto(&out.HealthCheck)
 	if in.Security != nil {
 		in, out := &in.Security, &out.Security
-		*out = new(SecuritySpec)
+		*out = new(ObjectStoreSecuritySpec)
 		(*in).DeepCopyInto(*out)
 	}
 	if in.HostNetwork != nil {

pkg/daemon/ceph/osd/kms/kms.go

@@ -209,29 +209,29 @@ func GetParam(kmsConfig map[string]string, param string) string {
 }
 
 // ValidateConnectionDetails validates mandatory KMS connection details
-func ValidateConnectionDetails(ctx context.Context, clusterdContext *clusterd.Context, securitySpec *cephv1.SecuritySpec, ns string) error {
+func ValidateConnectionDetails(ctx context.Context, clusterdContext *clusterd.Context, kms *cephv1.KeyManagementServiceSpec, ns string) error {
 	// Lookup mandatory connection details
 	for _, config := range kmsMandatoryConnectionDetails {
-		if GetParam(securitySpec.KeyManagementService.ConnectionDetails, config) == "" {
+		if GetParam(kms.ConnectionDetails, config) == "" {
 			return errors.Errorf("failed to validate kms config %q. cannot be empty", config)
 		}
 	}
 
 	// A token must be specified if token-auth is used
-	if !securitySpec.KeyManagementService.IsK8sAuthEnabled() && securitySpec.KeyManagementService.TokenSecretName == "" {
-		if !securitySpec.KeyManagementService.IsTokenAuthEnabled() {
+	if !kms.IsK8sAuthEnabled() && kms.TokenSecretName == "" {
+		if !kms.IsTokenAuthEnabled() {
 			return errors.New("failed to validate kms configuration (missing token in spec)")
 		}
 	}
 
 	// KMS provider must be specified
-	provider := GetParam(securitySpec.KeyManagementService.ConnectionDetails, Provider)
+	provider := GetParam(kms.ConnectionDetails, Provider)
 
 	// Validate potential token Secret presence
-	if securitySpec.KeyManagementService.IsTokenAuthEnabled() {
-		kmsToken, err := clusterdContext.Clientset.CoreV1().Secrets(ns).Get(ctx, securitySpec.KeyManagementService.TokenSecretName, metav1.GetOptions{})
+	if kms.IsTokenAuthEnabled() {
+		kmsToken, err := clusterdContext.Clientset.CoreV1().Secrets(ns).Get(ctx, kms.TokenSecretName, metav1.GetOptions{})
 		if err != nil {
-			return errors.Wrapf(err, "failed to fetch kms token secret %q", securitySpec.KeyManagementService.TokenSecretName)
+			return errors.Wrapf(err, "failed to fetch kms token secret %q", kms.TokenSecretName)
 		}
 
 		switch provider {
@@ -239,7 +239,7 @@ func ValidateConnectionDetails(ctx context.Context, clusterdContext *clusterd.Co
 			// Check for empty token
 			token, ok := kmsToken.Data[KMSTokenSecretNameKey]
 			if !ok || len(token) == 0 {
-				return errors.Errorf("failed to read k8s kms secret %q key %q (not found or empty)", KMSTokenSecretNameKey, securitySpec.KeyManagementService.TokenSecretName)
+				return errors.Errorf("failed to read k8s kms secret %q key %q (not found or empty)", KMSTokenSecretNameKey, kms.TokenSecretName)
 			}
 
 			// Set the env variable
@@ -252,10 +252,10 @@ func ValidateConnectionDetails(ctx context.Context, clusterdContext *clusterd.Co
 			for _, config := range kmsIBMKeyProtectMandatoryTokenDetails {
 				v, ok := kmsToken.Data[config]
 				if !ok || len(v) == 0 {
-					return errors.Errorf("failed to read k8s kms secret %q key %q (not found or empty)", config, securitySpec.KeyManagementService.TokenSecretName)
+					return errors.Errorf("failed to read k8s kms secret %q key %q (not found or empty)", config, kms.TokenSecretName)
 				}
 				// Append the token secret details to the connection details
-				securitySpec.KeyManagementService.ConnectionDetails[config] = strings.TrimSuffix(strings.TrimSpace(string(v)), "\n")
+				kms.ConnectionDetails[config] = strings.TrimSuffix(strings.TrimSpace(string(v)), "\n")
 			}
 		}
 	}
@@ -263,27 +263,27 @@ func ValidateConnectionDetails(ctx context.Context, clusterdContext *clusterd.Co
 	// Validate KMS provider connection details for each provider
 	switch provider {
 	case secrets.TypeVault:
-		err := validateVaultConnectionDetails(ctx, clusterdContext, ns, securitySpec.KeyManagementService.ConnectionDetails)
+		err := validateVaultConnectionDetails(ctx, clusterdContext, ns, kms.ConnectionDetails)
 		if err != nil {
 			return errors.Wrap(err, "failed to validate vault connection details")
 		}
 
-		secretEngine := securitySpec.KeyManagementService.ConnectionDetails[VaultSecretEngineKey]
+		secretEngine := kms.ConnectionDetails[VaultSecretEngineKey]
 		switch secretEngine {
 		case VaultKVSecretEngineKey:
 			// Append Backend Version if not already present
-			if GetParam(securitySpec.KeyManagementService.ConnectionDetails, vault.VaultBackendKey) == "" {
-				backendVersion, err := BackendVersion(ctx, clusterdContext, ns, securitySpec.KeyManagementService.ConnectionDetails)
+			if GetParam(kms.ConnectionDetails, vault.VaultBackendKey) == "" {
+				backendVersion, err := BackendVersion(ctx, clusterdContext, ns, kms.ConnectionDetails)
 				if err != nil {
 					return errors.Wrap(err, "failed to get backend version")
 				}
-				securitySpec.KeyManagementService.ConnectionDetails[vault.VaultBackendKey] = backendVersion
+				kms.ConnectionDetails[vault.VaultBackendKey] = backendVersion
 			}
 		}
 
 	case TypeIBM:
 		for _, config := range kmsIBMKeyProtectMandatoryConnectionDetails {
-			if GetParam(securitySpec.KeyManagementService.ConnectionDetails, config) == "" {
+			if GetParam(kms.ConnectionDetails, config) == "" {
 				return errors.Errorf("failed to validate kms config %q. cannot be empty", config)
 			}
 		}

pkg/daemon/ceph/osd/kms/kms_test.go

@@ -35,7 +35,7 @@ func TestValidateConnectionDetails(t *testing.T) {
 	ctx := context.TODO()
 	// Placeholder
 	clusterdContext := &clusterd.Context{Clientset: test.New(t, 3)}
-	securitySpec := &cephv1.SecuritySpec{KeyManagementService: cephv1.KeyManagementServiceSpec{ConnectionDetails: map[string]string{}}}
+	kms := &cephv1.KeyManagementServiceSpec{ConnectionDetails: map[string]string{}}
 	ns := "rook-ceph"
 	vaultSecret := &v1.Secret{
 		ObjectMeta: metav1.ObjectMeta{
@@ -56,22 +56,22 @@ func TestValidateConnectionDetails(t *testing.T) {
 			Namespace: ns,
 		},
 	}
-	ibmSecuritySpec := &cephv1.SecuritySpec{KeyManagementService: cephv1.KeyManagementServiceSpec{
+	ibmKMSSpec := &cephv1.KeyManagementServiceSpec{
 		ConnectionDetails: map[string]string{
 			"KMS_PROVIDER": TypeIBM,
 		},
-	}}
+	}
 
 	t.Run("no kms provider given", func(t *testing.T) {
-		err := ValidateConnectionDetails(ctx, clusterdContext, securitySpec, ns)
+		err := ValidateConnectionDetails(ctx, clusterdContext, kms, ns)
 		assert.Error(t, err, "")
 		assert.EqualError(t, err, "failed to validate kms config \"KMS_PROVIDER\". cannot be empty")
-		securitySpec.KeyManagementService.ConnectionDetails["KMS_PROVIDER"] = "vault"
+		kms.ConnectionDetails["KMS_PROVIDER"] = "vault"
 	})
 
 	t.Run("vault - no token object", func(t *testing.T) {
-		securitySpec.KeyManagementService.TokenSecretName = "vault-token"
-		err := ValidateConnectionDetails(ctx, clusterdContext, securitySpec, ns)
+		kms.TokenSecretName = "vault-token"
+		err := ValidateConnectionDetails(ctx, clusterdContext, kms, ns)
 		assert.Error(t, err, "")
 		assert.EqualError(t, err, "failed to fetch kms token secret \"vault-token\": secrets \"vault-token\" not found")
 	})
@@ -79,7 +79,7 @@ func TestValidateConnectionDetails(t *testing.T) {
 	t.Run("vault - token secret present but empty content", func(t *testing.T) {
 		_, err := clusterdContext.Clientset.CoreV1().Secrets(ns).Create(ctx, vaultSecret, metav1.CreateOptions{})
 		assert.NoError(t, err)
-		err = ValidateConnectionDetails(ctx, clusterdContext, securitySpec, ns)
+		err = ValidateConnectionDetails(ctx, clusterdContext, kms, ns)
 		assert.Error(t, err, "")
 		assert.EqualError(t, err, "failed to read k8s kms secret \"token\" key \"vault-token\" (not found or empty)")
 	})
@@ -88,7 +88,7 @@ func TestValidateConnectionDetails(t *testing.T) {
 		vaultSecret.Data = map[string][]byte{"foo": []byte("bar")}
 		_, err := clusterdContext.Clientset.CoreV1().Secrets(ns).Update(ctx, vaultSecret, metav1.UpdateOptions{})
 		assert.NoError(t, err)
-		err = ValidateConnectionDetails(ctx, clusterdContext, securitySpec, ns)
+		err = ValidateConnectionDetails(ctx, clusterdContext, kms, ns)
 		assert.Error(t, err, "")
 		assert.EqualError(t, err, "failed to read k8s kms secret \"token\" key \"vault-token\" (not found or empty)")
 	})
@@ -98,15 +98,15 @@ func TestValidateConnectionDetails(t *testing.T) {
 		vaultSecret.Data["token"] = []byte("token")
 		_, err := clusterdContext.Clientset.CoreV1().Secrets(ns).Update(ctx, vaultSecret, metav1.UpdateOptions{})
 		assert.NoError(t, err)
-		err = ValidateConnectionDetails(ctx, clusterdContext, securitySpec, ns)
+		err = ValidateConnectionDetails(ctx, clusterdContext, kms, ns)
 		assert.Error(t, err, "")
 		assert.EqualError(t, err, "failed to validate vault connection details: failed to find connection details \"VAULT_ADDR\"")
-		securitySpec.KeyManagementService.ConnectionDetails["VAULT_ADDR"] = "https://1.1.1.1:8200"
+		kms.ConnectionDetails["VAULT_ADDR"] = "https://1.1.1.1:8200"
 	})
 
 	t.Run("vault - TLS is configured but secrets do not exist", func(t *testing.T) {
-		securitySpec.KeyManagementService.ConnectionDetails["VAULT_CACERT"] = "vault-ca-secret"
-		err := ValidateConnectionDetails(ctx, clusterdContext, securitySpec, ns)
+		kms.ConnectionDetails["VAULT_CACERT"] = "vault-ca-secret"
+		err := ValidateConnectionDetails(ctx, clusterdContext, kms, ns)
 		assert.Error(t, err, "")
 		assert.EqualError(t, err, "failed to validate vault connection details: failed to find TLS connection details k8s secret \"vault-ca-secret\": secrets \"vault-ca-secret\" not found")
 	})
@@ -114,7 +114,7 @@ func TestValidateConnectionDetails(t *testing.T) {
 	t.Run("vault - TLS secret exists but empty key", func(t *testing.T) {
 		_, err := clusterdContext.Clientset.CoreV1().Secrets(ns).Create(ctx, tlsSecret, metav1.CreateOptions{})
 		assert.NoError(t, err)
-		err = ValidateConnectionDetails(ctx, clusterdContext, securitySpec, ns)
+		err = ValidateConnectionDetails(ctx, clusterdContext, kms, ns)
 		assert.Error(t, err, "")
 		assert.EqualError(t, err, "failed to validate vault connection details: failed to find TLS connection key \"cert\" for \"VAULT_CACERT\" in k8s secret \"vault-ca-secret\"")
 	})
@@ -123,7 +123,7 @@ func TestValidateConnectionDetails(t *testing.T) {
 		tlsSecret.Data = map[string][]byte{"cert": []byte("envnrevbnbvsbjkrtn")}
 		_, err := clusterdContext.Clientset.CoreV1().Secrets(ns).Update(ctx, tlsSecret, metav1.UpdateOptions{})
 		assert.NoError(t, err)
-		err = ValidateConnectionDetails(ctx, clusterdContext, securitySpec, ns)
+		err = ValidateConnectionDetails(ctx, clusterdContext, kms, ns)
 		assert.NoError(t, err, "")
 	})
 
@@ -145,34 +145,32 @@ func TestValidateConnectionDetails(t *testing.T) {
 		}); err != nil {
 			t.Fatal(err)
 		}
-		securitySpec := &cephv1.SecuritySpec{
-			KeyManagementService: cephv1.KeyManagementServiceSpec{
-				ConnectionDetails: map[string]string{
-					"VAULT_SECRET_ENGINE": "kv",
-					"KMS_PROVIDER":        "vault",
-					"VAULT_ADDR":          client.Address(),
-					"VAULT_BACKEND_PATH":  "rook",
-				},
-				TokenSecretName: "vault-token",
+		kms := &cephv1.KeyManagementServiceSpec{
+			ConnectionDetails: map[string]string{
+				"VAULT_SECRET_ENGINE": "kv",
+				"KMS_PROVIDER":        "vault",
+				"VAULT_ADDR":          client.Address(),
+				"VAULT_BACKEND_PATH":  "rook",
 			},
+			TokenSecretName: "vault-token",
 		}
-		err := ValidateConnectionDetails(ctx, clusterdContext, securitySpec, ns)
+		err := ValidateConnectionDetails(ctx, clusterdContext, kms, ns)
 		assert.NoError(t, err, "")
-		assert.Equal(t, securitySpec.KeyManagementService.ConnectionDetails["VAULT_BACKEND"], "v2")
+		assert.Equal(t, kms.ConnectionDetails["VAULT_BACKEND"], "v2")
 	})
 
 	t.Run("ibm kp - fail no token specified, only token is supported", func(t *testing.T) {
-		err := ValidateConnectionDetails(ctx, clusterdContext, ibmSecuritySpec, ns)
+		err := ValidateConnectionDetails(ctx, clusterdContext, ibmKMSSpec, ns)
 		assert.Error(t, err, "")
 		assert.EqualError(t, err, "failed to validate kms configuration (missing token in spec)")
-		ibmSecuritySpec.KeyManagementService.TokenSecretName = "ibm-token"
+		ibmKMSSpec.TokenSecretName = "ibm-token"
 
 	})
 
 	t.Run("ibm kp - token present but no key for service key", func(t *testing.T) {
 		_, err := clusterdContext.Clientset.CoreV1().Secrets(ns).Create(ctx, ibmSecret, metav1.CreateOptions{})
 		assert.NoError(t, err)
-		err = ValidateConnectionDetails(ctx, clusterdContext, ibmSecuritySpec, ns)
+		err = ValidateConnectionDetails(ctx, clusterdContext, ibmKMSSpec, ns)
 		assert.Error(t, err, "")
 		assert.EqualError(t, err, "failed to read k8s kms secret \"IBM_KP_SERVICE_API_KEY\" key \"ibm-token\" (not found or empty)")
 	})
@@ -181,18 +179,18 @@ func TestValidateConnectionDetails(t *testing.T) {
 		ibmSecret.Data["IBM_KP_SERVICE_API_KEY"] = []byte("foo")
 		_, err := clusterdContext.Clientset.CoreV1().Secrets(ns).Update(ctx, ibmSecret, metav1.UpdateOptions{})
 		assert.NoError(t, err)
-		err = ValidateConnectionDetails(ctx, clusterdContext, ibmSecuritySpec, ns)
+		err = ValidateConnectionDetails(ctx, clusterdContext, ibmKMSSpec, ns)
 		assert.Error(t, err, "")
 		assert.EqualError(t, err, "failed to validate kms config \"IBM_KP_SERVICE_INSTANCE_ID\". cannot be empty")
-		ibmSecuritySpec.KeyManagementService.ConnectionDetails["IBM_KP_SERVICE_INSTANCE_ID"] = "foo"
+		ibmKMSSpec.ConnectionDetails["IBM_KP_SERVICE_INSTANCE_ID"] = "foo"
 	})
 
 	t.Run("ibm kp - success", func(t *testing.T) {
-		err := ValidateConnectionDetails(ctx, clusterdContext, ibmSecuritySpec, ns)
+		err := ValidateConnectionDetails(ctx, clusterdContext, ibmKMSSpec, ns)
 		assert.NoError(t, err, "")
 		// IBM_KP_SERVICE_API_KEY must be appended to the details so that the client can be built with
 		// all the details
-		assert.Equal(t, ibmSecuritySpec.KeyManagementService.ConnectionDetails["IBM_KP_SERVICE_API_KEY"], "foo")
+		assert.Equal(t, ibmKMSSpec.ConnectionDetails["IBM_KP_SERVICE_API_KEY"], "foo")
 	})
 }
 

pkg/daemon/ceph/osd/kms/volumes.go

@@ -17,6 +17,8 @@ limitations under the License.
 package kms
 
 import (
+	"path"
+
 	"github.com/hashicorp/vault/api"
 	"github.com/libopenstorage/secrets"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -68,8 +70,15 @@ func VaultSecretVolumeAndMount(kmsVaultConfigFiles map[string]string, tokenSecre
 
 // VaultVolumeAndMount returns Vault volume and volume mount
 func VaultVolumeAndMount(kmsVaultConfigFiles map[string]string, tokenSecretName string) (v1.Volume, v1.VolumeMount) {
+	return VaultVolumeAndMountWithCustomName(kmsVaultConfigFiles, tokenSecretName, "")
+}
+
+func VaultVolumeAndMountWithCustomName(kmsVaultConfigFiles map[string]string, tokenSecretName, customName string) (v1.Volume, v1.VolumeMount) {
+	if len(kmsVaultConfigFiles) == 0 && len(tokenSecretName) == 0 {
+		return v1.Volume{}, v1.VolumeMount{}
+	}
 	v := v1.Volume{
-		Name: secrets.TypeVault,
+		Name: secrets.TypeVault + customName,
 		VolumeSource: v1.VolumeSource{
 			Projected: &v1.ProjectedVolumeSource{
 				Sources: VaultSecretVolumeAndMount(kmsVaultConfigFiles, tokenSecretName),
@@ -77,15 +86,18 @@ func VaultVolumeAndMount(kmsVaultConfigFiles map[string]string, tokenSecretName
 		},
 	}
 
+	mountPath := EtcVaultDir
+	if customName != "" {
+		mountPath = path.Join(mountPath, customName)
+	}
 	m := v1.VolumeMount{
-		Name:      secrets.TypeVault,
+		Name:      secrets.TypeVault + customName,
 		ReadOnly:  true,
-		MountPath: EtcVaultDir,
+		MountPath: mountPath,
 	}
 
 	return v, m
 }
-
 func tlsSecretPath(tlsOption string) string {
 	switch tlsOption {
 	case api.EnvVaultCACert:

pkg/daemon/ceph/osd/kms/volumes_test.go

@@ -17,9 +17,11 @@ limitations under the License.
 package kms
 
 import (
+	"path"
 	"reflect"
 	"testing"
 
+	"github.com/libopenstorage/secrets"
 	v1 "k8s.io/api/core/v1"
 )
 
@@ -96,3 +98,55 @@ func TestVaultSecretVolumeAndMount(t *testing.T) {
 		})
 	}
 }
+
+func TestVaultVolumeAndMountWithCustomName(t *testing.T) {
+	m := int32(0444)
+	type args struct {
+		config          map[string]string
+		tokenSecretName string
+		customName      string
+	}
+	tests := []struct {
+		name         string
+		args         args
+		wantVol      v1.Volume
+		wantVolMount v1.VolumeMount
+	}{
+		{"empty without custom name", args{config: map[string]string{}, tokenSecretName: "", customName: ""}, v1.Volume{}, v1.VolumeMount{}},
+		{"no kms related configs without custom name", args{config: map[string]string{"foo": "bar"}, tokenSecretName: "", customName: ""}, v1.Volume{Name: secrets.TypeVault, VolumeSource: v1.VolumeSource{Projected: &v1.ProjectedVolumeSource{Sources: []v1.VolumeProjection{}}}}, v1.VolumeMount{Name: secrets.TypeVault, ReadOnly: true, MountPath: EtcVaultDir}},
+		{"only cert passed without custom name", args{config: map[string]string{"VAULT_CACERT": "vault-ca-secret"}, tokenSecretName: "", customName: ""}, v1.Volume{Name: secrets.TypeVault, VolumeSource: v1.VolumeSource{Projected: &v1.ProjectedVolumeSource{Sources: []v1.VolumeProjection{
+			{Secret: &v1.SecretProjection{LocalObjectReference: v1.LocalObjectReference{Name: "vault-ca-secret"}, Items: []v1.KeyToPath{{Key: "cert", Path: "vault.ca", Mode: &m}}, Optional: nil}}}}}}, v1.VolumeMount{Name: secrets.TypeVault, ReadOnly: true, MountPath: EtcVaultDir},
+		},
+		{"only token passed without custom name", args{tokenSecretName: "vault-token", config: map[string]string{"foo": "bar"}, customName: ""}, v1.Volume{Name: secrets.TypeVault, VolumeSource: v1.VolumeSource{Projected: &v1.ProjectedVolumeSource{Sources: []v1.VolumeProjection{
+			{Secret: &v1.SecretProjection{LocalObjectReference: v1.LocalObjectReference{Name: "vault-token"}, Items: []v1.KeyToPath{{Key: "token", Path: "vault.token", Mode: &m}}, Optional: nil}}}}}}, v1.VolumeMount{Name: secrets.TypeVault, ReadOnly: true, MountPath: EtcVaultDir},
+		},
+		{"both token and cert passed without custom name", args{tokenSecretName: "vault-token", config: map[string]string{"VAULT_CACERT": "vault-ca-secret"}, customName: ""}, v1.Volume{Name: secrets.TypeVault, VolumeSource: v1.VolumeSource{Projected: &v1.ProjectedVolumeSource{Sources: []v1.VolumeProjection{
+			{Secret: &v1.SecretProjection{LocalObjectReference: v1.LocalObjectReference{Name: "vault-ca-secret"}, Items: []v1.KeyToPath{{Key: "cert", Path: "vault.ca", Mode: &m}}, Optional: nil}},
+			{Secret: &v1.SecretProjection{LocalObjectReference: v1.LocalObjectReference{Name: "vault-token"}, Items: []v1.KeyToPath{{Key: "token", Path: "vault.token", Mode: &m}}, Optional: nil}}}}}}, v1.VolumeMount{Name: secrets.TypeVault, ReadOnly: true, MountPath: EtcVaultDir},
+		},
+		{"empty with custom name", args{config: map[string]string{}, tokenSecretName: "", customName: "custom"}, v1.Volume{}, v1.VolumeMount{}},
+		{"no kms related configs with custom name", args{config: map[string]string{"foo": "bar"}, tokenSecretName: "", customName: "custom"}, v1.Volume{Name: secrets.TypeVault + "custom", VolumeSource: v1.VolumeSource{Projected: &v1.ProjectedVolumeSource{Sources: []v1.VolumeProjection{}}}}, v1.VolumeMount{Name: secrets.TypeVault + "custom", ReadOnly: true, MountPath: path.Join(EtcVaultDir, "custom")}},
+		{"only cert passed with custom name", args{config: map[string]string{"VAULT_CACERT": "vault-ca-secret"}, tokenSecretName: "", customName: "custom"}, v1.Volume{Name: secrets.TypeVault + "custom", VolumeSource: v1.VolumeSource{Projected: &v1.ProjectedVolumeSource{Sources: []v1.VolumeProjection{
+			{Secret: &v1.SecretProjection{LocalObjectReference: v1.LocalObjectReference{Name: "vault-ca-secret"}, Items: []v1.KeyToPath{{Key: "cert", Path: "vault.ca", Mode: &m}}, Optional: nil}}}}}}, v1.VolumeMount{Name: secrets.TypeVault + "custom", ReadOnly: true, MountPath: path.Join(EtcVaultDir, "custom")},
+		},
+		{"only token passed with custom name", args{tokenSecretName: "vault-token", config: map[string]string{"foo": "bar"}, customName: "custom"}, v1.Volume{Name: secrets.TypeVault + "custom", VolumeSource: v1.VolumeSource{Projected: &v1.ProjectedVolumeSource{Sources: []v1.VolumeProjection{
+			{Secret: &v1.SecretProjection{LocalObjectReference: v1.LocalObjectReference{Name: "vault-token"}, Items: []v1.KeyToPath{{Key: "token", Path: "vault.token", Mode: &m}}, Optional: nil}}}}}}, v1.VolumeMount{Name: secrets.TypeVault + "custom", ReadOnly: true, MountPath: path.Join(EtcVaultDir, "custom")},
+		},
+		{"both token and cert passed with custom name", args{tokenSecretName: "vault-token", config: map[string]string{"VAULT_CACERT": "vault-ca-secret"}, customName: "custom"}, v1.Volume{Name: secrets.TypeVault + "custom", VolumeSource: v1.VolumeSource{Projected: &v1.ProjectedVolumeSource{Sources: []v1.VolumeProjection{
+			{Secret: &v1.SecretProjection{LocalObjectReference: v1.LocalObjectReference{Name: "vault-ca-secret"}, Items: []v1.KeyToPath{{Key: "cert", Path: "vault.ca", Mode: &m}}, Optional: nil}},
+			{Secret: &v1.SecretProjection{LocalObjectReference: v1.LocalObjectReference{Name: "vault-token"}, Items: []v1.KeyToPath{{Key: "token", Path: "vault.token", Mode: &m}}, Optional: nil}}}}}}, v1.VolumeMount{Name: secrets.TypeVault + "custom", ReadOnly: true, MountPath: path.Join(EtcVaultDir, "custom")},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			gotVol, gotVolMount := VaultVolumeAndMountWithCustomName(tt.args.config, tt.args.tokenSecretName, tt.args.customName)
+			if !reflect.DeepEqual(gotVol, tt.wantVol) {
+				t.Errorf("VaultVolumeAndMountWithCustomName() = %v, want %v", gotVol, tt.wantVol)
+			}
+			if !reflect.DeepEqual(gotVolMount, tt.wantVolMount) {
+				t.Errorf("VaultVolumeAndMountWithCustomName() = %v, want %v", gotVolMount, tt.wantVolMount)
+			}
+
+		})
+	}
+}

pkg/operator/ceph/cluster/cluster.go

@@ -315,7 +315,7 @@ func preClusterStartValidation(cluster *cluster) error {
 	// Validate on-PVC cluster encryption KMS settings
 	if cluster.Spec.Storage.IsOnPVCEncrypted() && cluster.Spec.Security.KeyManagementService.IsEnabled() {
 		// Validate the KMS details
-		err := kms.ValidateConnectionDetails(cluster.ClusterInfo.Context, cluster.context, &cluster.Spec.Security, cluster.Namespace)
+		err := kms.ValidateConnectionDetails(cluster.ClusterInfo.Context, cluster.context, &cluster.Spec.Security.KeyManagementService, cluster.Namespace)
 		if err != nil {
 			return errors.Wrap(err, "failed to validate kms connection details")
 		}

pkg/operator/ceph/cluster/controller.go

@@ -555,7 +555,7 @@ func (c *ClusterController) deleteOSDEncryptionKeyFromKMS(currentCluster *cephv1
 	// We need to fetch the IBM_KP_SERVICE_API_KEY value
 	if currentCluster.Spec.Security.KeyManagementService.IsIBMKeyProtectKMS() {
 		// This will validate the connection details again and will add the IBM_KP_SERVICE_API_KEY to the spec
-		err = kms.ValidateConnectionDetails(ctx, c.context, &currentCluster.Spec.Security, currentCluster.Namespace)
+		err = kms.ValidateConnectionDetails(ctx, c.context, &currentCluster.Spec.Security.KeyManagementService, currentCluster.Namespace)
 		if err != nil {
 			return errors.Wrap(err, "failed to validate kms connection details to delete the secret")
 		}

pkg/operator/ceph/object/spec.go

@@ -19,7 +19,6 @@ package object
 import (
 	"fmt"
 	"path"
-	"reflect"
 	"strings"
 
 	"github.com/hashicorp/vault/api"
@@ -40,21 +39,37 @@ import (
 const (
 	readinessProbePath = "/swift/healthcheck"
 	serviceAccountName = "rook-ceph-rgw"
+	sseKMS             = "ssekms"
+	sseS3              = "sses3"
+	vaultPrefix        = "/v1/"
 	//nolint:gosec // since this is not leaking any hardcoded details
 	setupVaultTokenFile = `
 set -e
 
 VAULT_TOKEN_OLD_PATH=%s
 VAULT_TOKEN_NEW_PATH=%s
-
-cp --recursive --verbose $VAULT_TOKEN_OLD_PATH/..data/. $VAULT_TOKEN_NEW_PATH
-
-chmod --recursive --verbose 400 $VAULT_TOKEN_NEW_PATH/*
+if [ -d $VAULT_TOKEN_OLD_PATH/ssekms ]
+then
+cp --recursive --verbose $VAULT_TOKEN_OLD_PATH/ssekms/..data/. $VAULT_TOKEN_NEW_PATH/ssekms
+chmod --recursive --verbose 400 $VAULT_TOKEN_NEW_PATH/ssekms/*
+chmod --verbose 700 $VAULT_TOKEN_NEW_PATH/ssekms
+fi
+if [ -d $VAULT_TOKEN_OLD_PATH/sses3 ]
+then
+cp --recursive --verbose $VAULT_TOKEN_OLD_PATH/sses3/..data/. $VAULT_TOKEN_NEW_PATH/sses3
+chmod --recursive --verbose 400 $VAULT_TOKEN_NEW_PATH/sses3/*
+chmod --verbose 700 $VAULT_TOKEN_NEW_PATH/sses3
+fi
 chmod --verbose 700 $VAULT_TOKEN_NEW_PATH
 chown --recursive --verbose ceph:ceph $VAULT_TOKEN_NEW_PATH
 `
 )
 
+var (
+	cephVersionMinRGWSSES3     = cephver.CephVersion{Major: 17, Minor: 2, Extra: 3}
+	cephVersionMinRGWSSEKMSTLS = cephver.CephVersion{Major: 16, Minor: 2, Extra: 6}
+)
+
 func (c *clusterConfig) createDeployment(rgwConfig *rgwConfig) (*apps.Deployment, error) {
 	pod, err := c.makeRGWPodSpec(rgwConfig)
 	if err != nil {
@@ -95,9 +110,9 @@ func (c *clusterConfig) createDeployment(rgwConfig *rgwConfig) (*apps.Deployment
 }
 
 func (c *clusterConfig) makeRGWPodSpec(rgwConfig *rgwConfig) (v1.PodTemplateSpec, error) {
-	rgwDaemonContainer := c.makeDaemonContainer(rgwConfig)
-	if reflect.DeepEqual(rgwDaemonContainer, v1.Container{}) {
-		return v1.PodTemplateSpec{}, errors.New("got empty container for RGW daemon")
+	rgwDaemonContainer, err := c.makeDaemonContainer(rgwConfig)
+	if err != nil {
+		return v1.PodTemplateSpec{}, err
 	}
 
 	hostNetwork := c.store.Spec.IsHostNetwork(c.clusterSpec)
@@ -164,21 +179,32 @@ func (c *clusterConfig) makeRGWPodSpec(rgwConfig *rgwConfig) (v1.PodTemplateSpec
 	if err != nil {
 		return v1.PodTemplateSpec{}, err
 	}
-	if kmsEnabled {
-		if c.store.Spec.Security.KeyManagementService.IsTokenAuthEnabled() {
-			vaultFileVol, _ := kms.VaultVolumeAndMount(c.store.Spec.Security.KeyManagementService.ConnectionDetails,
-				c.store.Spec.Security.KeyManagementService.TokenSecretName)
-			tmpvolume := v1.Volume{
-				Name: rgwVaultVolumeName,
-				VolumeSource: v1.VolumeSource{
-					EmptyDir: &v1.EmptyDirVolumeSource{},
-				},
-			}
-
-			podSpec.Volumes = append(podSpec.Volumes, vaultFileVol, tmpvolume)
-			podSpec.InitContainers = append(podSpec.InitContainers,
-				c.vaultTokenInitContainer(rgwConfig))
+	s3Enabled, err := c.CheckRGWSSES3Enabled()
+	if err != nil {
+		return v1.PodTemplateSpec{}, err
+	}
+	if kmsEnabled || s3Enabled {
+		v := v1.Volume{
+			Name: rgwVaultVolumeName,
+			VolumeSource: v1.VolumeSource{
+				EmptyDir: &v1.EmptyDirVolumeSource{},
+			},
 		}
+		podSpec.Volumes = append(podSpec.Volumes, v)
+
+		if kmsEnabled && c.store.Spec.Security.KeyManagementService.IsTokenAuthEnabled() {
+			vaultFileVol, _ := kms.VaultVolumeAndMountWithCustomName(c.store.Spec.Security.KeyManagementService.ConnectionDetails,
+				c.store.Spec.Security.KeyManagementService.TokenSecretName, sseKMS)
+			podSpec.Volumes = append(podSpec.Volumes, vaultFileVol)
+		}
+		if s3Enabled && c.store.Spec.Security.ServerSideEncryptionS3.IsTokenAuthEnabled() {
+			vaultFileVol, _ := kms.VaultVolumeAndMountWithCustomName(c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails,
+				c.store.Spec.Security.ServerSideEncryptionS3.TokenSecretName, sseS3)
+			podSpec.Volumes = append(podSpec.Volumes, vaultFileVol)
+		}
+
+		podSpec.InitContainers = append(podSpec.InitContainers,
+			c.vaultTokenInitContainer(rgwConfig, kmsEnabled, s3Enabled))
 	}
 	c.store.Spec.Gateway.Placement.ApplyToPodSpec(&podSpec)
 
@@ -234,9 +260,19 @@ func (c *clusterConfig) createCaBundleUpdateInitContainer(rgwConfig *rgwConfig)
 // predefined value, so it won't work there. Hence the token file and certs (if present)
 // are copied to other volume from mounted secrets then ownership/permissions are changed
 // accordingly with help of an init container.
-func (c *clusterConfig) vaultTokenInitContainer(rgwConfig *rgwConfig) v1.Container {
-	_, srcVaultVolMount := kms.VaultVolumeAndMount(c.store.Spec.Security.KeyManagementService.ConnectionDetails, "")
+func (c *clusterConfig) vaultTokenInitContainer(rgwConfig *rgwConfig, kmsEnabled, s3Enabled bool) v1.Container {
+	var vaultVolMounts []v1.VolumeMount
+
 	tmpVaultMount := v1.VolumeMount{Name: rgwVaultVolumeName, MountPath: rgwVaultDirName}
+	vaultVolMounts = append(vaultVolMounts, tmpVaultMount)
+	if kmsEnabled {
+		_, ssekmsVaultVolMount := kms.VaultVolumeAndMountWithCustomName(c.store.Spec.Security.KeyManagementService.ConnectionDetails, "", sseKMS)
+		vaultVolMounts = append(vaultVolMounts, ssekmsVaultVolMount)
+	}
+	if s3Enabled {
+		_, sses3VaultVolMount := kms.VaultVolumeAndMountWithCustomName(c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails, "", sseS3)
+		vaultVolMounts = append(vaultVolMounts, sses3VaultVolMount)
+	}
 	return v1.Container{
 		Name: "vault-initcontainer-token-file-setup",
 		Command: []string{
@@ -247,7 +283,7 @@ func (c *clusterConfig) vaultTokenInitContainer(rgwConfig *rgwConfig) v1.Contain
 		},
 		Image: c.clusterSpec.CephVersion.Image,
 		VolumeMounts: append(
-			controller.DaemonVolumeMounts(c.DataPathMap, rgwConfig.ResourceName), srcVaultVolMount, tmpVaultMount),
+			controller.DaemonVolumeMounts(c.DataPathMap, rgwConfig.ResourceName), vaultVolMounts...),
 		Resources:       c.store.Spec.Gateway.Resources,
 		SecurityContext: controller.PodSecurityContext(),
 	}
@@ -263,7 +299,7 @@ func (c *clusterConfig) makeChownInitContainer(rgwConfig *rgwConfig) v1.Containe
 	)
 }
 
-func (c *clusterConfig) makeDaemonContainer(rgwConfig *rgwConfig) v1.Container {
+func (c *clusterConfig) makeDaemonContainer(rgwConfig *rgwConfig) (v1.Container, error) {
 	// start the rgw daemon in the foreground
 	container := v1.Container{
 		Name:  "rgw",
@@ -312,47 +348,42 @@ func (c *clusterConfig) makeDaemonContainer(rgwConfig *rgwConfig) v1.Container {
 	}
 	kmsEnabled, err := c.CheckRGWKMS()
 	if err != nil {
-		logger.Errorf("failed to enable KMS. %v", err)
-		return v1.Container{}
+		logger.Errorf("failed to enable SSE-KMS. %v", err)
+		return v1.Container{}, err
 	}
 	if kmsEnabled {
-		container.Args = append(container.Args,
-			cephconfig.NewFlag("rgw crypt s3 kms backend",
-				c.store.Spec.Security.KeyManagementService.ConnectionDetails[kms.Provider]),
-			cephconfig.NewFlag("rgw crypt vault addr",
-				c.store.Spec.Security.KeyManagementService.ConnectionDetails[api.EnvVaultAddress]),
-		)
+		logger.Debugf("enabliing SSE-KMS. %v", c.store.Spec.Security.KeyManagementService)
+		container.Args = append(container.Args, c.sseKMSDefaultOptions(kmsEnabled)...)
 		if c.store.Spec.Security.KeyManagementService.IsTokenAuthEnabled() {
-			container.Args = append(container.Args,
-				cephconfig.NewFlag("rgw crypt vault auth", kms.KMSTokenSecretNameKey),
-				cephconfig.NewFlag("rgw crypt vault token file",
-					path.Join(rgwVaultDirName, kms.VaultFileName)),
-				cephconfig.NewFlag("rgw crypt vault prefix", c.vaultPrefixRGW()),
-				cephconfig.NewFlag("rgw crypt vault secret engine",
-					c.store.Spec.Security.KeyManagementService.ConnectionDetails[kms.VaultSecretEngineKey]),
-			)
+			container.Args = append(container.Args, c.sseKMSVaultTokenOptions(kmsEnabled)...)
 		}
 		if c.store.Spec.Security.KeyManagementService.IsTLSEnabled() &&
-			c.clusterInfo.CephVersion.IsAtLeast(cephver.CephVersion{Major: 16, Minor: 2, Extra: 6}) {
-			container.Args = append(container.Args,
-				cephconfig.NewFlag("rgw crypt vault verify ssl", "true"))
-			if kms.GetParam(c.store.Spec.Security.KeyManagementService.ConnectionDetails, api.EnvVaultClientCert) != "" {
-				container.Args = append(container.Args,
-					cephconfig.NewFlag("rgw crypt vault ssl clientcert", path.Join(rgwVaultDirName, kms.VaultCertFileName)))
-			}
-			if kms.GetParam(c.store.Spec.Security.KeyManagementService.ConnectionDetails, api.EnvVaultClientKey) != "" {
-				container.Args = append(container.Args,
-					cephconfig.NewFlag("rgw crypt vault ssl clientkey", path.Join(rgwVaultDirName, kms.VaultKeyFileName)))
-			}
-			if kms.GetParam(c.store.Spec.Security.KeyManagementService.ConnectionDetails, api.EnvVaultCACert) != "" {
-				container.Args = append(container.Args,
-					cephconfig.NewFlag("rgw crypt vault ssl cacert", path.Join(rgwVaultDirName, kms.VaultCAFileName)))
-			}
+			c.clusterInfo.CephVersion.IsAtLeast(cephVersionMinRGWSSEKMSTLS) {
+			container.Args = append(container.Args, c.sseKMSVaultTLSOptions(kmsEnabled)...)
 		}
+	}
+
+	s3Enabled, err := c.CheckRGWSSES3Enabled()
+	if err != nil {
+		return v1.Container{}, err
+	}
+	if s3Enabled {
+		logger.Debugf("enabliing SSE-S3. %v", c.store.Spec.Security.ServerSideEncryptionS3)
+
+		container.Args = append(container.Args, c.sseS3DefaultOptions(s3Enabled)...)
+		if c.store.Spec.Security.ServerSideEncryptionS3.IsTokenAuthEnabled() {
+			container.Args = append(container.Args, c.sseS3VaultTokenOptions(s3Enabled)...)
+		}
+		if c.store.Spec.Security.ServerSideEncryptionS3.IsTLSEnabled() {
+			container.Args = append(container.Args, c.sseS3VaultTLSOptions(s3Enabled)...)
+		}
+	}
+
+	if s3Enabled || kmsEnabled {
 		vaultVolMount := v1.VolumeMount{Name: rgwVaultVolumeName, MountPath: rgwVaultDirName}
 		container.VolumeMounts = append(container.VolumeMounts, vaultVolMount)
 	}
-	return container
+	return container, nil
 }
 
 // configureReadinessProbe returns the desired readiness probe for a given daemon
@@ -525,14 +556,13 @@ func (c *clusterConfig) reconcileService(cephObjectStore *cephv1.CephObjectStore
 
 func (c *clusterConfig) vaultPrefixRGW() string {
 	secretEngine := c.store.Spec.Security.KeyManagementService.ConnectionDetails[kms.VaultSecretEngineKey]
-	vaultPrefixPath := "/v1/"
-
+	var vaultPrefixPath string
 	switch secretEngine {
 	case kms.VaultKVSecretEngineKey:
-		vaultPrefixPath = path.Join(vaultPrefixPath,
+		vaultPrefixPath = path.Join(vaultPrefix,
 			c.store.Spec.Security.KeyManagementService.ConnectionDetails[vault.VaultBackendPathKey], "/data")
 	case kms.VaultTransitSecretEngineKey:
-		vaultPrefixPath = path.Join(vaultPrefixPath, secretEngine)
+		vaultPrefixPath = path.Join(vaultPrefix, secretEngine)
 	}
 
 	return vaultPrefixPath
@@ -540,13 +570,13 @@ func (c *clusterConfig) vaultPrefixRGW() string {
 
 func (c *clusterConfig) CheckRGWKMS() (bool, error) {
 	if c.store.Spec.Security != nil && c.store.Spec.Security.KeyManagementService.IsEnabled() {
-		err := kms.ValidateConnectionDetails(c.clusterInfo.Context, c.context, c.store.Spec.Security, c.store.Namespace)
+		err := kms.ValidateConnectionDetails(c.clusterInfo.Context, c.context, &c.store.Spec.Security.KeyManagementService, c.store.Namespace)
 		if err != nil {
 			return false, err
 		}
 		secretEngine := c.store.Spec.Security.KeyManagementService.ConnectionDetails[kms.VaultSecretEngineKey]
 
-		// currently RGW supports kv(version 2) and transit secret engines in vault
+		// currently RGW supports kv(version 2) and transit secret engines in vault for sse:kms
 		switch secretEngine {
 		case kms.VaultKVSecretEngineKey:
 			kvVers := c.store.Spec.Security.KeyManagementService.ConnectionDetails[vault.VaultBackendKey]
@@ -571,6 +601,26 @@ func (c *clusterConfig) CheckRGWKMS() (bool, error) {
 	return false, nil
 }
 
+func (c *clusterConfig) CheckRGWSSES3Enabled() (bool, error) {
+	if c.store.Spec.Security != nil && c.store.Spec.Security.ServerSideEncryptionS3.IsEnabled() {
+		if !c.clusterInfo.CephVersion.IsAtLeast(cephVersionMinRGWSSES3) {
+			return false, errors.New("minimum ceph quincy is required for AWS-SSE:S3")
+		}
+		err := kms.ValidateConnectionDetails(c.clusterInfo.Context, c.context, &c.store.Spec.Security.ServerSideEncryptionS3, c.store.Namespace)
+		if err != nil {
+			return false, err
+		}
+
+		// currently RGW supports only transit secret engines in vault for sse:s3
+		if c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails[kms.VaultSecretEngineKey] != kms.VaultTransitSecretEngineKey {
+			return false, errors.New("vault secret engine is not transit")
+		}
+		return true, nil
+	}
+
+	return false, nil
+}
+
 func addPort(service *v1.Service, name string, port, destPort int32) {
 	if port == 0 || destPort == 0 {
 		return
@@ -672,3 +722,99 @@ func (c *clusterConfig) rgwTLSSecretType(secretName string) (v1.SecretType, erro
 func getDaemonName(rgwConfig *rgwConfig) string {
 	return fmt.Sprintf("ceph-%s", generateCephXUser(rgwConfig.ResourceName))
 }
+
+// Following apis set the RGW options if requested, since they are used in unit tests for validating different scenarios
+func (c *clusterConfig) sseKMSDefaultOptions(setOptions bool) []string {
+	if setOptions {
+		return []string{
+			cephconfig.NewFlag("rgw crypt s3 kms backend",
+				c.store.Spec.Security.KeyManagementService.ConnectionDetails[kms.Provider]),
+			cephconfig.NewFlag("rgw crypt vault addr",
+				c.store.Spec.Security.KeyManagementService.ConnectionDetails[api.EnvVaultAddress]),
+		}
+	}
+	return []string{}
+}
+
+func (c *clusterConfig) sseS3DefaultOptions(setOptions bool) []string {
+	if setOptions {
+		return []string{
+			cephconfig.NewFlag("rgw crypt sse s3 backend",
+				c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails[kms.Provider]),
+			cephconfig.NewFlag("rgw crypt sse s3 vault addr",
+				c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails[api.EnvVaultAddress]),
+		}
+	}
+	return []string{}
+}
+
+func (c *clusterConfig) sseKMSVaultTokenOptions(setOptions bool) []string {
+	if setOptions {
+		return []string{
+			cephconfig.NewFlag("rgw crypt vault auth", kms.KMSTokenSecretNameKey),
+			cephconfig.NewFlag("rgw crypt vault token file",
+				path.Join(rgwVaultDirName, sseKMS, kms.VaultFileName)),
+			cephconfig.NewFlag("rgw crypt vault prefix", c.vaultPrefixRGW()),
+			cephconfig.NewFlag("rgw crypt vault secret engine",
+				c.store.Spec.Security.KeyManagementService.ConnectionDetails[kms.VaultSecretEngineKey]),
+		}
+	}
+	return []string{}
+}
+
+func (c *clusterConfig) sseS3VaultTokenOptions(setOptions bool) []string {
+	if setOptions {
+		return []string{
+			cephconfig.NewFlag("rgw crypt sse s3 vault auth", kms.KMSTokenSecretNameKey),
+			cephconfig.NewFlag("rgw crypt sse s3 vault token file",
+				path.Join(rgwVaultDirName, sseS3, kms.VaultFileName)),
+			cephconfig.NewFlag("rgw crypt sse s3 vault prefix",
+				path.Join(vaultPrefix, c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails[kms.VaultSecretEngineKey])),
+			cephconfig.NewFlag("rgw crypt sse s3 vault secret engine",
+				c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails[kms.VaultSecretEngineKey]),
+		}
+	}
+	return []string{}
+}
+
+func (c *clusterConfig) sseKMSVaultTLSOptions(setOptions bool) []string {
+	var rgwOptions []string
+	if setOptions {
+		rgwOptions = append(rgwOptions, cephconfig.NewFlag("rgw crypt vault verify ssl", "true"))
+
+		if kms.GetParam(c.store.Spec.Security.KeyManagementService.ConnectionDetails, api.EnvVaultClientCert) != "" {
+			rgwOptions = append(rgwOptions,
+				cephconfig.NewFlag("rgw crypt vault ssl clientcert", path.Join(rgwVaultDirName, sseKMS, kms.VaultCertFileName)))
+		}
+		if kms.GetParam(c.store.Spec.Security.KeyManagementService.ConnectionDetails, api.EnvVaultClientKey) != "" {
+			rgwOptions = append(rgwOptions,
+				cephconfig.NewFlag("rgw crypt vault ssl clientkey", path.Join(rgwVaultDirName, sseKMS, kms.VaultKeyFileName)))
+		}
+		if kms.GetParam(c.store.Spec.Security.KeyManagementService.ConnectionDetails, api.EnvVaultCACert) != "" {
+			rgwOptions = append(rgwOptions,
+				cephconfig.NewFlag("rgw crypt vault ssl cacert", path.Join(rgwVaultDirName, sseKMS, kms.VaultCAFileName)))
+		}
+	}
+	return rgwOptions
+}
+
+func (c *clusterConfig) sseS3VaultTLSOptions(setOptions bool) []string {
+	var rgwOptions []string
+	if setOptions {
+		rgwOptions = append(rgwOptions, cephconfig.NewFlag("rgw crypt sse s3 vault verify ssl", "true"))
+
+		if kms.GetParam(c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails, api.EnvVaultClientCert) != "" {
+			rgwOptions = append(rgwOptions,
+				cephconfig.NewFlag("rgw crypt sse s3 vault ssl clientcert", path.Join(rgwVaultDirName, sseS3, kms.VaultCertFileName)))
+		}
+		if kms.GetParam(c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails, api.EnvVaultClientKey) != "" {
+			rgwOptions = append(rgwOptions,
+				cephconfig.NewFlag("rgw crypt sse s3 vault ssl clientkey", path.Join(rgwVaultDirName, sseS3, kms.VaultKeyFileName)))
+		}
+		if kms.GetParam(c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails, api.EnvVaultCACert) != "" {
+			rgwOptions = append(rgwOptions,
+				cephconfig.NewFlag("rgw crypt sse s3 vault ssl cacert", path.Join(rgwVaultDirName, sseS3, kms.VaultCAFileName)))
+		}
+	}
+	return rgwOptions
+}

pkg/operator/ceph/object/spec_test.go

@@ -33,11 +33,41 @@ import (
 	exectest "github.com/rook/rook/pkg/util/exec/test"
 	"github.com/stretchr/testify/assert"
 	v1 "k8s.io/api/core/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 )
 
+func configureSSE(t *testing.T, c *clusterConfig, kms bool, s3 bool) {
+	c.store.Spec.Security = &cephv1.ObjectStoreSecuritySpec{}
+	if kms {
+		c.store.Spec.Security.KeyManagementService = cephv1.KeyManagementServiceSpec{ConnectionDetails: map[string]string{}}
+		c.store.Spec.Security.KeyManagementService.TokenSecretName = "vault-token"
+		c.store.Spec.Security.KeyManagementService.ConnectionDetails["KMS_PROVIDER"] = "vault"
+		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_ADDR"] = "https://1.1.1.1:8200"
+	}
+	if s3 {
+		c.store.Spec.Security.ServerSideEncryptionS3 = cephv1.KeyManagementServiceSpec{ConnectionDetails: map[string]string{}}
+		c.store.Spec.Security.ServerSideEncryptionS3.TokenSecretName = "vault-token"
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["KMS_PROVIDER"] = "vault"
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["VAULT_ADDR"] = "https://1.1.1.1:8200"
+	}
+	s := &v1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "vault-token",
+			Namespace: c.store.Namespace,
+		},
+		Data: map[string][]byte{
+			"token": []byte("myt-otkenbenvqrev"),
+		},
+	}
+	_, err := c.context.Clientset.CoreV1().Secrets(c.store.Namespace).Create(c.clusterInfo.Context, s, metav1.CreateOptions{})
+	if err != nil && !k8serrors.IsAlreadyExists(err) {
+		assert.Error(t, err)
+	}
+}
+
 func TestPodSpecs(t *testing.T) {
 	store := simpleStore()
 	store.Spec.Gateway.Resources = v1.ResourceRequirements{
@@ -93,8 +123,8 @@ func TestPodSpecs(t *testing.T) {
 	t.Run(("check rgw ConfigureProbe"), func(t *testing.T) {
 		c.store.Spec.HealthCheck.StartupProbe = &cephv1.ProbeSpec{Disabled: false, Probe: &v1.Probe{InitialDelaySeconds: 1000}}
 		c.store.Spec.HealthCheck.LivenessProbe = &cephv1.ProbeSpec{Disabled: false, Probe: &v1.Probe{InitialDelaySeconds: 900}}
-		deployment := c.makeDaemonContainer(rgwConfig)
-		assert.NotNil(t, deployment)
+		deployment, err := c.makeDaemonContainer(rgwConfig)
+		assert.NoError(t, err)
 		assert.NotNil(t, c.store.Spec.HealthCheck.LivenessProbe)
 		assert.NotNil(t, c.store.Spec.HealthCheck.StartupProbe)
 		assert.Equal(t, int32(900), deployment.LivenessProbe.InitialDelaySeconds)
@@ -131,7 +161,7 @@ func TestSSLPodSpec(t *testing.T) {
 		context:     context,
 		rookVersion: "rook/rook:myversion",
 		clusterSpec: &cephv1.ClusterSpec{
-			CephVersion: cephv1.CephVersionSpec{Image: "quay.io/ceph/ceph:v15"},
+			CephVersion: cephv1.CephVersionSpec{Image: "quay.io/ceph/ceph:v17.2.3"},
 			Network: cephv1.NetworkSpec{
 				HostNetwork: true,
 			},
@@ -229,7 +259,6 @@ func TestSSLPodSpec(t *testing.T) {
 
 	assert.True(t, s.Spec.HostNetwork)
 	assert.Equal(t, v1.DNSClusterFirstWithHostNet, s.Spec.DNSPolicy)
-
 }
 
 func TestValidateSpec(t *testing.T) {
@@ -451,29 +480,12 @@ func TestCheckRGWKMS(t *testing.T) {
 	setupTest := func() *clusterConfig {
 		context := &clusterd.Context{Clientset: test.New(t, 3)}
 		store := simpleStore()
-		store.Spec.Security = &cephv1.SecuritySpec{KeyManagementService: cephv1.KeyManagementServiceSpec{ConnectionDetails: map[string]string{}}}
 		return &clusterConfig{
 			context:     context,
 			store:       store,
 			clusterInfo: &client.ClusterInfo{Context: ctx},
 		}
 	}
-	configureKMS := func(c *clusterConfig) {
-		c.store.Spec.Security.KeyManagementService.TokenSecretName = "vault-token"
-		c.store.Spec.Security.KeyManagementService.ConnectionDetails["KMS_PROVIDER"] = "vault"
-		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_ADDR"] = "https://1.1.1.1:8200"
-		s := &v1.Secret{
-			ObjectMeta: metav1.ObjectMeta{
-				Name:      c.store.Spec.Security.KeyManagementService.TokenSecretName,
-				Namespace: c.store.Namespace,
-			},
-			Data: map[string][]byte{
-				"token": []byte("myt-otkenbenvqrev"),
-			},
-		}
-		_, err := c.context.Clientset.CoreV1().Secrets(c.store.Namespace).Create(ctx, s, metav1.CreateOptions{})
-		assert.NoError(t, err)
-	}
 
 	t.Run("KMS is disabled", func(t *testing.T) {
 		c := setupTest()
@@ -484,7 +496,7 @@ func TestCheckRGWKMS(t *testing.T) {
 
 	t.Run("Vault Secret Engine is missing", func(t *testing.T) {
 		c := setupTest()
-		configureKMS(c)
+		configureSSE(t, c, true, false)
 		b, err := c.CheckRGWKMS()
 		assert.False(t, b)
 		assert.Error(t, err)
@@ -492,7 +504,7 @@ func TestCheckRGWKMS(t *testing.T) {
 
 	t.Run("Vault Secret Engine is kv with v1", func(t *testing.T) {
 		c := setupTest()
-		configureKMS(c)
+		configureSSE(t, c, true, false)
 		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_SECRET_ENGINE"] = "kv"
 		b, err := c.CheckRGWKMS()
 		assert.False(t, b)
@@ -501,7 +513,7 @@ func TestCheckRGWKMS(t *testing.T) {
 
 	t.Run("Vault Secret Engine is kv with v2", func(t *testing.T) {
 		c := setupTest()
-		configureKMS(c)
+		configureSSE(t, c, true, false)
 		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_SECRET_ENGINE"] = "kv"
 		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_BACKEND"] = "v2"
 		b, err := c.CheckRGWKMS()
@@ -511,7 +523,7 @@ func TestCheckRGWKMS(t *testing.T) {
 
 	t.Run("Vault Secret Engine is transit", func(t *testing.T) {
 		c := setupTest()
-		configureKMS(c)
+		configureSSE(t, c, true, false)
 		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_SECRET_ENGINE"] = "transit"
 		b, err := c.CheckRGWKMS()
 		assert.True(t, b)
@@ -520,7 +532,7 @@ func TestCheckRGWKMS(t *testing.T) {
 
 	t.Run("TLS is configured but secrets do not exist", func(t *testing.T) {
 		c := setupTest()
-		configureKMS(c)
+		configureSSE(t, c, true, false)
 		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_SECRET_ENGINE"] = "transit"
 		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_CACERT"] = "vault-ca-secret"
 		b, err := c.CheckRGWKMS()
@@ -531,7 +543,7 @@ func TestCheckRGWKMS(t *testing.T) {
 
 	t.Run("TLS secret exists but empty key", func(t *testing.T) {
 		c := setupTest()
-		configureKMS(c)
+		configureSSE(t, c, true, false)
 		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_SECRET_ENGINE"] = "transit"
 		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_CACERT"] = "vault-ca-secret"
 		tlsSecret := &v1.Secret{
@@ -550,7 +562,7 @@ func TestCheckRGWKMS(t *testing.T) {
 
 	t.Run("TLS config is valid", func(t *testing.T) {
 		c := setupTest()
-		configureKMS(c)
+		configureSSE(t, c, true, false)
 		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_SECRET_ENGINE"] = "transit"
 		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_CACERT"] = "vault-ca-secret"
 		tlsSecret := &v1.Secret{
@@ -568,6 +580,114 @@ func TestCheckRGWKMS(t *testing.T) {
 	})
 }
 
+func TestCheckRGWSSES3Enabled(t *testing.T) {
+	ctx := context.TODO()
+	setupTest := func() *clusterConfig {
+		context := &clusterd.Context{Clientset: test.New(t, 3)}
+		store := simpleStore()
+		return &clusterConfig{
+			context:     context,
+			store:       store,
+			clusterInfo: &client.ClusterInfo{Context: ctx, CephVersion: cephver.CephVersion{Major: 17, Minor: 2, Extra: 3}},
+			clusterSpec: &cephv1.ClusterSpec{
+				CephVersion: cephv1.CephVersionSpec{Image: "quay.io/ceph/ceph:v17.2.3"},
+			},
+		}
+	}
+
+	t.Run("KMS is disabled", func(t *testing.T) {
+		c := setupTest()
+		b, err := c.CheckRGWSSES3Enabled()
+		assert.False(t, b)
+		assert.NoError(t, err)
+	})
+
+	t.Run("Vault Secret Engine is missing", func(t *testing.T) {
+		c := setupTest()
+		configureSSE(t, c, false, true)
+		b, err := c.CheckRGWSSES3Enabled()
+		assert.False(t, b)
+		assert.Error(t, err)
+	})
+
+	t.Run("Vault Secret Engine is kv with v1", func(t *testing.T) {
+		c := setupTest()
+		configureSSE(t, c, false, true)
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["VAULT_SECRET_ENGINE"] = "kv"
+		b, err := c.CheckRGWSSES3Enabled()
+		assert.False(t, b)
+		assert.Error(t, err)
+	})
+
+	t.Run("Vault Secret Engine is kv with v2", func(t *testing.T) {
+		c := setupTest()
+		configureSSE(t, c, false, true)
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["VAULT_SECRET_ENGINE"] = "kv"
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["VAULT_BACKEND"] = "v2"
+		b, err := c.CheckRGWSSES3Enabled()
+		assert.False(t, b)
+		assert.Error(t, err)
+	})
+
+	t.Run("Vault Secret Engine is transit", func(t *testing.T) {
+		c := setupTest()
+		configureSSE(t, c, false, true)
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["VAULT_SECRET_ENGINE"] = "transit"
+		b, err := c.CheckRGWSSES3Enabled()
+		assert.True(t, b)
+		assert.NoError(t, err)
+	})
+
+	t.Run("TLS is configured but secrets do not exist", func(t *testing.T) {
+		c := setupTest()
+		configureSSE(t, c, false, true)
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["VAULT_SECRET_ENGINE"] = "transit"
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["VAULT_CACERT"] = "vault-ca-secret"
+		b, err := c.CheckRGWSSES3Enabled()
+		assert.False(t, b)
+		assert.Error(t, err, "")
+		assert.EqualError(t, err, "failed to validate vault connection details: failed to find TLS connection details k8s secret \"vault-ca-secret\": secrets \"vault-ca-secret\" not found")
+	})
+
+	t.Run("TLS secret exists but empty key", func(t *testing.T) {
+		c := setupTest()
+		configureSSE(t, c, false, true)
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["VAULT_SECRET_ENGINE"] = "transit"
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["VAULT_CACERT"] = "vault-ca-secret"
+		tlsSecret := &v1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "vault-ca-secret",
+				Namespace: c.store.Namespace,
+			},
+		}
+		_, err := c.context.Clientset.CoreV1().Secrets(c.store.Namespace).Create(context.TODO(), tlsSecret, metav1.CreateOptions{})
+		assert.NoError(t, err)
+		b, err := c.CheckRGWSSES3Enabled()
+		assert.False(t, b)
+		assert.Error(t, err, "")
+		assert.EqualError(t, err, "failed to validate vault connection details: failed to find TLS connection key \"cert\" for \"VAULT_CACERT\" in k8s secret \"vault-ca-secret\"")
+	})
+
+	t.Run("TLS config is valid", func(t *testing.T) {
+		c := setupTest()
+		configureSSE(t, c, false, true)
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["VAULT_SECRET_ENGINE"] = "transit"
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["VAULT_CACERT"] = "vault-ca-secret"
+		tlsSecret := &v1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "vault-ca-secret",
+				Namespace: c.store.Namespace,
+			},
+		}
+		tlsSecret.Data = map[string][]byte{"cert": []byte("envnrevbnbvsbjkrtn")}
+		_, err := c.context.Clientset.CoreV1().Secrets(c.store.Namespace).Create(context.TODO(), tlsSecret, metav1.CreateOptions{})
+		assert.NoError(t, err)
+		b, err := c.CheckRGWSSES3Enabled()
+		assert.True(t, b)
+		assert.NoError(t, err, "")
+	})
+}
+
 func TestGetDaemonName(t *testing.T) {
 	context := &clusterd.Context{Clientset: test.New(t, 3)}
 	store := simpleStore()
@@ -648,3 +768,168 @@ func TestMakeRGWPodSpec(t *testing.T) {
 		})
 	}
 }
+
+func TestAWSServerSideEncryption(t *testing.T) {
+	ctx := context.TODO()
+	// Placeholder
+	context := &clusterd.Context{Clientset: test.New(t, 3)}
+
+	store := simpleStore()
+	info := clienttest.CreateTestClusterInfo(1)
+	info.CephVersion = cephver.CephVersion{Major: 17, Minor: 2, Extra: 3}
+	info.Namespace = store.Namespace
+	data := cephconfig.NewStatelessDaemonDataPathMap(cephconfig.RgwType, "default", "rook-ceph", "/var/lib/rook/")
+
+	c := &clusterConfig{
+		clusterInfo: info,
+		store:       store,
+		context:     context,
+		rookVersion: "rook/rook:myversion",
+		clusterSpec: &cephv1.ClusterSpec{
+			CephVersion: cephv1.CephVersionSpec{Image: "quay.io/ceph/ceph:v17.3"},
+			Network: cephv1.NetworkSpec{
+				HostNetwork: true,
+			},
+		},
+		DataPathMap: data,
+	}
+
+	resourceName := fmt.Sprintf("%s-%s", AppName, c.store.Name)
+	rgwConfig := &rgwConfig{
+		ResourceName: resourceName,
+		DaemonID:     "default",
+	}
+	checkRGWOptions := func(allRGWOptions, optionsNeeded []string) bool {
+		if len(optionsNeeded) == 0 {
+			return false
+		}
+		optionMap := make(map[string]bool)
+		for _, option := range allRGWOptions {
+			optionMap[option] = true
+		}
+
+		for _, option := range optionsNeeded {
+			if _, found := optionMap[option]; !found {
+				return false
+			}
+		}
+
+		return true
+	}
+
+	t.Run("SecuritySpec is not configured no options will be configured", func(t *testing.T) {
+		rgwContainer, err := c.makeDaemonContainer(rgwConfig)
+		assert.NoError(t, err)
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseKMSDefaultOptions(false)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseS3DefaultOptions(false)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseKMSVaultTokenOptions(false)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseS3VaultTokenOptions(false)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseKMSVaultTLSOptions(false)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseS3VaultTLSOptions(false)))
+	})
+
+	configureSSE(t, c, true, false)
+	t.Run("Invalid Security Spec configured, no options will be configured", func(t *testing.T) {
+		rgwContainer, err := c.makeDaemonContainer(rgwConfig)
+		assert.Error(t, err)
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseKMSDefaultOptions(false)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseS3DefaultOptions(false)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseKMSVaultTokenOptions(false)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseS3VaultTokenOptions(false)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseKMSVaultTLSOptions(false)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseS3VaultTLSOptions(false)))
+	})
+
+	t.Run("Security Spec configured with kms only,so kms options will be configured", func(t *testing.T) {
+		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_SECRET_ENGINE"] = "transit"
+		rgwContainer, err := c.makeDaemonContainer(rgwConfig)
+		assert.NoError(t, err)
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseKMSDefaultOptions(true)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseS3DefaultOptions(false)))
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseKMSVaultTokenOptions(true)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseS3VaultTokenOptions(false)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseKMSVaultTLSOptions(false)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseS3VaultTLSOptions(false)))
+	})
+
+	t.Run("Security Spec configured with s3 only, so s3 options will be configured", func(t *testing.T) {
+		configureSSE(t, c, false, true)
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["VAULT_SECRET_ENGINE"] = "transit"
+		rgwContainer, err := c.makeDaemonContainer(rgwConfig)
+		assert.NoError(t, err)
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseKMSDefaultOptions(false)))
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseS3DefaultOptions(true)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseKMSVaultTokenOptions(false)))
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseS3VaultTokenOptions(true)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseKMSVaultTLSOptions(false)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseS3VaultTLSOptions(false)))
+	})
+
+	t.Run("Security Spec configured with both kms and s3 settings, so both kms and s3 options will be configured", func(t *testing.T) {
+		configureSSE(t, c, true, true)
+		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_SECRET_ENGINE"] = "transit"
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["VAULT_SECRET_ENGINE"] = "transit"
+		rgwContainer, err := c.makeDaemonContainer(rgwConfig)
+		assert.NoError(t, err)
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseKMSDefaultOptions(true)))
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseS3DefaultOptions(true)))
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseKMSVaultTokenOptions(true)))
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseS3VaultTokenOptions(true)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseKMSVaultTLSOptions(false)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseS3VaultTLSOptions(false)))
+	})
+
+	tlsSecret := &v1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "vault-ca-secret",
+			Namespace: c.store.Namespace,
+		},
+	}
+	tlsSecret.Data = map[string][]byte{"cert": []byte("envnrevbnbvsbjkrtn")}
+	_, err := c.context.Clientset.CoreV1().Secrets(c.store.Namespace).Create(ctx, tlsSecret, metav1.CreateOptions{})
+	assert.NoError(t, err)
+
+	t.Run("Security Spec configured with kms along with TLS, so kms including TLS options will be configured", func(t *testing.T) {
+		configureSSE(t, c, true, false)
+		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_SECRET_ENGINE"] = "transit"
+		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_CACERT"] = "vault-ca-secret"
+		rgwContainer, err := c.makeDaemonContainer(rgwConfig)
+		assert.NoError(t, err)
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseKMSDefaultOptions(true)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseS3DefaultOptions(false)))
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseKMSVaultTokenOptions(true)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseS3VaultTokenOptions(false)))
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseKMSVaultTLSOptions(true)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseS3VaultTLSOptions(false)))
+	})
+
+	t.Run("Security Spec configured with s3 along with TLS, so s3 including TLS options will be configured", func(t *testing.T) {
+		configureSSE(t, c, false, true)
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["VAULT_SECRET_ENGINE"] = "transit"
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["VAULT_CACERT"] = "vault-ca-secret"
+		rgwContainer, err := c.makeDaemonContainer(rgwConfig)
+		assert.NoError(t, err)
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseKMSDefaultOptions(false)))
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseS3DefaultOptions(true)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseKMSVaultTokenOptions(false)))
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseS3VaultTokenOptions(true)))
+		assert.False(t, checkRGWOptions(rgwContainer.Args, c.sseKMSVaultTLSOptions(true)))
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseS3VaultTLSOptions(true)))
+	})
+
+	t.Run("Security Spec configured both kms and s3 aloong with TLS, all the serverr side encryption related options will be configured", func(t *testing.T) {
+		configureSSE(t, c, true, true)
+		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_SECRET_ENGINE"] = "transit"
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["VAULT_SECRET_ENGINE"] = "transit"
+		c.store.Spec.Security.KeyManagementService.ConnectionDetails["VAULT_CACERT"] = "vault-ca-secret"
+		c.store.Spec.Security.ServerSideEncryptionS3.ConnectionDetails["VAULT_CACERT"] = "vault-ca-secret"
+		rgwContainer, err := c.makeDaemonContainer(rgwConfig)
+		assert.NoError(t, err)
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseKMSDefaultOptions(true)))
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseS3DefaultOptions(true)))
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseKMSVaultTokenOptions(true)))
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseS3VaultTokenOptions(true)))
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseKMSVaultTLSOptions(true)))
+		assert.True(t, checkRGWOptions(rgwContainer.Args, c.sseS3VaultTLSOptions(true)))
+	})
+}