diff --git a/pkg/operator/ceph/controller/cluster_info.go b/pkg/operator/ceph/controller/cluster_info.go
index 9e3fb75a8..010c4fbaa 100644
--- a/pkg/operator/ceph/controller/cluster_info.go
+++ b/pkg/operator/ceph/controller/cluster_info.go
@@ -168,6 +168,10 @@ func CreateOrLoadClusterInfo(clusterdContext *clusterd.Context, context context.
 	if len(clusterInfo.CSIDriverSpec.ReadAffinity.CrushLocationLabels) == 0 {
 		clusterInfo.CSIDriverSpec.ReadAffinity.CrushLocationLabels = strings.Split(topology.GetDefaultTopologyLabels(), ",")
 	}
+	// if the csi spec does not specify mount options, apply them from the operator env vars
+	if cephClusterSpec.CSI.CephFS.KernelMountOptions == "" {
+		clusterInfo.CSIDriverSpec.CephFS.KernelMountOptions = k8sutil.GetOperatorSetting("CSI_CEPHFS_KERNEL_MOUNT_OPTIONS", "")
+	}
 
 	// If an admin key was provided we don't need to load the other resources
 	// Some people might want to give the admin key
diff --git a/pkg/operator/ceph/controller/cluster_info_test.go b/pkg/operator/ceph/controller/cluster_info_test.go
index d95394dac..2034199a5 100644
--- a/pkg/operator/ceph/controller/cluster_info_test.go
+++ b/pkg/operator/ceph/controller/cluster_info_test.go
@@ -139,4 +139,25 @@ func TestCreateClusterSecrets(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, "testid", info.CephCred.Username)
 	assert.Equal(t, "testkey", info.CephCred.Secret)
+	assert.Equal(t, "", info.CSIDriverSpec.CephFS.KernelMountOptions)
+
+	// Verify the csi settings are loaded from an env var
+	t.Setenv("CSI_CEPHFS_KERNEL_MOUNT_OPTIONS", "ms_mode=crc")
+	info, _, _, err = CreateOrLoadClusterInfo(context, ctx, namespace, ownerInfo, cephClusterSpec)
+	assert.NoError(t, err)
+	assert.Equal(t, "ms_mode=crc", info.CSIDriverSpec.CephFS.KernelMountOptions)
+
+	// Verify the csi settings are loaded from the cephClusterSpec if specified and override the env var
+	cephClusterSpec.CSI = cephv1.CSIDriverSpec{
+		ReadAffinity: cephv1.ReadAffinitySpec{
+			Enabled:             true,
+			CrushLocationLabels: []string{"kubernetes.io/hostname"},
+		},
+		CephFS: cephv1.CSICephFSSpec{
+			KernelMountOptions: "ms_mode=secure",
+		},
+	}
+	info, _, _, err = CreateOrLoadClusterInfo(context, ctx, namespace, ownerInfo, cephClusterSpec)
+	assert.NoError(t, err)
+	assert.Equal(t, "ms_mode=secure", info.CSIDriverSpec.CephFS.KernelMountOptions)
 }