Merge pull request #11749 from Rakshith-R/enc-key-rotation

Implement OSD encryption key rotation

.github/workflows/canary-integration-test.yml

@@ -696,11 +696,6 @@ jobs:
           kubectl -n rook-ceph get secrets
           sudo lsblk
 
-      - name: test osd deployment removal and re-hydration
-        run: |
-          kubectl -n rook-ceph delete deploy/rook-ceph-osd-0
-          tests/scripts/github-action-helper.sh wait_for_ceph_to_be_ready osd 2
-
       - name: teardown cluster with cleanup policy
         run: |
           kubectl -n rook-ceph patch cephcluster rook-ceph --type merge -p '{"spec":{"cleanupPolicy":{"confirmation":"yes-really-destroy-data"}}}'
@@ -806,6 +801,7 @@ jobs:
           cat tests/manifests/test-on-pvc-db.yaml >> tests/manifests/test-cluster-on-pvc-encrypted.yaml
           cat tests/manifests/test-on-pvc-wal.yaml >> tests/manifests/test-cluster-on-pvc-encrypted.yaml
           kubectl create -f tests/manifests/test-cluster-on-pvc-encrypted.yaml
+          kubectl patch -n rook-ceph cephcluster rook-ceph --type merge -p '{"spec":{"security":{"keyRotation":{"enabled": true, "schedule":"*/1 * * * *"}}}}'
           tests/scripts/github-action-helper.sh deploy_manifest_with_local_build deploy/examples/toolbox.yaml
 
       - name: wait for prepare pod
@@ -817,6 +813,14 @@ jobs:
           kubectl -n rook-ceph get pods
           kubectl -n rook-ceph get secrets
 
+      - name: wait and verify key rotation
+        run: tests/scripts/github-action-helper.sh verify_key_rotation
+
+      - name: test osd deployment removal and re-hydration
+        run: |
+          kubectl -n rook-ceph delete deploy/rook-ceph-osd-0
+          tests/scripts/github-action-helper.sh wait_for_ceph_to_be_ready osd 1
+
       - name: collect common logs
         if: always()
         uses: ./.github/workflows/collect-logs

Documentation/Storage-Configuration/Advanced/key-management-system.md

@@ -3,6 +3,7 @@ title: Key Management System
 ---
 
 Rook has the ability to encrypt OSDs of clusters running on PVC via the flag (`encrypted: true`) in your `storageClassDeviceSets` [template](#pvc-based-cluster).
+Rook also has the ability to rotate encryption keys of OSDs using a cron job per OSD.
 By default, the Key Encryption Keys (also known as Data Encryption Keys) are stored in a Kubernetes Secret.
 However, if a Key Management System exists Rook is capable of using it.
 
@@ -12,6 +13,12 @@ The `security` section contains settings related to encryption of the cluster.
   * `kms`: Key Management System settings
     * `connectionDetails`: the list of parameters representing kms connection details
     * `tokenSecretName`: the name of the Kubernetes Secret containing the kms authentication token
+  * `keyRotation`: Key Rotation settings
+    * `enabled`: whether key rotation is enabled or not, default is `false`
+    * `schedule`: the schedule, written in [cron format](https://en.wikipedia.org/wiki/Cron), with which key rotation [CronJob](https://kubernetes.io/docs/concepts/workloads/controllers/cron-jobs/) is created, default value is `"@weekly"`.
+
+!!! note
+    Currently key rotation is only supported for the default type, where the Key Encryption Keys are stored in a Kubernetes Secret.
 
 Supported KMS providers:
 

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))
+	}
+}

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"]

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

@@ -121,6 +121,7 @@ rules:
   - create
   - update
   - delete
+  - deletecollection
 # The Rook operator must be able to watch all ceph.rook.io resources to reconcile them.
 - apiGroups: ["ceph.rook.io"]
   resources:
@@ -207,6 +208,14 @@ rules:
   - update
   - delete
   - deletecollection
+- apiGroups:
+  - apps
+  resources:
+  # This is to add osd deployment owner ref on key rotation
+  # cron jobs.
+  - deployments/finalizers
+  verbs:
+  - update
 - apiGroups:
   - healthchecking.openshift.io
   resources:

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

@@ -2548,6 +2548,19 @@ spec:
                   description: Security represents security settings
                   nullable: true
                   properties:
+                    keyRotation:
+                      description: KeyRotation defines options for Key Rotation.
+                      nullable: true
+                      properties:
+                        enabled:
+                          default: false
+                          description: Enabled represents whether the key rotation is enabled.
+                          type: boolean
+                        schedule:
+                          default: '@weekly'
+                          description: Schedule represents the cron schedule for key rotation.
+                          type: string
+                      type: object
                     kms:
                       description: KeyManagementService is the main Key Management option
                       nullable: true
@@ -12986,6 +12999,19 @@ spec:
                   description: Security represents security settings
                   nullable: true
                   properties:
+                    keyRotation:
+                      description: KeyRotation defines options for Key Rotation.
+                      nullable: true
+                      properties:
+                        enabled:
+                          default: false
+                          description: Enabled represents whether the key rotation is enabled.
+                          type: boolean
+                        schedule:
+                          default: '@weekly'
+                          description: Schedule represents the cron schedule for key rotation.
+                          type: string
+                      type: object
                     kms:
                       description: KeyManagementService is the main Key Management option
                       nullable: true

deploy/examples/cluster-on-pvc.yaml

@@ -185,6 +185,15 @@ spec:
     pgHealthCheckTimeout: 0
   # security oriented settings
   # security:
+  # Settings to enable key rotation for KEK(Key Encryption Key).
+  # Currently, this is supported only for the default encryption type,
+  # using kubernetes secrets.
+  #   keyRotation:
+  #     enabled: true
+  #     # The schedule, written in [cron format](https://en.wikipedia.org/wiki/Cron),
+  #     # with which key rotation [CronJob](https://kubernetes.io/docs/concepts/workloads/controllers/cron-jobs/)
+  #     # is created. The default value is `"@weekly"`.
+  #     schedule: "@monthly"
   # To enable the KMS configuration properly don't forget to uncomment the Secret at the end of the file
   #   kms:
   #     # name of the config map containing all the kms connection details

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"]

deploy/examples/common.yaml

@@ -260,6 +260,7 @@ rules:
       - create
       - update
       - delete
+      - deletecollection
   # The Rook operator must be able to watch all ceph.rook.io resources to reconcile them.
   - apiGroups: ["ceph.rook.io"]
     resources:
@@ -346,6 +347,14 @@ rules:
       - update
       - delete
       - deletecollection
+  - apiGroups:
+      - apps
+    resources:
+      # This is to add osd deployment owner ref on key rotation
+      # cron jobs.
+      - deployments/finalizers
+    verbs:
+      - update
   - apiGroups:
       - healthchecking.openshift.io
     resources:
@@ -801,10 +810,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"]

deploy/examples/crds.yaml

@@ -2546,6 +2546,19 @@ spec:
                   description: Security represents security settings
                   nullable: true
                   properties:
+                    keyRotation:
+                      description: KeyRotation defines options for Key Rotation.
+                      nullable: true
+                      properties:
+                        enabled:
+                          default: false
+                          description: Enabled represents whether the key rotation is enabled.
+                          type: boolean
+                        schedule:
+                          default: '@weekly'
+                          description: Schedule represents the cron schedule for key rotation.
+                          type: string
+                      type: object
                     kms:
                       description: KeyManagementService is the main Key Management option
                       nullable: true
@@ -12978,6 +12991,19 @@ spec:
                   description: Security represents security settings
                   nullable: true
                   properties:
+                    keyRotation:
+                      description: KeyRotation defines options for Key Rotation.
+                      nullable: true
+                      properties:
+                        enabled:
+                          default: false
+                          description: Enabled represents whether the key rotation is enabled.
+                          type: boolean
+                        schedule:
+                          default: '@weekly'
+                          description: Schedule represents the cron schedule for key rotation.
+                          type: string
+                      type: object
                     kms:
                       description: KeyManagementService is the main Key Management option
                       nullable: true

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

@@ -42,6 +42,11 @@ func GetMonAnnotations(a AnnotationsSpec) Annotations {
 	return mergeAllAnnotationsWithKey(a, KeyMon)
 }
 
+// GetKeyRotationAnnotations returns the annotations for the key rotation job
+func GetKeyRotationAnnotations(a AnnotationsSpec) Annotations {
+	return mergeAllAnnotationsWithKey(a, KeyRotation)
+}
+
 // GetOSDPrepareAnnotations returns the annotations for the OSD service
 func GetOSDPrepareAnnotations(a AnnotationsSpec) Annotations {
 	return mergeAllAnnotationsWithKey(a, KeyOSDPrepare)

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

@@ -24,6 +24,7 @@ const (
 	KeyMonArbiter      KeyType = "arbiter"
 	KeyMgr             KeyType = "mgr"
 	KeyOSDPrepare      KeyType = "prepareosd"
+	KeyRotation        KeyType = "keyrotation"
 	KeyOSD             KeyType = "osd"
 	KeyCleanup         KeyType = "cleanup"
 	KeyMonitoring      KeyType = "monitoring"

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

@@ -48,6 +48,11 @@ func GetMonLabels(a LabelsSpec) Labels {
 	return mergeAllLabelsWithKey(a, KeyMon)
 }
 
+// GetKeyRotationLabels returns labels for the key Rotation job
+func GetKeyRotationLabels(a LabelsSpec) Labels {
+	return mergeAllLabelsWithKey(a, KeyRotation)
+}
+
 // GetOSDPrepareLabels returns the Labels for the OSD prepare job
 func GetOSDPrepareLabels(a LabelsSpec) Labels {
 	return mergeAllLabelsWithKey(a, KeyOSDPrepare)

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

@@ -242,6 +242,10 @@ type SecuritySpec struct {
 	// +optional
 	// +nullable
 	KeyManagementService KeyManagementServiceSpec `json:"kms,omitempty"`
+	// KeyRotation defines options for Key Rotation.
+	// +optional
+	// +nullable
+	KeyRotation KeyRotationSpec `json:"keyRotation,omitempty"`
 }
 
 // ObjectStoreSecuritySpec is spec to define security features like encryption
@@ -268,6 +272,18 @@ type KeyManagementServiceSpec struct {
 	TokenSecretName string `json:"tokenSecretName,omitempty"`
 }
 
+// KeyRotationSpec represents the settings for Key Rotation.
+type KeyRotationSpec struct {
+	// Enabled represents whether the key rotation is enabled.
+	// +optional
+	// +kubebuilder:default=false
+	Enabled bool `json:"enabled,omitempty"`
+	// Schedule represents the cron schedule for key rotation.
+	// +optional
+	// +kubebuilder:default="@weekly"
+	Schedule string `json:"schedule,omitempty"`
+}
+
 // CephVersionSpec represents the settings for the Ceph version that Rook is orchestrating.
 type CephVersionSpec struct {
 	// Image is the container image used to launch the ceph daemons, such as quay.io/ceph/ceph:<tag>

pkg/daemon/ceph/osd/encryption.go

@@ -30,6 +30,7 @@ import (
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/rook/rook/pkg/daemon/ceph/osd/kms"
 	oposd "github.com/rook/rook/pkg/operator/ceph/cluster/osd"
+	"github.com/rook/rook/pkg/util"
 )
 
 const (
@@ -173,6 +174,123 @@ func openEncryptedDevice(context *clusterd.Context, disk, target, passphrase str
 	return nil
 }
 
+// removeEncryptionKeySlot removes the given key slot from the target disk.
+// This function ignores error indicating that the given key slot is not active.
+func removeEncryptionKeySlot(context *clusterd.Context, disk, passphrase, slot string) error {
+	passphraseFile, err := util.CreateTempFile(passphrase)
+	if err != nil {
+		return errors.Wrapf(err, "failed to create passphrase file")
+	}
+	defer os.Remove(passphraseFile.Name())
+
+	args := []string{
+		"--verbose",
+		fmt.Sprintf("--key-file=%s", passphraseFile.Name()),
+		"luksKillSlot",
+		disk,
+		slot,
+	}
+	output, err := context.Executor.ExecuteCommandWithTimeout(luksOpenCmdTimeOut,
+		cryptsetupBinary, args...)
+	// ignore the error if the key slot is not active.
+	if err != nil && !strings.Contains(err.Error()+output, fmt.Sprintf("Keyslot %s is not active", slot)) {
+		return errors.Wrapf(err, "failed to remove key slot %q of encrypted device %q: %q",
+			slot, disk, output)
+	}
+
+	return nil
+}
+
+// ensureEncryptionKey ensures the given key is in the given slot of the target disk.
+// If the error, received from lukChangeKey cmd, shows that the key did not match,
+// the function will return false and no error signify that the given key is not in the slot.
+func ensureEncryptionKey(context *clusterd.Context, disk, passphrase, slot string) (bool, error) {
+	passphraseFile, err := util.CreateTempFile(passphrase)
+	if err != nil {
+		return false, errors.Wrapf(err, "failed to create passphrase file")
+	}
+	defer os.Remove(passphraseFile.Name())
+
+	args := []string{
+		"--verbose",
+		fmt.Sprintf("--key-file=%s", passphraseFile.Name()),
+		fmt.Sprintf("--key-slot=%s", slot),
+		"luksChangeKey",
+		disk,
+		passphraseFile.Name(),
+	}
+	output, err := context.Executor.ExecuteCommandWithTimeout(luksOpenCmdTimeOut,
+		cryptsetupBinary, args...)
+	if err != nil {
+		if strings.Contains(err.Error()+output, "No key available with this passphrase") {
+			// ignore the error if the key does not match the one in key slot and return false
+			// to signify no match.
+			return false, nil
+		}
+
+		return false, errors.Wrapf(err, "failed to ensure passphrase in slot %q of encrypted device %q: %q",
+			slot, disk, output)
+	}
+
+	return true, nil
+}
+
+// addEncryptionKey adds a new key to the given slot of the target disk.
+// If the given slot is filled:
+// - a check is done to see if the key in the give slot is the same as the new key
+//   - if the key is the same, nothing is done.
+//   - else, the slot is wiped and the new key is added.
+func addEncryptionKey(context *clusterd.Context, disk, passphrase, newPassphrase, slot string) error {
+	passphraseFile, err := util.CreateTempFile(passphrase)
+	if err != nil {
+		return errors.Wrapf(err, "failed to create passphrase file")
+	}
+	defer os.Remove(passphraseFile.Name())
+
+	newPassphraseFile, err := util.CreateTempFile(newPassphrase)
+	if err != nil {
+		return errors.Wrapf(err, "failed to create new passphrase file")
+	}
+	defer os.Remove(newPassphraseFile.Name())
+
+	args := []string{
+		"--verbose",
+		fmt.Sprintf("--key-file=%s", passphraseFile.Name()),
+		fmt.Sprintf("--key-slot=%s", slot),
+		"luksAddKey",
+		disk,
+		newPassphraseFile.Name(),
+	}
+	output, err := context.Executor.ExecuteCommandWithTimeout(luksOpenCmdTimeOut,
+		cryptsetupBinary, args...)
+	if err == nil {
+		return nil
+	}
+	if strings.Contains(err.Error()+output, fmt.Sprintf("Key slot %s is full", slot)) {
+		// run ensureEncryptionKey to make sure the newPassphrase is the one that is set.
+		matched, err := ensureEncryptionKey(context, disk, newPassphrase, slot)
+		if err != nil {
+			return errors.Wrapf(err, "failed to ensure passphrase in slot %q of encrypted device %q", slot, disk)
+		}
+		// if newPassphrase is not one in the slot, then remove the key slot and
+		// add add the newPassphrase to it.
+		if !matched {
+			err = removeEncryptionKeySlot(context, disk, newPassphrase, slot)
+			if err != nil {
+				return errors.Wrapf(err, "failed to remove key slot %q of encrypted device %q", slot, disk)
+			}
+			err = addEncryptionKey(context, disk, passphrase, newPassphrase, slot)
+			if err != nil {
+				return errors.Wrapf(err, "failed to add new passphrase to slot %q encrypted device %q", slot, disk)
+			}
+		}
+		return nil
+	}
+
+	return errors.Wrapf(err, "failed to add new passphrase to encrypted device %q: %q",
+		disk, output)
+}
+
 func removeEncryptedDevice(context *clusterd.Context, target string) error {
 	args := []string{"remove", "--force", target}
 	output, err := context.Executor.ExecuteCommandWithTimeout(removeEncryptedDeviceCmdTimeOut, "dmsetup", args...)

pkg/daemon/ceph/osd/encryption_test.go

@@ -17,7 +17,9 @@ limitations under the License.
 package osd
 
 import (
+	"strings"
 	"testing"
+	"time"
 
 	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/clusterd"
@@ -83,6 +85,153 @@ func TestCloseEncryptedDevice(t *testing.T) {
 	assert.NoError(t, err)
 }
 
+func TestRemoveEncryptionKeySlot(t *testing.T) {
+	t.Run("given key slot is active", func(t *testing.T) {
+		executor := &exectest.MockExecutor{}
+		executor.MockExecuteCommandWithTimeout = func(duration time.Duration, command string, args ...string) (string, error) {
+			logger.Infof("%s %v", command, args)
+			if command == "cryptsetup" && args[0] == "--verbose" &&
+				strings.HasPrefix(args[1], "--key-file=") && args[2] == "luksKillSlot" {
+				return "success", nil
+			}
+
+			return "", errors.Errorf("unknown command %s %s", command, args)
+		}
+
+		context := &clusterd.Context{Executor: executor}
+		err := removeEncryptionKeySlot(context, "/dev/mapper/ceph-43e9efed-0676-4731-b75a-a4c42ece1bb1-xvdbr-block-dmcrypt",
+			"passphrase", "0")
+		assert.NoError(t, err)
+	})
+	t.Run("given key slot is not active", func(t *testing.T) {
+		executor := &exectest.MockExecutor{}
+		executor.MockExecuteCommandWithTimeout = func(duration time.Duration, command string, args ...string) (string, error) {
+			logger.Infof("%s %v", command, args)
+
+			return "Keyslot 0 is not active", errors.New("Failed")
+		}
+
+		context := &clusterd.Context{Executor: executor}
+		err := removeEncryptionKeySlot(context, "/dev/mapper/ceph-43e9efed-0676-4731-b75a-a4c42ece1bb1-xvdbr-block-dmcrypt",
+			"passphrase", "0")
+		assert.NoError(t, err)
+	})
+}
+
+func TestEnsureEncryptionKey(t *testing.T) {
+	t.Run("given passphrase matches with the one in key slot", func(t *testing.T) {
+		executor := &exectest.MockExecutor{}
+		executor.MockExecuteCommandWithTimeout = func(duration time.Duration, command string, args ...string) (string, error) {
+			logger.Infof("%s %v", command, args)
+			if command == "cryptsetup" && args[0] == "--verbose" &&
+				strings.HasPrefix(args[1], "--key-file=") && strings.HasPrefix(args[2], "--key-slot=") &&
+				args[3] == "luksChangeKey" {
+				return "success", nil
+			}
+
+			return "", errors.Errorf("unknown command %s %s", command, args)
+		}
+
+		context := &clusterd.Context{Executor: executor}
+		matched, err := ensureEncryptionKey(context, "/dev/mapper/ceph-43e9efed-0676-4731-b75a-a4c42ece1bb1-xvdbr-block-dmcrypt",
+			"passphrase", "0")
+		assert.NoError(t, err)
+		assert.True(t, matched)
+	})
+
+	t.Run("given passphrase does not matching with the one in key slot", func(t *testing.T) {
+		executor := &exectest.MockExecutor{}
+
+		executor.MockExecuteCommandWithTimeout = func(duration time.Duration, command string, args ...string) (string, error) {
+			logger.Infof("%s %v", command, args)
+			return "No key available with this passphrase", errors.New("Failed")
+		}
+
+		context := &clusterd.Context{Executor: executor}
+		matched, err := ensureEncryptionKey(context, "/dev/mapper/ceph-43e9efed-0676-4731-b75a-a4c42ece1bb1-xvdbr-block-dmcrypt",
+			"passphrase", "0")
+		assert.NoError(t, err)
+		assert.False(t, matched)
+	})
+}
+
+func TestAddEncryptionKey(t *testing.T) {
+	t.Run("given key slot is empty.", func(t *testing.T) {
+		executor := &exectest.MockExecutor{}
+		executor.MockExecuteCommandWithTimeout = func(duration time.Duration, command string, args ...string) (string, error) {
+			logger.Infof("%s %v", command, args)
+			if command == "cryptsetup" && args[0] == "--verbose" &&
+				strings.HasPrefix(args[1], "--key-file=") && strings.HasPrefix(args[2], "--key-slot=") &&
+				args[3] == "luksAddKey" {
+				return "success", nil
+			}
+
+			return "", errors.Errorf("unknown command %s %s", command, args)
+		}
+
+		context := &clusterd.Context{Executor: executor}
+		err := addEncryptionKey(context, "/dev/mapper/ceph-43e9efed-0676-4731-b75a-a4c42ece1bb1-xvdbr-block-dmcrypt",
+			"passphrase", "newPassphrase", "0")
+		assert.NoError(t, err)
+	})
+
+	t.Run("given key slot is already filled with newPassphrase.", func(t *testing.T) {
+		executor := &exectest.MockExecutor{}
+		executor.MockExecuteCommandWithTimeout = func(duration time.Duration, command string, args ...string) (string, error) {
+			logger.Infof("%s %v", command, args)
+			if command == "cryptsetup" && args[0] == "--verbose" &&
+				strings.HasPrefix(args[1], "--key-file=") && strings.HasPrefix(args[2], "--key-slot=") &&
+				args[3] == "luksAddKey" {
+				return "Key slot 0 is full", errors.New("Failed")
+			}
+			if command == "cryptsetup" && args[0] == "--verbose" &&
+				strings.HasPrefix(args[1], "--key-file=") && strings.HasPrefix(args[2], "--key-slot=") &&
+				args[3] == "luksChangeKey" {
+				return "success", nil
+			}
+
+			return "", errors.Errorf("unknown command %s %s", command, args)
+		}
+
+		context := &clusterd.Context{Executor: executor}
+		err := addEncryptionKey(context, "/dev/mapper/ceph-43e9efed-0676-4731-b75a-a4c42ece1bb1-xvdbr-block-dmcrypt",
+			"passphrase", "newPassphrase", "0")
+		assert.NoError(t, err)
+	})
+
+	t.Run("given key slot is already filled but does not match with newPassphrase", func(t *testing.T) {
+		failedOnce := false
+		executor := &exectest.MockExecutor{}
+		executor.MockExecuteCommandWithTimeout = func(duration time.Duration, command string, args ...string) (string, error) {
+			logger.Infof("%s %v", command, args)
+			if command == "cryptsetup" && args[0] == "--verbose" &&
+				strings.HasPrefix(args[1], "--key-file=") && strings.HasPrefix(args[2], "--key-slot=") &&
+				args[3] == "luksAddKey" {
+				if failedOnce {
+					return "success", nil
+				}
+				failedOnce = true
+				return "Key slot 0 is full", errors.New("Failed")
+			}
+			if command == "cryptsetup" && args[0] == "--verbose" &&
+				strings.HasPrefix(args[1], "--key-file=") && strings.HasPrefix(args[2], "--key-slot=") &&
+				args[3] == "luksChangeKey" {
+				return "No key available with this passphrase", errors.New("Failed")
+			}
+			if command == "cryptsetup" && args[0] == "--verbose" &&
+				strings.HasPrefix(args[1], "--key-file=") && args[2] == "luksKillSlot" {
+				return "success", nil
+			}
+
+			return "", errors.Errorf("unknown command %s %s", command, args)
+		}
+		context := &clusterd.Context{Executor: executor}
+		err := addEncryptionKey(context, "/dev/mapper/ceph-43e9efed-0676-4731-b75a-a4c42ece1bb1-xvdbr-block-dmcrypt",
+			"passphrase", "newPassphrase", "0")
+		assert.NoError(t, err)
+	})
+}
+
 func TestDmsetupVersion(t *testing.T) {
 	dmsetupOutput := `
 Library version:   1.02.154 (2018-12-07)

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
+}

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

@@ -54,6 +54,24 @@ func (c *Config) storeSecretInKubernetes(pvcName, key string) error {
 	return nil
 }
 
+// updateSecretInKubernetes updates the dmcrypt key in a Kubernetes Secret
+func (c *Config) updateSecretInKubernetes(pvcName, key string) error {
+	secretName := GenerateOSDEncryptionSecretName(pvcName)
+	secret, err := c.context.Clientset.CoreV1().Secrets(c.ClusterInfo.Namespace).Get(c.ClusterInfo.Context, secretName, metav1.GetOptions{})
+	if err != nil {
+		return errors.Wrapf(err, "failed to get secret %q", secretName)
+	}
+
+	secret.StringData = map[string]string{OsdEncryptionSecretNameKeyName: key}
+	// Update the Kubernetes Secret
+	_, err = c.context.Clientset.CoreV1().Secrets(c.ClusterInfo.Namespace).Update(c.ClusterInfo.Context, secret, metav1.UpdateOptions{})
+	if err != nil {
+		return errors.Wrapf(err, "failed to update ceph osd encryption key for pvc %q", pvcName)
+	}
+
+	return nil
+}
+
 // getKubernetesSecret returns key value from secret.
 func (c *Config) getKubernetesSecret(pvcName string) (string, error) {
 	secretName := GenerateOSDEncryptionSecretName(pvcName)

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

@@ -151,8 +151,16 @@ func (c *Config) PutSecret(secretName, secretValue string) error {
 // GetSecret returns an encrypted key from a KMS
 func (c *Config) GetSecret(secretName string) (string, error) {
 	var value string
+	if c.IsK8s() {
+		// Retrieve the secret from Kubernetes Secrets
+		value, err := c.getKubernetesSecret(secretName)
+		if err != nil {
+			return "", errors.Wrap(err, "failed to get secret from kubernetes secret")
+		}
+		return value, nil
+	}
 	if c.IsVault() {
-		// Store the secret in Vault
+		// Retrieve the secret from Vault
 		v, err := InitVault(c.ClusterInfo.Context, c.context, c.ClusterInfo.Namespace, c.clusterSpec.Security.KeyManagementService.ConnectionDetails)
 		if err != nil {
 			return "", errors.Wrap(err, "failed to init vault")
@@ -195,6 +203,22 @@ func (c *Config) GetSecret(secretName string) (string, error) {
 	return value, nil
 }
 
+// UpdateSecret updates the encrypted key in a KMS
+func (c *Config) UpdateSecret(secretName, secretValue string) error {
+	// If Kubernetes Secret KMS is selected (default)
+	if c.IsK8s() {
+		// Update the secret in Kubernetes Secrets
+		err := c.updateSecretInKubernetes(secretName, secretValue)
+		if err != nil {
+			return errors.Wrap(err, "failed to update secret in kubernetes secret")
+		}
+
+		return nil
+	}
+
+	return errors.Errorf("update secret is not supported for the %q KMS", c.Provider)
+}
+
 // DeleteSecret deletes an encrypted key from a KMS
 func (c *Config) DeleteSecret(secretName string) error {
 	if c.IsVault() {

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")

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)

pkg/operator/ceph/cluster/osd/key_rotation.go

@@ -0,0 +1,284 @@
+/*
+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 (
+	"fmt"
+	"path/filepath"
+
+	"github.com/pkg/errors"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	kms "github.com/rook/rook/pkg/daemon/ceph/osd/kms"
+	"github.com/rook/rook/pkg/operator/ceph/controller"
+	"github.com/rook/rook/pkg/operator/k8sutil"
+	batch "k8s.io/api/batch/v1"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	ctrl "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+)
+
+const (
+	keyRotationCronJobAppName    = "rook-ceph-osd-key-rotation"
+	keyRotationCronJobAppNameFmt = "rook-ceph-osd-key-rotation-%d"
+)
+
+// keyRotationCronJobName returns the name of the key rotation cron job for the given OSD ID.
+func keyRotationCronJobName(osdID int) string {
+	return fmt.Sprintf(keyRotationCronJobAppNameFmt, osdID)
+}
+
+// applyKeyRotationPlacement applies the placement settings for the key rotation job
+// so that it is scheduled on the same node as the OSD for which the key rotation is scheduled.
+func applyKeyRotationPlacement(spec *v1.PodSpec, labels map[string]string) {
+	spec.TopologySpreadConstraints = nil
+	if spec.Affinity == nil {
+		spec.Affinity = &v1.Affinity{}
+	} else {
+		spec.Affinity.PodAntiAffinity = nil
+	}
+	spec.Affinity.PodAffinity = &v1.PodAffinity{
+		RequiredDuringSchedulingIgnoredDuringExecution: []v1.PodAffinityTerm{
+			{
+				LabelSelector: &metav1.LabelSelector{
+					MatchLabels: labels,
+				},
+				TopologyKey: v1.LabelHostname,
+			},
+		},
+	}
+}
+
+// getKeyRotationContainer returns the container spec for the key rotation job.
+func (c *Cluster) getKeyRotationContainer(osdProps osdProperties, volumeMounts []v1.VolumeMount, devices []string) (v1.Container, error) {
+	envVars := c.getConfigEnvVars(osdProps, k8sutil.DataDir, true)
+
+	// enable debug logging
+	envVars = append(envVars, setDebugLogLevelEnvVar(true))
+	envVars = append(envVars, v1.EnvVar{Name: "ROOK_CEPH_VERSION", Value: c.clusterInfo.CephVersion.CephVersionFormatted()})
+	envVars = append(envVars, kms.ConfigToEnvVar(c.spec)...)
+
+	// run privileged always since we always mount /dev
+	privileged := true
+	runAsUser := int64(0)
+	runAsNonRoot := false
+	readOnlyRootFilesystem := false
+
+	args := []string{"key-management", "rotate-key", osdProps.pvc.ClaimName}
+	args = append(args, devices...)
+
+	osdProvisionContainer := v1.Container{
+		Args:            args,
+		Name:            keyRotationCronJobAppName,
+		Image:           c.rookVersion,
+		ImagePullPolicy: controller.GetContainerImagePullPolicy(c.spec.CephVersion.ImagePullPolicy),
+		VolumeMounts:    volumeMounts,
+		Env:             envVars,
+		EnvFrom:         getEnvFromSources(),
+		SecurityContext: &v1.SecurityContext{
+			Privileged:             &privileged,
+			RunAsUser:              &runAsUser,
+			RunAsNonRoot:           &runAsNonRoot,
+			ReadOnlyRootFilesystem: &readOnlyRootFilesystem,
+		},
+		Resources: osdProps.resources,
+	}
+
+	return osdProvisionContainer, nil
+}
+
+// getKeyRotationPodTemplateSpec returns the pod template spec for the key rotation job.
+func (c *Cluster) getKeyRotationPodTemplateSpec(osdProps osdProperties, osd OSDInfo, restart v1.RestartPolicy) (*v1.PodTemplateSpec, error) {
+	// create a volume on /dev so the pod can access devices on the host
+	devVolume := v1.Volume{Name: "devices", VolumeSource: v1.VolumeSource{HostPath: &v1.HostPathVolumeSource{Path: "/dev"}}}
+	udevVolume := v1.Volume{Name: "udev", VolumeSource: v1.VolumeSource{HostPath: &v1.HostPathVolumeSource{Path: "/run/udev"}}}
+	hostPathType := v1.HostPathDirectory
+	hostPath := filepath.Join(c.spec.DataDirHostPath, c.clusterInfo.Namespace, osdProps.pvc.ClaimName, fmt.Sprintf("ceph-%d", osd.ID))
+	hostPathVolume := v1.Volume{
+		Name: "bridge",
+		VolumeSource: v1.VolumeSource{
+			HostPath: &v1.HostPathVolumeSource{
+				Path: hostPath,
+				Type: &hostPathType,
+			},
+		},
+	}
+	devicesBasePath := "/var/lib/ceph/osd/"
+	volumes := []v1.Volume{
+		udevVolume,
+		devVolume,
+		hostPathVolume,
+	}
+	volumeMounts := []v1.VolumeMount{
+		{Name: "devices", MountPath: "/dev"},
+		{Name: "udev", MountPath: "/run/udev"},
+		{Name: "bridge", MountPath: devicesBasePath},
+	}
+
+	devices := []string{encryptionBlockDestinationCopy(devicesBasePath, bluestoreBlockName)}
+	if osdProps.metadataPVC.ClaimName != "" {
+		devices = append(devices, encryptionBlockDestinationCopy(devicesBasePath, bluestoreMetadataName))
+	}
+	if osdProps.walPVC.ClaimName != "" {
+		devices = append(devices, encryptionBlockDestinationCopy(devicesBasePath, bluestoreWalName))
+	}
+
+	keyRotationContainer, err := c.getKeyRotationContainer(osdProps, volumeMounts, devices)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to generate key rotation container")
+	}
+
+	podTemplateSpec := v1.PodTemplateSpec{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: keyRotationCronJobAppName,
+			Labels: map[string]string{
+				k8sutil.AppAttr:     keyRotationCronJobAppName,
+				k8sutil.ClusterAttr: c.clusterInfo.Namespace,
+			},
+			Annotations: map[string]string{},
+		},
+		Spec: v1.PodSpec{
+			ServiceAccountName: serviceAccountName,
+			Containers: []v1.Container{
+				keyRotationContainer,
+			},
+			RestartPolicy:     restart,
+			Volumes:           volumes,
+			HostNetwork:       c.spec.Network.IsHost(),
+			PriorityClassName: cephv1.GetOSDPriorityClassName(c.spec.PriorityClassNames),
+			SchedulerName:     osdProps.schedulerName,
+		},
+	}
+	if c.spec.Network.IsHost() {
+		podTemplateSpec.Spec.DNSPolicy = v1.DNSClusterFirstWithHostNet
+	} else if c.spec.Network.IsMultus() {
+		if err := k8sutil.ApplyMultus(c.spec.Network, &podTemplateSpec.ObjectMeta); err != nil {
+			return nil, err
+		}
+	}
+
+	cephv1.GetKeyRotationAnnotations(c.spec.Annotations).ApplyToObjectMeta(&podTemplateSpec.ObjectMeta)
+	cephv1.GetKeyRotationLabels(c.spec.Labels).ApplyToObjectMeta(&podTemplateSpec.ObjectMeta)
+
+	c.applyAllPlacementIfNeeded(&podTemplateSpec.Spec)
+	// apply storageClassDeviceSets.Placement
+	osdProps.placement.ApplyToPodSpec(&podTemplateSpec.Spec)
+	applyKeyRotationPlacement(&podTemplateSpec.Spec, c.getOSDLabels(osd, osdProps.crushHostname, osdProps.portable))
+
+	// cryptsetup synchronizes with udev on host through semaphore
+	podTemplateSpec.Spec.HostIPC = true
+
+	k8sutil.RemoveDuplicateEnvVars(&podTemplateSpec.Spec)
+	return &podTemplateSpec, nil
+}
+
+// makeKeyRotationCronJob creates a key rotation cron job for the given OSD.
+func (c *Cluster) makeKeyRotationCronJob(pvcName string, osd OSDInfo, osdProps osdProperties) (*batch.CronJob, error) {
+	podSpec, err := c.getKeyRotationPodTemplateSpec(osdProps, osd, v1.RestartPolicyOnFailure)
+	if err != nil {
+		return nil, err
+	}
+	c.applyResourcesToAllContainers(&podSpec.Spec, cephv1.GetOSDResources(c.spec.Resources, osd.DeviceClass))
+	cronJob := &batch.CronJob{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        keyRotationCronJobName(osd.ID),
+			Namespace:   c.clusterInfo.Namespace,
+			Labels:      podSpec.Labels,
+			Annotations: podSpec.Annotations,
+		},
+		Spec: batch.CronJobSpec{
+			ConcurrencyPolicy: batch.ForbidConcurrent,
+			Schedule:          c.spec.Security.KeyRotation.Schedule,
+			JobTemplate: batch.JobTemplateSpec{
+				Spec: batch.JobSpec{
+					Template: *podSpec,
+				},
+			},
+		},
+	}
+
+	return cronJob, nil
+}
+
+// reconcileKeyRotationCronJobs reconciles the key rotation cron jobs for the OSDs.
+func (c *Cluster) reconcileKeyRotationCronJob() error {
+	if !c.spec.Security.KeyRotation.Enabled {
+		listOpts := metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s", k8sutil.AppAttr, keyRotationCronJobAppName)}
+		err := c.context.Clientset.BatchV1().
+			CronJobs(c.clusterInfo.Namespace).
+			DeleteCollection(c.clusterInfo.Context,
+				metav1.DeleteOptions{},
+				listOpts)
+		if err != nil {
+			if client.IgnoreNotFound(err) != nil {
+				return errors.Wrap(err, "failed to delete key rotation cron jobs")
+			}
+			return nil
+		}
+		logger.Debugf("successfully deleted key rotation cron jobs")
+
+		return nil
+	}
+
+	// Get the list of OSDs backed by pvc.
+	osdListOpts := metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s,%s", k8sutil.AppAttr, AppName, OSDOverPVCLabelKey)}
+	deployments, err := c.context.Clientset.AppsV1().Deployments(c.clusterInfo.Namespace).List(c.clusterInfo.Context, osdListOpts)
+	if err != nil {
+		return errors.Wrap(err, "failed to query existing OSD deployments")
+	}
+
+	logger.Debugf("found %d osd deployments", len(deployments.Items))
+	for i := range deployments.Items {
+		osdDep := deployments.Items[i]
+		osd, err := c.getOSDInfo(&osdDep)
+		if err != nil {
+			return errors.Wrapf(err, "failed to get osd info for osd %q", osdDep.Name)
+		}
+		pvcName := osdDep.Labels[OSDOverPVCLabelKey]
+		if pvcName == "" {
+			return errors.Errorf("pvc name label %q for osd %q is empty",
+				OSDOverPVCLabelKey, osdDep.Name)
+		}
+		osdProps, err := c.getOSDPropsForPVC(pvcName, osd.DeviceClass)
+		if err != nil {
+			return errors.Wrapf(err, "failed to generate config for osd %q", osdDep.Name)
+		}
+		if !osdProps.encrypted {
+			continue
+		}
+
+		logger.Infof("starting OSD key rotation cron job for osd %q", osd.ID)
+		cj, err := c.makeKeyRotationCronJob(pvcName, osd, osdProps)
+		if err != nil {
+			return errors.Wrap(err, "failed to make key rotation cron job")
+		}
+
+		err = ctrl.SetOwnerReference(&osdDep, cj, c.context.Client.Scheme())
+		if err != nil {
+			return errors.Wrapf(err, "failed to set controllerReference on cron job %q", cj.Name)
+		}
+
+		_, err = k8sutil.CreateOrUpdateCronJob(c.clusterInfo.Context, c.context.Clientset, cj)
+		if err != nil {
+			return errors.Wrapf(err, "failed to create or update key rotation cron job %q", cj.Name)
+		}
+		logger.Infof("started OSD key rotation cron job %q", cj.Name)
+	}
+	logger.Infof("successfully started OSD key rotation cron jobs")
+
+	return nil
+}

pkg/operator/ceph/cluster/osd/key_rotation_test.go

@@ -0,0 +1,116 @@
+/*
+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 (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func Test_keyRotationCronJobName(t *testing.T) {
+	type args struct {
+		osdID int
+	}
+	tests := []struct {
+		name string
+		args args
+		want string
+	}{
+		{
+			name: "OSD ID 0",
+			args: args{
+				osdID: 0,
+			},
+			want: fmt.Sprintf(keyRotationCronJobAppNameFmt, 0),
+		},
+		{
+			name: "OSD ID 1",
+			args: args{
+				osdID: 1,
+			},
+			want: fmt.Sprintf(keyRotationCronJobAppNameFmt, 1),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := keyRotationCronJobName(tt.args.osdID); got != tt.want {
+				t.Errorf("keyRotationCronJobName() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_applyKeyRotationPlacement(t *testing.T) {
+	type args struct {
+		spec   *v1.PodSpec
+		labels map[string]string
+	}
+	tests := []struct {
+		name string
+		args args
+	}{
+		{
+			name: "test 1",
+			args: args{
+				spec: &v1.PodSpec{
+					Affinity: &v1.Affinity{
+						PodAffinity:     &v1.PodAffinity{},
+						PodAntiAffinity: &v1.PodAntiAffinity{},
+					},
+					TopologySpreadConstraints: []v1.TopologySpreadConstraint{},
+				},
+				labels: map[string]string{
+					"app":    "rook-ceph-osd",
+					"osd-id": "0",
+				},
+			},
+		},
+		{
+			name: "Affinity is nil",
+			args: args{
+				spec: &v1.PodSpec{
+					Affinity:                  nil,
+					TopologySpreadConstraints: []v1.TopologySpreadConstraint{},
+				},
+				labels: map[string]string{
+					"app":    "rook-ceph-osd",
+					"osd-id": "0",
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		currentTT := tt
+		t.Run(currentTT.name, func(t *testing.T) {
+			applyKeyRotationPlacement(currentTT.args.spec, currentTT.args.labels)
+			assert.Nil(t, currentTT.args.spec.Affinity.PodAntiAffinity)
+			assert.Nil(t, currentTT.args.spec.TopologySpreadConstraints)
+			assert.Equal(t, currentTT.args.spec.Affinity.PodAffinity, &v1.PodAffinity{
+				RequiredDuringSchedulingIgnoredDuringExecution: []v1.PodAffinityTerm{
+					{
+						LabelSelector: &metav1.LabelSelector{MatchLabels: currentTT.args.labels},
+						TopologyKey:   v1.LabelHostname,
+					},
+				},
+			})
+		})
+	}
+}

pkg/operator/ceph/cluster/osd/osd.go

@@ -241,6 +241,11 @@ func (c *Cluster) Start() error {
 	// The following block is used to apply any command(s) required by an upgrade
 	c.applyUpgradeOSDFunctionality()
 
+	err = c.reconcileKeyRotationCronJob()
+	if err != nil {
+		return errors.Wrapf(err, "failed to reconcile key rotation cron jobs")
+	}
+
 	logger.Infof("finished running OSDs in namespace %q", namespace)
 	return nil
 }

pkg/operator/k8sutil/deployment.go

@@ -522,6 +522,17 @@ func CreateDeployment(ctx context.Context, clientset kubernetes.Interface, dep *
 	return clientset.AppsV1().Deployments(dep.Namespace).Create(ctx, dep, metav1.CreateOptions{})
 }
 
+// createCronJob creates a cron job with a last applied hash annotation added
+func createCronJob(ctx context.Context, clientset kubernetes.Interface, cj *batchv1.CronJob) (*batchv1.CronJob, error) {
+	// Set hash annotation to the newly generated deployment
+	err := patch.DefaultAnnotator.SetLastAppliedAnnotation(cj)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to set hash annotation on cron job %q", cj.Name)
+	}
+
+	return clientset.BatchV1().CronJobs(cj.Namespace).Create(ctx, cj, metav1.CreateOptions{})
+}
+
 func CreateOrUpdateDeployment(ctx context.Context, clientset kubernetes.Interface, dep *appsv1.Deployment) (*appsv1.Deployment, error) {
 	newDep, err := CreateDeployment(ctx, clientset, dep)
 	if err != nil {
@@ -536,6 +547,20 @@ func CreateOrUpdateDeployment(ctx context.Context, clientset kubernetes.Interfac
 	return newDep, nil
 }
 
+func CreateOrUpdateCronJob(ctx context.Context, clientset kubernetes.Interface, cj *batchv1.CronJob) (*batchv1.CronJob, error) {
+	newCJ, err := createCronJob(ctx, clientset, cj)
+	if err != nil {
+		if k8serrors.IsAlreadyExists(err) {
+			// annotation was added in CreateCronJob to cj passed by reference
+			newCJ, err = clientset.BatchV1().CronJobs(cj.Namespace).Update(ctx, cj, metav1.UpdateOptions{})
+		}
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to create or update cronjob %q: %+v", cj.Name, cj)
+		}
+	}
+	return newCJ, nil
+}
+
 func maxInt32Ptr(a, b *int32) *int32 {
 	var ret int32
 	if a == nil && b == nil {

tests/scripts/github-action-helper.sh

@@ -329,6 +329,31 @@ function wait_for_ceph_to_be_ready() {
   kubectl -n rook-ceph get pods
 }
 
+function verify_key_rotation(){
+  pvc_name=$(kubectl get pvc -n rook-ceph -l ceph.rook.io/setIndex=0 -o jsonpath='{.items[0].metadata.name}')
+  old_key=$(kubectl -n rook-ceph get secrets -l "pvc_name=$pvc_name" -o jsonpath='{.items[0].data.'dmcrypt-key'}' | base64 --decode)
+  runtime="3 minutes"
+  endtime=$(date -ud "$runtime" +%s)
+  while [[ $(date -u +%s) -le $endtime ]]
+  do
+      echo "Time Now: `date +%H:%M:%S`"
+      new_key=$(kubectl -n rook-ceph get secrets -l "pvc_name=$pvc_name" -o jsonpath='{.items[0].data.'dmcrypt-key'}' | base64 --decode)
+      if [ "$old_key" != "$new_key" ]; then
+        echo "encryption passphrase is successfully rotated"
+        exit 0
+      fi
+      echo "encryption passphrase is not rotated, sleeping for 10 seconds"
+      sleep 10s
+  done
+  new_key=$(kubectl -n rook-ceph get secrets -l "pvc_name=$pvc_name" -o jsonpath='{.items[0].data.'dmcrypt-key'}' | base64 --decode)
+  if [ "$old_key" == "$new_key" ]; then
+    echo "encryption passphrase is not rotated"
+    exit 1
+  else
+    echo "encryption passphrase is successfully rotated"
+  fi
+}
+
 function check_ownerreferences() {
   curl -L https://github.com/kubernetes-sigs/kubectl-check-ownerreferences/releases/download/v0.2.0/kubectl-check-ownerreferences-linux-amd64.tar.gz -o kubectl-check-ownerreferences-linux-amd64.tar.gz
   tar xzvf kubectl-check-ownerreferences-linux-amd64.tar.gz