cephfs: fscrypt encryption support

Add Ceph FS fscrypt support, similar to the RBD/ext4 fscrypt
integration. Supports encrypted PVCs, snapshots and clones.

Requires kernel and Ceph MDS support that is currently not in any
stable release.

Signed-off-by: Marcel Lauhoff <marcel.lauhoff@suse.com>

internal/cephfs/controllerserver.go

@@ -26,6 +26,7 @@ import (
 	"github.com/ceph/ceph-csi/internal/cephfs/store"
 	fsutil "github.com/ceph/ceph-csi/internal/cephfs/util"
 	csicommon "github.com/ceph/ceph-csi/internal/csi-common"
+	"github.com/ceph/ceph-csi/internal/kms"
 	"github.com/ceph/ceph-csi/internal/util"
 	"github.com/ceph/ceph-csi/internal/util/k8s"
 	"github.com/ceph/ceph-csi/internal/util/log"
@@ -66,18 +67,29 @@ func (cs *ControllerServer) createBackingVolume(
 	ctx context.Context,
 	volOptions,
 	parentVolOpt *store.VolumeOptions,
-	pvID *store.VolumeIdentifier,
+	vID, pvID *store.VolumeIdentifier,
 	sID *store.SnapshotIdentifier,
+	secrets map[string]string,
 ) error {
 	var err error
 	volClient := core.NewSubVolume(volOptions.GetConnection(),
 		&volOptions.SubVolume, volOptions.ClusterID, cs.ClusterName, cs.SetMetadata)
 
 	if sID != nil {
-		return cs.createBackingVolumeFromSnapshotSource(ctx, volOptions, parentVolOpt, volClient, sID)
+		err = parentVolOpt.CopyEncryptionConfig(volOptions, sID.SnapshotID, vID.VolumeID)
+		if err != nil {
+			return status.Error(codes.Internal, err.Error())
+		}
+
+		return cs.createBackingVolumeFromSnapshotSource(ctx, volOptions, parentVolOpt, volClient, sID, secrets)
 	}
 
 	if parentVolOpt != nil {
+		err = parentVolOpt.CopyEncryptionConfig(volOptions, pvID.VolumeID, vID.VolumeID)
+		if err != nil {
+			return status.Error(codes.Internal, err.Error())
+		}
+
 		return cs.createBackingVolumeFromVolumeSource(ctx, parentVolOpt, volClient, pvID)
 	}
 
@@ -96,6 +108,7 @@ func (cs *ControllerServer) createBackingVolumeFromSnapshotSource(
 	parentVolOpt *store.VolumeOptions,
 	volClient core.SubVolumeClient,
 	sID *store.SnapshotIdentifier,
+	secrets map[string]string,
 ) error {
 	if err := cs.OperationLocks.GetRestoreLock(sID.SnapshotID); err != nil {
 		log.ErrorLog(ctx, err.Error())
@@ -105,7 +118,7 @@ func (cs *ControllerServer) createBackingVolumeFromSnapshotSource(
 	defer cs.OperationLocks.ReleaseRestoreLock(sID.SnapshotID)
 
 	if volOptions.BackingSnapshot {
-		if err := store.AddSnapshotBackedVolumeRef(ctx, volOptions, cs.ClusterName, cs.SetMetadata); err != nil {
+		if err := store.AddSnapshotBackedVolumeRef(ctx, volOptions, cs.ClusterName, cs.SetMetadata, secrets); err != nil {
 			log.ErrorLog(ctx, "failed to create snapshot-backed volume from snapshot %s: %v",
 				sID.FsSnapshotName, err)
 
@@ -162,7 +175,8 @@ func (cs *ControllerServer) checkContentSource(
 	switch volumeSource.Type.(type) {
 	case *csi.VolumeContentSource_Snapshot:
 		snapshotID := req.VolumeContentSource.GetSnapshot().GetSnapshotId()
-		volOpt, _, sid, err := store.NewSnapshotOptionsFromID(ctx, snapshotID, cr, cs.ClusterName, cs.SetMetadata)
+		volOpt, _, sid, err := store.NewSnapshotOptionsFromID(ctx, snapshotID, cr,
+			req.GetSecrets(), cs.ClusterName, cs.SetMetadata)
 		if err != nil {
 			if errors.Is(err, cerrors.ErrSnapNotFound) {
 				return nil, nil, nil, status.Error(codes.NotFound, err.Error())
@@ -294,6 +308,7 @@ func (cs *ControllerServer) CreateVolume(
 
 		return nil, status.Error(codes.Internal, err.Error())
 	}
+
 	// TODO return error message if requested vol size greater than found volume return error
 
 	metadata := k8s.GetVolumeMetadata(req.GetParameters())
@@ -370,7 +385,7 @@ func (cs *ControllerServer) CreateVolume(
 	}()
 
 	// Create a volume
-	err = cs.createBackingVolume(ctx, volOptions, parentVol, pvID, sID)
+	err = cs.createBackingVolume(ctx, volOptions, parentVol, vID, pvID, sID, req.GetSecrets())
 	if err != nil {
 		if cerrors.IsCloneRetryError(err) {
 			return nil, status.Error(codes.Aborted, err.Error())
@@ -529,7 +544,7 @@ func (cs *ControllerServer) DeleteVolume(
 	}
 	defer cr.DeleteCredentials()
 
-	if err := cs.cleanUpBackingVolume(ctx, volOptions, vID, cr); err != nil {
+	if err := cs.cleanUpBackingVolume(ctx, volOptions, vID, cr, secrets); err != nil {
 		return nil, err
 	}
 
@@ -547,7 +562,19 @@ func (cs *ControllerServer) cleanUpBackingVolume(
 	volOptions *store.VolumeOptions,
 	volID *store.VolumeIdentifier,
 	cr *util.Credentials,
+	secrets map[string]string,
 ) error {
+	if volOptions.IsEncrypted() && volOptions.Encryption.KMS.RequiresDEKStore() == kms.DEKStoreIntegrated {
+		// Only remove DEK when the KMS stores it itself. On
+		// GetSecret enabled KMS the DEKs are stored by
+		// fscrypt on the volume that is going to be deleted anyway.
+		log.DebugLog(ctx, "going to remove DEK for integrated store %q (fscrypt)", volOptions.Encryption.GetID())
+		if err := volOptions.Encryption.RemoveDEK(volID.VolumeID); err != nil {
+			log.WarningLog(ctx, "failed to clean the passphrase for volume %q (file encryption): %s",
+				volOptions.VolID, err)
+		}
+	}
+
 	if !volOptions.BackingSnapshot {
 		// Regular volumes need to be purged.
 
@@ -585,7 +612,7 @@ func (cs *ControllerServer) cleanUpBackingVolume(
 	}
 
 	snapParentVolOptions, _, snapID, err := store.NewSnapshotOptionsFromID(ctx,
-		volOptions.BackingSnapshotID, cr, cs.ClusterName, cs.SetMetadata)
+		volOptions.BackingSnapshotID, cr, secrets, cs.ClusterName, cs.SetMetadata)
 	if err != nil {
 		absorbErrs := []error{
 			util.ErrPoolNotFound,
@@ -874,6 +901,14 @@ func (cs *ControllerServer) CreateSnapshot(
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
+	// Use same encryption KMS than source volume and copy the passphrase. The passphrase becomes
+	// available under the snapshot id for CreateVolume to use this snap as a backing volume
+	snapVolOptions := store.VolumeOptions{}
+	err = parentVolOptions.CopyEncryptionConfig(&snapVolOptions, sourceVolID, sID.SnapshotID)
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
 	return &csi.CreateSnapshotResponse{
 		Snapshot: &csi.Snapshot{
 			SizeBytes:      info.BytesQuota,
@@ -991,7 +1026,8 @@ func (cs *ControllerServer) DeleteSnapshot(
 	}
 	defer cs.OperationLocks.ReleaseDeleteLock(snapshotID)
 
-	volOpt, snapInfo, sid, err := store.NewSnapshotOptionsFromID(ctx, snapshotID, cr, cs.ClusterName, cs.SetMetadata)
+	volOpt, snapInfo, sid, err := store.NewSnapshotOptionsFromID(ctx, snapshotID, cr,
+		req.GetSecrets(), cs.ClusterName, cs.SetMetadata)
 	if err != nil {
 		switch {
 		case errors.Is(err, util.ErrPoolNotFound):

internal/cephfs/nodeserver.go

@@ -30,6 +30,7 @@ import (
 	fsutil "github.com/ceph/ceph-csi/internal/cephfs/util"
 	csicommon "github.com/ceph/ceph-csi/internal/csi-common"
 	"github.com/ceph/ceph-csi/internal/util"
+	"github.com/ceph/ceph-csi/internal/util/fscrypt"
 	"github.com/ceph/ceph-csi/internal/util/log"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
@@ -88,7 +89,7 @@ func (ns *NodeServer) getVolumeOptions(
 			return nil, status.Error(codes.Internal, err.Error())
 		}
 
-		volOptions, _, err = store.NewVolumeOptionsFromStaticVolume(string(volID), volContext)
+		volOptions, _, err = store.NewVolumeOptionsFromStaticVolume(string(volID), volContext, volSecrets)
 		if err != nil {
 			if !errors.Is(err, cerrors.ErrNonStaticVolume) {
 				return nil, status.Error(codes.Internal, err.Error())
@@ -118,6 +119,39 @@ func validateSnapshotBackedVolCapability(volCap *csi.VolumeCapability) error {
 	return nil
 }
 
+// maybeUnlockFileEncryption unlocks fscrypt on stagingTargetPath, if volOptions enable encryption.
+func maybeUnlockFileEncryption(
+	ctx context.Context,
+	volOptions *store.VolumeOptions,
+	stagingTargetPath string,
+	volID fsutil.VolumeID,
+) error {
+	if volOptions.IsEncrypted() {
+		log.DebugLog(ctx, "cephfs: unlocking fscrypt on volume %q path %s", volID, stagingTargetPath)
+
+		return fscrypt.Unlock(ctx, volOptions.Encryption, stagingTargetPath, string(volID))
+	}
+
+	return nil
+}
+
+// maybeInitializeFileEncryption initializes KMS and node specifics, if volContext enables encryption.
+func maybeInitializeFileEncryption(
+	ctx context.Context,
+	mnt mounter.VolumeMounter,
+	volOptions *store.VolumeOptions,
+) error {
+	if volOptions.IsEncrypted() {
+		if _, isFuse := mnt.(*mounter.FuseMounter); isFuse {
+			return errors.New("FUSE mounter does not support encryption")
+		}
+
+		return fscrypt.InitializeNode(ctx)
+	}
+
+	return nil
+}
+
 // NodeStageVolume mounts the volume to a staging path on the node.
 func (ns *NodeServer) NodeStageVolume(
 	ctx context.Context,
@@ -170,6 +204,11 @@ func (ns *NodeServer) NodeStageVolume(
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
+	err = maybeInitializeFileEncryption(ctx, mnt, volOptions)
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
 	// Check if the volume is already mounted
 
 	if err = ns.tryRestoreFuseMountInNodeStage(ctx, mnt, stagingTargetPath); err != nil {
@@ -185,6 +224,9 @@ func (ns *NodeServer) NodeStageVolume(
 
 	if isMnt {
 		log.DebugLog(ctx, "cephfs: volume %s is already mounted to %s, skipping", volID, stagingTargetPath)
+		if err = maybeUnlockFileEncryption(ctx, volOptions, stagingTargetPath, volID); err != nil {
+			return nil, status.Error(codes.Internal, err.Error())
+		}
 
 		return &csi.NodeStageVolumeResponse{}, nil
 	}
@@ -205,6 +247,10 @@ func (ns *NodeServer) NodeStageVolume(
 
 	log.DebugLog(ctx, "cephfs: successfully mounted volume %s to %s", volID, stagingTargetPath)
 
+	if err = maybeUnlockFileEncryption(ctx, volOptions, stagingTargetPath, volID); err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
 	if _, isFuse := mnt.(*mounter.FuseMounter); isFuse {
 		// FUSE mount recovery needs NodeStageMountinfo records.
 
@@ -452,6 +498,16 @@ func (ns *NodeServer) NodePublishVolume(
 	}
 
 	// It's not, mount now
+	encrypted, err := store.IsEncrypted(ctx, req.GetVolumeContext())
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+	if encrypted {
+		stagingTargetPath = fscrypt.AppendEncyptedSubdirectory(stagingTargetPath)
+		if err = fscrypt.IsDirectoryUnlocked(stagingTargetPath, "ceph"); err != nil {
+			return nil, status.Error(codes.Internal, err.Error())
+		}
+	}
 
 	if err = mounter.BindMount(
 		ctx,

internal/cephfs/store/backingsnapshot.go

@@ -36,6 +36,7 @@ func AddSnapshotBackedVolumeRef(
 	volOptions *VolumeOptions,
 	clusterName string,
 	setMetadata bool,
+	secrets map[string]string,
 ) error {
 	ioctx, err := volOptions.conn.GetIoctx(volOptions.MetadataPool)
 	if err != nil {
@@ -98,7 +99,7 @@ func AddSnapshotBackedVolumeRef(
 	// deleting the backing snapshot. Make sure the snapshot still exists by
 	// trying to retrieve it again.
 	_, _, _, err = NewSnapshotOptionsFromID(ctx,
-		volOptions.BackingSnapshotID, volOptions.conn.Creds, clusterName, setMetadata)
+		volOptions.BackingSnapshotID, volOptions.conn.Creds, secrets, clusterName, setMetadata)
 	if err != nil {
 		log.ErrorLog(ctx, "failed to get backing snapshot %s: %v", volOptions.BackingSnapshotID, err)
 	}

internal/cephfs/store/fsjournal.go

@@ -90,8 +90,10 @@ func CheckVolExists(ctx context.Context,
 	}
 	defer j.Destroy()
 
+	kmsID, encryptionType := getEncryptionConfig(volOptions)
+
 	imageData, err := j.CheckReservation(
-		ctx, volOptions.MetadataPool, volOptions.RequestName, volOptions.NamePrefix, "", "", util.EncryptionTypeNone)
+		ctx, volOptions.MetadataPool, volOptions.RequestName, volOptions.NamePrefix, "", kmsID, encryptionType)
 	if err != nil {
 		return nil, err
 	}
@@ -249,6 +251,14 @@ func updateTopologyConstraints(volOpts *VolumeOptions) error {
 	return nil
 }
 
+func getEncryptionConfig(volOptions *VolumeOptions) (string, util.EncryptionType) {
+	if volOptions.IsEncrypted() {
+		return volOptions.Encryption.GetID(), util.EncryptionTypeFile
+	}
+
+	return "", util.EncryptionTypeNone
+}
+
 // ReserveVol is a helper routine to request a UUID reservation for the CSI VolumeName and,
 // to generate the volume identifier for the reserved UUID.
 func ReserveVol(ctx context.Context, volOptions *VolumeOptions, secret map[string]string) (*VolumeIdentifier, error) {
@@ -276,10 +286,13 @@ func ReserveVol(ctx context.Context, volOptions *VolumeOptions, secret map[strin
 	}
 	defer j.Destroy()
 
+	kmsID, encryptionType := getEncryptionConfig(volOptions)
+
 	imageUUID, vid.FsSubvolName, err = j.ReserveName(
 		ctx, volOptions.MetadataPool, util.InvalidPoolID,
 		volOptions.MetadataPool, util.InvalidPoolID, volOptions.RequestName,
-		volOptions.NamePrefix, "", "", volOptions.ReservedID, "", volOptions.BackingSnapshotID, util.EncryptionTypeNone)
+		volOptions.NamePrefix, "", kmsID, volOptions.ReservedID, volOptions.Owner,
+		volOptions.BackingSnapshotID, encryptionType)
 	if err != nil {
 		return nil, err
 	}
@@ -319,10 +332,13 @@ func ReserveSnap(
 	}
 	defer j.Destroy()
 
+	kmsID, encryptionType := getEncryptionConfig(volOptions)
+
 	imageUUID, vid.FsSnapshotName, err = j.ReserveName(
 		ctx, volOptions.MetadataPool, util.InvalidPoolID,
 		volOptions.MetadataPool, util.InvalidPoolID, snap.RequestName,
-		snap.NamePrefix, parentSubVolName, "", snap.ReservedID, "", "", util.EncryptionTypeNone)
+		snap.NamePrefix, parentSubVolName, kmsID, snap.ReservedID, "",
+		volOptions.Owner, encryptionType)
 	if err != nil {
 		return nil, err
 	}
@@ -390,8 +406,10 @@ func CheckSnapExists(
 	}
 	defer j.Destroy()
 
+	kmsID, encryptionType := getEncryptionConfig(volOptions)
+
 	snapData, err := j.CheckReservation(
-		ctx, volOptions.MetadataPool, snap.RequestName, snap.NamePrefix, volOptions.VolID, "", util.EncryptionTypeNone)
+		ctx, volOptions.MetadataPool, snap.RequestName, snap.NamePrefix, volOptions.VolID, kmsID, encryptionType)
 	if err != nil {
 		return nil, nil, err
 	}

internal/cephfs/store/volumeoptions.go

@@ -29,10 +29,16 @@ import (
 	"github.com/ceph/ceph-csi/internal/cephfs/core"
 	cerrors "github.com/ceph/ceph-csi/internal/cephfs/errors"
 	fsutil "github.com/ceph/ceph-csi/internal/cephfs/util"
+	kmsapi "github.com/ceph/ceph-csi/internal/kms"
 	"github.com/ceph/ceph-csi/internal/util"
+	"github.com/ceph/ceph-csi/internal/util/k8s"
 	"github.com/ceph/ceph-csi/internal/util/log"
 )
 
+const (
+	cephfsDefaultEncryptionType = util.EncryptionTypeFile
+)
+
 type VolumeOptions struct {
 	core.SubVolume
 
@@ -55,6 +61,11 @@ type VolumeOptions struct {
 	Topology             map[string]string
 	FscID                int64
 
+	// Encryption provides access to optional VolumeEncryption functions
+	Encryption *util.VolumeEncryption
+	// Owner is the creator (tenant, Kubernetes Namespace) of the volume
+	Owner string
+
 	// conn is a connection to the Ceph cluster obtained from a ConnPool
 	conn *util.ClusterConnection
 
@@ -84,6 +95,9 @@ func (vo *VolumeOptions) Destroy() {
 	if vo.conn != nil {
 		vo.conn.Destroy()
 	}
+	if vo.IsEncrypted() {
+		vo.Encryption.Destroy()
+	}
 }
 
 func validateNonEmptyField(field, fieldName string) error {
@@ -219,6 +233,7 @@ func NewVolumeOptions(
 	opts.ClusterID = clusterData.ClusterID
 	opts.Monitors = strings.Join(clusterData.Monitors, ",")
 	opts.SubvolumeGroup = clusterData.CephFS.SubvolumeGroup
+	opts.Owner = k8s.GetOwner(volOptions)
 
 	if err = extractOptionalOption(&opts.Pool, "pool", volOptions); err != nil {
 		return nil, err
@@ -248,6 +263,10 @@ func NewVolumeOptions(
 		return nil, err
 	}
 
+	if err = opts.InitKMS(ctx, volOptions, req.GetSecrets()); err != nil {
+		return nil, fmt.Errorf("failed to init KMS: %w", err)
+	}
+
 	if backingSnapshotBool != "" {
 		if opts.BackingSnapshot, err = strconv.ParseBool(backingSnapshotBool); err != nil {
 			return nil, fmt.Errorf("failed to parse backingSnapshot: %w", err)
@@ -294,7 +313,7 @@ func NewVolumeOptions(
 
 		opts.BackingSnapshotID = req.GetVolumeContentSource().GetSnapshot().GetSnapshotId()
 
-		err = opts.populateVolumeOptionsFromBackingSnapshot(ctx, cr, clusterName, setMetadata)
+		err = opts.populateVolumeOptionsFromBackingSnapshot(ctx, cr, req.GetSecrets(), clusterName, setMetadata)
 		if err != nil {
 			return nil, err
 		}
@@ -382,6 +401,7 @@ func NewVolumeOptionsFromVolID(
 	}
 	volOptions.RequestName = imageAttributes.RequestName
 	vid.FsSubvolName = imageAttributes.ImageName
+	volOptions.Owner = imageAttributes.Owner
 
 	if volOpt != nil {
 		if err = extractOptionalOption(&volOptions.Pool, "pool", volOpt); err != nil {
@@ -403,6 +423,10 @@ func NewVolumeOptionsFromVolID(
 		if err = extractMounter(&volOptions.Mounter, volOpt); err != nil {
 			return nil, nil, err
 		}
+
+		if err = volOptions.InitKMS(ctx, volOpt, secrets); err != nil {
+			return nil, nil, err
+		}
 	}
 
 	if imageAttributes.BackingSnapshotID != "" || volOptions.BackingSnapshotID != "" {
@@ -414,11 +438,18 @@ func NewVolumeOptionsFromVolID(
 	volOptions.SubVolume.VolID = vid.FsSubvolName
 
 	if volOptions.BackingSnapshot {
-		err = volOptions.populateVolumeOptionsFromBackingSnapshot(ctx, cr, clusterName, setMetadata)
+		err = volOptions.populateVolumeOptionsFromBackingSnapshot(ctx, cr, secrets, clusterName, setMetadata)
 	} else {
 		err = volOptions.populateVolumeOptionsFromSubvolume(ctx, clusterName, setMetadata)
 	}
 
+	if volOpt == nil && imageAttributes.KmsID != "" && volOptions.Encryption == nil {
+		err = volOptions.ConfigureEncryption(ctx, imageAttributes.KmsID, secrets)
+		if err != nil {
+			return &volOptions, &vid, err
+		}
+	}
+
 	return &volOptions, &vid, err
 }
 
@@ -447,6 +478,7 @@ func (vo *VolumeOptions) populateVolumeOptionsFromSubvolume(
 func (vo *VolumeOptions) populateVolumeOptionsFromBackingSnapshot(
 	ctx context.Context,
 	cr *util.Credentials,
+	secrets map[string]string,
 	clusterName string,
 	setMetadata bool,
 ) error {
@@ -471,7 +503,7 @@ func (vo *VolumeOptions) populateVolumeOptionsFromBackingSnapshot(
 	}
 
 	parentBackingSnapVolOpts, _, snapID, err := NewSnapshotOptionsFromID(ctx,
-		vo.BackingSnapshotID, cr, clusterName, setMetadata)
+		vo.BackingSnapshotID, cr, secrets, clusterName, setMetadata)
 	if err != nil {
 		return fmt.Errorf("failed to retrieve backing snapshot %s: %w", vo.BackingSnapshotID, err)
 	}
@@ -576,6 +608,11 @@ func NewVolumeOptionsFromMonitorList(
 		return nil, nil, err
 	}
 
+	opts.Owner = k8s.GetOwner(options)
+	if err = opts.InitKMS(context.TODO(), options, secrets); err != nil {
+		return nil, nil, err
+	}
+
 	vid.FsSubvolName = volID
 	vid.VolumeID = volID
 
@@ -591,7 +628,7 @@ func NewVolumeOptionsFromMonitorList(
 // detected to be a statically provisioned volume.
 func NewVolumeOptionsFromStaticVolume(
 	volID string,
-	options map[string]string,
+	options, secrets map[string]string,
 ) (*VolumeOptions, *VolumeIdentifier, error) {
 	var (
 		opts      VolumeOptions
@@ -625,6 +662,7 @@ func NewVolumeOptionsFromStaticVolume(
 	opts.ClusterID = clusterData.ClusterID
 	opts.Monitors = strings.Join(clusterData.Monitors, ",")
 	opts.SubvolumeGroup = clusterData.CephFS.SubvolumeGroup
+	opts.Owner = k8s.GetOwner(options)
 
 	if err = extractOption(&opts.RootPath, "rootPath", options); err != nil {
 		return nil, nil, err
@@ -650,6 +688,10 @@ func NewVolumeOptionsFromStaticVolume(
 		return nil, nil, err
 	}
 
+	if err = opts.InitKMS(context.TODO(), options, secrets); err != nil {
+		return nil, nil, err
+	}
+
 	vid.FsSubvolName = opts.RootPath
 	vid.VolumeID = volID
 
@@ -666,6 +708,7 @@ func NewSnapshotOptionsFromID(
 	ctx context.Context,
 	snapID string,
 	cr *util.Credentials,
+	secrets map[string]string,
 	clusterName string,
 	setMetadata bool,
 ) (*VolumeOptions, *core.SnapshotInfo, *SnapshotIdentifier, error) {
@@ -739,8 +782,16 @@ func NewSnapshotOptionsFromID(
 	sid.FsSubvolName = imageAttributes.SourceName
 
 	volOptions.SubVolume.VolID = sid.FsSubvolName
+	volOptions.Owner = imageAttributes.Owner
 	vol := core.NewSubVolume(volOptions.conn, &volOptions.SubVolume, volOptions.ClusterID, clusterName, setMetadata)
 
+	if imageAttributes.KmsID != "" && volOptions.Encryption == nil {
+		err = volOptions.ConfigureEncryption(ctx, imageAttributes.KmsID, secrets)
+		if err != nil {
+			return &volOptions, nil, &sid, err
+		}
+	}
+
 	subvolInfo, err := vol.GetSubVolumeInfo(ctx)
 	if err != nil {
 		return &volOptions, nil, &sid, err
@@ -788,3 +839,139 @@ func GenSnapFromOptions(ctx context.Context, req *csi.CreateSnapshotRequest) (*S
 
 	return cephfsSnap, nil
 }
+
+func parseEncryptionOpts(volOptions map[string]string) (string, util.EncryptionType, error) {
+	var (
+		err              error
+		ok               bool
+		encrypted, kmsID string
+	)
+	encrypted, ok = volOptions["encrypted"]
+	if !ok {
+		return "", util.EncryptionTypeNone, nil
+	}
+	kmsID, err = util.FetchEncryptionKMSID(encrypted, volOptions["encryptionKMSID"])
+	if err != nil {
+		return "", util.EncryptionTypeInvalid, err
+	}
+
+	encType := util.FetchEncryptionType(volOptions, cephfsDefaultEncryptionType)
+
+	return kmsID, encType, nil
+}
+
+// IsEncrypted returns true if volOptions enables file encryption.
+func IsEncrypted(ctx context.Context, volOptions map[string]string) (bool, error) {
+	_, encType, err := parseEncryptionOpts(volOptions)
+	if err != nil {
+		return false, err
+	}
+
+	return encType == util.EncryptionTypeFile, nil
+}
+
+// CopyEncryptionConfig copies passphrases and initializes a fresh
+// Encryption struct if necessary from (vo, vID) to (cp, cpVID).
+func (vo *VolumeOptions) CopyEncryptionConfig(cp *VolumeOptions, vID, cpVID string) error {
+	var err error
+
+	if !vo.IsEncrypted() {
+		return nil
+	}
+
+	if vID == cpVID {
+		return fmt.Errorf("BUG: %v and %v have the same VolID %q "+
+			"set!? Call stack: %s", vo, cp, vID, util.CallStack())
+	}
+
+	if cp.Encryption == nil {
+		cp.Encryption, err = util.NewVolumeEncryption(vo.Encryption.GetID(), vo.Encryption.KMS)
+		if errors.Is(err, util.ErrDEKStoreNeeded) {
+			_, err := vo.Encryption.KMS.GetSecret("")
+			if errors.Is(err, kmsapi.ErrGetSecretUnsupported) {
+				return err
+			}
+		}
+	}
+
+	if vo.Encryption.KMS.RequiresDEKStore() == kmsapi.DEKStoreIntegrated {
+		passphrase, err := vo.Encryption.GetCryptoPassphrase(vID)
+		if err != nil {
+			return fmt.Errorf("failed to fetch passphrase for %q (%+v): %w",
+				vID, vo, err)
+		}
+
+		err = cp.Encryption.StoreCryptoPassphrase(cpVID, passphrase)
+		if err != nil {
+			return fmt.Errorf("failed to store passphrase for %q (%+v): %w",
+				cpVID, cp, err)
+		}
+	}
+
+	return nil
+}
+
+// ConfigureEncryption initializes the Ceph CSI key management from
+// kmsID and credentials. Sets vo.Encryption on success.
+func (vo *VolumeOptions) ConfigureEncryption(
+	ctx context.Context,
+	kmsID string,
+	credentials map[string]string,
+) error {
+	kms, err := kmsapi.GetKMS(vo.Owner, kmsID, credentials)
+	if err != nil {
+		log.ErrorLog(ctx, "get KMS failed %+v: %v", vo, err)
+
+		return err
+	}
+
+	vo.Encryption, err = util.NewVolumeEncryption(kmsID, kms)
+
+	if errors.Is(err, util.ErrDEKStoreNeeded) {
+		// fscrypt uses secrets directly from the KMS.
+		// Therefore we do not support an additional DEK
+		// store. Since not all "metadata" KMS support
+		// GetSecret, test for support here. Postpone any
+		// other error handling
+		_, err := vo.Encryption.KMS.GetSecret("")
+		if errors.Is(err, kmsapi.ErrGetSecretUnsupported) {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// InitKMS initialized the Ceph CSI key management by parsing the
+// configuration from volume options + credentials. Sets vo.Encryption
+// on success.
+func (vo *VolumeOptions) InitKMS(
+	ctx context.Context,
+	volOptions, credentials map[string]string,
+) error {
+	var err error
+
+	kmsID, encType, err := parseEncryptionOpts(volOptions)
+	if err != nil {
+		return err
+	}
+
+	if encType == util.EncryptionTypeNone {
+		return nil
+	}
+
+	if encType != util.EncryptionTypeFile {
+		return fmt.Errorf("unsupported encryption type %v. only supported type is 'file'", encType)
+	}
+
+	err = vo.ConfigureEncryption(ctx, kmsID, credentials)
+	if err != nil {
+		return fmt.Errorf("invalid encryption kms configuration: %w", err)
+	}
+
+	return nil
+}
+
+func (vo *VolumeOptions) IsEncrypted() bool {
+	return vo.Encryption != nil
+}