Adds per volume encryption with Vault integration

- adds proposal document for PVC encryption from PR448 - adds per-volume encription by generating encryption passphrase for each volume and storing it in a KMS - adds HashiCorp Vault integration as a KMS for encryption passphrases - avoids encrypting volume second time if it was already encrypted but no file system created - avoids unnecessary checks if volume is a mapped device when encryption was not requested - prevents resizing encrypted volumes (it is not currently supported) - prevents creating snapshots from encrypted volumes to prevent attack on encryption key (security guard until re-encryption of volumes implemented) Signed-off-by: Vasyl Purchel vasyl.purchel@workday.com Signed-off-by: Andrea Baglioni andrea.baglioni@workday.com Fixes #420 Fixes #744
2024-11-22 06:10:22 +00:00 · 2020-01-29 11:44:45 +00:00 · 2020-01-29 11:44:45 +00:00 · 419ad0dd8e
commit 419ad0dd8e
parent 1adef00c86
26 changed files with 1210 additions and 102 deletions
--- a/charts/ceph-csi-rbd/templates/encryptionkms-configmap.yaml
+++ b/charts/ceph-csi-rbd/templates/encryptionkms-configmap.yaml
@ -0,0 +1,14 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: {{ .Values.kmsConfigMapName | quote }}
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app: {{ include "ceph-csi-rbd.name" . }}
+    chart: {{ include "ceph-csi-rbd.chart" . }}
+    component: {{ .Values.nodeplugin.name }}
+    release: {{ .Release.Name }}
+    heritage: {{ .Release.Service }}
+data:
+  config.json: |-
+{{ toJson .Values.encryptionKMSConfig | indent 4 -}}
--- a/charts/ceph-csi-rbd/templates/nodeplugin-daemonset.yaml
+++ b/charts/ceph-csi-rbd/templates/nodeplugin-daemonset.yaml
@ -121,6 +121,8 @@ spec:
              readOnly: true
            - name: ceph-csi-config
              mountPath: /etc/ceph-csi-config/
+            - name: ceph-csi-encryption-kms-config
+              mountPath: /etc/ceph-csi-encryption-kms-config/
            - name: plugin-dir
              mountPath: /var/lib/kubelet/plugins
              mountPropagation: "Bidirectional"
@ -189,6 +191,9 @@ spec:
        - name: ceph-csi-config
          configMap:
            name: {{ .Values.configMapName | quote }}
+        - name: ceph-csi-encryption-kms-config
+          configMap:
+            name: {{ .Values.kmsConfigMapName | quote }}
        - name: keys-tmp-dir
          emptyDir: {
            medium: "Memory"
--- a/charts/ceph-csi-rbd/templates/provisioner-deployment.yaml
+++ b/charts/ceph-csi-rbd/templates/provisioner-deployment.yaml
@ -148,6 +148,8 @@ spec:
              readOnly: true
            - name: ceph-csi-config
              mountPath: /etc/ceph-csi-config/
+            - name: ceph-csi-encryption-kms-config
+              mountPath: /etc/ceph-csi-encryption-kms-config/
            - name: keys-tmp-dir
              mountPath: /tmp/csi/keys
          resources:
@ -193,6 +195,9 @@ spec:
        - name: ceph-csi-config
          configMap:
            name: {{ .Values.configMapName | quote }}
+        - name: ceph-csi-encryption-kms-config
+          configMap:
+            name: {{ .Values.kmsConfigMapName | quote }}
        - name: keys-tmp-dir
          emptyDir: {
            medium: "Memory"
--- a/charts/ceph-csi-rbd/values.yaml
+++ b/charts/ceph-csi-rbd/values.yaml
@ -27,6 +27,14 @@ serviceAccounts:
 #       - "<MONValue2>"
 csiConfig: []

+# Configuration for the encryption KMS
+# Ref: https://github.com/ceph/ceph-csi/blob/master/docs/deploy-rbd.md
+# Example:
+# encryptionKMSConfig:
+#   - encryptionKMSID: "<kms-id>"
+#     <kms-specific-configs>
+encryptionKMSConfig: []
+
 nodeplugin:
  name: nodeplugin
  # if you are using rbd-nbd client set this value to OnDelete
@ -248,3 +256,5 @@ pluginDir: /var/lib/kubelet/plugins
 driverName: rbd.csi.ceph.com
 # Name of the configmap used for state
 configMapName: ceph-csi-config-rbd
+# Name of the configmap used for encryption kms configuration
+kmsConfigMapName: ceph-csi-encryption-kms-config
--- a/deploy/rbd/kubernetes/csi-rbdplugin-provisioner.yaml
+++ b/deploy/rbd/kubernetes/csi-rbdplugin-provisioner.yaml
@ -140,6 +140,8 @@ spec:
              readOnly: true
            - name: ceph-csi-config
              mountPath: /etc/ceph-csi-config/
+            - name: ceph-csi-encryption-kms-config
+              mountPath: /etc/ceph-csi-encryption-kms-config/
            - name: keys-tmp-dir
              mountPath: /tmp/csi/keys
        - name: liveness-prometheus
@ -179,6 +181,9 @@ spec:
        - name: ceph-csi-config
          configMap:
            name: ceph-csi-config
+        - name: ceph-csi-encryption-kms-config
+          configMap:
+            name: ceph-csi-encryption-kms-config
        - name: keys-tmp-dir
          emptyDir: {
            medium: "Memory"
--- a/deploy/rbd/kubernetes/csi-rbdplugin.yaml
+++ b/deploy/rbd/kubernetes/csi-rbdplugin.yaml
@ -96,6 +96,8 @@ spec:
              readOnly: true
            - name: ceph-csi-config
              mountPath: /etc/ceph-csi-config/
+            - name: ceph-csi-encryption-kms-config
+              mountPath: /etc/ceph-csi-encryption-kms-config/
            - name: plugin-dir
              mountPath: /var/lib/kubelet/plugins
              mountPropagation: "Bidirectional"
@ -158,6 +160,9 @@ spec:
        - name: ceph-csi-config
          configMap:
            name: ceph-csi-config
+        - name: ceph-csi-encryption-kms-config
+          configMap:
+            name: ceph-csi-encryption-kms-config
        - name: keys-tmp-dir
          emptyDir: {
            medium: "Memory"
--- a/docs/deploy-rbd.md
+++ b/docs/deploy-rbd.md
@ -1,4 +1,3 @@
-
 # CSI RBD Plugin

 The RBD CSI plugin is able to provision new RBD images and
@ -55,6 +54,8 @@ make image-cephcsi
 | `csi.storage.k8s.io/provisioner-secret-namespace`, `csi.storage.k8s.io/node-stage-secret-namespace` | yes (for Kubernetes) | namespaces of the above Secret objects                                                                                                                                                                                  |
 | `mounter`                                                                                           | no                   | if set to `rbd-nbd`, use `rbd-nbd` on nodes that have `rbd-nbd` and `nbd` kernel modules to map rbd images                                                                                                              |
 | `encrypted`                                                                                         | no                   | disabled by default, use `"true"` to enable LUKS encryption on pvc and `"false"` to disable it. **Do not change for existing storageclasses**                                                                           |
+| `encryptionKMS`                                                                                     | no                   | specifies key management system for encrypytion. Currently supports `vault`                                                                                                                                             |
+| `encryptionKMSID`                                                                                   | no                   | required if `encryptionKMS` is set to `vault` to specify a unique identifier for vault configuration                                                                                                                    |

 **NOTE:** An accompanying CSI configuration file, needs to be provided to the
 running pods. Refer to [Creating CSI configuration](../examples/README.md#creating-csi-configuration)
@ -196,8 +197,10 @@ and `csi.storage.k8s.io/node-stage-secret-namespace`).
 * volume is attached to provisioner container
 * on first time attachment
  (no file system on the attached device, checked with blkid)
+  * new passphrase is generated and stored in selected KMS if KMS is in use
  * device is encrypted with LUKS using a passphrase from K8s secrets
  * image-meta updated to "encrypted" in Ceph
+* passphrase is retrieved from selected KMS if KMS is in use
 * device is open and device path is changed to use a mapper device
 * mapper device is used instead of original one with usual workflow

@ -205,6 +208,7 @@ and `csi.storage.k8s.io/node-stage-secret-namespace`).

 * mapper device closed and device path changed to original volume path
 * volume is detached as usual
+* passphrase removed from KMS if needed (with failures ignored)

 ### Encryption configuration

@ -213,6 +217,38 @@ secrets under `encryptionPassphrase` key and switch `encrypted` option in
 StorageClass to `"true"`. This is not supported for storage classes that already
 have PVs provisioned.

+### Encryption KMS configuration
+
+To further improve security robustness it is possible to use unique passphrases
+generated for each volume and stored in a Key Management System (KMS). Currently
+HashiCorp Vault is the only KMS supported.
+
+To use Vault as KMS set `encryptionKMS` to `vault` and `encryptionKMSID` to a
+unique identifier for Vault configuration. You will also need to create vault
+configuration similar to the [example](../examples/rbd/kms-config.yaml)
+and use same `encryptionKMSID`. In order for ceph-csi to be able to access the
+configuration you will need to have it mounted to csi-rbdplugin containers in
+both daemonset (so kms client can be instantiated to encrypt/decrypt volumes)
+and deployment pods (so kms client can be instantiated to delete passphrase on
+volume delete) `ceph-csi-encryption-kms-config` config map.
+
+#### Configuring HashiCorp Vault
+
+Using Vault as KMS you need to configure Kubernetes authentication method as
+described in [official
+documentation](https://www.vaultproject.io/docs/auth/kubernetes.html).
+
+If token reviewer is used, you will need to configure service account for
+that also like in [example](../examples/rbd/csi-vaulttokenreview-rbac.yaml) to
+be able to review jwt tokens.
+
+Configure a role(s) for service accounts used for ceph-csi:
+
+* provisioner service account (`rbd-csi-provisioner`) requires only **delete**
+  permissions to delete passphrases on pvc delete
+* nodeplugin service account (`rbd-csi-nodeplugin`) requires **create** and
+  **read** permissions to save new keys and retrieve existing
+
 ### Encryption prerequisites

 In order for encryption to work you need to make sure that `dm-crypt` kernel
--- a/docs/design/proposals/encrypted-pvc.md
+++ b/docs/design/proposals/encrypted-pvc.md
@ -0,0 +1,131 @@
+# Encrypted Persistent Volume Claims
+
+## Proposal
+
+Subject of this proposal is to add support for encryption of RBD volumes in
+Ceph-CSI.
+
+Some but not all the benefits of this approach:
+
+* guarantee encryption in transit to rbd without using messenger v2
+* extra security layer to application with special regulatory needs
+* at rest encryption can be disabled to selectively allow encryption only where
+  required
+
+## Document Terminology
+
+* volume encryption: encryption of a volume attached by rbd
+* encryption at rest: encryption of physical disk done by ceph
+* LUKS: Linux Unified Key Setup: stores all of the needed setup information for
+  dm-crypt on the disk
+* dm-crypt: linux kernel device-mapper crypto target
+* cryptsetup: the command line tool to interface with dm-crypt
+
+## Proposed Solution
+
+The proposed solution in this document, is to address the volume encryption
+requirement by using dm-crypt module through cryptsetup cli interface.
+
+### Implementation Summary
+
+* Encryption is implemented using cryptsetup with LUKS extension.
+  A good introduction to LUKS and dm-crypt in general can be found
+  [here](https://wiki.archlinux.org/index.php/Dm-crypt/Device_encryption#Encrypting_devices_with_cryptsetup)
+  Functions to implement necessary interaction are implemented in a separate
+  `cryptsetup.go` file.
+  * LuksFormat
+  * LuksOpen
+  * LuksClose
+  * LuksStatus
+
+* `CreateVolume`: refactored to prepare for encryption (tag image that it
+  requires encryption later), before returning, if encrypted volume option is
+  set.
+* `NodeStageVolume`: refactored to call `encryptDevice` method on the very first
+  volume attach request
+* `NodeStageVolume`: refactored to open encrypted device (`openEncryptedDevice`)
+* `openEncryptedDevice`: looks up for a passphrase matching the volume id,
+  returns the new device path in the form: `/dev/mapper/luks-<volume_id>`.
+  On the woker node where the attach is scheduled:
+
+  ```shell
+  $ lsblk
+  NAME                            MAJ:MIN RM  SIZE RO TYPE  MOUNTPOINT
+  sda                               8:0    0   10G  0 disk
+  └─sda1                            8:1    0   10G  0 part  /
+  sdb                               8:16   0   20G  0 disk
+  rbd0                            253:0    0    1G  0 disk
+  └─luks-pvc-8a710f4c934811e9 252:0    0 1020M  0 crypt /var/lib/kubelet/pods/9eaceaef-936c-11e9-b396-005056af3de0/volumes/kubernetes.io~csi/pvc-8a710f4c934811e9/mount
+  ```
+
+* `detachRBDDevice`: calls `LuksClose` function to remove the LUKS mapping
+  before detaching the volume.
+
+* StorageClass extended with following parameters:
+  1. `encrypted` ("true" or "false")
+  1. `encryptionKMS` (string representing kms of choice)
+  ceph-csi plugin may support different kms vendors with different type of
+  authentication
+  1. `encryptionKMSID` (string representing kms configuration)
+
+* New KMS Configuration created.
+
+#### Annotated YAML for RBD StorageClass
+
+```yaml
+apiVersion: storage.k8s.io/v1
+kind: StorageClass
+metadata:
+   name: csi-rbd
+provisioner: rbd.csi.ceph.com
+parameters:
+   # String representing Ceph cluster configuration
+   clusterID: <cluster-id>
+   # ceph pool
+   pool: rbd
+
+   # RBD image features, CSI creates image with image-format 2
+   # CSI RBD currently supports only `layering` feature.
+   imageFeatures: layering
+
+   # The secrets have to contain Ceph credentials with required access
+   # to the 'pool'.
+   csi.storage.k8s.io/provisioner-secret-name: csi-rbd-secret
+   csi.storage.k8s.io/provisioner-secret-namespace: default
+   csi.storage.k8s.io/controller-expand-secret-name: csi-rbd-secret
+   csi.storage.k8s.io/controller-expand-secret-namespace: default
+   csi.storage.k8s.io/node-stage-secret-name: csi-rbd-secret
+   csi.storage.k8s.io/node-stage-secret-namespace: default
+   # Specify the filesystem type of the volume. If not specified,
+   # csi-provisioner will set default as `ext4`.
+   csi.storage.k8s.io/fstype: ext4
+
+   # Encrypt volumes
+   encrypted: "true"
+
+   # The type of kms we want to connect to: Barbican, aws kms or others can be
+   # supported
+   encryptionKMS: vault
+   # String representing a KMS configuration
+   encryptionKMSID: <kms-id>
+
+reclaimPolicy: Delete
+```
+
+And kms configuration:
+
+```yaml
+---
+apiVersion: v1
+kind: ConfigMap
+data:
+  config.json: |-
+    [
+      {
+        "kmsID": "<kms-id>",
+        kms specific config...
+      }
+    ]
+metadata:
+  name: ceph-csi-encryption-kms-config
+```
--- a/e2e/deploy-vault.go
+++ b/e2e/deploy-vault.go
@ -0,0 +1,54 @@
+package e2e
+
+import (
+	. "github.com/onsi/gomega" // nolint
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/kubernetes/test/e2e/framework"
+	e2elog "k8s.io/kubernetes/test/e2e/framework/log"
+)
+
+var (
+	vaultExamplePath = "../examples/kms/vault/"
+	vaultServicePath = "vault.yaml"
+	vaultPSPPath     = "vault-psp.yaml"
+	vaultRBACPath    = "csi-vaulttokenreview-rbac.yaml"
+	vaultConfigPath  = "kms-config.yaml"
+)
+
+func deployVault(c kubernetes.Interface, deployTimeout int) {
+	framework.RunKubectlOrDie("create", "-f", vaultExamplePath+vaultServicePath)
+	framework.RunKubectlOrDie("create", "-f", vaultExamplePath+vaultPSPPath)
+	framework.RunKubectlOrDie("create", "-f", vaultExamplePath+vaultRBACPath)
+	framework.RunKubectlOrDie("create", "-f", vaultExamplePath+vaultConfigPath)
+
+	opt := metav1.ListOptions{
+		LabelSelector: "app=vault",
+	}
+
+	pods, err := c.CoreV1().Pods("default").List(opt)
+	Expect(err).Should(BeNil())
+	Expect(len(pods.Items)).Should(Equal(1))
+	name := pods.Items[0].Name
+	err = waitForPodInRunningState(name, "default", c, deployTimeout)
+	Expect(err).Should(BeNil())
+}
+
+func deleteVault() {
+	_, err := framework.RunKubectl("delete", "-f", vaultExamplePath+vaultServicePath)
+	if err != nil {
+		e2elog.Logf("failed to delete vault statefull set %v", err)
+	}
+	_, err = framework.RunKubectl("delete", "-f", vaultExamplePath+vaultRBACPath)
+	if err != nil {
+		e2elog.Logf("failed to delete vault statefull set %v", err)
+	}
+	_, err = framework.RunKubectl("delete", "-f", vaultExamplePath+vaultConfigPath)
+	if err != nil {
+		e2elog.Logf("failed to delete vault config map %v", err)
+	}
+	_, err = framework.RunKubectl("delete", "-f", vaultExamplePath+vaultPSPPath)
+	if err != nil {
+		e2elog.Logf("failed to delete vault psp %v", err)
+	}
+}
--- a/e2e/rbd.go
+++ b/e2e/rbd.go
@ -76,7 +76,7 @@ var _ = Describe("RBD", func() {
 		deployRBDPlugin()
 		createRBDStorageClass(f.ClientSet, f, make(map[string]string))
 		createRBDSecret(f.ClientSet, f)
-
+		deployVault(f.ClientSet, deployTimeout)
 	})

 	AfterEach(func() {
@ -91,6 +91,7 @@ var _ = Describe("RBD", func() {
 		deleteResource(rbdExamplePath + "secret.yaml")
 		deleteResource(rbdExamplePath + "storageclass.yaml")
 		// deleteResource(rbdExamplePath + "snapshotclass.yaml")
+		deleteVault()
 	})

 	Context("Test RBD CSI", func() {
@ -135,7 +136,20 @@ var _ = Describe("RBD", func() {
 			By("create a PVC and Bind it to an app with encrypted RBD volume", func() {
 				deleteResource(rbdExamplePath + "storageclass.yaml")
 				createRBDStorageClass(f.ClientSet, f, map[string]string{"encrypted": "true"})
-				validateEncryptedPVCAndAppBinding(pvcPath, appPath, f)
+				validateEncryptedPVCAndAppBinding(pvcPath, appPath, "", f)
+				deleteResource(rbdExamplePath + "storageclass.yaml")
+				createRBDStorageClass(f.ClientSet, f, make(map[string]string))
+			})
+
+			By("create a PVC and Bind it to an app with encrypted RBD volume with Vault KMS", func() {
+				deleteResource(rbdExamplePath + "storageclass.yaml")
+				scOpts := map[string]string{
+					"encrypted":       "true",
+					"encryptionKMS":   "vault",
+					"encryptionKMSID": "vault-test",
+				}
+				createRBDStorageClass(f.ClientSet, f, scOpts)
+				validateEncryptedPVCAndAppBinding(pvcPath, appPath, "vault", f)
 				deleteResource(rbdExamplePath + "storageclass.yaml")
 				createRBDStorageClass(f.ClientSet, f, make(map[string]string))
 			})
--- a/e2e/utils.go
+++ b/e2e/utils.go
@ -32,6 +32,8 @@ import (

 const (
 	rookNS        = "rook-ceph"
+	vaultAddr     = "http://vault.default.svc.cluster.local:8200"
+	vaultSecretNs = "/secret/ceph-csi/" // nolint: gosec, #nosec
 )

 var poll = 2 * time.Second
@ -108,14 +110,14 @@ func waitForDeploymentComplete(name, ns string, c clientset.Interface, t int) er
 	return nil
 }

-func execCommandInPod(f *framework.Framework, c, ns string, opt *metav1.ListOptions) (string, string) {
+func getCommandInPodOpts(f *framework.Framework, c, ns string, opt *metav1.ListOptions) framework.ExecOptions {
 	cmd := []string{"/bin/sh", "-c", c}
 	podList, err := f.PodClientNS(ns).List(*opt)
 	framework.ExpectNoError(err)
 	Expect(podList.Items).NotTo(BeNil())
 	Expect(err).Should(BeNil())

-	podPot := framework.ExecOptions{
+	return framework.ExecOptions{
 		Command:            cmd,
 		PodName:            podList.Items[0].Name,
 		Namespace:          ns,
@ -125,6 +127,10 @@ func execCommandInPod(f *framework.Framework, c, ns string, opt *metav1.ListOpti
 		CaptureStderr:      true,
 		PreserveWhitespace: true,
 	}
+}
+
+func execCommandInPod(f *framework.Framework, c, ns string, opt *metav1.ListOptions) (string, string) {
+	podPot := getCommandInPodOpts(f, c, ns, opt)
 	stdOut, stdErr, err := f.ExecWithOptions(podPot)
 	if stdErr != "" {
 		e2elog.Logf("stdErr occurred: %v", stdErr)
@ -133,6 +139,15 @@ func execCommandInPod(f *framework.Framework, c, ns string, opt *metav1.ListOpti
 	return stdOut, stdErr
 }

+func execCommandInPodAndAllowFail(f *framework.Framework, c, ns string, opt *metav1.ListOptions) (string, string) {
+	podPot := getCommandInPodOpts(f, c, ns, opt)
+	stdOut, stdErr, err := f.ExecWithOptions(podPot)
+	if err != nil {
+		e2elog.Logf("command %s failed: %v", c, err)
+	}
+	return stdOut, stdErr
+}
+
 func getMons(ns string, c kubernetes.Interface) []string {
 	opt := metav1.ListOptions{
 		LabelSelector: "app=rook-ceph-mon",
@ -557,39 +572,22 @@ func validatePVCAndAppBinding(pvcPath, appPath string, f *framework.Framework) {
 	}
 }

-func getImageIDFromPVC(pvcNamespace, pvcName string, f *framework.Framework) (string, error) {
+func getRBDImageIds(pvcNamespace, pvcName string, f *framework.Framework) (string, string, error) {
 	c := f.ClientSet.CoreV1()
 	pvc, err := c.PersistentVolumeClaims(pvcNamespace).Get(pvcName, metav1.GetOptions{})
 	if err != nil {
-		return "", err
+		return "", "", err
 	}

 	pv, err := c.PersistentVolumes().Get(pvc.Spec.VolumeName, metav1.GetOptions{})
 	if err != nil {
-		return "", err
+		return "", "", err
 	}

 	imageIDRegex := regexp.MustCompile(`(\w+\-?){5}$`)
 	imageID := imageIDRegex.FindString(pv.Spec.CSI.VolumeHandle)

-	return imageID, nil
-}
-func getRBDImageSpec(pvcNamespace, pvcName string, f *framework.Framework) (string, error) {
-	imageID, err := getImageIDFromPVC(pvcNamespace, pvcName, f)
-	if err != nil {
-		return "", err
-	}
-
-	return fmt.Sprintf("replicapool/csi-vol-%s", imageID), nil
-}
-
-func getCephFSVolumeName(pvcNamespace, pvcName string, f *framework.Framework) (string, error) {
-	imageID, err := getImageIDFromPVC(pvcNamespace, pvcName, f)
-	if err != nil {
-		return "", err
-	}
-
-	return fmt.Sprintf("csi-vol-%s", imageID), nil
+	return fmt.Sprintf("csi-vol-%s", imageID), pv.Spec.CSI.VolumeHandle, nil
 }

 func getImageMeta(rbdImageSpec, metaKey string, f *framework.Framework) (string, error) {
@ -616,13 +614,31 @@ func getMountType(appName, appNamespace, mountPath string, f *framework.Framewor
 	return strings.TrimSpace(stdOut), nil
 }

-func validateEncryptedPVCAndAppBinding(pvcPath, appPath string, f *framework.Framework) {
+// readVaultSecret method will execute few commands to try read the secret for
+// specified key from inside the vault container:
+//  * authenticate with vault and ignore any stdout (we do not need output)
+//  * issue get request for particular key
+// resulting in stdOut (first entry in tuple) - output that contains the key
+// or stdErr (second entry in tuple) - error getting the key
+func readVaultSecret(key string, f *framework.Framework) (string, string) {
+	loginCmd := fmt.Sprintf("vault login -address=%s sample_root_token_id > /dev/null", vaultAddr)
+	readSecret := fmt.Sprintf("vault kv get -address=%s %s%s", vaultAddr, vaultSecretNs, key)
+	cmd := fmt.Sprintf("%s && %s", loginCmd, readSecret)
+	opt := metav1.ListOptions{
+		LabelSelector: "app=vault",
+	}
+	stdOut, stdErr := execCommandInPodAndAllowFail(f, cmd, "default", &opt)
+	return strings.TrimSpace(stdOut), strings.TrimSpace(stdErr)
+}
+
+func validateEncryptedPVCAndAppBinding(pvcPath, appPath, kms string, f *framework.Framework) {
 	pvc, app := createPVCAndAppBinding(pvcPath, appPath, f)

-	rbdImageSpec, err := getRBDImageSpec(pvc.Namespace, pvc.Name, f)
+	rbdImageID, rbdImageHandle, err := getRBDImageIds(pvc.Namespace, pvc.Name, f)
 	if err != nil {
 		Fail(err.Error())
 	}
+	rbdImageSpec := fmt.Sprintf("replicapool/%s", rbdImageID)
 	encryptedState, err := getImageMeta(rbdImageSpec, ".rbd.csi.ceph.com/encrypted", f)
 	if err != nil {
 		Fail(err.Error())
@ -636,10 +652,26 @@ func validateEncryptedPVCAndAppBinding(pvcPath, appPath string, f *framework.Fra
 	}
 	Expect(mountType).To(Equal("crypt"))

+	if kms == "vault" {
+		// check new passphrase created
+		_, stdErr := readVaultSecret(rbdImageHandle, f)
+		if stdErr != "" {
+			Fail(fmt.Sprintf("failed to read passphrase from vault: %s", stdErr))
+		}
+	}
+
 	err = deletePVCAndApp("", f, pvc, app)
 	if err != nil {
 		Fail(err.Error())
 	}
+
+	if kms == "vault" {
+		// check new passphrase created
+		stdOut, _ := readVaultSecret(rbdImageHandle, f)
+		if stdOut != "" {
+			Fail(fmt.Sprintf("passphrase found in vault while should be deleted: %s", stdOut))
+		}
+	}
 }

 func deletePodWithLabel(label string) error {
@ -794,7 +826,7 @@ func validateNormalUserPVCAccess(pvcPath string, f *framework.Framework) {
 // }

 func deleteBackingCephFSVolume(f *framework.Framework, pvc *v1.PersistentVolumeClaim) error {
-	volname, err := getCephFSVolumeName(pvc.Namespace, pvc.Name, f)
+	volname, _, err := getRBDImageIds(pvc.Namespace, pvc.Name, f)
 	if err != nil {
 		return err
 	}
--- a/examples/kms/vault/csi-vaulttokenreview-rbac.yaml
+++ b/examples/kms/vault/csi-vaulttokenreview-rbac.yaml
@ -0,0 +1,28 @@
+---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: rbd-csi-vault-token-review
+---
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: rbd-csi-vault-token-review
+rules:
+  - apiGroups: ["authentication.k8s.io"]
+    resources: ["tokenreviews"]
+    verbs: ["create", "get", "list"]
+
+---
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: rbd-csi-vault-token-review
+subjects:
+  - kind: ServiceAccount
+    name: rbd-csi-vault-token-review
+    namespace: default
+roleRef:
+  kind: ClusterRole
+  name: rbd-csi-vault-token-review
+  apiGroup: rbac.authorization.k8s.io
--- a/examples/kms/vault/kms-config.yaml
+++ b/examples/kms/vault/kms-config.yaml
@ -0,0 +1,18 @@
+---
+apiVersion: v1
+kind: ConfigMap
+data:
+  config.json: |-
+    [
+      {
+        "encryptionKMSID": "vault-test",
+        "vaultAddress": "http://vault.default.svc.cluster.local:8200",
+        "vaultAuthPath": "/v1/auth/kubernetes/login",
+        "vaultRole": "csi-kubernetes",
+        "vaultPassphraseRoot": "/v1/secret",
+        "vaultPassphrasePath": "ceph-csi/",
+        "vaultCAVerify": false
+      }
+    ]
+metadata:
+  name: ceph-csi-encryption-kms-config
--- a/examples/kms/vault/vault-psp.yaml
+++ b/examples/kms/vault/vault-psp.yaml
@ -0,0 +1,47 @@
+---
+apiVersion: policy/v1beta1
+kind: PodSecurityPolicy
+metadata:
+  name: rbd-csi-vault-token-review-psp
+spec:
+  fsGroup:
+    rule: RunAsAny
+  runAsUser:
+    rule: RunAsAny
+  seLinux:
+    rule: RunAsAny
+  supplementalGroups:
+    rule: RunAsAny
+  volumes:
+    - 'configMap'
+    - 'secret'
+
+---
+kind: Role
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  # replace with non-default namespace name
+  namespace: default
+  name: rbd-csi-vault-token-review-psp
+rules:
+  - apiGroups: ['policy']
+    resources: ['podsecuritypolicies']
+    verbs: ['use']
+    resourceNames: ['rbd-csi-vault-token-review-psp']
+
+---
+kind: RoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: rbd-csi-vault-token-review-psp
+  # replace with non-default namespace name
+  namespace: default
+subjects:
+  - kind: ServiceAccount
+    name: rbd-csi-vault-token-review
+    # replace with non-default namespace name
+    namespace: default
+roleRef:
+  kind: Role
+  name: rbd-csi-vault-token-review-psp
+  apiGroup: rbac.authorization.k8s.io
--- a/examples/kms/vault/vault.yaml
+++ b/examples/kms/vault/vault.yaml
@ -0,0 +1,150 @@
+# HashiCorp Vault configuration for minikube
+# This is not part of ceph-csi project, used only
+# for e2e testing of integration with such KMS
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: vault
+  labels:
+    app: vault-api
+spec:
+  ports:
+    - name: vault-api
+      port: 8200
+  clusterIP: None
+  selector:
+    app: vault
+    role: server
+
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: vault
+  labels:
+    app: vault
+    role: server
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: vault
+      role: server
+  template:
+    metadata:
+      labels:
+        app: vault
+        role: server
+    spec:
+      containers:
+        - name: vault
+          image: vault
+          securityContext:
+            runAsUser: 100
+          env:
+            - name: VAULT_DEV_ROOT_TOKEN_ID
+              value: sample_root_token_id
+            - name: SKIP_SETCAP
+              value: any
+          livenessProbe:
+            exec:
+              command:
+                - pidof
+                - vault
+            initialDelaySeconds: 5
+            timeoutSeconds: 2
+          ports:
+            - containerPort: 8200
+              name: vault-api
+---
+apiVersion: v1
+items:
+  - apiVersion: v1
+    data:
+      init-vault.sh: |
+        set -x -e
+
+        timeout 300 sh -c 'until vault status; do sleep 5; done'
+
+        # login into vault to retrieve token
+        vault login ${VAULT_DEV_ROOT_TOKEN_ID}
+
+        # enable kubernetes auth method under specific path:
+        vault auth enable -path="/${CLUSTER_IDENTIFIER}" kubernetes
+
+        # write configuration to use your cluster
+        vault write auth/${CLUSTER_IDENTIFIER}/config \
+          token_reviewer_jwt=@${SERVICE_ACCOUNT_TOKEN_PATH}/token \
+          kubernetes_host="${K8S_HOST}" \
+          kubernetes_ca_cert=@${SERVICE_ACCOUNT_TOKEN_PATH}/ca.crt
+
+        # create policy to use keys related to the cluster
+        vault policy write "${CLUSTER_IDENTIFIER}" - << EOS
+        path "secret/data/ceph-csi/*" {
+          capabilities = ["create", "update", "delete", "read"]
+        }
+
+        path "secret/metadata/ceph-csi/*" {
+          capabilities = ["read", "delete"]
+        }
+        EOS
+
+        # create a role
+        vault write "auth/${CLUSTER_IDENTIFIER}/role/${PLUGIN_ROLE}" \
+            bound_service_account_names="${SERVICE_ACCOUNTS}" \
+            bound_service_account_namespaces="${SERVICE_ACCOUNTS_NAMESPACE}" \
+            policies="${CLUSTER_IDENTIFIER}"
+    kind: ConfigMap
+    metadata:
+      creationTimestamp: null
+      name: init-scripts
+kind: List
+metadata: {}
+
+---
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: vault-init-job
+spec:
+  parallelism: 1
+  completions: 1
+  template:
+    metadata:
+      name: vault-init-job
+    spec:
+      serviceAccount: rbd-csi-vault-token-review
+      volumes:
+        - name: init-scripts-volume
+          configMap:
+            name: init-scripts
+      containers:
+        - name: vault-init-job
+          image: vault
+          volumeMounts:
+            - mountPath: /init-scripts
+              name: init-scripts-volume
+          env:
+            - name: HOME
+              value: /tmp
+            - name: CLUSTER_IDENTIFIER
+              value: kubernetes
+            - name: SERVICE_ACCOUNT_TOKEN_PATH
+              value: /var/run/secrets/kubernetes.io/serviceaccount
+            - name: K8S_HOST
+              value: https://kubernetes.default.svc.cluster.local
+            - name: PLUGIN_ROLE
+              value: csi-kubernetes
+            - name: SERVICE_ACCOUNTS
+              value: rbd-csi-nodeplugin,rbd-csi-provisioner
+            - name: SERVICE_ACCOUNTS_NAMESPACE
+              value: default
+            - name: VAULT_ADDR
+              value: http://vault.default.svc.cluster.local:8200/
+            - name: VAULT_DEV_ROOT_TOKEN_ID
+              value: sample_root_token_id
+          command:
+            - /bin/sh
+            - /init-scripts/init-vault.sh
+      restartPolicy: Never
--- a/examples/rbd/storageclass.yaml
+++ b/examples/rbd/storageclass.yaml
@ -43,6 +43,13 @@ parameters:
   # By default it is disabled. Valid values are “true” or “false”.
   # A string is expected here, i.e. “true”, not true.
   # encrypted: "true"
+
+   # Use external key management system for encryption passphrases
+   # encryptionKMS: vault
+
+   # String representing KMS configuration. Should be unique and match ID in
+   # KMS ConfigMap. The ID is only used for correlation to config map entry.
+   # encryptionKMSID: <kms-config-id>
 reclaimPolicy: Delete
 allowVolumeExpansion: true
 mountOptions:
--- a/pkg/cephfs/fsjournal.go
+++ b/pkg/cephfs/fsjournal.go
@ -58,7 +58,7 @@ func checkVolExists(ctx context.Context, volOptions *volumeOptions, secret map[s
 	defer cr.DeleteCredentials()

 	imageUUID, err := volJournal.CheckReservation(ctx, volOptions.Monitors, cr,
-		volOptions.MetadataPool, volOptions.RequestName, "")
+		volOptions.MetadataPool, volOptions.RequestName, "", "")
 	if err != nil {
 		return nil, err
 	}
@ -116,7 +116,7 @@ func reserveVol(ctx context.Context, volOptions *volumeOptions, secret map[strin
 	defer cr.DeleteCredentials()

 	imageUUID, err := volJournal.ReserveName(ctx, volOptions.Monitors, cr,
-		volOptions.MetadataPool, volOptions.RequestName, "")
+		volOptions.MetadataPool, volOptions.RequestName, "", "")
 	if err != nil {
 		return nil, err
 	}
--- a/pkg/cephfs/volumeoptions.go
+++ b/pkg/cephfs/volumeoptions.go
@ -221,7 +221,7 @@ func newVolumeOptionsFromVolID(ctx context.Context, volID string, volOpt, secret
 		return nil, nil, err
 	}

-	volOptions.RequestName, _, err = volJournal.GetObjectUUIDData(ctx, volOptions.Monitors, cr,
+	volOptions.RequestName, _, _, err = volJournal.GetObjectUUIDData(ctx, volOptions.Monitors, cr,
 		volOptions.MetadataPool, vi.ObjectUUID, false)
 	if err != nil {
 		return nil, nil, err
--- a/pkg/rbd/controllerserver.go
+++ b/pkg/rbd/controllerserver.go
@ -95,7 +95,7 @@ func (cs *ControllerServer) parseVolCreateRequest(ctx context.Context, req *csi.
 	}

 	// if it's NOT SINGLE_NODE_WRITER and it's BLOCK we'll set the parameter to ignore the in-use checks
-	rbdVol, err := genVolFromVolumeOptions(ctx, req.GetParameters(), nil, (isMultiNode && isBlock), false)
+	rbdVol, err := genVolFromVolumeOptions(ctx, req.GetParameters(), req.GetSecrets(), (isMultiNode && isBlock), false)
 	if err != nil {
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
@ -343,7 +343,7 @@ func (cs *ControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
 	defer cs.VolumeLocks.Release(volumeID)

 	rbdVol := &rbdVolume{}
-	if err := genVolFromVolID(ctx, rbdVol, volumeID, cr); err != nil {
+	if err = genVolFromVolID(ctx, rbdVol, volumeID, cr, req.GetSecrets()); err != nil {
 		// If error is ErrInvalidVolID it could be a version 1.0.0 or lower volume, attempt
 		// to process it as such
 		if _, ok := err.(ErrInvalidVolID); ok {
@ -377,7 +377,7 @@ func (cs *ControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
 		}
 		defer cs.VolumeLocks.Release(rbdVol.RequestName)

-		if err := undoVolReservation(ctx, rbdVol, cr); err != nil {
+		if err = undoVolReservation(ctx, rbdVol, cr); err != nil {
 			return nil, status.Error(codes.Internal, err.Error())
 		}
 		return &csi.DeleteVolumeResponse{}, nil
@ -393,18 +393,24 @@ func (cs *ControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol

 	// Deleting rbd image
 	klog.V(4).Infof(util.Log(ctx, "deleting image %s"), rbdVol.RbdImageName)
-	if err := deleteImage(ctx, rbdVol, cr); err != nil {
+	if err = deleteImage(ctx, rbdVol, cr); err != nil {
 		klog.Errorf(util.Log(ctx, "failed to delete rbd image: %s/%s with error: %v"),
 			rbdVol.Pool, rbdVol.RbdImageName, err)
 		return nil, status.Error(codes.Internal, err.Error())
 	}

-	if err := undoVolReservation(ctx, rbdVol, cr); err != nil {
+	if err = undoVolReservation(ctx, rbdVol, cr); err != nil {
 		klog.Errorf(util.Log(ctx, "failed to remove reservation for volume (%s) with backing image (%s) (%s)"),
 			rbdVol.RequestName, rbdVol.RbdImageName, err)
 		return nil, status.Error(codes.Internal, err.Error())
 	}

+	if rbdVol.Encrypted {
+		if err = rbdVol.KMS.DeletePassphrase(rbdVol.VolID); err != nil {
+			klog.V(3).Infof(util.Log(ctx, "failed to clean the passphrase for volume %s: %s"), rbdVol.VolID, err)
+		}
+	}
+
 	return &csi.DeleteVolumeResponse{}, nil
 }

@ -447,7 +453,7 @@ func (cs *ControllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateS

 	// Fetch source volume information
 	rbdVol := new(rbdVolume)
-	err = genVolFromVolID(ctx, rbdVol, req.GetSourceVolumeId(), cr)
+	err = genVolFromVolID(ctx, rbdVol, req.GetSourceVolumeId(), cr, req.GetSecrets())
 	if err != nil {
 		if _, ok := err.(ErrImageNotFound); ok {
 			return nil, status.Errorf(codes.NotFound, "source Volume ID %s not found", req.GetSourceVolumeId())
@ -455,6 +461,12 @@ func (cs *ControllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateS
 		return nil, status.Errorf(codes.Internal, err.Error())
 	}

+	// TODO: re-encrypt snapshot with a new passphrase
+	if rbdVol.Encrypted {
+		return nil, status.Errorf(codes.Unimplemented, "source Volume %s is encrypted, "+
+			"snapshotting is not supported currently", rbdVol.VolID)
+	}
+
 	// Check if source volume was created with required image features for snaps
 	if !hasSnapshotFeature(rbdVol.ImageFeatures) {
 		return nil, status.Errorf(codes.InvalidArgument, "volume(%s) has not snapshot feature(layering)", req.GetSourceVolumeId())
@ -698,7 +710,7 @@ func (cs *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi
 	defer cr.DeleteCredentials()

 	rbdVol := &rbdVolume{}
-	err = genVolFromVolID(ctx, rbdVol, volID, cr)
+	err = genVolFromVolID(ctx, rbdVol, volID, cr, req.GetSecrets())
 	if err != nil {
 		if _, ok := err.(ErrImageNotFound); ok {
 			return nil, status.Errorf(codes.NotFound, "volume ID %s not found", volID)
@ -706,6 +718,11 @@ func (cs *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi
 		return nil, status.Errorf(codes.Internal, err.Error())
 	}

+	if rbdVol.Encrypted {
+		return nil, status.Errorf(codes.InvalidArgument, "encrypted volumes do not support resize (%s/%s)",
+			rbdVol.Pool, rbdVol.RbdImageName)
+	}
+
 	// always round up the request size in bytes to the nearest MiB/GiB
 	volSize := util.RoundOffBytes(req.GetCapacityRange().GetRequiredBytes())

--- a/pkg/rbd/nodeserver.go
+++ b/pkg/rbd/nodeserver.go
@ -116,6 +116,7 @@ func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVol
 	volOptions.VolID = req.GetVolumeId()

 	isMounted := false
+	isEncrypted := false
 	isStagePathCreated := false
 	devicePath := ""

@ -127,7 +128,7 @@ func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVol
 	}
 	defer func() {
 		if err != nil {
-			ns.undoStagingTransaction(ctx, stagingParentPath, devicePath, volID, isStagePathCreated, isMounted)
+			ns.undoStagingTransaction(ctx, stagingParentPath, devicePath, volID, isStagePathCreated, isMounted, isEncrypted)
 		}
 	}()

@ -140,10 +141,11 @@ func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVol
 		req.GetVolumeId(), volOptions.Pool, devicePath)

 	if volOptions.Encrypted {
-		devicePath, err = ns.processEncryptedDevice(ctx, volOptions, devicePath, cr, req.GetSecrets())
+		devicePath, err = ns.processEncryptedDevice(ctx, volOptions, devicePath, cr)
 		if err != nil {
 			return nil, status.Error(codes.Internal, err.Error())
 		}
+		isEncrypted = true
 	}

 	err = ns.createStageMountPoint(ctx, stagingTargetPath, isBlock)
@ -170,7 +172,7 @@ func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVol
 	return &csi.NodeStageVolumeResponse{}, nil
 }

-func (ns *NodeServer) undoStagingTransaction(ctx context.Context, stagingParentPath, devicePath, volID string, isStagePathCreated, isMounted bool) {
+func (ns *NodeServer) undoStagingTransaction(ctx context.Context, stagingParentPath, devicePath, volID string, isStagePathCreated, isMounted, isEncrypted bool) {
 	var err error

 	stagingTargetPath := stagingParentPath + "/" + volID
@ -193,7 +195,7 @@ func (ns *NodeServer) undoStagingTransaction(ctx context.Context, stagingParentP

 	// Unmapping rbd device
 	if devicePath != "" {
-		err = detachRBDDevice(ctx, devicePath, volID)
+		err = detachRBDDevice(ctx, devicePath, volID, isEncrypted)
 		if err != nil {
 			klog.Errorf(util.Log(ctx, "failed to unmap rbd device: %s for volume %s with error: %v"), devicePath, volID, err)
 			// continue on failure to delete the stash file, as kubernetes will fail to delete the staging path otherwise
@ -510,7 +512,7 @@ func (ns *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstag

 	// Unmapping rbd device
 	imageSpec := imgInfo.Pool + "/" + imgInfo.ImageName
-	if err = detachRBDImageOrDeviceSpec(ctx, imageSpec, true, imgInfo.NbdAccess, req.GetVolumeId()); err != nil {
+	if err = detachRBDImageOrDeviceSpec(ctx, imageSpec, true, imgInfo.NbdAccess, imgInfo.Encrypted, req.GetVolumeId()); err != nil {
 		klog.Errorf(util.Log(ctx, "error unmapping volume (%s) from staging path (%s): (%v)"), req.GetVolumeId(), stagingTargetPath, err)
 		return nil, status.Error(codes.Internal, err.Error())
 	}
@ -526,6 +528,7 @@ func (ns *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstag
 	return &csi.NodeUnstageVolumeResponse{}, nil
 }

+// NodeExpandVolume resizes rbd volumes
 func (ns *NodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolumeRequest) (*csi.NodeExpandVolumeResponse, error) {
 	volumeID := req.GetVolumeId()
 	if volumeID == "" {
@ -620,7 +623,7 @@ func (ns *NodeServer) NodeGetCapabilities(ctx context.Context, req *csi.NodeGetC
 	}, nil
 }

-func (ns *NodeServer) processEncryptedDevice(ctx context.Context, volOptions *rbdVolume, devicePath string, cr *util.Credentials, secrets map[string]string) (string, error) {
+func (ns *NodeServer) processEncryptedDevice(ctx context.Context, volOptions *rbdVolume, devicePath string, cr *util.Credentials) (string, error) {
 	imageSpec := volOptions.Pool + "/" + volOptions.RbdImageName
 	encrypted, err := util.CheckRbdImageEncrypted(ctx, cr, volOptions.Monitors, imageSpec)
 	if err != nil {
@ -637,20 +640,31 @@ func (ns *NodeServer) processEncryptedDevice(ctx context.Context, volOptions *rb
 		if err != nil {
 			return "", fmt.Errorf("failed to get disk format for path %s, error: %v", devicePath, err)
 		}
-		if existingFormat != "" {
-			return "", fmt.Errorf("can not encrypt rbdImage %s that already has file system: %s",
-				imageSpec, existingFormat)
-		}
-		err = encryptDevice(ctx, volOptions, secrets, cr, devicePath)
+
+		switch existingFormat {
+		case "":
+			err = encryptDevice(ctx, volOptions, cr, devicePath)
 			if err != nil {
 				return "", fmt.Errorf("failed to encrypt rbd image %s: %v", imageSpec, err)
 			}
+		case "crypt":
+			klog.Warningf(util.Log(ctx, "rbd image %s is encrypted, but encryption state was not updated"),
+				imageSpec)
+			err = util.SaveRbdImageEncryptionStatus(
+				ctx, cr, volOptions.Monitors, imageSpec, rbdImageEncrypted)
+			if err != nil {
+				return "", fmt.Errorf("failed to update encryption state for rbd image %s", imageSpec)
+			}
+		default:
+			return "", fmt.Errorf("can not encrypt rbdImage %s that already has file system: %s",
+				imageSpec, existingFormat)
+		}
 	} else if encrypted != rbdImageEncrypted {
 		return "", fmt.Errorf("rbd image %s found mounted with unexpected encryption status %s",
 			imageSpec, encrypted)
 	}

-	devicePath, err = openEncryptedDevice(ctx, volOptions, devicePath, secrets)
+	devicePath, err = openEncryptedDevice(ctx, volOptions, devicePath)
 	if err != nil {
 		return "", err
 	}
@ -658,8 +672,8 @@ func (ns *NodeServer) processEncryptedDevice(ctx context.Context, volOptions *rb
 	return devicePath, nil
 }

-func encryptDevice(ctx context.Context, rbdVol *rbdVolume, secret map[string]string, cr *util.Credentials, devicePath string) error {
-	passphrase, err := util.GetCryptoPassphrase(secret)
+func encryptDevice(ctx context.Context, rbdVol *rbdVolume, cr *util.Credentials, devicePath string) error {
+	passphrase, err := util.GetCryptoPassphrase(ctx, rbdVol.VolID, rbdVol.KMS)
 	if err != nil {
 		klog.Errorf(util.Log(ctx, "failed to get crypto passphrase for %s/%s: %v"),
 			rbdVol.Pool, rbdVol.RbdImageName, err)
@ -678,8 +692,8 @@ func encryptDevice(ctx context.Context, rbdVol *rbdVolume, secret map[string]str
 	return err
 }

-func openEncryptedDevice(ctx context.Context, volOptions *rbdVolume, devicePath string, secrets map[string]string) (string, error) {
-	passphrase, err := util.GetCryptoPassphrase(secrets)
+func openEncryptedDevice(ctx context.Context, volOptions *rbdVolume, devicePath string) (string, error) {
+	passphrase, err := util.GetCryptoPassphrase(ctx, volOptions.VolID, volOptions.KMS)
 	if err != nil {
 		klog.Errorf(util.Log(ctx, "failed to get passphrase for encrypted device %s/%s: %v"),
 			volOptions.Pool, volOptions.RbdImageName, err)
--- a/pkg/rbd/rbd_attach.go
+++ b/pkg/rbd/rbd_attach.go
@ -231,7 +231,7 @@ func createPath(ctx context.Context, volOpt *rbdVolume, cr *util.Credentials) (s
 		klog.Warningf(util.Log(ctx, "rbd: map error %v, rbd output: %s"), err, string(output))
 		// unmap rbd image if connection timeout
 		if strings.Contains(err.Error(), rbdMapConnectionTimeout) {
-			detErr := detachRBDImageOrDeviceSpec(ctx, imagePath, true, isNbd, volOpt.VolID)
+			detErr := detachRBDImageOrDeviceSpec(ctx, imagePath, true, isNbd, volOpt.Encrypted, volOpt.VolID)
 			if detErr != nil {
 				klog.Warningf(util.Log(ctx, "rbd: %s unmap error %v"), imagePath, detErr)
 			}
@ -266,20 +266,21 @@ func waitForrbdImage(ctx context.Context, backoff wait.Backoff, volOptions *rbdV
 	return err
 }

-func detachRBDDevice(ctx context.Context, devicePath, volumeID string) error {
+func detachRBDDevice(ctx context.Context, devicePath, volumeID string, encrypted bool) error {
 	nbdType := false
 	if strings.HasPrefix(devicePath, "/dev/nbd") {
 		nbdType = true
 	}

-	return detachRBDImageOrDeviceSpec(ctx, devicePath, false, nbdType, volumeID)
+	return detachRBDImageOrDeviceSpec(ctx, devicePath, false, nbdType, encrypted, volumeID)
 }

 // detachRBDImageOrDeviceSpec detaches an rbd imageSpec or devicePath, with additional checking
 // when imageSpec is used to decide if image is already unmapped
-func detachRBDImageOrDeviceSpec(ctx context.Context, imageOrDeviceSpec string, isImageSpec, ndbType bool, volumeID string) error {
+func detachRBDImageOrDeviceSpec(ctx context.Context, imageOrDeviceSpec string, isImageSpec, ndbType, encrypted bool, volumeID string) error {
 	var output []byte

+	if encrypted {
 		mapperFile, mapperPath := util.VolumeMapper(volumeID)
 		mappedDevice, mapper, err := util.DeviceEncryptionStatus(ctx, mapperPath)
 		if err != nil {
@ -291,12 +292,13 @@ func detachRBDImageOrDeviceSpec(ctx context.Context, imageOrDeviceSpec string, i
 			// mapper found, so it is open Luks device
 			err = util.CloseEncryptedVolume(ctx, mapperFile)
 			if err != nil {
-			klog.Warningf(util.Log(ctx, "error closing LUKS device on %s, %s: %s"),
+				klog.Errorf(util.Log(ctx, "error closing LUKS device on %s, %s: %s"),
 					mapperPath, imageOrDeviceSpec, err)
 				return err
 			}
 			imageOrDeviceSpec = mappedDevice
 		}
+	}

 	accessType := accessTypeKRbd
 	if ndbType {
@ -304,7 +306,7 @@ func detachRBDImageOrDeviceSpec(ctx context.Context, imageOrDeviceSpec string, i
 	}
 	options := []string{"unmap", "--device-type", accessType, imageOrDeviceSpec}

-	output, err = execCommand(rbd, options)
+	output, err := execCommand(rbd, options)
 	if err != nil {
 		// Messages for krbd and nbd differ, hence checking either of them for missing mapping
 		// This is not applicable when a device path is passed in
--- a/pkg/rbd/rbd_journal.go
+++ b/pkg/rbd/rbd_journal.go
@ -115,7 +115,7 @@ func checkSnapExists(ctx context.Context, rbdSnap *rbdSnapshot, cr *util.Credent
 	}

 	snapUUID, err := snapJournal.CheckReservation(ctx, rbdSnap.Monitors, cr, rbdSnap.Pool,
-		rbdSnap.RequestName, rbdSnap.RbdImageName)
+		rbdSnap.RequestName, rbdSnap.RbdImageName, "")
 	if err != nil {
 		return false, err
 	}
@ -162,8 +162,12 @@ func checkVolExists(ctx context.Context, rbdVol *rbdVolume, cr *util.Credentials
 		return false, err
 	}

+	encryptionKmsConfig := ""
+	if rbdVol.Encrypted {
+		encryptionKmsConfig = rbdVol.KMS.KmsConfig()
+	}
 	imageUUID, err := volJournal.CheckReservation(ctx, rbdVol.Monitors, cr, rbdVol.Pool,
-		rbdVol.RequestName, "")
+		rbdVol.RequestName, "", encryptionKmsConfig)
 	if err != nil {
 		return false, err
 	}
@ -211,7 +215,7 @@ func checkVolExists(ctx context.Context, rbdVol *rbdVolume, cr *util.Credentials
 // volume ID for the generated name
 func reserveSnap(ctx context.Context, rbdSnap *rbdSnapshot, cr *util.Credentials) error {
 	snapUUID, err := snapJournal.ReserveName(ctx, rbdSnap.Monitors, cr, rbdSnap.Pool,
-		rbdSnap.RequestName, rbdSnap.RbdImageName)
+		rbdSnap.RequestName, rbdSnap.RbdImageName, "")
 	if err != nil {
 		return err
 	}
@ -233,8 +237,12 @@ func reserveSnap(ctx context.Context, rbdSnap *rbdSnapshot, cr *util.Credentials
 // reserveVol is a helper routine to request a rbdVolume name reservation and generate the
 // volume ID for the generated name
 func reserveVol(ctx context.Context, rbdVol *rbdVolume, cr *util.Credentials) error {
+	encryptionKmsConfig := ""
+	if rbdVol.Encrypted {
+		encryptionKmsConfig = rbdVol.KMS.KmsConfig()
+	}
 	imageUUID, err := volJournal.ReserveName(ctx, rbdVol.Monitors, cr, rbdVol.Pool,
-		rbdVol.RequestName, "")
+		rbdVol.RequestName, "", encryptionKmsConfig)
 	if err != nil {
 		return err
 	}
--- a/pkg/rbd/rbd_util.go
+++ b/pkg/rbd/rbd_util.go
@ -86,6 +86,7 @@ type rbdVolume struct {
 	VolSize            int64  `json:"volSize"`
 	DisableInUseChecks bool   `json:"disableInUseChecks"`
 	Encrypted          bool
+	KMS                util.EncryptionKMS
 }

 // rbdSnapshot represents a CSI snapshot and its RBD snapshot specifics
@ -306,7 +307,7 @@ func genSnapFromSnapID(ctx context.Context, rbdSnap *rbdSnapshot, snapshotID str
 		return err
 	}

-	rbdSnap.RequestName, rbdSnap.RbdImageName, err = snapJournal.GetObjectUUIDData(ctx, rbdSnap.Monitors,
+	rbdSnap.RequestName, rbdSnap.RbdImageName, _, err = snapJournal.GetObjectUUIDData(ctx, rbdSnap.Monitors,
 		cr, rbdSnap.Pool, vi.ObjectUUID, true)
 	if err != nil {
 		return err
@ -319,7 +320,7 @@ func genSnapFromSnapID(ctx context.Context, rbdSnap *rbdSnapshot, snapshotID str

 // genVolFromVolID generates a rbdVolume structure from the provided identifier, updating
 // the structure with elements from on-disk image metadata as well
-func genVolFromVolID(ctx context.Context, rbdVol *rbdVolume, volumeID string, cr *util.Credentials) error {
+func genVolFromVolID(ctx context.Context, rbdVol *rbdVolume, volumeID string, cr *util.Credentials, secrets map[string]string) error {
 	var (
 		options map[string]string
 		vi      util.CSIIdentifier
@ -350,11 +351,23 @@ func genVolFromVolID(ctx context.Context, rbdVol *rbdVolume, volumeID string, cr
 		return err
 	}

-	rbdVol.RequestName, _, err = volJournal.GetObjectUUIDData(ctx, rbdVol.Monitors, cr,
-		rbdVol.Pool, vi.ObjectUUID, false)
+	kmsConfig := ""
+	rbdVol.RequestName, _, kmsConfig, err = volJournal.GetObjectUUIDData(
+		ctx, rbdVol.Monitors, cr, rbdVol.Pool, vi.ObjectUUID, false)
 	if err != nil {
 		return err
 	}
+	if kmsConfig != "" {
+		rbdVol.Encrypted = true
+		kmsOpts, kmsConfigParseErr := util.GetKMSConfig(kmsConfig)
+		if kmsConfigParseErr != nil {
+			return kmsConfigParseErr
+		}
+		rbdVol.KMS, err = util.GetKMS(kmsOpts, secrets)
+		if err != nil {
+			return err
+		}
+	}

 	err = updateVolWithImageInfo(ctx, rbdVol, cr)

@ -447,6 +460,7 @@ func genVolFromVolumeOptions(ctx context.Context, volOptions, credentials map[st
 	var (
 		ok        bool
 		err       error
+		encrypted string
 	)

 	rbdVol := &rbdVolume{}
@ -493,13 +507,20 @@ func genVolFromVolumeOptions(ctx context.Context, volOptions, credentials map[st
 	}

 	rbdVol.Encrypted = false
-	encrypted, ok := volOptions["encrypted"]
+	encrypted, ok = volOptions["encrypted"]
 	if ok {
 		rbdVol.Encrypted, err = strconv.ParseBool(encrypted)
 		if err != nil {
 			return nil, fmt.Errorf(
 				"invalid value set in 'encrypted': %s (should be \"true\" or \"false\")", encrypted)
 		}
+
+		if rbdVol.Encrypted {
+			rbdVol.KMS, err = util.GetKMS(volOptions, credentials)
+			if err != nil {
+				return nil, fmt.Errorf("invalid encryption kms configuration: %s", err)
+			}
+		}
 	}

 	return rbdVol, nil
@ -763,6 +784,7 @@ type rbdImageMetadataStash struct {
 	Pool      string `json:"pool"`
 	ImageName string `json:"image"`
 	NbdAccess bool   `json:"accessType"`
+	Encrypted bool   `json:"encrypted"`
 }

 // file name in which image metadata is stashed
@ -772,9 +794,10 @@ const stashFileName = "image-meta.json"
 // JSON format
 func stashRBDImageMetadata(volOptions *rbdVolume, path string) error {
 	var imgMeta = rbdImageMetadataStash{
-		Version:   1, // Stash a v1 for now, in case of changes later, there are no checks for this at present
+		Version:   2, // there are no checks for this at present
 		Pool:      volOptions.Pool,
 		ImageName: volOptions.RbdImageName,
+		Encrypted: volOptions.Encrypted,
 	}

 	imgMeta.NbdAccess = false
--- a/pkg/util/crypto.go
+++ b/pkg/util/crypto.go
@ -18,12 +18,15 @@ package util

 import (
 	"context"
+	"encoding/base64"
 	"fmt"
 	"path"
 	"strings"

 	"github.com/pkg/errors"

+	"crypto/rand"
+
 	"k8s.io/klog"
 )

@ -36,8 +39,123 @@ const (

 	// Encryption passphrase location in K8s secrets
 	encryptionPassphraseKey = "encryptionPassphrase"
+
+	// kmsConfigPath is the location of the vault config file
+	kmsConfigPath = "/etc/ceph-csi-encryption-kms-config/config.json"
+
+	// Passphrase size - 20 bytes is 160 bits to satisfy:
+	// https://tools.ietf.org/html/rfc6749#section-10.10
+	encryptionPassphraseSize = 20
 )

+// EncryptionKMS provides external Key Management System for encryption
+// passphrases storage
+type EncryptionKMS interface {
+	GetPassphrase(key string) (string, error)
+	SavePassphrase(key, value string) error
+	DeletePassphrase(key string) error
+	KmsConfig() string
+}
+
+// MissingPassphrase is an error instructing to generate new passphrase
+type MissingPassphrase struct {
+	error
+}
+
+// SecretsKMS is default KMS implementation that means no KMS is in use
+type SecretsKMS struct {
+	passphrase string
+}
+
+func initSecretsKMS(secrets map[string]string) (EncryptionKMS, error) {
+	passphraseValue, ok := secrets[encryptionPassphraseKey]
+	if !ok {
+		return nil, errors.New("missing encryption passphrase in secrets")
+	}
+	return SecretsKMS{passphrase: passphraseValue}, nil
+}
+
+// KmsConfig returns KMS configuration: "<kms-type>|<kms-id>"
+func (kms SecretsKMS) KmsConfig() string {
+	return "secrets|kubernetes"
+}
+
+// GetPassphrase returns passphrase from Kubernetes secrets
+func (kms SecretsKMS) GetPassphrase(key string) (string, error) {
+	return kms.passphrase, nil
+}
+
+// SavePassphrase is not implemented
+func (kms SecretsKMS) SavePassphrase(key, value string) error {
+	return fmt.Errorf("save new passphrase is not implemented for Kubernetes secrets")
+}
+
+// DeletePassphrase is doing nothing as no new passphrases are saved with
+// SecretsKMS
+func (kms SecretsKMS) DeletePassphrase(key string) error {
+	return nil
+}
+
+// GetKMS returns an instance of Key Management System
+func GetKMS(opts, secrets map[string]string) (EncryptionKMS, error) {
+	kmsType, ok := opts["encryptionKMS"]
+	if !ok || kmsType == "" || kmsType == "secrets" {
+		return initSecretsKMS(secrets)
+	}
+	if kmsType == "vault" {
+		return InitVaultKMS(opts, secrets)
+	}
+	return nil, fmt.Errorf("unknown encryption KMS type %s", kmsType)
+}
+
+// GetKMSConfig returns required keys for KMS to instantiate from it's config
+// - map with kms type and ID keys
+// - error if format is invalid
+func GetKMSConfig(config string) (map[string]string, error) {
+	kmsConfigParts := strings.Split(config, "|")
+	if len(kmsConfigParts) != 2 {
+		return make(map[string]string), fmt.Errorf("failed to parse encryption KMS "+
+			"configuration from config string, expected <type>|<id>, got: %s", config)
+	}
+	return map[string]string{
+		"encryptionKMS":   kmsConfigParts[0],
+		"encryptionKMSID": kmsConfigParts[1],
+	}, nil
+}
+
+// GetCryptoPassphrase Retrieves passphrase to encrypt volume
+func GetCryptoPassphrase(ctx context.Context, volumeID string, kms EncryptionKMS) (string, error) {
+	passphrase, err := kms.GetPassphrase(volumeID)
+	if err == nil {
+		return passphrase, nil
+	}
+	if _, ok := err.(MissingPassphrase); ok {
+		klog.V(4).Infof(Log(ctx, "Encryption passphrase is missing for %s. Generating a new one"),
+			volumeID)
+		passphrase, err = generateNewEncryptionPassphrase()
+		if err != nil {
+			return "", fmt.Errorf("failed to generate passphrase for %s: %s", volumeID, err)
+		}
+		err = kms.SavePassphrase(volumeID, passphrase)
+		if err != nil {
+			return "", fmt.Errorf("failed to save the passphrase for %s: %s", volumeID, err)
+		}
+		return passphrase, nil
+	}
+	klog.Errorf(Log(ctx, "failed to get encryption passphrase for %s: %s"), volumeID, err)
+	return "", err
+}
+
+// generateNewEncryptionPassphrase generates a random passphrase for encryption
+func generateNewEncryptionPassphrase() (string, error) {
+	bytesPassphrase := make([]byte, encryptionPassphraseSize)
+	_, err := rand.Read(bytesPassphrase)
+	if err != nil {
+		return "", err
+	}
+	return base64.URLEncoding.EncodeToString(bytesPassphrase), nil
+}
+
 // VolumeMapper returns file name and it's path to where encrypted device should be open
 func VolumeMapper(volumeID string) (mapperFile, mapperFilePath string) {
 	mapperFile = mapperFilePrefix + volumeID
@ -45,15 +163,6 @@ func VolumeMapper(volumeID string) (mapperFile, mapperFilePath string) {
 	return mapperFile, mapperFilePath
 }

-// GetCryptoPassphrase Retrieves passphrase to encrypt volume
-func GetCryptoPassphrase(secrets map[string]string) (string, error) {
-	val, ok := secrets[encryptionPassphraseKey]
-	if !ok {
-		return "", errors.New("missing encryption passphrase in secrets")
-	}
-	return val, nil
-}
-
 // EncryptVolume encrypts provided device with LUKS
 func EncryptVolume(ctx context.Context, devicePath, passphrase string) error {
 	klog.V(4).Infof(Log(ctx, "Encrypting device %s with LUKS"), devicePath)
--- a/pkg/util/vault.go
+++ b/pkg/util/vault.go
@ -0,0 +1,341 @@
+/*
+Copyright 2019 The Ceph-CSI Authors.
+
+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 util
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/json"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"strings"
+)
+
+const (
+	// path to service account token that will be used to authenticate with Vault
+	// #nosec
+	serviceAccountTokenPath = "/var/run/secrets/kubernetes.io/serviceaccount/token"
+
+	// vault configuration defaults
+	vaultDefaultAuthPath       = "/v1/auth/kubernetes/login"
+	vaultDefaultRole           = "csi-kubernetes"
+	vaultDefaultNamespace      = ""
+	vaultDefaultPassphraseRoot = "/v1/secret"
+	vaultDefaultPassphrasePath = ""
+
+	// vault request headers
+	vaultTokenHeader     = "X-Vault-Token" // nolint: gosec, #nosec
+	vaultNamespaceHeader = "X-Vault-Namespace"
+)
+
+/*
+kmsKMS represents a Hashicorp Vault KMS configuration
+
+Example JSON structure in the KMS config is,
+[
+	{
+		"encryptionKMSID": "local_vault_unique_identifier",
+		"vaultAddress": "https://127.0.0.1:8500",
+		"vaultAuthPath": "/v1/auth/kubernetes/login",
+		"vaultRole": "csi-kubernetes",
+		"vaultNamespace": "",
+		"vaultPassphraseRoot": "/v1/secret",
+		"vaultPassphrasePath": "",
+		"vaultCAVerify": true,
+		"vaultCAFromSecret": "vault-ca"
+	},
+	...
+]
+*/
+type VaultKMS struct {
+	EncryptionKMSID     string `json:"encryptionKMSID"`
+	VaultAddress        string `json:"vaultAddress"`
+	VaultAuthPath       string `json:"vaultAuthPath"`
+	VaultRole           string `json:"vaultRole"`
+	VaultNamespace      string `json:"vaultNamespace"`
+	VaultPassphraseRoot string `json:"vaultPassphraseRoot"`
+	VaultPassphrasePath string `json:"vaultPassphrasePath"`
+	VaultCAVerify       bool   `json:"vaultCAVerify"`
+	VaultCAFromSecret   string `json:"vaultCAFromSecret"`
+	vaultCA             *x509.CertPool
+}
+
+// InitVaultKMS returns an interface to HashiCorp Vault KMS
+func InitVaultKMS(opts, secrets map[string]string) (EncryptionKMS, error) {
+	var config []VaultKMS
+
+	vaultID, ok := opts["encryptionKMSID"]
+	if !ok {
+		return nil, fmt.Errorf("missing encryptionKMSID for vault as encryption KMS")
+	}
+
+	// #nosec
+	content, err := ioutil.ReadFile(kmsConfigPath)
+	if err != nil {
+		return nil, fmt.Errorf("error fetching vault configuration for vault ID (%s): (%s)",
+			vaultID, err)
+	}
+
+	err = json.Unmarshal(content, &config)
+	if err != nil {
+		return nil, fmt.Errorf("unmarshal failed: %v. raw buffer response: %s",
+			err, string(content))
+	}
+
+	for i := range config {
+		vault := &config[i]
+		if vault.EncryptionKMSID != vaultID {
+			continue
+		}
+		if vault.VaultAddress == "" {
+			return nil, fmt.Errorf("missing vaultAddress for vault as encryption KMS")
+		}
+		if vault.VaultAuthPath == "" {
+			vault.VaultAuthPath = vaultDefaultAuthPath
+		}
+		if vault.VaultRole == "" {
+			vault.VaultRole = vaultDefaultRole
+		}
+		if vault.VaultNamespace == "" {
+			vault.VaultNamespace = vaultDefaultNamespace
+		}
+		if vault.VaultPassphraseRoot == "" {
+			vault.VaultPassphraseRoot = vaultDefaultPassphraseRoot
+		}
+		if vault.VaultPassphrasePath == "" {
+			vault.VaultPassphrasePath = vaultDefaultPassphrasePath
+		}
+		if vault.VaultCAFromSecret != "" {
+			caPEM, ok := secrets[vault.VaultCAFromSecret]
+			if !ok {
+				return nil, fmt.Errorf("missing vault CA in secret %s", vault.VaultCAFromSecret)
+			}
+			roots := x509.NewCertPool()
+			ok = roots.AppendCertsFromPEM([]byte(caPEM))
+			if !ok {
+				return nil, fmt.Errorf("failed loading CA bundle for vault from secret %s",
+					vault.VaultCAFromSecret)
+			}
+			vault.vaultCA = roots
+		}
+		return vault, nil
+	}
+
+	return nil, fmt.Errorf("missing configuration for vault ID (%s)", vaultID)
+}
+
+// KmsConfig returns KMS configuration: "<kms-type>|<kms-id>"
+func (kms *VaultKMS) KmsConfig() string {
+	return fmt.Sprintf("vault|%s", kms.EncryptionKMSID)
+}
+
+// GetPassphrase returns passphrase from Vault
+func (kms *VaultKMS) GetPassphrase(key string) (string, error) {
+	var passphrase string
+	resp, err := kms.request("GET", kms.getKeyDataURI(key), nil)
+	if err != nil {
+		return "", fmt.Errorf("failed to retrieve passphrase for %s from vault: %s",
+			key, err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode == 404 {
+		return "", MissingPassphrase{fmt.Errorf("passphrase for %s not found", key)}
+	}
+	err = kms.processError(resp, fmt.Sprintf("get passphrase for %s", key))
+	if err != nil {
+		return "", err
+	}
+
+	// parse resp as JSON and retrieve vault token
+	var result map[string]interface{}
+	err = json.NewDecoder(resp.Body).Decode(&result)
+	if err != nil {
+		return "", fmt.Errorf("failed parsing passphrase for %s from response: %s",
+			key, err)
+	}
+	data, ok := result["data"].(map[string]interface{})
+	if !ok {
+		return "", fmt.Errorf("failed parsing data for get passphrase request for %s", key)
+	}
+	data, ok = data["data"].(map[string]interface{})
+	if !ok {
+		return "", fmt.Errorf("failed parsing data.data for get passphrase request for %s", key)
+	}
+	passphrase, ok = data["passphrase"].(string)
+	if !ok {
+		return "", fmt.Errorf("failed parsing passphrase for get passphrase request for %s", key)
+	}
+
+	return passphrase, nil
+}
+
+// SavePassphrase saves new passphrase in Vault
+func (kms *VaultKMS) SavePassphrase(key, value string) error {
+	data, err := json.Marshal(map[string]map[string]string{
+		"data": {
+			"passphrase": value,
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("passphrase request data is broken: %s", err)
+	}
+
+	resp, err := kms.request("POST", kms.getKeyDataURI(key), data)
+	if err != nil {
+		return fmt.Errorf("failed to POST passphrase for %s to vault: %s", key, err)
+	}
+	defer resp.Body.Close()
+	err = kms.processError(resp, "save passphrase")
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// DeletePassphrase deletes passphrase from Vault
+func (kms *VaultKMS) DeletePassphrase(key string) error {
+	vaultToken, err := kms.getAccessToken()
+	if err != nil {
+		return fmt.Errorf("could not retrieve vault token to delete the passphrase at %s: %s",
+			key, err)
+	}
+
+	resp, err := kms.send("DELETE", kms.getKeyMetadataURI(key), &vaultToken, nil)
+	if err != nil {
+		return fmt.Errorf("delete passphrase at %s request to vault failed: %s", key, err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != 404 {
+		err = kms.processError(resp, "delete passphrase")
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (kms *VaultKMS) getKeyDataURI(key string) string {
+	return kms.VaultPassphraseRoot + "/data/" + kms.VaultPassphrasePath + key
+}
+
+func (kms *VaultKMS) getKeyMetadataURI(key string) string {
+	return kms.VaultPassphraseRoot + "/metadata/" + kms.VaultPassphrasePath + key
+}
+
+/*
+getVaultAccessToken retrieves vault token using kubernetes authentication:
+ 1. read jwt service account token from well known location
+ 2. request token from vault using service account jwt token
+Vault will verify service account jwt token with Kubernetes and return token
+if the requester is allowed
+*/
+func (kms *VaultKMS) getAccessToken() (string, error) {
+	saToken, err := ioutil.ReadFile(serviceAccountTokenPath)
+	if err != nil {
+		return "", fmt.Errorf("service account token could not be read: %s", err)
+	}
+	data, err := json.Marshal(map[string]string{
+		"role": kms.VaultRole,
+		"jwt":  string(saToken),
+	})
+	if err != nil {
+		return "", fmt.Errorf("vault token request data is broken: %s", err)
+	}
+	resp, err := kms.send("POST", kms.VaultAuthPath, nil, data)
+	if err != nil {
+		return "", fmt.Errorf("failed to retrieve vault token: %s", err)
+	}
+	defer resp.Body.Close()
+
+	err = kms.processError(resp, "retrieve vault token")
+	if err != nil {
+		return "", err
+	}
+	// parse resp as JSON and retrieve vault token
+	var result map[string]interface{}
+	err = json.NewDecoder(resp.Body).Decode(&result)
+	if err != nil {
+		return "", fmt.Errorf("failed parsing vaultToken from response: %s", err)
+	}
+
+	auth, ok := result["auth"].(map[string]interface{})
+	if !ok {
+		return "", fmt.Errorf("failed parsing vault token auth data")
+	}
+	vaultToken, ok := auth["client_token"].(string)
+	if !ok {
+		return "", fmt.Errorf("failed parsing vault client_token")
+	}
+
+	return vaultToken, nil
+}
+
+func (kms *VaultKMS) processError(resp *http.Response, action string) error {
+	if resp.StatusCode >= 200 || resp.StatusCode < 300 {
+		return nil
+	}
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return fmt.Errorf("failed to %s (%v), error body parsing failed: %s",
+			action, resp.StatusCode, err)
+	}
+	return fmt.Errorf("failed to %s (%v): %s", action, resp.StatusCode, body)
+}
+
+func (kms *VaultKMS) request(method, path string, data []byte) (*http.Response, error) {
+	vaultToken, err := kms.getAccessToken()
+	if err != nil {
+		return nil, err
+	}
+
+	return kms.send(method, path, &vaultToken, data)
+}
+
+func (kms *VaultKMS) send(method, path string, token *string, data []byte) (*http.Response, error) {
+	tlsConfig := &tls.Config{}
+	if !kms.VaultCAVerify {
+		tlsConfig.InsecureSkipVerify = true
+	}
+	if kms.vaultCA != nil {
+		tlsConfig.RootCAs = kms.vaultCA
+	}
+	netTransport := &http.Transport{TLSClientConfig: tlsConfig}
+	client := &http.Client{Transport: netTransport}
+
+	var dataToSend io.Reader
+	if data != nil {
+		dataToSend = strings.NewReader(string(data))
+	}
+
+	req, err := http.NewRequest(method, kms.VaultAddress+path, dataToSend)
+	if err != nil {
+		return nil, fmt.Errorf("could not create a Vault request: %s", err)
+	}
+
+	if kms.VaultNamespace != "" {
+		req.Header.Set(vaultNamespaceHeader, kms.VaultNamespace)
+	}
+	if token != nil {
+		req.Header.Set(vaultTokenHeader, *token)
+	}
+
+	return client.Do(req)
+}
--- a/pkg/util/voljournal.go
+++ b/pkg/util/voljournal.go
@ -118,6 +118,9 @@ type CSIJournal struct {

 	// namespace in which the RADOS objects are stored, default is no namespace
 	namespace string
+
+	// encryptKMS in which encryption passphrase was saved, default is no encryption
+	encryptKMSKey string
 }

 // CSIVolumeJournal returns an instance of volume keys
@ -130,6 +133,7 @@ func NewCSIVolumeJournal() *CSIJournal {
 		namingPrefix:            "csi-vol-",
 		cephSnapSourceKey:       "",
 		namespace:               "",
+		encryptKMSKey:           "csi.volume.encryptKMS",
 	}
 }

@ -143,6 +147,7 @@ func NewCSISnapshotJournal() *CSIJournal {
 		namingPrefix:            "csi-snap-",
 		cephSnapSourceKey:       "csi.source",
 		namespace:               "",
+		encryptKMSKey:           "csi.volume.encryptKMS",
 	}
 }

@ -176,7 +181,7 @@ Return values:
 	there was no reservation found
 	- error: non-nil in case of any errors
 */
-func (cj *CSIJournal) CheckReservation(ctx context.Context, monitors string, cr *Credentials, pool, reqName, parentName string) (string, error) {
+func (cj *CSIJournal) CheckReservation(ctx context.Context, monitors string, cr *Credentials, pool, reqName, parentName, encryptionKmsConfig string) (string, error) {
 	var snapSource bool

 	if parentName != "" {
@ -199,7 +204,7 @@ func (cj *CSIJournal) CheckReservation(ctx context.Context, monitors string, cr
 		return "", err
 	}

-	savedReqName, savedReqParentName, err := cj.GetObjectUUIDData(ctx, monitors, cr, pool,
+	savedReqName, savedReqParentName, savedKms, err := cj.GetObjectUUIDData(ctx, monitors, cr, pool,
 		objUUID, snapSource)
 	if err != nil {
 		// error should specifically be not found, for image to be absent, any other error
@ -219,6 +224,14 @@ func (cj *CSIJournal) CheckReservation(ctx context.Context, monitors string, cr
 			reqName, objUUID, savedReqName)
 	}

+	if encryptionKmsConfig != "" {
+		if savedKms != encryptionKmsConfig {
+			return "", fmt.Errorf("internal state inconsistent, omap encryption KMS"+
+				" mismatch, request KMS (%s) volume UUID (%s) volume omap KMS (%s)",
+				encryptionKmsConfig, objUUID, savedKms)
+		}
+	}
+
 	if snapSource {
 		// check if source UUID key points back to the parent volume passed in
 		if savedReqParentName != parentName {
@ -310,7 +323,7 @@ Return values:
 	- string: Contains the UUID that was reserved for the passed in reqName
 	- error: non-nil in case of any errors
 */
-func (cj *CSIJournal) ReserveName(ctx context.Context, monitors string, cr *Credentials, pool, reqName, parentName string) (string, error) {
+func (cj *CSIJournal) ReserveName(ctx context.Context, monitors string, cr *Credentials, pool, reqName, parentName, encryptionKmsConfig string) (string, error) {
 	var snapSource bool

 	if parentName != "" {
@ -355,6 +368,14 @@ func (cj *CSIJournal) ReserveName(ctx context.Context, monitors string, cr *Cred
 		return "", err
 	}

+	if encryptionKmsConfig != "" {
+		err = SetOMapKeyValue(ctx, monitors, cr, pool, cj.namespace, cj.cephUUIDDirectoryPrefix+volUUID,
+			cj.encryptKMSKey, encryptionKmsConfig)
+		if err != nil {
+			return "", err
+		}
+	}
+
 	if snapSource {
 		// Update UUID directory to store source volume UUID in case of snapshots
 		err = SetOMapKeyValue(ctx, monitors, cr, pool, cj.namespace, cj.cephUUIDDirectoryPrefix+volUUID,
@ -372,30 +393,42 @@ GetObjectUUIDData fetches all keys from a UUID directory
 Return values:
 	- string: Contains the request name for the passed in UUID
 	- string: Contains the parent image name for the passed in UUID, if it is a snapshot
+	- string: Contains encryption KMS, if it is an encrypted image
 	- error: non-nil in case of any errors
 */
-func (cj *CSIJournal) GetObjectUUIDData(ctx context.Context, monitors string, cr *Credentials, pool, objectUUID string, snapSource bool) (string, string, error) {
+func (cj *CSIJournal) GetObjectUUIDData(ctx context.Context, monitors string, cr *Credentials, pool, objectUUID string, snapSource bool) (string, string, string, error) {
 	var sourceName string

 	if snapSource && cj.cephSnapSourceKey == "" {
 		err := errors.New("invalid request, cephSnapSourceKey is nil")
-		return "", "", err
+		return "", "", "", err
 	}

 	// TODO: fetch all omap vals in one call, than make multiple listomapvals
 	requestName, err := GetOMapValue(ctx, monitors, cr, pool, cj.namespace,
 		cj.cephUUIDDirectoryPrefix+objectUUID, cj.csiNameKey)
 	if err != nil {
-		return "", "", err
+		return "", "", "", err
+	}
+
+	encryptionKmsConfig := ""
+	encryptionKmsConfig, err = GetOMapValue(ctx, monitors, cr, pool, cj.namespace,
+		cj.cephUUIDDirectoryPrefix+objectUUID, cj.encryptKMSKey)
+	if err != nil {
+		if _, ok := err.(ErrKeyNotFound); !ok {
+			klog.Errorf(Log(ctx, "=> GetObjectUUIDData encryptedKMS failed: %s (%s)"), cj.cephUUIDDirectoryPrefix+objectUUID, err)
+			return "", "", "", err
+		}
+		// ErrKeyNotFound means no encryption KMS was used
 	}

 	if snapSource {
 		sourceName, err = GetOMapValue(ctx, monitors, cr, pool, cj.namespace,
 			cj.cephUUIDDirectoryPrefix+objectUUID, cj.cephSnapSourceKey)
 		if err != nil {
-			return "", "", err
+			return "", "", "", err
 		}
 	}

-	return requestName, sourceName, nil
+	return requestName, sourceName, encryptionKmsConfig, nil
 }