Merge pull request #2071 from andyzhangx/wi-token-refresh

fix: workload identity token refresh issue

Author: andyzhangx

charts/latest/blob-csi-driver-v0.0.0.tgz

@@ -12,5 +12,7 @@ spec:
   volumeLifecycleModes:
     - Persistent
     - Ephemeral
+  requiresRepublish: {{ .Values.feature.requiresRepublish }}
   tokenRequests:
     - audience: api://AzureADTokenExchange
+      expirationSeconds: 3600

charts/latest/blob-csi-driver/templates/csi-blob-driver.yaml

@@ -165,6 +165,7 @@ node:
 
 feature:
   fsGroupPolicy: ReadWriteOnceWithFSType
+  requiresRepublish: true
   enableGetVolumeStats: false
 
 driver:

charts/latest/blob-csi-driver/values.yaml

@@ -10,5 +10,7 @@ spec:
   volumeLifecycleModes:
     - Persistent
     - Ephemeral
+  requiresRepublish: true
   tokenRequests:
     - audience: api://AzureADTokenExchange
+      expirationSeconds: 3600

deploy/csi-blob-driver.yaml

@@ -23,6 +23,7 @@ import (
 	"flag"
 	"fmt"
 	"os"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"sync"
@@ -161,7 +162,6 @@ const (
 	tagValueDelimiterField = "tagvaluedelimiter"
 
 	DefaultTokenAudience = "api://AzureADTokenExchange" //nolint:gosec // G101 ignore this!
-
 )
 
 var (
@@ -171,7 +171,8 @@ var (
 
 	// azcopyCloneVolumeOptions used in volume cloning between different storage account and --check-length to false because volume data may be in changing state, copy volume is not same as current source volume,
 	// set --s2s-preserve-access-tier=false to avoid BlobAccessTierNotSupportedForAccountType error in azcopy
-	azcopyCloneVolumeOptions = []string{"--recursive", "--check-length=false", "--s2s-preserve-access-tier=false", "--log-level=ERROR"}
+	azcopyCloneVolumeOptions  = []string{"--recursive", "--check-length=false", "--s2s-preserve-access-tier=false", "--log-level=ERROR"}
+	defaultAzureOAuthTokenDir = "/var/lib/kubelet/plugins/" + DefaultDriverName
 )
 
 // DriverOptions defines driver parameters specified in driver deployment
@@ -587,13 +588,17 @@ func (d *Driver) GetAuthEnv(ctx context.Context, volumeID, protocol string, attr
 			if err != nil {
 				return rgName, accountName, accountKey, containerName, authEnv, err
 			}
+			azureOAuthTokenFile := filepath.Join(defaultAzureOAuthTokenDir, clientID+accountName)
+			if err := os.WriteFile(azureOAuthTokenFile, []byte(workloadIdentityToken), 0600); err != nil {
+				return rgName, accountName, accountKey, containerName, authEnv, fmt.Errorf("failed to write workload identity token file %s: %v", azureOAuthTokenFile, err)
+			}
 
 			authEnv = append(authEnv, "AZURE_STORAGE_SPN_CLIENT_ID="+clientID)
 			if tenantID != "" {
 				authEnv = append(authEnv, "AZURE_STORAGE_SPN_TENANT_ID="+tenantID)
 			}
-			authEnv = append(authEnv, "WORKLOAD_IDENTITY_TOKEN="+workloadIdentityToken)
-
+			authEnv = append(authEnv, "AZURE_OAUTH_TOKEN_FILE="+azureOAuthTokenFile)
+			klog.V(2).Infof("workload identity auth: %v", authEnv)
 			return rgName, accountName, accountKey, containerName, authEnv, err
 		}
 		klog.V(2).Infof("clientID(%s) is specified, use service account token to get account key", clientID)

pkg/blob/blob.go

@@ -266,12 +266,6 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 		return &csi.NodeStageVolumeResponse{}, nil
 	}
 
-	mc := metrics.NewMetricContext(blobCSIDriverName, "node_stage_volume", d.cloud.ResourceGroup, "", d.Name)
-	isOperationSucceeded := false
-	defer func() {
-		mc.ObserveOperationWithResult(isOperationSucceeded, VolumeID, volumeID)
-	}()
-
 	var serverAddress, storageEndpointSuffix, protocol, ephemeralVolMountOptions string
 	var ephemeralVol, isHnsEnabled bool
 
@@ -327,16 +321,22 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "Could not mount target %q: %v", targetPath, err)
 	}
+
+	_, accountName, _, containerName, authEnv, err := d.GetAuthEnv(ctx, volumeID, protocol, attrib, secrets)
+	if err != nil && !mnt {
+		return nil, status.Errorf(codes.Internal, "%v", err)
+	}
+
 	if mnt {
 		klog.V(2).Infof("NodeStageVolume: volume %s is already mounted on %s", volumeID, targetPath)
-		isOperationSucceeded = true
 		return &csi.NodeStageVolumeResponse{}, nil
 	}
 
-	_, accountName, _, containerName, authEnv, err := d.GetAuthEnv(ctx, volumeID, protocol, attrib, secrets)
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "%v", err)
-	}
+	mc := metrics.NewMetricContext(blobCSIDriverName, "node_stage_volume", d.cloud.ResourceGroup, "", d.Name)
+	isOperationSucceeded := false
+	defer func() {
+		mc.ObserveOperationWithResult(isOperationSucceeded, VolumeID, volumeID)
+	}()
 
 	// replace pv/pvc name namespace metadata in subDir
 	containerName = replaceWithMap(containerName, containerNameReplaceMap)

pkg/blob/nodeserver.go

@@ -243,6 +243,12 @@ func TestNodePublishVolume(t *testing.T) {
 				// Mock NodeStageVolume to return success
 				d.cloud.ResourceGroup = "rg"
 				d.enableBlobMockMount = true
+				// Create the directory for token file
+				defaultAzureOAuthTokenDir = "./blob.csi.azure.com/"
+				_ = makeDir(defaultAzureOAuthTokenDir)
+			},
+			cleanup: func(_ *Driver) {
+				_ = os.RemoveAll(defaultAzureOAuthTokenDir)
 			},
 			req: &csi.NodePublishVolumeRequest{
 				VolumeCapability:  &csi.VolumeCapability{AccessMode: &volumeCap},
@@ -253,6 +259,7 @@ func TestNodePublishVolume(t *testing.T) {
 					serviceAccountTokenField: `{"api://AzureADTokenExchange":{"token":"test-token","expirationTimestamp":"2023-01-01T00:00:00Z"}}`,
 					mountWithWITokenField:    "true",
 					clientIDField:            "client-id-value",
+					storageAccountNameField:  "test-account",
 				},
 			},
 			expectedErr: nil,