diff --git a/cmd/rook/secret.go b/cmd/rook/secret.go
index 84068b209..e3eaf6b46 100644
--- a/cmd/rook/secret.go
+++ b/cmd/rook/secret.go
@@ -23,7 +23,9 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/rook/rook/cmd/rook/rook"
+	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/rook/rook/pkg/daemon/ceph/osd"
 	"github.com/rook/rook/pkg/daemon/ceph/osd/kms"
 	operator "github.com/rook/rook/pkg/operator/ceph"
 	"github.com/rook/rook/pkg/operator/k8sutil"
@@ -37,16 +39,18 @@ var KeyManagementCmd = &cobra.Command{
 	Use:   "key-management",
 	Short: "key-management interacts with a given Key Management System and perform actions.",
 	Long: `The secret sub-command helps interacting with Key Management System.
-	It can perform various actions such as retrieving the content of a Key Encryption Key.`,
+	It can perform various actions such as retrieving the content of a Key Encryption Key and
+	rotating Key Encryption Key.`,
 }
 
 func init() {
 	KeyManagementCmd.AddCommand(
 		cliGetSecret(),
+		cliRotateSecret(),
 	)
 }
 
-func startSecret() *kms.Config {
+func startSecret() (*kms.Config, *clusterd.Context) {
 	// Initialize the context
 	ctx, cancel := signal.NotifyContext(context.Background(), operator.ShutdownSignals...)
 	defer cancel()
@@ -71,13 +75,15 @@ func startSecret() *kms.Config {
 		rook.TerminateFatal(errors.Wrapf(err, "failed to get ceph cluster in namespace %q", namespace))
 	}
 
-	// Validate connection details
-	err = kms.ValidateConnectionDetails(ctx, context, &cephCluster.Spec.Security.KeyManagementService, namespace)
-	if err != nil {
-		rook.TerminateFatal(errors.Wrap(err, "failed to validate kms connection details"))
+	if cephCluster.Spec.Security.KeyManagementService.IsEnabled() {
+		// Validate connection details
+		err = kms.ValidateConnectionDetails(ctx, context, &cephCluster.Spec.Security.KeyManagementService, namespace)
+		if err != nil {
+			rook.TerminateFatal(errors.Wrap(err, "failed to validate kms connection details"))
+		}
 	}
 
-	return kms.NewConfig(context, &cephCluster.Spec, clusterInfo)
+	return kms.NewConfig(context, &cephCluster.Spec, clusterInfo), context
 }
 
 // cliGetSecret is the Cobra CLI call
@@ -98,7 +104,7 @@ func getSecret(cmd *cobra.Command, args []string) {
 
 	secretName := args[0]
 	secretPath := args[1]
-	keyManagementService := startSecret()
+	keyManagementService, _ := startSecret()
 	keyManagementService.ClusterInfo.Context = ctx
 
 	// Fetch the secret
@@ -113,3 +119,33 @@ func getSecret(cmd *cobra.Command, args []string) {
 		rook.TerminateFatal(errors.Wrapf(err, "failed to write secret %q file to %q", secretName, secretPath))
 	}
 }
+
+// cliRotateSecret is the Cobra CLI call
+func cliRotateSecret() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "rotate-key kms-secret-key data-device [metadata-device] [wal-device]",
+		Short: "Rotate key",
+		Args:  cobra.RangeArgs(2, 4),
+		Run:   rotateSecret,
+	}
+	return cmd
+}
+
+// rotateSecret rotates the Key Encryption Key of a given OSD.
+// It accepts the name of the secret to rotate, and the path to the
+// encrypted devices.
+func rotateSecret(cmd *cobra.Command, args []string) {
+	rook.SetLogLevel()
+	// Initialize the context
+	ctx, cancel := signal.NotifyContext(context.Background(), operator.ShutdownSignals...)
+	defer cancel()
+	secretName := args[0]
+	devicePaths := args[1:]
+	keyManagementService, context := startSecret()
+	keyManagementService.ClusterInfo.Context = ctx
+
+	err := osd.RotateKeyEncryptionKey(context, keyManagementService, secretName, devicePaths)
+	if err != nil {
+		rook.TerminateFatal(errors.Wrapf(err, "failed to rotate secret %q", secretName))
+	}
+}
diff --git a/deploy/charts/library/templates/_cluster-role.tpl b/deploy/charts/library/templates/_cluster-role.tpl
index 61b10509c..13de8de58 100644
--- a/deploy/charts/library/templates/_cluster-role.tpl
+++ b/deploy/charts/library/templates/_cluster-role.tpl
@@ -9,10 +9,10 @@ metadata:
   namespace: {{ .Release.Namespace }} # namespace:cluster
 rules:
   # this is needed for rook's "key-management" CLI to fetch the vault token from the secret when
-  # validating the connection details
+  # validating the connection details and for key rotation operations.
   - apiGroups: [""]
     resources: ["secrets"]
-    verbs: ["get"]
+    verbs: ["get","update"]
   - apiGroups: [""]
     resources: ["configmaps"]
     verbs: ["get", "list", "watch", "create", "update", "delete"]
diff --git a/deploy/examples/common-second-cluster.yaml b/deploy/examples/common-second-cluster.yaml
index f84633397..684868e6c 100644
--- a/deploy/examples/common-second-cluster.yaml
+++ b/deploy/examples/common-second-cluster.yaml
@@ -166,6 +166,11 @@ metadata:
   name: rook-ceph-osd
   namespace: $NAMESPACE
 rules:
+  # this is needed for rook's "key-management" CLI to fetch the vault token from the secret when
+  # validating the connection details and for key rotation operations.
+  - apiGroups: [""]
+    resources: ["secrets"]
+    verbs: ["get", "update"]
   - apiGroups: [""]
     resources: ["configmaps"]
     verbs: ["get", "list", "watch", "create", "update", "delete"]
diff --git a/deploy/examples/common.yaml b/deploy/examples/common.yaml
index 7247ec65e..c27c858e3 100644
--- a/deploy/examples/common.yaml
+++ b/deploy/examples/common.yaml
@@ -801,10 +801,10 @@ metadata:
   namespace: rook-ceph # namespace:cluster
 rules:
   # this is needed for rook's "key-management" CLI to fetch the vault token from the secret when
-  # validating the connection details
+  # validating the connection details and for key rotation operations.
   - apiGroups: [""]
     resources: ["secrets"]
-    verbs: ["get"]
+    verbs: ["get", "update"]
   - apiGroups: [""]
     resources: ["configmaps"]
     verbs: ["get", "list", "watch", "create", "update", "delete"]
diff --git a/pkg/daemon/ceph/osd/key_rotation.go b/pkg/daemon/ceph/osd/key_rotation.go
new file mode 100644
index 000000000..0ba6dd638
--- /dev/null
+++ b/pkg/daemon/ceph/osd/key_rotation.go
@@ -0,0 +1,97 @@
+/*
+Copyright 2023 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package osd
+
+import (
+	"github.com/pkg/errors"
+	"github.com/rook/rook/pkg/clusterd"
+	"github.com/rook/rook/pkg/daemon/ceph/osd/kms"
+	oposd "github.com/rook/rook/pkg/operator/ceph/cluster/osd"
+)
+
+const (
+	slotZero string = "0"
+	slotOne  string = "1"
+)
+
+func RotateKeyEncryptionKey(context *clusterd.Context, kms *kms.Config, secretName string, devicePaths []string) error {
+	logger.Info("fetching the current key")
+	// Fetch the currentKey.
+	currentKey, err := kms.GetSecret(secretName)
+	if err != nil {
+		return errors.Wrapf(err, "failed to get secret %q", secretName)
+	}
+
+	// Ensure currentKey is in slot 1.
+	for _, devicePath := range devicePaths {
+		logger.Infof("adding the current key to slot %q of the device %q", slotOne, devicePath)
+		err = addEncryptionKey(context, devicePath, currentKey, currentKey, slotOne)
+		if err != nil {
+			return errors.Wrapf(err, "failed to add the current key to slot %q of the device", slotOne)
+		}
+	}
+
+	logger.Info("generating new key")
+	// Generate new key.
+	newKey, err := oposd.GenerateDmCryptKey()
+	if err != nil {
+		return errors.Wrapf(err, "failed to generate new key")
+	}
+
+	// Add newKey to slot 0.
+	for _, devicePath := range devicePaths {
+		logger.Infof("removing key slot %q, if found, of the device %q", slotZero, devicePath)
+		err = removeEncryptionKeySlot(context, devicePath, currentKey, slotZero)
+		if err != nil {
+			return errors.Wrapf(err, "failed to remove key slot %q of the device %q", slotZero, devicePath)
+		}
+		logger.Infof("adding new key to slot %q of the device %q", slotZero, devicePath)
+		err = addEncryptionKey(context, devicePath, currentKey, newKey, slotZero)
+		if err != nil {
+			return errors.Wrapf(err, "failed to add new key to slot %q of the device %q", slotZero, devicePath)
+		}
+	}
+
+	logger.Info("updating the new key in the KMS")
+	// Update new key.
+	err = kms.UpdateSecret(secretName, newKey)
+	if err != nil {
+		return errors.Wrapf(err, "failed to update secret %q with new key", secretName)
+	}
+
+	logger.Info("fetching the key from the KMS to verify it.")
+	// Fetch key to verify its the new key.
+	keyInKMS, err := kms.GetSecret(secretName)
+	if err != nil {
+		return errors.Wrapf(err, "failed to get secret %q", secretName)
+	}
+	if keyInKMS != newKey {
+		return errors.New("failed to verify the new key in the KMS, the fetched key is not the same as the new key")
+	}
+
+	// Remove old key from slot 1.
+	for _, devicePath := range devicePaths {
+		logger.Infof("removing the old key from the slot %q of the device %q", slotOne, devicePath)
+		err = removeEncryptionKeySlot(context, devicePath, newKey, slotOne)
+		if err != nil {
+			return errors.Wrapf(err, "failed to remove the old key from the slot %q of the device %q", slotOne, devicePath)
+		}
+	}
+	logger.Infof("Successfully rotated the key")
+
+	return nil
+}
diff --git a/pkg/operator/ceph/cluster/osd/config.go b/pkg/operator/ceph/cluster/osd/config.go
index ef57ee279..7f97ab501 100644
--- a/pkg/operator/ceph/cluster/osd/config.go
+++ b/pkg/operator/ceph/cluster/osd/config.go
@@ -58,7 +58,7 @@ func encryptionBlockDestinationCopy(mountPath, blockType string) string {
 	return path.Join(mountPath, blockType) + "-tmp"
 }
 
-func generateDmCryptKey() (string, error) {
+func GenerateDmCryptKey() (string, error) {
 	key, err := mgr.GenerateRandomBytes(dmCryptKeySize)
 	if err != nil {
 		return "", errors.Wrap(err, "failed to generate random bytes")
diff --git a/pkg/operator/ceph/cluster/osd/create.go b/pkg/operator/ceph/cluster/osd/create.go
index bf38a71ae..9d5d97d3e 100644
--- a/pkg/operator/ceph/cluster/osd/create.go
+++ b/pkg/operator/ceph/cluster/osd/create.go
@@ -187,7 +187,7 @@ func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig, errs *provi
 
 		if osdProps.encrypted {
 			// create encryption Kubernetes Secret if the PVC is encrypted
-			key, err := generateDmCryptKey()
+			key, err := GenerateDmCryptKey()
 			if err != nil {
 				errMsg := fmt.Sprintf("failed to generate dmcrypt key for osd claim %q. %v", osdProps.pvc.ClaimName, err)
 				errs.addError(errMsg)