From 669fed2531ebdf3a1805014e5865050bb9fede14 Mon Sep 17 00:00:00 2001
From: Elijah Rodriguez-Beltran <elijahrb@google.com>
Date: Tue, 11 Aug 2026 22:46:47 +0000
Subject: [PATCH] Adopt pod-identity for authenticated CSI controller
 communication.

CONV=08b81b20-9977-45df-b98e-9e340bc55e2d
---
 .../servicednssigner/servicednssigner.go      |   4 +
 hack/install-ate.sh                           |  22 +-
 hack/install-demo-counter.sh                  |   9 +-
 hack/setup-csi-hostpath-kind.sh               |  40 ++-
 .../deploy/csi-hostpath-testing.yaml          |  96 ++++-
 internal/volume/csi/client.go                 |  20 +-
 internal/volume/csi/client_test.go            |   2 +-
 internal/volume/csi/plugin.go                 |  75 +++-
 internal/volume/csi/plugin_test.go            |  14 +-
 internal/volume/csi/tls_test.go               | 334 ++++++++++++++++++
 .../generated/ate.dev_csidriverconfigs.yaml   |  20 ++
 manifests/ate-install/kind/kustomization.yaml |   1 +
 pkg/api/v1alpha1/csidriverconfig_types.go     |  21 ++
 pkg/api/v1alpha1/zz_generated.deepcopy.go     |  22 +-
 14 files changed, 628 insertions(+), 52 deletions(-)
 create mode 100644 internal/volume/csi/tls_test.go

diff --git a/cmd/podcertcontroller/internal/servicednssigner/servicednssigner.go b/cmd/podcertcontroller/internal/servicednssigner/servicednssigner.go
index 1a995b20d..36f908e71 100644
--- a/cmd/podcertcontroller/internal/servicednssigner/servicednssigner.go
+++ b/cmd/podcertcontroller/internal/servicednssigner/servicednssigner.go
@@ -113,6 +113,10 @@ func (h *Impl) MakeCert(ctx context.Context, pcr *certsv1beta1.PodCertificateReq
 			continue
 		}
 
+		if len(svc.Spec.Selector) == 0 {
+			continue
+		}
+
 		// Find the set of pods that the service selects.
 		matchedPods, err := h.kc.CoreV1().Pods(pcr.ObjectMeta.Namespace).List(ctx, metav1.ListOptions{
 			LabelSelector: metav1.FormatLabelSelector(&metav1.LabelSelector{MatchLabels: svc.Spec.Selector}),
diff --git a/hack/install-ate.sh b/hack/install-ate.sh
index 5ace949c2..f1b13f47b 100755
--- a/hack/install-ate.sh
+++ b/hack/install-ate.sh
@@ -471,14 +471,6 @@ deploy_ate_system() {
   # schemas and RBAC (role.yaml has no other apply path).
   deploy_crds
 
-  if [[ "${SETUP_CSI:-false}" == "true" ]]; then
-    if [[ "${ATE_INSTALL_KIND:-false}" == "true" ]]; then
-      setup_csi
-    else
-      echo "Warning: CSI setup is only supported for Kind local installations. Skipping."
-    fi
-  fi
-
   # Enforce per-class SandboxConfig asset requirements (applied before any
   # SandboxConfig so the defaults below are validated too).
   run_kubectl apply -f manifests/ate-install/sandboxconfig-validation.yaml
@@ -514,6 +506,18 @@ deploy_ate_system() {
     run_kubectl apply -f manifests/ate-install/postgres.yaml
   fi
 
+  # CSI setup must run after podcertificate-controller is ready and trust bundles
+  # exist. The ghostunnel sidecar uses projected podCertificate and clusterTrustBundle
+  # volumes which cannot be fulfilled until podcertcontroller is actively signing,
+  # otherwise rollout of csi-hostpath-socat times out.
+  if [[ "${SETUP_CSI:-false}" == "true" ]]; then
+    if [[ "${ATE_INSTALL_KIND:-false}" == "true" ]]; then
+      setup_csi
+    else
+      echo "Warning: CSI setup is only supported for Kind local installations. Skipping."
+    fi
+  fi
+
   local manifests=""
   manifests="$(render_ate_system_manifests)"
   echo "${manifests}" | run_kubectl apply -f -
@@ -787,7 +791,7 @@ done
 # flag they configure (e.g. --benchmark-worker-count before/after
 # --deploy-benchmarks). The dispatch loop below also accepts these flags but
 # treats them as no-ops since the value is already captured here.
-SETUP_CSI=false
+SETUP_CSI="${SETUP_CSI:-false}"
 BENCHMARK_WORKER_COUNT=1
 BENCHMARK_SANDBOX_CLASS=gvisor
 prescan_args=("$@")
diff --git a/hack/install-demo-counter.sh b/hack/install-demo-counter.sh
index c2248271e..54e7191a9 100644
--- a/hack/install-demo-counter.sh
+++ b/hack/install-demo-counter.sh
@@ -43,9 +43,16 @@ demo-counter_deploy() {
   local ext_vol_mount_cmd=("-e" "/\${EXTERNAL_VOLUME_MOUNTS}/d")
   local ext_vol_spec_cmd=("-e" "/\${EXTERNAL_VOLUMES}/d")
   if [[ "${with_external_volume}" == "true" ]]; then
+    # csi-hostpath-sc only exists when hack/setup-csi-hostpath-kind.sh has run (via SETUP_CSI=true).
+    # Otherwise fall back to the default "standard" StorageClass.
+    local storage_class="standard"
+    if [[ "${SETUP_CSI:-false}" == "true" ]]; then
+      storage_class="csi-hostpath-sc"
+    fi
+
     validate_cmd=("-e" "s|\${VALIDATE_EXISTING_FILE_PATH_ARG}|    - --validate-existing-file-path=/external-data/test.txt|g")
     ext_vol_mount_cmd=("-e" "s|\${EXTERNAL_VOLUME_MOUNTS}|    - name: external-data\n      mountPath: /external-data|g")
-    ext_vol_spec_cmd=("-e" "s|\${EXTERNAL_VOLUMES}|  - name: external-data\n    externalVolumeTemplate:\n      capacity: 1Gi\n      storageClassName: standard|g")
+    ext_vol_spec_cmd=("-e" "s|\${EXTERNAL_VOLUMES}|  - name: external-data\n    externalVolumeTemplate:\n      capacity: 1Gi\n      storageClassName: ${storage_class}|g")
   fi
 
   sed -e "s|\${BUCKET_NAME}|${BUCKET_NAME}|g" \
diff --git a/hack/setup-csi-hostpath-kind.sh b/hack/setup-csi-hostpath-kind.sh
index 74365b394..d2a8af06e 100755
--- a/hack/setup-csi-hostpath-kind.sh
+++ b/hack/setup-csi-hostpath-kind.sh
@@ -54,6 +54,25 @@ else
   echo "Warning: Kind node ${KIND_NODE} not running. Skipping directory cleanup."
 fi
 
+# 2 Expose CSI Controller over TCP Service.
+# This Service must be created before the driver pod deploys for the servicednssigner
+# because servicednssigner to issue a certificate.
+echo "Exposing CSI Controller over TCP Service..."
+cat <<EOF | kubectl apply -f -
+apiVersion: v1
+kind: Service
+metadata:
+  name: csi-hostpath-controller
+  namespace: default
+spec:
+  selector:
+    app.kubernetes.io/name: csi-hostpath-socat
+  ports:
+  - port: 50051
+    targetPort: 10000
+    name: grpc
+EOF
+
 # 3. Deploy the CSI Hostpath Driver
 echo "Deploying CSI Hostpath Driver..."
 # The apply might fail at snapshotclass due to missing CRDs. We catch and ignore this.
@@ -88,23 +107,6 @@ spec:
           type: DirectoryOrCreate
 "
 
-# 5. Create the TCP Service mapping port 50051 to socat port 10000
-echo "Exposing CSI Controller over TCP Service..."
-cat <<EOF | kubectl apply -f -
-apiVersion: v1
-kind: Service
-metadata:
-  name: csi-hostpath-controller
-  namespace: default
-spec:
-  selector:
-    app.kubernetes.io/name: csi-hostpath-socat
-  ports:
-  - port: 50051
-    targetPort: 10000
-    name: grpc
-EOF
-
 # 6. Create the StorageClass
 echo "Creating csi-hostpath-sc StorageClass..."
 kubectl apply -f "${DRIVER_DIR}/examples/csi-storageclass.yaml"
@@ -125,6 +127,10 @@ spec:
   driverName: hostpath.csi.k8s.io
   controllerEndpoint: tcp://csi-hostpath-controller.default.svc.cluster.local:50051
   nodeSocketOverride: unix:///var/lib/kubelet/plugins/csi-hostpath/csi.sock
+  tls:
+    enabled: true
+    usePodIdentity: true
+    serverName: csi-hostpath-controller.default.svc
 EOF
 
 # 9. Restart atelet to recreate image cache directories if they were wiped
diff --git a/hack/third_party/csi-driver-host-path/deploy/csi-hostpath-testing.yaml b/hack/third_party/csi-driver-host-path/deploy/csi-hostpath-testing.yaml
index 1a9d12c7a..ad056b222 100644
--- a/hack/third_party/csi-driver-host-path/deploy/csi-hostpath-testing.yaml
+++ b/hack/third_party/csi-driver-host-path/deploy/csi-hostpath-testing.yaml
@@ -7,6 +7,65 @@
 #
 # This is useful for testing with csi-sanity or csc.
 
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: csi-hostpath-envoy-config
+  namespace: default
+  labels:
+    app.kubernetes.io/instance: hostpath.csi.k8s.io
+    app.kubernetes.io/part-of: csi-driver-host-path
+    app.kubernetes.io/name: csi-hostpath-socat
+    app.kubernetes.io/component: envoy
+data:
+  envoy.yaml: |
+    admin:
+      address:
+        socket_address: { address: 127.0.0.1, port_value: 15000 }
+    static_resources:
+      listeners:
+      - name: csi_tls_listener
+        address:
+          socket_address: { address: 0.0.0.0, port_value: 10000 }
+        filter_chains:
+        - transport_socket:
+            name: envoy.transport_sockets.tls
+            typed_config:
+              "@type": type.googleapis.com/envoy.extensions.transport_sockets.tls.v3.DownstreamTlsContext
+              require_client_certificate: true
+              common_tls_context:
+                alpn_protocols:
+                - h2
+                tls_certificates:
+                - certificate_chain: { filename: /run/servicedns/credential-bundle.pem }
+                  private_key: { filename: /run/servicedns/credential-bundle.pem }
+                  watched_directory: { path: /run/servicedns }
+                validation_context:
+                  trusted_ca: { filename: /run/podidentity-ca/trust-bundle.pem }
+                  watched_directory: { path: /run/podidentity-ca }
+                  match_typed_subject_alt_names:
+                  - san_type: URI
+                    matcher:
+                      exact: "spiffe://cluster.local/ns/ate-system/sa/ate-api-server"
+          filters:
+          - name: envoy.filters.network.tcp_proxy
+            typed_config:
+              "@type": type.googleapis.com/envoy.extensions.filters.network.tcp_proxy.v3.TcpProxy
+              stat_prefix: csi_proxy
+              cluster: csi_unix_socket
+      clusters:
+      - name: csi_unix_socket
+        connect_timeout: 0.25s
+        type: STATIC
+        load_assignment:
+          cluster_name: csi_unix_socket
+          endpoints:
+          - lb_endpoints:
+            - endpoint:
+                address:
+                  pipe:
+                    path: /csi/csi.sock
+---
 apiVersion: v1
 kind: Service
 metadata:
@@ -65,23 +124,50 @@ spec:
                   - hostpath.csi.k8s.io
             topologyKey: kubernetes.io/hostname
       containers:
-        - name: socat
-          image: registry.k8s.io/sig-storage/hostpathplugin:v1.17.1
+        - name: envoy
+          image: envoyproxy/envoy:v1.34-latest
           command:
-          - socat
+            - "/usr/local/bin/envoy"
           args:
-            - tcp-listen:10000,fork,reuseaddr
-            - unix-connect:/csi/csi.sock
+            - "-c"
+            - "/etc/envoy/envoy.yaml"
           securityContext:
             # This is necessary only for systems with SELinux, where
             # non-privileged sidecar containers cannot access unix domain socket
             # created by privileged CSI driver container.
             privileged: true
           volumeMounts:
+          - mountPath: /etc/envoy
+            name: envoy-config
           - mountPath: /csi
             name: socket-dir
+          - mountPath: /run/servicedns
+            name: servicedns-certs
+            readOnly: true
+          - mountPath: /run/podidentity-ca
+            name: podidentity-ca
+            readOnly: true
       volumes:
         - hostPath:
             path: /var/lib/kubelet/plugins/csi-hostpath
             type: DirectoryOrCreate
           name: socket-dir
+        - name: envoy-config
+          configMap:
+            name: csi-hostpath-envoy-config
+        - name: servicedns-certs
+          projected:
+            sources:
+            - podCertificate:
+                signerName: servicedns.podcert.ate.dev/identity
+                keyType: ECDSAP256
+                credentialBundlePath: credential-bundle.pem
+        - name: podidentity-ca
+          projected:
+            sources:
+            - clusterTrustBundle:
+                signerName: podidentity.podcert.ate.dev/identity
+                labelSelector:
+                  matchLabels:
+                    podcert.ate.dev/canarying: live
+                path: trust-bundle.pem
diff --git a/internal/volume/csi/client.go b/internal/volume/csi/client.go
index 7a8c5fcae..19ff30627 100644
--- a/internal/volume/csi/client.go
+++ b/internal/volume/csi/client.go
@@ -15,12 +15,13 @@
 package csi
 
 import (
+	"crypto/tls"
 	"fmt"
-	"log/slog"
 	"net/url"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 )
 
@@ -57,21 +58,20 @@ func parseEndpoint(endpoint string) (string, string, error) {
 
 // NewCSIClient establishes a gRPC connection to the CSI driver over UDS or TCP
 // and returns a client initialized with Identity, Controller, and Node service clients.
-func NewCSIClient(endpoint string) (*Client, error) {
-	scheme, target, err := parseEndpoint(endpoint)
+func NewCSIClient(endpoint string, tlsCfg *tls.Config) (*Client, error) {
+	_, target, err := parseEndpoint(endpoint)
 	if err != nil {
 		return nil, err
 	}
 
-	if scheme == "tcp" {
-		slog.Warn("CSI connection is unencrypted over TCP! (TODO: Implement TLS)", slog.String("endpoint", target))
+	var creds credentials.TransportCredentials
+	if tlsCfg != nil {
+		creds = credentials.NewTLS(tlsCfg)
+	} else {
+		creds = insecure.NewCredentials()
 	}
 
-	dialOpts := []grpc.DialOption{
-		grpc.WithTransportCredentials(insecure.NewCredentials()),
-	}
-
-	conn, err := grpc.NewClient(target, dialOpts...)
+	conn, err := grpc.NewClient(target, grpc.WithTransportCredentials(creds))
 	if err != nil {
 		return nil, fmt.Errorf("failed to dial CSI endpoint %q: %w", target, err)
 	}
diff --git a/internal/volume/csi/client_test.go b/internal/volume/csi/client_test.go
index 521bd3681..1e69af2b6 100644
--- a/internal/volume/csi/client_test.go
+++ b/internal/volume/csi/client_test.go
@@ -138,7 +138,7 @@ func TestNewCSIClient_TCP(t *testing.T) {
 
 	// Connect to the mock server using NewCSIClient
 	endpoint := "tcp://" + lis.Addr().String()
-	client, err := NewCSIClient(endpoint)
+	client, err := NewCSIClient(endpoint, nil)
 	if err != nil {
 		t.Fatalf("failed to create CSI client: %v", err)
 	}
diff --git a/internal/volume/csi/plugin.go b/internal/volume/csi/plugin.go
index 53b3fb3ec..e8cd9a844 100644
--- a/internal/volume/csi/plugin.go
+++ b/internal/volume/csi/plugin.go
@@ -16,13 +16,17 @@ package csi
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"fmt"
 	"log/slog"
 	"os"
 	"path/filepath"
 
 	"github.com/agent-substrate/substrate/internal/ateompath"
+	"github.com/agent-substrate/substrate/internal/credbundle"
 	"github.com/agent-substrate/substrate/internal/volume"
+	v1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
 	listersv1alpha1 "github.com/agent-substrate/substrate/pkg/client/listers/api/v1alpha1"
 	"github.com/container-storage-interface/spec/lib/go/csi"
 	"google.golang.org/grpc/codes"
@@ -30,6 +34,23 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 )
 
+const (
+	// DefaultClientCertPath is the path to the pod identity credential bundle.
+	DefaultClientCertPath = "/run/podidentity.podcert.ate.dev/credential-bundle.pem"
+	// DefaultCACertPath is the path to the servicedns trust bundle.
+	DefaultCACertPath = "/run/servicedns.podcert.ate.dev/trust-bundle.pem"
+)
+
+type tlsPaths struct {
+	clientCert string
+	caCert     string
+}
+
+var defaultTLSPaths = tlsPaths{
+	clientCert: DefaultClientCertPath,
+	caCert:     DefaultCACertPath,
+}
+
 // Plugin implements volume.VolumePluginWorkerPlane using the CSI Client.
 type Plugin struct {
 	client           *Client
@@ -250,6 +271,10 @@ func getStandardCapabilities() []*csi.VolumeCapability {
 
 // NewCSIPlugin establishes a CSI client and returns a verified Plugin instance.
 func NewCSIPlugin(ctx context.Context, lister listersv1alpha1.CSIDriverConfigLister, driverName string, isController bool) (*Plugin, error) {
+	return newCSIPlugin(ctx, lister, driverName, isController, defaultTLSPaths)
+}
+
+func newCSIPlugin(ctx context.Context, lister listersv1alpha1.CSIDriverConfigLister, driverName string, isController bool, paths tlsPaths) (*Plugin, error) {
 	if lister == nil {
 		return nil, fmt.Errorf("missing csiDriverConfigLister")
 	}
@@ -269,7 +294,17 @@ func NewCSIPlugin(ctx context.Context, lister listersv1alpha1.CSIDriverConfigLis
 	default:
 		endpoint = "unix://" + ateompath.KubeletPluginSocketPath(driverName)
 	}
-	csiClient, err := NewCSIClient(endpoint)
+
+	var tlsCfg *tls.Config
+	if isController {
+		var err error
+		tlsCfg, err = resolveTLSConfig(cfg, paths)
+		if err != nil {
+			return nil, fmt.Errorf("failed to resolve TLS config for %q: %w", driverName, err)
+		}
+	}
+
+	csiClient, err := NewCSIClient(endpoint, tlsCfg)
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize CSI client from endpoint %q: %w", endpoint, err)
 	}
@@ -288,3 +323,41 @@ func NewCSIPlugin(ctx context.Context, lister listersv1alpha1.CSIDriverConfigLis
 
 	return csiPlugin, nil
 }
+
+func resolveTLSConfig(cfg *v1alpha1.CSIDriverConfig, paths tlsPaths) (*tls.Config, error) {
+	if cfg == nil || cfg.Spec.TLS == nil || !cfg.Spec.TLS.Enabled {
+		return nil, nil
+	}
+
+	tlsCfg := cfg.Spec.TLS
+
+	if !tlsCfg.UsePodIdentity {
+		// TODO: Support manual certificates loaded from Secrets specified in the config.
+		return nil, fmt.Errorf("only pod identity TLS is supported in this configuration")
+	}
+
+	// Load CA Pool
+	caPoolBytes, err := os.ReadFile(paths.caCert)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read CA cert from %q: %w", paths.caCert, err)
+	}
+	caPool := x509.NewCertPool()
+	if !caPool.AppendCertsFromPEM(caPoolBytes) {
+		return nil, fmt.Errorf("failed to parse CA certs from %q", paths.caCert)
+	}
+
+	return &tls.Config{
+		MinVersion: tls.VersionTLS13,
+		RootCAs:    caPool,
+		ServerName: tlsCfg.ServerName,
+		// NextProtos configures ALPN h2 for gRPC over TLS.
+		NextProtos: []string{"h2"},
+		// Load Client Certificate dynamically.
+		// In Substrate's Pod Identity model, certificates are projected into the pod
+		// as files by the kubelet (via Substrate's podcertcontroller).
+		// Kubelet handles the rotation of these files on disk.
+		// credbundle.ClientLoader monitors these files and automatically reloads
+		// them when they change, ensuring rotation is picked up on subsequent handshakes.
+		GetClientCertificate: credbundle.ClientLoader(paths.clientCert),
+	}, nil
+}
diff --git a/internal/volume/csi/plugin_test.go b/internal/volume/csi/plugin_test.go
index 6f50d9540..79b5a75d9 100644
--- a/internal/volume/csi/plugin_test.go
+++ b/internal/volume/csi/plugin_test.go
@@ -175,7 +175,7 @@ func TestPlugin_CreateVolume(t *testing.T) {
 	endpoint, cleanup := startMockCSIDriver(t, driver)
 	defer cleanup()
 
-	client, err := NewCSIClient(endpoint)
+	client, err := NewCSIClient(endpoint, nil)
 	if err != nil {
 		t.Fatalf("failed to create CSI client: %v", err)
 	}
@@ -199,7 +199,7 @@ func TestPlugin_DeleteVolume(t *testing.T) {
 	endpoint, cleanup := startMockCSIDriver(t, driver)
 	defer cleanup()
 
-	client, err := NewCSIClient(endpoint)
+	client, err := NewCSIClient(endpoint, nil)
 	if err != nil {
 		t.Fatalf("failed to create CSI client: %v", err)
 	}
@@ -219,7 +219,7 @@ func TestPlugin_AttachVolume(t *testing.T) {
 	endpoint, cleanup := startMockCSIDriver(t, driver)
 	defer cleanup()
 
-	client, err := NewCSIClient(endpoint)
+	client, err := NewCSIClient(endpoint, nil)
 	if err != nil {
 		t.Fatalf("failed to create CSI client: %v", err)
 	}
@@ -248,7 +248,7 @@ func TestPlugin_DetachVolume(t *testing.T) {
 	endpoint, cleanup := startMockCSIDriver(t, driver)
 	defer cleanup()
 
-	client, err := NewCSIClient(endpoint)
+	client, err := NewCSIClient(endpoint, nil)
 	if err != nil {
 		t.Fatalf("failed to create CSI client: %v", err)
 	}
@@ -277,7 +277,7 @@ func TestPlugin_MountVolume(t *testing.T) {
 	endpoint, cleanup := startMockCSIDriver(t, driver)
 	defer cleanup()
 
-	client, err := NewCSIClient(endpoint)
+	client, err := NewCSIClient(endpoint, nil)
 	if err != nil {
 		t.Fatalf("failed to create CSI client: %v", err)
 	}
@@ -324,7 +324,7 @@ func TestPlugin_UnmountVolume(t *testing.T) {
 	endpoint, cleanup := startMockCSIDriver(t, driver)
 	defer cleanup()
 
-	client, err := NewCSIClient(endpoint)
+	client, err := NewCSIClient(endpoint, nil)
 	if err != nil {
 		t.Fatalf("failed to create CSI client: %v", err)
 	}
@@ -376,7 +376,7 @@ func TestClient_Identity(t *testing.T) {
 	endpoint, cleanup := startMockCSIDriver(t, driver)
 	defer cleanup()
 
-	client, err := NewCSIClient(endpoint)
+	client, err := NewCSIClient(endpoint, nil)
 	if err != nil {
 		t.Fatalf("failed to create CSI client: %v", err)
 	}
diff --git a/internal/volume/csi/tls_test.go b/internal/volume/csi/tls_test.go
new file mode 100644
index 000000000..5c24ed78d
--- /dev/null
+++ b/internal/volume/csi/tls_test.go
@@ -0,0 +1,334 @@
+// Copyright 2026 Google LLC
+//
+// 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 csi
+
+import (
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"fmt"
+	"math/big"
+	"net"
+	"os"
+	"path/filepath"
+	"slices"
+	"testing"
+	"time"
+
+	"github.com/agent-substrate/substrate/pkg/api/v1alpha1"
+	listersv1alpha1 "github.com/agent-substrate/substrate/pkg/client/listers/api/v1alpha1"
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"k8s.io/apimachinery/pkg/labels"
+)
+
+const mockDriverName = "mock-driver"
+
+// testCA is a self-signed CA that issues the server and client certificates for
+// one test.
+type testCA struct {
+	cert *x509.Certificate
+	key  *ecdsa.PrivateKey
+}
+
+func newTestCA(t *testing.T) *testCA {
+	t.Helper()
+	key := newKey(t)
+	tmpl := &x509.Certificate{
+		SerialNumber:          big.NewInt(1),
+		Subject:               pkix.Name{CommonName: "test-ca"},
+		NotBefore:             time.Now().Add(-time.Hour),
+		NotAfter:              time.Now().Add(time.Hour),
+		IsCA:                  true,
+		KeyUsage:              x509.KeyUsageCertSign,
+		BasicConstraintsValid: true,
+	}
+	cert, err := x509.ParseCertificate(createCert(t, tmpl, tmpl, &key.PublicKey, key))
+	if err != nil {
+		t.Fatalf("parse CA certificate: %v", err)
+	}
+	return &testCA{cert: cert, key: key}
+}
+
+// certPEM returns the CA certificate as a trust bundle would hold it.
+func (ca *testCA) certPEM() []byte {
+	return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: ca.cert.Raw})
+}
+
+// pool returns a CertPool trusting this CA and nothing else.
+func (ca *testCA) pool() *x509.CertPool {
+	pool := x509.NewCertPool()
+	pool.AppendCertsFromPEM(ca.certPEM())
+	return pool
+}
+
+// serverCert issues a serving certificate valid for dnsName.
+func (ca *testCA) serverCert(t *testing.T, dnsName string) tls.Certificate {
+	t.Helper()
+	key := newKey(t)
+	der := createCert(t, &x509.Certificate{
+		SerialNumber: big.NewInt(2),
+		Subject:      pkix.Name{CommonName: "test-server"},
+		NotBefore:    time.Now().Add(-time.Hour),
+		NotAfter:     time.Now().Add(time.Hour),
+		DNSNames:     []string{dnsName},
+		KeyUsage:     x509.KeyUsageDigitalSignature,
+		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+	}, ca.cert, &key.PublicKey, ca.key)
+	return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key}
+}
+
+// clientBundle issues a client certificate in credential-bundle layout: a PKCS#8
+// PRIVATE KEY block followed by the CERTIFICATE block
+func (ca *testCA) clientBundle(t *testing.T) []byte {
+	t.Helper()
+	key := newKey(t)
+	der := createCert(t, &x509.Certificate{
+		SerialNumber: big.NewInt(3),
+		Subject:      pkix.Name{CommonName: "test-client"},
+		NotBefore:    time.Now().Add(-time.Hour),
+		NotAfter:     time.Now().Add(time.Hour),
+		KeyUsage:     x509.KeyUsageDigitalSignature,
+		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth},
+	}, ca.cert, &key.PublicKey, ca.key)
+
+	keyDER, err := x509.MarshalPKCS8PrivateKey(key)
+	if err != nil {
+		t.Fatalf("marshal client key: %v", err)
+	}
+	return append(
+		pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER}),
+		pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})...,
+	)
+}
+
+func newKey(t *testing.T) *ecdsa.PrivateKey {
+	t.Helper()
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		t.Fatalf("generate key: %v", err)
+	}
+	return key
+}
+
+func createCert(t *testing.T, tmpl, parent *x509.Certificate, pub *ecdsa.PublicKey, signer *ecdsa.PrivateKey) []byte {
+	t.Helper()
+	der, err := x509.CreateCertificate(rand.Reader, tmpl, parent, pub, signer)
+	if err != nil {
+		t.Fatalf("create certificate %q: %v", tmpl.Subject.CommonName, err)
+	}
+	return der
+}
+
+// writeCreds lays out a credential bundle and a trust bundle on disk the way the
+// projected pod-identity volumes do, and returns the paths to read them from.
+func writeCreds(t *testing.T, clientBundle, trustBundle []byte) tlsPaths {
+	t.Helper()
+	dir := t.TempDir()
+	paths := tlsPaths{
+		clientCert: filepath.Join(dir, "credential-bundle.pem"),
+		caCert:     filepath.Join(dir, "trust-bundle.pem"),
+	}
+	writeFile(t, paths.clientCert, clientBundle)
+	writeFile(t, paths.caCert, trustBundle)
+	return paths
+}
+
+func writeFile(t *testing.T, path string, data []byte) {
+	t.Helper()
+	if err := os.WriteFile(path, data, 0600); err != nil {
+		t.Fatalf("write %s: %v", path, err)
+	}
+}
+
+// startTLSServer serves the CSI Identity service over mTLS and returns its address.
+func startTLSServer(t *testing.T, serverCert tls.Certificate, clientCAs *x509.CertPool) string {
+	t.Helper()
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	srv := grpc.NewServer(grpc.Creds(credentials.NewTLS(&tls.Config{
+		Certificates: []tls.Certificate{serverCert},
+		ClientAuth:   tls.RequireAndVerifyClientCert,
+		ClientCAs:    clientCAs,
+		MinVersion:   tls.VersionTLS13,
+	})))
+	csi.RegisterIdentityServer(srv, &mockCSIDriver{})
+	go func() { _ = srv.Serve(lis) }()
+
+	t.Cleanup(srv.Stop)
+	return lis.Addr().String()
+}
+
+func driverConfig(endpoint string, tlsCfg *v1alpha1.CSIDriverTLSConfig) *v1alpha1.CSIDriverConfig {
+	return &v1alpha1.CSIDriverConfig{
+		Spec: v1alpha1.CSIDriverConfigSpec{
+			DriverName:         mockDriverName,
+			ControllerEndpoint: endpoint,
+			TLS:                tlsCfg,
+		},
+	}
+}
+
+// dialPlugin builds a controller plugin pointed at addr. newCSIPlugin issues a
+// GetPluginInfo RPC before returning, so a nil error means the mTLS handshake
+// and a real call both succeeded.
+func dialPlugin(t *testing.T, addr, serverName string, paths tlsPaths) (*Plugin, error) {
+	t.Helper()
+	cfg := driverConfig("tcp://"+addr, &v1alpha1.CSIDriverTLSConfig{
+		Enabled:        true,
+		UsePodIdentity: true,
+		ServerName:     serverName,
+	})
+	return newCSIPlugin(t.Context(), &mockLister{cfg: cfg}, mockDriverName, true /*isController*/, paths)
+}
+
+// mockLister serves a single CSIDriverConfig under mockDriverName.
+type mockLister struct{ cfg *v1alpha1.CSIDriverConfig }
+
+func (m *mockLister) List(labels.Selector) ([]*v1alpha1.CSIDriverConfig, error) {
+	return []*v1alpha1.CSIDriverConfig{m.cfg}, nil
+}
+
+func (m *mockLister) Get(name string) (*v1alpha1.CSIDriverConfig, error) {
+	if name != mockDriverName {
+		return nil, fmt.Errorf("no CSIDriverConfig named %q", name)
+	}
+	return m.cfg, nil
+}
+
+var _ listersv1alpha1.CSIDriverConfigLister = (*mockLister)(nil)
+
+func TestMTLSSucceeds(t *testing.T) {
+	t.Parallel()
+	ca := newTestCA(t)
+	addr := startTLSServer(t, ca.serverCert(t, "localhost"), ca.pool())
+	paths := writeCreds(t, ca.clientBundle(t), ca.certPEM())
+
+	plugin, err := dialPlugin(t, addr, "localhost", paths)
+	if err != nil {
+		t.Fatalf("newCSIPlugin over mTLS: %v", err)
+	}
+	t.Cleanup(func() { plugin.client.Close() })
+}
+
+// The server presents a certificate from a CA the client does not trust.
+func TestMTLSRejectsUntrustedServerCA(t *testing.T) {
+	t.Parallel()
+	serverCA, unauthClientCA := newTestCA(t), newTestCA(t)
+	addr := startTLSServer(t, serverCA.serverCert(t, "localhost"), unauthClientCA.pool())
+	paths := writeCreds(t, unauthClientCA.clientBundle(t), unauthClientCA.certPEM())
+
+	if _, err := dialPlugin(t, addr, "localhost", paths); err == nil {
+		t.Fatal("newCSIPlugin accepted a server certificate from an untrusted CA")
+	}
+}
+
+// The server's certificate is trusted but was issued for a different name.
+func TestMTLSRejectsServerNameMismatch(t *testing.T) {
+	t.Parallel()
+	ca := newTestCA(t)
+	addr := startTLSServer(t, ca.serverCert(t, "localhost"), ca.pool())
+	paths := writeCreds(t, ca.clientBundle(t), ca.certPEM())
+
+	if _, err := dialPlugin(t, addr, "wrong.example", paths); err == nil {
+		t.Fatal("newCSIPlugin accepted a server certificate issued for another name")
+	}
+}
+
+// The client presents a certificate from a CA the server does not accept, which
+// is what proves the connection is mutually authenticated and not one-way TLS.
+func TestMTLSRejectsUntrustedClientCert(t *testing.T) {
+	t.Parallel()
+	ca, otherCA := newTestCA(t), newTestCA(t)
+	addr := startTLSServer(t, ca.serverCert(t, "localhost"), otherCA.pool())
+	paths := writeCreds(t, ca.clientBundle(t), ca.certPEM())
+
+	if _, err := dialPlugin(t, addr, "localhost", paths); err == nil {
+		t.Fatal("newCSIPlugin succeeded with a client certificate the server should reject")
+	}
+}
+
+func TestResolveTLSConfigDisabled(t *testing.T) {
+	t.Parallel()
+	for _, tc := range []struct {
+		name string
+		cfg  *v1alpha1.CSIDriverConfig
+	}{
+		{"nil config", nil},
+		{"no TLS block", driverConfig("", nil)},
+		{"TLS disabled", driverConfig("", &v1alpha1.CSIDriverTLSConfig{Enabled: false})},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := resolveTLSConfig(tc.cfg, defaultTLSPaths)
+			if err != nil {
+				t.Fatalf("resolveTLSConfig: %v", err)
+			}
+			if got != nil {
+				t.Errorf("resolveTLSConfig = %v, want nil", got)
+			}
+		})
+	}
+}
+
+func TestResolveTLSConfigRejectsManualCerts(t *testing.T) {
+	t.Parallel()
+	cfg := driverConfig("", &v1alpha1.CSIDriverTLSConfig{Enabled: true, UsePodIdentity: false})
+
+	if _, err := resolveTLSConfig(cfg, defaultTLSPaths); err == nil {
+		t.Fatal("resolveTLSConfig accepted usePodIdentity=false; manual certificates are unsupported")
+	}
+}
+
+func TestResolveTLSConfigFields(t *testing.T) {
+	t.Parallel()
+	const serverName = "my-service.default.svc"
+	ca := newTestCA(t)
+	paths := writeCreds(t, ca.clientBundle(t), ca.certPEM())
+
+	got, err := resolveTLSConfig(driverConfig("", &v1alpha1.CSIDriverTLSConfig{
+		Enabled:        true,
+		UsePodIdentity: true,
+		ServerName:     serverName,
+	}), paths)
+	if err != nil {
+		t.Fatalf("resolveTLSConfig: %v", err)
+	}
+
+	if got.MinVersion != tls.VersionTLS13 {
+		t.Errorf("MinVersion = %d, want %d", got.MinVersion, tls.VersionTLS13)
+	}
+	if got.InsecureSkipVerify {
+		t.Error("InsecureSkipVerify = true, want false")
+	}
+	if got.RootCAs == nil {
+		t.Error("RootCAs = nil, want the trust bundle")
+	}
+	if got.GetClientCertificate == nil {
+		t.Error("GetClientCertificate = nil, want the credential-bundle loader")
+	}
+	if got.ServerName != serverName {
+		t.Errorf("ServerName = %q, want %q", got.ServerName, serverName)
+	}
+	if !slices.Equal(got.NextProtos, []string{"h2"}) {
+		t.Errorf("NextProtos = %v, want [h2]", got.NextProtos)
+	}
+}
diff --git a/manifests/ate-install/generated/ate.dev_csidriverconfigs.yaml b/manifests/ate-install/generated/ate.dev_csidriverconfigs.yaml
index 16885bf03..31bdc43f8 100644
--- a/manifests/ate-install/generated/ate.dev_csidriverconfigs.yaml
+++ b/manifests/ate-install/generated/ate.dev_csidriverconfigs.yaml
@@ -82,6 +82,26 @@ spec:
                   on the worker nodes. If empty, ATE defaults to unix:///var/lib/kubelet/plugins/[DriverName]/csi.sock.
                 pattern: ^unix://.+$
                 type: string
+              tls:
+                description: TLS configures TLS/mTLS for the connection to the ControllerEndpoint.
+                properties:
+                  enabled:
+                    description: Enabled controls whether TLS is used.
+                    type: boolean
+                  serverName:
+                    description: ServerName override for TLS verification.
+                    type: string
+                  usePodIdentity:
+                    description: UsePodIdentity indicates whether to reuse Substrate's
+                      Pod Identity (SPIFFE) certificates.
+                    type: boolean
+                required:
+                - enabled
+                type: object
+                x-kubernetes-validations:
+                - message: tls.usePodIdentity must be true when tls.enabled is true;
+                    manual certificates are not yet supported
+                  rule: '!self.enabled || (has(self.usePodIdentity) && self.usePodIdentity)'
             required:
             - controllerEndpoint
             - driverName
diff --git a/manifests/ate-install/kind/kustomization.yaml b/manifests/ate-install/kind/kustomization.yaml
index c6f9e5e51..b853a0308 100644
--- a/manifests/ate-install/kind/kustomization.yaml
+++ b/manifests/ate-install/kind/kustomization.yaml
@@ -56,3 +56,4 @@ patches:
               env:
               - name: OTEL_TRACES_SAMPLER
                 value: parentbased_always_on
+
diff --git a/pkg/api/v1alpha1/csidriverconfig_types.go b/pkg/api/v1alpha1/csidriverconfig_types.go
index 857a52f8a..02d095245 100644
--- a/pkg/api/v1alpha1/csidriverconfig_types.go
+++ b/pkg/api/v1alpha1/csidriverconfig_types.go
@@ -43,6 +43,27 @@ type CSIDriverConfigSpec struct {
 	// +optional
 	// +kubebuilder:validation:Pattern=`^unix://.+$`
 	NodeSocketOverride string `json:"nodeSocketOverride,omitempty"`
+
+	// TLS configures TLS/mTLS for the connection to the ControllerEndpoint.
+	// +optional
+	TLS *CSIDriverTLSConfig `json:"tls,omitempty"`
+}
+
+// CSIDriverTLSConfig holds TLS and mTLS configuration for CSI driver connections.
+// +kubebuilder:validation:XValidation:rule="!self.enabled || (has(self.usePodIdentity) && self.usePodIdentity)",message="tls.usePodIdentity must be true when tls.enabled is true; manual certificates are not yet supported"
+type CSIDriverTLSConfig struct {
+	// Enabled controls whether TLS is used.
+	// +required
+	Enabled bool `json:"enabled"`
+
+	// TODO: Add alternative support for manual certs by adding SecretReference fields.
+	// UsePodIdentity indicates whether to reuse Substrate's Pod Identity (SPIFFE) certificates.
+	// +optional
+	UsePodIdentity bool `json:"usePodIdentity,omitempty"`
+
+	// ServerName override for TLS verification.
+	// +optional
+	ServerName string `json:"serverName,omitempty"`
 }
 
 // CSIDriverConfig is the Schema for the csidriverconfigs API
diff --git a/pkg/api/v1alpha1/zz_generated.deepcopy.go b/pkg/api/v1alpha1/zz_generated.deepcopy.go
index 8b54a4259..33827fe96 100644
--- a/pkg/api/v1alpha1/zz_generated.deepcopy.go
+++ b/pkg/api/v1alpha1/zz_generated.deepcopy.go
@@ -166,7 +166,7 @@ func (in *CSIDriverConfig) DeepCopyInto(out *CSIDriverConfig) {
 	*out = *in
 	out.TypeMeta = in.TypeMeta
 	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
-	out.Spec = in.Spec
+	in.Spec.DeepCopyInto(&out.Spec)
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CSIDriverConfig.
@@ -222,6 +222,11 @@ func (in *CSIDriverConfigList) DeepCopyObject() runtime.Object {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *CSIDriverConfigSpec) DeepCopyInto(out *CSIDriverConfigSpec) {
 	*out = *in
+	if in.TLS != nil {
+		in, out := &in.TLS, &out.TLS
+		*out = new(CSIDriverTLSConfig)
+		**out = **in
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CSIDriverConfigSpec.
@@ -234,6 +239,21 @@ func (in *CSIDriverConfigSpec) DeepCopy() *CSIDriverConfigSpec {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *CSIDriverTLSConfig) DeepCopyInto(out *CSIDriverTLSConfig) {
+	*out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CSIDriverTLSConfig.
+func (in *CSIDriverTLSConfig) DeepCopy() *CSIDriverTLSConfig {
+	if in == nil {
+		return nil
+	}
+	out := new(CSIDriverTLSConfig)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Container) DeepCopyInto(out *Container) {
 	*out = *in