Add LUKS encryption support for cloud volumes via CDH

Integrate with CDH's secure_mount TTRPC API to support LUKS2 encrypted
block volumes in Peer Pods. When a StorageClass specifies encrypt-type
and kbs-key-id, the interceptor inside the PodVM delegates to CDH which
handles LUKS formatting/opening with keys fetched from KBS after remote
attestation.

Changes:
- Extend CloudVolumeAnnotation with EncryptType and KeyID fields
- Proxy reads encrypt-type/kbs-key-id from mountInfo.json metadata
- Add CDH TTRPC client using generated proto types, with context-aware
  connection retry logic (10 attempts, respects CreateContainer cancel)
- Interceptor branches on EncryptType: CDH path vs plain formatAndMount
- Pass predictable mapperName to CDH for reliable cryptsetup cleanup;
  fallback to /proc/mounts lookup if name is unavailable
- Validate encrypt type (only luks/luks2 accepted, always send luks2
  to CDH since that is what CDH accepts)
- isLuks detection via magic-byte read distinguishes fresh disks from
  existing LUKS volumes; errors are hard failures to prevent CDH from
  reformatting an encrypted disk due to a transient read error
- Resolve mapper name before unmount and skip cryptsetup close if
  unmount fails
- unmountCloudVolumes handles cryptsetup close for encrypted mounts
- Unit tests for annotation flow, proto round-trip, encrypt type
  validation, CDH request format, encrypted mount tracking

Signed-off-by: Mohammed Adnan <muhammedadnan50007@gmail.com>

Author: mohammedadnan21

src/cloud-api-adaptor/pkg/adaptor/proxy/cloud_volumes_test.go

@@ -365,3 +365,94 @@ func TestCloudVolumes_SkipsInvalidMountInfoJSON(t *testing.T) {
 	_, ok := req.OCI.Annotations["io.confidentialcontainers.org.cloud_volumes"]
 	assert.False(t, ok, "annotation should not be set with invalid JSON")
 }
+
+func TestCloudVolumes_EncryptionAnnotation(t *testing.T) {
+	dir := t.TempDir()
+	overrideKataDirectVolumesDir(t, dir)
+
+	service, cleanup := setupMockAgentAndService(t)
+	defer cleanup()
+
+	podUID := "pod-uid-enc-333"
+	volPath := "/var/lib/kubelet/pods/" + podUID + "/volumes/kubernetes.io~csi/pvc-encrypted/mount"
+
+	writeTestMountInfo(t, dir, volPath, map[string]interface{}{
+		"device": "/subscriptions/sub/disks/csi-vol-pvc-encrypted",
+		"fstype": "ext4",
+		"metadata": map[string]interface{}{
+			"cloud-volume-path": "/subscriptions/sub/disks/csi-vol-pvc-encrypted",
+			"encrypt-type":      "LUKS",
+			"kbs-key-id":        "default/key/volume-enc-key",
+		},
+	})
+
+	req := newCreateContainerRequest("test-encrypted-vol").
+		withAnnotations(map[string]string{
+			"io.kubernetes.cri.sandbox-uid": podUID,
+		}).
+		withMounts(&pb.Mount{
+			Destination: "/mnt/secret",
+			Source:      volPath,
+			Type:        "bind",
+		}).
+		build()
+
+	_, err := service.CreateContainer(context.Background(), req)
+	require.NoError(t, err)
+
+	cvJSON, ok := req.OCI.Annotations[util.CloudVolumesAnnotationKey]
+	require.True(t, ok, "cloud_volumes annotation should be set")
+
+	var cloudVolumes map[string]util.CloudVolumeAnnotation
+	require.NoError(t, json.Unmarshal([]byte(cvJSON), &cloudVolumes))
+
+	require.Contains(t, cloudVolumes, "vol-0")
+	vol := cloudVolumes["vol-0"]
+	assert.Equal(t, "/mnt/secret", vol.MountPoint)
+	assert.Equal(t, "ext4", vol.FSType)
+	assert.Equal(t, "0", vol.LUN)
+	assert.Equal(t, "/subscriptions/sub/disks/csi-vol-pvc-encrypted", vol.DiskID)
+	assert.Equal(t, "LUKS", vol.EncryptType)
+	assert.Equal(t, "default/key/volume-enc-key", vol.KeyID)
+}
+
+func TestCloudVolumes_NoEncryptionParamsWhenAbsent(t *testing.T) {
+	dir := t.TempDir()
+	overrideKataDirectVolumesDir(t, dir)
+
+	service, cleanup := setupMockAgentAndService(t)
+	defer cleanup()
+
+	podUID := "pod-uid-plain-444"
+	volPath := "/var/lib/kubelet/pods/" + podUID + "/volumes/kubernetes.io~csi/pvc-plain/mount"
+
+	writeTestMountInfo(t, dir, volPath, map[string]interface{}{
+		"device": "vol-abc123def",
+		"fstype": "xfs",
+	})
+
+	req := newCreateContainerRequest("test-plain-vol").
+		withAnnotations(map[string]string{
+			"io.kubernetes.cri.sandbox-uid": podUID,
+		}).
+		withMounts(&pb.Mount{
+			Destination: "/mnt/data",
+			Source:      volPath,
+			Type:        "bind",
+		}).
+		build()
+
+	_, err := service.CreateContainer(context.Background(), req)
+	require.NoError(t, err)
+
+	cvJSON, ok := req.OCI.Annotations[util.CloudVolumesAnnotationKey]
+	require.True(t, ok, "cloud_volumes annotation should be set")
+
+	var cloudVolumes map[string]util.CloudVolumeAnnotation
+	require.NoError(t, json.Unmarshal([]byte(cvJSON), &cloudVolumes))
+
+	require.Contains(t, cloudVolumes, "vol-0")
+	vol := cloudVolumes["vol-0"]
+	assert.Equal(t, "", vol.EncryptType)
+	assert.Equal(t, "", vol.KeyID)
+}

src/cloud-api-adaptor/pkg/adaptor/proxy/service.go

@@ -139,10 +139,18 @@ func (s *proxyService) CreateContainer(ctx context.Context, req *pb.CreateContai
 			}
 
 			diskID := ""
+			encryptType := ""
+			keyID := ""
 			if md, ok := mountInfo["metadata"].(map[string]interface{}); ok {
 				if cp, ok := md["cloud-volume-path"].(string); ok && cp != "" {
 					diskID = cp
 				}
+				if et, ok := md["encrypt-type"].(string); ok {
+					encryptType = et
+				}
+				if kid, ok := md["kbs-key-id"].(string); ok {
+					keyID = kid
+				}
 			}
 			if diskID == "" {
 				if d, ok := mountInfo["device"].(string); ok {
@@ -174,12 +182,14 @@ func (s *proxyService) CreateContainer(ctx context.Context, req *pb.CreateContai
 
 			volKey := fmt.Sprintf("vol-%d", canonicalIdx)
 			cloudVolumes[volKey] = util.CloudVolumeAnnotation{
-				MountPoint: mountDest,
-				FSType:     fsType,
-				LUN:        fmt.Sprintf("%d", canonicalIdx),
-				DiskID:     diskID,
+				MountPoint:  mountDest,
+				FSType:      fsType,
+				LUN:         fmt.Sprintf("%d", canonicalIdx),
+				DiskID:      diskID,
+				EncryptType: encryptType,
+				KeyID:       keyID,
 			}
-			logger.Printf("Detected cloud volume %s -> %s (lun=%d, disk=%s, fs=%s)", volKey, mountDest, canonicalIdx, diskID, fsType)
+			logger.Printf("Detected cloud volume %s -> %s (lun=%d, disk=%s, fs=%s, encrypt=%s)", volKey, mountDest, canonicalIdx, diskID, fsType, encryptType)
 			canonicalIdx++
 		}
 	}

src/cloud-api-adaptor/pkg/forwarder/interceptor/cdh_client.go

@@ -0,0 +1,76 @@
+// (C) Copyright Confidential Containers Contributors
+// SPDX-License-Identifier: Apache-2.0
+
+package interceptor
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	"github.com/containerd/ttrpc"
+
+	"github.com/confidential-containers/cloud-api-adaptor/src/cloud-api-adaptor/pkg/forwarder/interceptor/cdhpb"
+)
+
+const cdhServiceName = "api.SecureMountService"
+
+type cdhClient struct {
+	conn   net.Conn
+	client *ttrpc.Client
+}
+
+func newCDHClient(ctx context.Context, socketPath string) (*cdhClient, error) {
+	const maxAttempts = 10
+	const retryDelay = 2 * time.Second
+
+	dialer := &net.Dialer{Timeout: 5 * time.Second}
+
+	var conn net.Conn
+	var err error
+	for attempt := 1; attempt <= maxAttempts; attempt++ {
+		conn, err = dialer.DialContext(ctx, "unix", socketPath)
+		if err == nil {
+			break
+		}
+		if attempt == maxAttempts {
+			break
+		}
+		logger.Printf("CDH socket %s not ready (attempt %d/%d): %v", socketPath, attempt, maxAttempts, err)
+		select {
+		case <-ctx.Done():
+			return nil, fmt.Errorf("dialing CDH socket %s: %w", socketPath, ctx.Err())
+		case <-time.After(retryDelay):
+		}
+	}
+	if err != nil {
+		return nil, fmt.Errorf("dialing CDH socket %s after %d attempts: %w", socketPath, maxAttempts, err)
+	}
+
+	client := ttrpc.NewClient(conn)
+	return &cdhClient{conn: conn, client: client}, nil
+}
+
+func (c *cdhClient) close() {
+	c.conn.Close()
+	if err := c.client.Close(); err != nil {
+		logger.Printf("WARNING: closing CDH ttrpc client: %v", err)
+	}
+}
+
+func (c *cdhClient) secureMount(ctx context.Context, volumeType string, options map[string]string, flags []string, mountPoint string) error {
+	req := &cdhpb.SecureMountRequest{
+		VolumeType: volumeType,
+		Options:    options,
+		Flags:      flags,
+		MountPoint: mountPoint,
+	}
+	resp := &cdhpb.SecureMountResponse{}
+
+	if err := c.client.Call(ctx, cdhServiceName, "SecureMount", req, resp); err != nil {
+		return fmt.Errorf("CDH SecureMount RPC failed: %w", err)
+	}
+
+	return nil
+}