Merge pull request #8005 from leseb/luks-label

ceph: add ceph cluster fsid to LUKS header

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

@@ -16,6 +16,16 @@ limitations under the License.
 
 package v1
 
+import (
+	"strings"
+
+	"github.com/hashicorp/vault/api"
+)
+
+var (
+	VaultTLSConnectionDetails = []string{api.EnvVaultCACert, api.EnvVaultClientCert, api.EnvVaultClientKey}
+)
+
 // IsEnabled return whether a KMS is configured
 func (kms *KeyManagementServiceSpec) IsEnabled() bool {
 	return len(kms.ConnectionDetails) != 0
@@ -25,3 +35,22 @@ func (kms *KeyManagementServiceSpec) IsEnabled() bool {
 func (kms *KeyManagementServiceSpec) IsTokenAuthEnabled() bool {
 	return kms.TokenSecretName != ""
 }
+
+// IsTLSEnabled return KMS TLS details are configured
+func (kms *KeyManagementServiceSpec) IsTLSEnabled() bool {
+	for _, tlsOption := range VaultTLSConnectionDetails {
+		tlsSecretName := getParam(kms.ConnectionDetails, tlsOption)
+		if tlsSecretName != "" {
+			return true
+		}
+	}
+	return false
+}
+
+// getParam returns the value of the KMS config option
+func getParam(kmsConfig map[string]string, param string) string {
+	if val, ok := kmsConfig[param]; ok && val != "" {
+		return strings.TrimSpace(val)
+	}
+	return ""
+}

pkg/daemon/ceph/osd/daemon.go

@@ -305,8 +305,15 @@ func getAvailableDevices(context *clusterd.Context, agent *OsdAgent) (*DeviceOsd
 		// cannot detect that correctly
 		// see: https://tracker.ceph.com/issues/43585
 		if device.Filesystem != "" {
-			logger.Infof("skipping device %q because it contains a filesystem %q", device.Name, device.Filesystem)
-			continue
+			// Allow further inspection of that device before skipping it
+			if device.Filesystem == "crypto_LUKS" && agent.pvcBacked {
+				if isCephEncryptedBlock(context, agent.clusterInfo.FSID, device.Name) {
+					logger.Infof("encrypted disk %q is an OSD part of this cluster, considering it", device.Name)
+				} else {
+					logger.Infof("skipping device %q because it contains a filesystem %q", device.Name, device.Filesystem)
+					continue
+				}
+			}
 		}
 
 		// If we detect a partition we have to make sure that ceph-volume will be able to consume it

pkg/daemon/ceph/osd/encryption.go

@@ -17,7 +17,10 @@ limitations under the License.
 package osd
 
 import (
+	"fmt"
 	"os"
+	"regexp"
+	"strings"
 
 	"github.com/pkg/errors"
 	v1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -32,6 +35,10 @@ const (
 	dmsetupBinary    = "dmsetup"
 )
 
+var (
+	luksLabelCephFSID = regexp.MustCompile("ceph_fsid=(.*)")
+)
+
 func closeEncryptedDevice(context *clusterd.Context, dmName string) error {
 	args := []string{"--verbose", "luksClose", dmName}
 	cryptsetupOut, err := context.Executor.ExecuteCommandWithCombinedOutput(cryptsetupBinary, args...)
@@ -74,3 +81,59 @@ func setKEKinEnv(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo)
 
 	return nil
 }
+
+func setLUKSLabelAndSubsystem(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, disk string) error {
+	// The PVC info is a nice to have
+	pvcName := os.Getenv(oposd.PVCNameEnvVarName)
+	if pvcName == "" {
+		return errors.Errorf("failed to find %q environment variable", oposd.PVCNameEnvVarName)
+	}
+	subsystem := fmt.Sprintf("ceph_fsid=%s", clusterInfo.FSID)
+	label := fmt.Sprintf("pvc_name=%s", pvcName)
+
+	logger.Infof("setting LUKS subsystem to %q and label to %q to disk %q", subsystem, label, disk)
+	args := []string{"config", disk, "--subsystem", subsystem, "--label", label}
+	output, err := context.Executor.ExecuteCommandWithCombinedOutput(cryptsetupBinary, args...)
+	if err != nil {
+		return errors.Wrapf(err, "failed to set subsystem %q and label %q to encrypted device %q. is your distro built with LUKS1 as a default?. %s", subsystem, label, disk, output)
+	}
+
+	logger.Infof("successfully set LUKS subsystem to %q and label to %q to disk %q", subsystem, label, disk)
+	return nil
+}
+
+func dumpLUKS(context *clusterd.Context, disk string) (string, error) {
+	args := []string{"luksDump", disk}
+	cryptsetupOut, err := context.Executor.ExecuteCommandWithCombinedOutput(cryptsetupBinary, args...)
+	if err != nil {
+		return "", errors.Wrapf(err, "failed to dump LUKS header for disk %q. %s", disk, cryptsetupOut)
+	}
+
+	return cryptsetupOut, nil
+}
+
+func isCephEncryptedBlock(context *clusterd.Context, currentClusterFSID string, disk string) bool {
+	metadata, err := dumpLUKS(context, disk)
+	if err != nil {
+		logger.Errorf("failed to determine if the encrypted block %q is from our cluster. %v", disk, err)
+		return false
+	}
+
+	// Now we parse the CLI output
+	// JSON output is only available with cryptsetup 2.4.x - https://gitlab.com/cryptsetup/cryptsetup/-/issues/511
+	ceph_fsid := luksLabelCephFSID.FindString(metadata)
+	if ceph_fsid == "" {
+		logger.Error("failed to find ceph_fsid in the LUKS header, the encrypted disk is not from a ceph cluster")
+		return false
+	}
+
+	// is it an OSD from our cluster?
+	currentDiskCephFSID := strings.SplitAfter(ceph_fsid, "=")[1]
+	if currentDiskCephFSID != currentClusterFSID {
+		logger.Errorf("encrypted disk %q is part of a different ceph cluster %q", disk, currentDiskCephFSID)
+		return false
+	}
+
+	return true
+
+}

pkg/daemon/ceph/osd/encryption_test.go

@@ -25,6 +25,48 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+var (
+	luksDump = `LUKS header information
+Version:        2
+Epoch:          13
+Metadata area:  12288 bytes
+UUID:           a97525ee-7c30-4f70-89ac-e56d48907cc5
+Label:          pvc_name=set1-data-0lmdjp
+Subsystem:      ceph_fsid=811e7dc0-ea13-4951-b000-24a8565d0735
+Flags:          (no flags)
+
+Data segments:
+  0: crypt
+        offset: 2097152 [bytes]
+        length: (whole device)
+        cipher: aes-xts-plain64
+        sector: 512 [bytes]
+
+Keyslots:
+  0: luks2
+        Key:        256 bits
+        Priority:   normal
+        Cipher:     aes-xts-plain64
+        PBKDF:      pbkdf2
+        Hash:       sha256
+        Iterations: 583190
+        Salt:       4f 9d 0d 0b 83 41 2f 47 b4 1f 6b 35 df 89 e0 33
+                    c8 bd 27 60 22 a5 f5 02 62 94 a9 92 12 2a 4f c0
+        AF stripes: 4000
+        Area offset:32768 [bytes]
+        Area length:131072 [bytes]
+        Digest ID:  0
+Tokens:
+Digests:
+  0: pbkdf2
+        Hash:       sha256
+        Iterations: 36127
+        Salt:       db 98 33 3a d4 15 b6 6c 48 63 6d 7b 33 b0 7e cd
+                    ef 90 8d 81 46 37 78 b4 82 37 3b 84 e8 e7 d8 1b
+        Digest:     6d 86 96 05 99 4f a9 48 87 54
+                    5c ef 4b 99 3b 9d fa 0b 8f 8a`
+)
+
 func TestCloseEncryptedDevice(t *testing.T) {
 	executor := &exectest.MockExecutor{}
 	executor.MockExecuteCommandWithCombinedOutput = func(command string, args ...string) (string, error) {
@@ -60,3 +102,25 @@ Driver version:    4.40.0
 	err := dmsetupVersion(context)
 	assert.NoError(t, err)
 }
+
+func TestIsCephEncryptedBlock(t *testing.T) {
+	executor := &exectest.MockExecutor{}
+	executor.MockExecuteCommandWithCombinedOutput = func(command string, args ...string) (string, error) {
+		logger.Infof("%s %v", command, args)
+		if command == cryptsetupBinary && args[0] == "luksDump" {
+			return luksDump, nil
+		}
+
+		return "", errors.Errorf("unknown command %s %s", command, args)
+	}
+	context := &clusterd.Context{Executor: executor}
+
+	t.Run("different fsid", func(t *testing.T) {
+		isCephEncryptedBlock := isCephEncryptedBlock(context, "foo", "/dev/sda1")
+		assert.False(t, isCephEncryptedBlock)
+	})
+	t.Run("same cluster", func(t *testing.T) {
+		isCephEncryptedBlock := isCephEncryptedBlock(context, "811e7dc0-ea13-4951-b000-24a8565d0735", "/dev/sda1")
+		assert.True(t, isCephEncryptedBlock)
+	})
+}

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

@@ -53,7 +53,7 @@ func vaultTokenEnvVarFromSecret(tokenSecretName string) v1.EnvVar {
 func vaultTLSEnvVarFromSecret(kmsConfig map[string]string) []v1.EnvVar {
 	vaultTLSEnvVar := []v1.EnvVar{}
 
-	for _, tlsOption := range vaultTLSConnectionDetails {
+	for _, tlsOption := range cephv1.VaultTLSConnectionDetails {
 		tlsSecretName := GetParam(kmsConfig, tlsOption)
 		if tlsSecretName != "" {
 			vaultTLSEnvVar = append(vaultTLSEnvVar, v1.EnvVar{Name: tlsOption, Value: path.Join(EtcVaultDir, tlsSecretPath(tlsOption))})
@@ -73,7 +73,7 @@ func VaultConfigToEnvVar(spec cephv1.ClusterSpec) []v1.EnvVar {
 	}
 	for k, v := range spec.Security.KeyManagementService.ConnectionDetails {
 		// Skip TLS and token env var to avoid env being set multiple times
-		toSkip := append(vaultTLSConnectionDetails, api.EnvVaultToken)
+		toSkip := append(cephv1.VaultTLSConnectionDetails, api.EnvVaultToken)
 		if client.StringInSlice(k, toSkip) {
 			continue
 		}

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

@@ -25,6 +25,7 @@ import (
 	"github.com/libopenstorage/secrets"
 	"github.com/libopenstorage/secrets/vault"
 	"github.com/pkg/errors"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -42,7 +43,6 @@ const (
 
 var (
 	vaultMandatoryConnectionDetails = []string{api.EnvVaultAddress}
-	vaultTLSConnectionDetails       = []string{api.EnvVaultCACert, api.EnvVaultClientCert, api.EnvVaultClientKey}
 )
 
 /* VAULT API INTERNAL VALUES
@@ -98,7 +98,7 @@ func InitVault(context *clusterd.Context, namespace string, config map[string]st
 
 func configTLS(clusterdContext *clusterd.Context, namespace string, config map[string]string) (map[string]string, error) {
 	ctx := context.TODO()
-	for _, tlsOption := range vaultTLSConnectionDetails {
+	for _, tlsOption := range cephv1.VaultTLSConnectionDetails {
 		tlsSecretName := GetParam(config, tlsOption)
 		if tlsSecretName == "" {
 			continue
@@ -209,7 +209,7 @@ func validateVaultConnectionDetails(clusterdContext *clusterd.Context, ns string
 	}
 
 	// Validate potential TLS configuration
-	for _, tlsOption := range vaultTLSConnectionDetails {
+	for _, tlsOption := range cephv1.VaultTLSConnectionDetails {
 		tlsSecretName := GetParam(kmsConfig, tlsOption)
 		if tlsSecretName != "" {
 			// Fetch the secret

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

@@ -19,6 +19,7 @@ package kms
 import (
 	"github.com/hashicorp/vault/api"
 	"github.com/libopenstorage/secrets"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	v1 "k8s.io/api/core/v1"
 )
 
@@ -45,7 +46,7 @@ func TLSSecretVolumeAndMount(config map[string]string) []v1.VolumeProjection {
 	mode := int32(0400)
 
 	// Vault TLS Secrets
-	for _, tlsOption := range vaultTLSConnectionDetails {
+	for _, tlsOption := range cephv1.VaultTLSConnectionDetails {
 		tlsSecretName := GetParam(config, tlsOption)
 		if tlsSecretName != "" {
 			projectionSecret := &v1.SecretProjection{Items: []v1.KeyToPath{{Key: tlsSecretKeyToCheck(tlsOption), Path: tlsSecretPath(tlsOption), Mode: &mode}}}

pkg/daemon/ceph/osd/volume.go

@@ -1009,6 +1009,14 @@ func GetCephVolumeRawOSDs(context *clusterd.Context, clusterInfo *client.Cluster
 
 		// If this is an encrypted OSD
 		if os.Getenv(oposd.CephVolumeEncryptedKeyEnvVarName) != "" {
+			// // Set subsystem and label for recovery and detection
+			// We use /mnt/<pvc_name> since LUKS label/subsystem must be applied on the main block device, not the resulting encrypted dm
+			mainBlock := fmt.Sprintf("/mnt/%s", os.Getenv(oposd.PVCNameEnvVarName))
+			err = setLUKSLabelAndSubsystem(context, clusterInfo, mainBlock)
+			if err != nil {
+				return nil, errors.Wrapf(err, "failed to set subsystem and label to encrypted device %q for osd %d", mainBlock, osdID)
+			}
+
 			// Close encrypted device
 			err = closeEncryptedDevice(context, block)
 			if err != nil {

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

@@ -144,6 +144,8 @@ fi
 	openEncryptedBlock = `
 set -xe
 
+CEPH_FSID=%s
+PVC_NAME=%s
 KEY_FILE_PATH=%s
 BLOCK_PATH=%s
 DM_NAME=%s
@@ -158,6 +160,12 @@ function open_encrypted_block {
 	rm -f "$KEY_FILE_PATH"
 }
 
+# This is done for upgraded clusters that did not have the subsystem and label set by the prepare job
+function set_luks_subsystem_and_label {
+	echo "setting LUKS label and subsystem"
+	cryptsetup config $BLOCK_PATH --subsystem ceph_fsid="$CEPH_FSID" --label pvc_name="$PVC_NAME"
+}
+
 if [ -b "$DM_PATH" ]; then
 	echo "Encrypted device $BLOCK_PATH already opened at $DM_PATH"
 	for field in $(dmsetup table "$DM_NAME"); do
@@ -174,6 +182,13 @@ if [ -b "$DM_PATH" ]; then
 else
 	open_encrypted_block
 fi
+
+# Setting label and subsystem on LUKS1 is not supported and the command will fail
+if cryptsetup luksDump $BLOCK_PATH|grep -qEs "Version:.*2"; then
+	set_luks_subsystem_and_label
+else
+	echo "LUKS version is not 2 so not setting label and subsystem"
+fi
 `
 	// #nosec G101 no leak just variable names
 	getKEKFromVaultWithToken = `
@@ -183,7 +198,7 @@ set -e
 KEK_NAME=%s
 KEY_PATH=%s
 CURL_PAYLOAD=$(mktemp)
-ARGS=(--silent --show-error --request GET --header "X-Vault-Token: ${VAULT_TOKEN}")
+ARGS=(--silent --show-error --request GET --header "X-Vault-Token: ${VAULT_TOKEN//[$'\t\r\n']}")
 PYTHON_DATA_PARSE="['data']"
 
 # If a vault namespace is set
@@ -344,7 +359,10 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo, provisionC
 		if osdProps.encrypted && osd.CVMode == "raw" {
 			encryptedVol, _ := c.getEncryptionVolume(osdProps)
 			volumes = append(volumes, encryptedVol)
-			if c.spec.Security.KeyManagementService.IsEnabled() {
+			// We don't need to pass the Volume with projection for TLS when TLS is not enabled
+			// Somehow when this happens and we try to update a deployment spec it fails with:
+			//  ValidationError(Pod.spec.volumes[7].projected): missing required field "sources"
+			if c.spec.Security.KeyManagementService.IsEnabled() && c.spec.Security.KeyManagementService.IsTLSEnabled() {
 				encryptedVol, _ := kms.VaultVolumeAndMount(c.spec.Security.KeyManagementService.ConnectionDetails)
 				volumes = append(volumes, encryptedVol)
 			}
@@ -869,7 +887,7 @@ func (c *Cluster) generateEncryptionOpenBlockContainer(resources v1.ResourceRequ
 		Command: []string{
 			"/bin/bash",
 			"-c",
-			fmt.Sprintf(openEncryptedBlock, encryptionKeyPath(), encryptionBlockDestinationCopy(mountPath, blockType), encryptionDMName(pvcName, cryptBlockType), encryptionDMPath(pvcName, cryptBlockType)),
+			fmt.Sprintf(openEncryptedBlock, c.clusterInfo.FSID, pvcName, encryptionKeyPath(), encryptionBlockDestinationCopy(mountPath, blockType), encryptionDMName(pvcName, cryptBlockType), encryptionDMPath(pvcName, cryptBlockType)),
 		},
 		VolumeMounts:    []v1.VolumeMount{getPvcOSDBridgeMountActivate(mountPath, volumeMountPVCName), getDeviceMapperMount()},
 		SecurityContext: PrivilegedContext(),
@@ -907,8 +925,10 @@ func (c *Cluster) getPVCEncryptionOpenInitContainerActivate(mountPath string, os
 				getKEKFromKMSContainer.VolumeMounts = append(getKEKFromKMSContainer.VolumeMounts, volMount)
 
 				// Now let's see if there is a TLS config we need to mount as well
-				_, vaultVolMount := kms.VaultVolumeAndMount(c.spec.Security.KeyManagementService.ConnectionDetails)
-				getKEKFromKMSContainer.VolumeMounts = append(getKEKFromKMSContainer.VolumeMounts, vaultVolMount)
+				if c.spec.Security.KeyManagementService.IsTLSEnabled() {
+					_, vaultVolMount := kms.VaultVolumeAndMount(c.spec.Security.KeyManagementService.ConnectionDetails)
+					getKEKFromKMSContainer.VolumeMounts = append(getKEKFromKMSContainer.VolumeMounts, vaultVolMount)
+				}
 
 				// Add the container to the list of containers
 				containers = append(containers, getKEKFromKMSContainer)

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

@@ -383,8 +383,7 @@ func testPodDevices(t *testing.T, dataDir, deviceName string, allDevices bool) {
 	assert.Equal(t, 1, len(deployment.Spec.Template.Spec.Containers))
 	cont = deployment.Spec.Template.Spec.Containers[0]
 	assert.Equal(t, 7, len(cont.VolumeMounts), cont.VolumeMounts)
-	assert.Equal(t, 10, len(deployment.Spec.Template.Spec.Volumes), deployment.Spec.Template.Spec.Volumes)                                     // One more than the encryption with k8s for the kek get init container
-	assert.Equal(t, 0, len(deployment.Spec.Template.Spec.Volumes[7].VolumeSource.Projected.Sources), deployment.Spec.Template.Spec.Volumes[0]) // 0 since we have no tls secrets
+	assert.Equal(t, 9, len(deployment.Spec.Template.Spec.Volumes), deployment.Spec.Template.Spec.Volumes) // One more than the encryption with k8s for the kek get init container
 
 	// Test with encrypted OSD on PVC with RAW with KMS with TLS
 	osdProp.encrypted = true

tests/manifests/test-cluster-on-pvc-encrypted.yaml

@@ -14,7 +14,7 @@ spec:
           requests:
             storage: 5Gi
   cephVersion:
-    image: ceph/ceph:v14
+    image: ceph/ceph:v15
   dashboard:
     enabled: false
   network:

tests/scripts/github-action-helper.sh

@@ -124,7 +124,7 @@ function deploy_cluster() {
 }
 
 function wait_for_prepare_pod() {
-    timeout 180 sh -c 'until kubectl -n rook-ceph logs -f $(kubectl -n rook-ceph get pod -l app=rook-ceph-osd-prepare -o jsonpath='{.items[*].metadata.name}'); do sleep 5; done' || true
+    timeout 180 sh -c 'until kubectl -n rook-ceph logs -f job/$(kubectl -n rook-ceph get job -l app=rook-ceph-osd-prepare -o jsonpath='{.items[0].metadata.name}'); do sleep 5; done' || true
     timeout 60 sh -c 'until kubectl -n rook-ceph logs $(kubectl -n rook-ceph get pod -l app=rook-ceph-osd,ceph_daemon_id=0 -o jsonpath='{.items[*].metadata.name}') --all-containers; do echo "waiting for osd container" && sleep 1; done' || true
     kubectl -n rook-ceph describe job/$(kubectl -n rook-ceph get pod -l app=rook-ceph-osd-prepare -o jsonpath='{.items[*].metadata.name}') || true
     kubectl -n rook-ceph describe deploy/rook-ceph-osd-0 || true

tests/scripts/validate_cluster.sh

@@ -105,10 +105,14 @@ function display_status {
   $EXEC_COMMAND osd dump > test/ceph-osd-dump.txt
   $EXEC_COMMAND report > test/ceph-report.txt
 
-  kubectl -n rook-ceph logs "$(kubectl -n rook-ceph -l app=rook-ceph-operator get pods -o jsonpath='{.items[*].metadata.name}')" > test/operator.txt
-  kubectl -n rook-ceph get pods > test/pods-list.txt
-  kubectl -n rook-ceph describe job/"$(kubectl -n rook-ceph get pod -l app=rook-ceph-osd-prepare -o jsonpath='{.items[*].metadata.name}')" > test/osd-prepare.txt
-  kubectl -n rook-ceph describe deploy/rook-ceph-osd-0 > test/osd-deploy.txt
+  kubectl -n rook-ceph logs deploy/rook-ceph-operator > test/operator-logs.txt
+  kubectl -n rook-ceph get pods -o wide > test/pods-list.txt
+  kubectl -n rook-ceph describe job/"$(kubectl -n rook-ceph get job -l app=rook-ceph-osd-prepare -o jsonpath='{.items[*].metadata.name}')" > test/osd-prepare-describe.txt
+  kubectl -n rook-ceph log job/"$(kubectl -n rook-ceph get job -l app=rook-ceph-osd-prepare -o jsonpath='{.items[*].metadata.name}')" > test/osd-prepare-logs.txt
+  kubectl -n rook-ceph describe deploy/rook-ceph-osd-0 > test/rook-ceph-osd-0-describe.txt
+  kubectl -n rook-ceph describe deploy/rook-ceph-osd-1 > test/rook-ceph-osd-1-describe.txt
+  kubectl -n rook-ceph logs deploy/rook-ceph-osd-0 --all-containers > test/rook-ceph-osd-0-logs.txt
+  kubectl -n rook-ceph logs deploy/rook-ceph-osd-1 --all-containers > test/rook-ceph-osd-1-logs.txt
   kubectl get all -n rook-ceph -o wide > test/cluster-wide.txt
   kubectl get all -n rook-ceph -o yaml > test/cluster-yaml.txt
   kubectl -n rook-ceph get cephcluster -o yaml > test/cephcluster.txt