fix(common/redis): re-AUTH new pool connections with fresh Entra token

common/component/redis/entraid_onconnect_test.go

@@ -0,0 +1,208 @@
+/*
+Copyright 2021 The Dapr 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 redis
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	v8 "github.com/go-redis/redis/v8"
+	v9 "github.com/redis/go-redis/v9"
+)
+
+// fakeTokenCredential is a stub azcore.TokenCredential for exercising the
+// Entra ID auth path without contacting Azure.
+type fakeTokenCredential struct {
+	token string
+	err   error
+}
+
+func (f fakeTokenCredential) GetToken(context.Context, policy.TokenRequestOptions) (azcore.AccessToken, error) {
+	if f.err != nil {
+		return azcore.AccessToken{}, f.err
+	}
+	return azcore.AccessToken{Token: f.token, ExpiresOn: time.Now().Add(time.Hour)}, nil
+}
+
+// entraIDSettings returns Settings wired as InitEntraIDCredential would leave
+// them after a successful init, but with a stubbed credential.
+func entraIDSettings(redisType string, failover bool) *Settings {
+	return &Settings{
+		Host:                   "localhost:6379",
+		RedisType:              redisType,
+		Failover:               failover,
+		UseEntraID:             true,
+		entraIDUsername:        "00000000-0000-0000-0000-000000000000",
+		entraIDTokenCredential: fakeTokenCredential{token: "fresh-token"},
+	}
+}
+
+// v8OnConnectAndPassword extracts the OnConnect callback and Password configured
+// on the underlying go-redis v8 client, across the node/cluster variants.
+func v8OnConnectAndPassword(t *testing.T, c RedisClient) (onConnect any, password string) {
+	t.Helper()
+	switch client := c.(v8Client).client.(type) {
+	case *v8.Client:
+		return client.Options().OnConnect, client.Options().Password
+	case *v8.ClusterClient:
+		return client.Options().OnConnect, client.Options().Password
+	default:
+		t.Fatalf("unexpected v8 client type %T", client)
+		return nil, ""
+	}
+}
+
+// v9OnConnectAndPassword extracts the OnConnect callback and Password configured
+// on the underlying go-redis v9 client, across the node/cluster variants.
+func v9OnConnectAndPassword(t *testing.T, c RedisClient) (onConnect any, password string) {
+	t.Helper()
+	switch client := c.(v9Client).client.(type) {
+	case *v9.Client:
+		return client.Options().OnConnect, client.Options().Password
+	case *v9.ClusterClient:
+		return client.Options().OnConnect, client.Options().Password
+	default:
+		t.Fatalf("unexpected v9 client type %T", client)
+		return nil, ""
+	}
+}
+
+func TestEntraIDOnConnectWiring(t *testing.T) {
+	cases := []struct {
+		name      string
+		redisType string
+		failover  bool
+		newV8     func(*Settings) (RedisClient, error)
+		newV9     func(*Settings) (RedisClient, error)
+	}{
+		{name: "node", redisType: NodeType, newV8: newV8Client, newV9: newV9Client},
+		{name: "cluster", redisType: ClusterType, newV8: newV8Client, newV9: newV9Client},
+		{name: "failover", redisType: NodeType, failover: true, newV8: newV8FailoverClient, newV9: newV9FailoverClient},
+		{name: "failover-cluster", redisType: ClusterType, failover: true, newV8: newV8FailoverClient, newV9: newV9FailoverClient},
+	}
+
+	for _, tc := range cases {
+		t.Run("v8/"+tc.name+"/entraID-enabled", func(t *testing.T) {
+			c, err := tc.newV8(entraIDSettings(tc.redisType, tc.failover))
+			require.NoError(t, err)
+			require.NotNil(t, c)
+			defer c.Close()
+
+			onConnect, password := v8OnConnectAndPassword(t, c)
+			assert.NotNil(t, onConnect, "OnConnect must be wired when UseEntraID is true")
+			assert.Empty(t, password, "static Password must not be set when UseEntraID is true")
+		})
+
+		t.Run("v8/"+tc.name+"/entraID-disabled", func(t *testing.T) {
+			s := entraIDSettings(tc.redisType, tc.failover)
+			s.UseEntraID = false
+			c, err := tc.newV8(s)
+			require.NoError(t, err)
+			require.NotNil(t, c)
+			defer c.Close()
+
+			onConnect, _ := v8OnConnectAndPassword(t, c)
+			assert.Nil(t, onConnect, "OnConnect must not be wired when UseEntraID is false")
+		})
+
+		t.Run("v9/"+tc.name+"/entraID-enabled", func(t *testing.T) {
+			c, err := tc.newV9(entraIDSettings(tc.redisType, tc.failover))
+			require.NoError(t, err)
+			require.NotNil(t, c)
+			defer c.Close()
+
+			onConnect, password := v9OnConnectAndPassword(t, c)
+			assert.NotNil(t, onConnect, "OnConnect must be wired when UseEntraID is true")
+			assert.Empty(t, password, "static Password must not be set when UseEntraID is true")
+		})
+
+		t.Run("v9/"+tc.name+"/entraID-disabled", func(t *testing.T) {
+			s := entraIDSettings(tc.redisType, tc.failover)
+			s.UseEntraID = false
+			c, err := tc.newV9(s)
+			require.NoError(t, err)
+			require.NotNil(t, c)
+			defer c.Close()
+
+			onConnect, _ := v9OnConnectAndPassword(t, c)
+			assert.Nil(t, onConnect, "OnConnect must not be wired when UseEntraID is false")
+		})
+	}
+}
+
+func TestEntraIDFetchAuthArgs(t *testing.T) {
+	t.Run("returns username and fresh token", func(t *testing.T) {
+		s := entraIDSettings(NodeType, false)
+		user, pass, err := s.entraIDFetchAuthArgs(context.Background())
+		require.NoError(t, err)
+		assert.Equal(t, "00000000-0000-0000-0000-000000000000", user)
+		assert.Equal(t, "fresh-token", pass)
+	})
+
+	t.Run("errors when credential not initialized", func(t *testing.T) {
+		s := &Settings{UseEntraID: true, entraIDUsername: "oid"}
+		_, _, err := s.entraIDFetchAuthArgs(context.Background())
+		require.Error(t, err)
+	})
+
+	t.Run("errors when username (OID) not initialized", func(t *testing.T) {
+		s := &Settings{UseEntraID: true, entraIDTokenCredential: fakeTokenCredential{token: "fresh-token"}}
+		_, _, err := s.entraIDFetchAuthArgs(context.Background())
+		require.Error(t, err)
+	})
+
+	t.Run("propagates token acquisition error", func(t *testing.T) {
+		s := &Settings{UseEntraID: true, entraIDUsername: "oid", entraIDTokenCredential: fakeTokenCredential{err: errors.New("boom")}}
+		_, _, err := s.entraIDFetchAuthArgs(context.Background())
+		require.Error(t, err)
+	})
+}
+
+// TestEntraIDOnConnectCallback invokes the OnConnect callbacks themselves (not just
+// asserting they are wired) to confirm they drive authentication through a fresh-token
+// fetch and surface a credential failure as a dial error. A failing credential makes the
+// callback return before it touches the (nil) connection, so this exercises the
+// connect-time auth path end-to-end without a live Redis server.
+func TestEntraIDOnConnectCallback(t *testing.T) {
+	t.Run("v8 surfaces token-fetch failure as a dial error", func(t *testing.T) {
+		s := &Settings{UseEntraID: true, entraIDUsername: "oid", entraIDTokenCredential: fakeTokenCredential{err: errors.New("boom")}}
+		err := entraIDOnConnectV8(s)(context.Background(), nil)
+		require.Error(t, err)
+	})
+
+	t.Run("v9 surfaces token-fetch failure as a dial error", func(t *testing.T) {
+		s := &Settings{UseEntraID: true, entraIDUsername: "oid", entraIDTokenCredential: fakeTokenCredential{err: errors.New("boom")}}
+		err := entraIDOnConnectV9(s)(context.Background(), nil)
+		require.Error(t, err)
+	})
+
+	t.Run("v8 surfaces missing-OID as a dial error", func(t *testing.T) {
+		s := &Settings{UseEntraID: true, entraIDTokenCredential: fakeTokenCredential{token: "fresh-token"}}
+		err := entraIDOnConnectV8(s)(context.Background(), nil)
+		require.Error(t, err)
+	})
+
+	t.Run("v9 surfaces missing-OID as a dial error", func(t *testing.T) {
+		s := &Settings{UseEntraID: true, entraIDTokenCredential: fakeTokenCredential{token: "fresh-token"}}
+		err := entraIDOnConnectV9(s)(context.Background(), nil)
+		require.Error(t, err)
+	})
+}

common/component/redis/redis.go

@@ -176,9 +176,8 @@ func ParseClientFromProperties(properties map[string]string, componentType metad
 	}
 
 	var tokenExpires *time.Time
-	var tokenCredential *azcore.TokenCredential
 	if settings.UseEntraID {
-		tokenExpires, tokenCredential, err = settings.GetEntraIDCredentialAndSetInitialTokenAsPassword(ctx, &properties)
+		tokenExpires, err = settings.InitEntraIDCredential(ctx, &properties)
 		if err != nil {
 			return nil, nil, err
 		}
@@ -221,19 +220,22 @@ func ParseClientFromProperties(properties map[string]string, componentType metad
 		return nil, nil, fmt.Errorf("redis client configuration error: %w", err)
 	}
 
-	// start the token refresh goroutine
+	// start the token refresh goroutine — periodically issues AUTH ACL with a fresh token
+	// before the prior one expires. Note that AUTH is per-connection, so this re-authenticates
+	// pooled connections as they are exercised rather than every open connection at once. It is
+	// complementary to the OnConnect hook, which authenticates each brand-new pool connection.
 
 	if settings.UseEntraID {
-		StartEntraIDTokenRefreshBackgroundRoutine(c, settings.Username, *tokenExpires, tokenCredential, logger)
+		StartEntraIDTokenRefreshBackgroundRoutine(c, &settings, *tokenExpires, logger)
 	}
 	if settings.UseOIDC {
 		StartOIDCTokenRefreshBackgroundRoutine(c, settings.Username, oidcTokenExpiry, oidcTokenSource, logger)
 	}
 	return c, &settings, nil
 }
 
-func StartEntraIDTokenRefreshBackgroundRoutine(client RedisClient, username string, nextExpiration time.Time, cred *azcore.TokenCredential, logger *kitlogger.Logger) {
-	go func(cred *azcore.TokenCredential, username string, logger *kitlogger.Logger) {
+func StartEntraIDTokenRefreshBackgroundRoutine(client RedisClient, s *Settings, nextExpiration time.Time, logger *kitlogger.Logger) {
+	go func(s *Settings, logger *kitlogger.Logger) {
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
 		backoffConfig := kitretry.DefaultConfig()
@@ -260,9 +262,7 @@ func StartEntraIDTokenRefreshBackgroundRoutine(client RedisClient, username stri
 				tokenErr := kitretry.NotifyRecover(
 					func() error {
 						var innerTokenErr error
-						token, innerTokenErr = (*cred).GetToken(ctx, policy.TokenRequestOptions{
-							Scopes: []string{"https://redis.azure.com/.default"},
-						})
+						token, innerTokenErr = s.entraIDToken(ctx)
 						return innerTokenErr
 					},
 					backoffManager,
@@ -283,7 +283,7 @@ func StartEntraIDTokenRefreshBackgroundRoutine(client RedisClient, username stri
 				backoffManager = backoffConfig.NewBackOffWithContext(ctx)
 				authErr := kitretry.NotifyRecover(
 					func() error {
-						innerAuthErr := client.AuthACL(ctx, username, token.Token)
+						innerAuthErr := client.AuthACL(ctx, s.entraIDUsername, token.Token)
 						return innerAuthErr
 					},
 					backoffManager,
@@ -306,46 +306,75 @@ func StartEntraIDTokenRefreshBackgroundRoutine(client RedisClient, username stri
 				tokenRefreshDuration = time.Until(token.ExpiresOn.Add(-refreshGracePeriod))
 			}
 		}
-	}(cred, username, logger)
+	}(s, logger)
 }
 
-func (s *Settings) GetEntraIDCredentialAndSetInitialTokenAsPassword(ctx context.Context, properties *map[string]string) (*time.Time, *azcore.TokenCredential, error) {
+// InitEntraIDCredential validates Entra ID configuration, acquires the Azure SDK token
+// credential, and parses the initial access token's "oid" claim to derive the Redis ACL
+// username. The credential and OID are stored on the Settings as unexported fields for
+// later use by:
+//
+//  1. The OnConnect hook installed on the underlying go-redis client (see v8client.go /
+//     v9client.go), which AUTHs every newly-dialed pool connection with a freshly-acquired
+//     token.
+//  2. StartEntraIDTokenRefreshBackgroundRoutine, which periodically re-AUTHs existing
+//     long-lived connections via AUTH ACL before the prior token expires.
+//
+// In contrast to the previous behavior of this function (which snapshot the initial token
+// into Settings.Password), nothing is written to Settings.Username or Settings.Password.
+// That snapshot is the source of dapr/components-contrib#3554: when go-redis later opens a
+// new pool connection, it sends the static AUTH from Options.Password — which has long
+// since expired — and Redis returns WRONGPASS. The OnConnect hook fixes this by always
+// fetching a fresh token at connect time.
+//
+// Returns the initial token's expiration, used to schedule the first refresh. The credential
+// and OID are stored on the Settings for the refresh goroutine and OnConnect hook to use.
+func (s *Settings) InitEntraIDCredential(ctx context.Context, properties *map[string]string) (*time.Time, error) {
 	if len(s.Password) > 0 || len(s.Username) > 0 {
-		return nil, nil, errors.New(
-			"redis client configuration error: username or password must not be specified when using Entra ID authentication",
-		)
+		return nil, errors.New(
+			"redis client configuration error: username or password must not be specified when using Entra ID authentication")
 	}
 	envSettings, err := azure.NewEnvironmentSettings(*properties)
 	if err != nil {
-		return nil, nil, fmt.Errorf("redis client configuration error: %w", err)
+		return nil, fmt.Errorf("redis client configuration error: %w", err)
 	}
 	cred, err := envSettings.GetTokenCredential()
 	if err != nil {
-		return nil, nil, fmt.Errorf("redis client configuration error: %w", err)
+		return nil, fmt.Errorf("redis client configuration error: %w", err)
 	}
 
 	token, err := cred.GetToken(ctx, policy.TokenRequestOptions{
-		Scopes: []string{"https://redis.azure.com/.default"},
+		Scopes: []string{entraIDRedisScope},
 	})
 	if err != nil {
-		return nil, nil, fmt.Errorf("redis client configuration error: %w", err)
+		return nil, fmt.Errorf("redis client configuration error: %w", err)
 	}
 
-	s.Password = token.Token
-
-	// This token has already been validated by EntraID. We use insecure parsing to get the object ID.
+	// Parse the initial token without verification or validation to extract the OID claim
+	// used as the Redis ACL username. Signature/expiry are intentionally not checked here
+	// (WithVerify(false), WithValidate(false)): the token was just issued by Entra ID and we
+	// are only inspecting a claim — Redis itself does the authoritative validation at AUTH.
 	parsedToken, err := jwt.ParseString(token.Token, jwt.WithVerify(false), jwt.WithValidate(false))
 	if err != nil {
-		return nil, nil, fmt.Errorf("redis client configuration error: %w", err)
+		return nil, fmt.Errorf("redis client configuration error: %w", err)
 	}
-	objectID, found := parsedToken.Get("oid")
-
-	if found {
-		s.Username = objectID.(string)
-	} else {
-		return nil, nil, errors.New("redis client configuration error: could not parse object ID from Auth token")
+	objectIDRaw, found := parsedToken.Get("oid")
+	if !found {
+		return nil, errors.New("redis client configuration error: could not parse object ID from Auth token")
+	}
+	objectID, ok := objectIDRaw.(string)
+	if !ok {
+		return nil, errors.New("redis client configuration error: object ID claim is not a string")
 	}
-	return &token.ExpiresOn, &cred, nil
+	// Fail fast on an empty OID: AUTH ACL requires a non-empty username, so an empty
+	// claim would otherwise surface as a confusing connect-time error rather than here.
+	if objectID == "" {
+		return nil, errors.New("redis client configuration error: object ID claim is empty in Auth token")
+	}
+
+	s.entraIDUsername = objectID
+	s.entraIDTokenCredential = cred
+	return &token.ExpiresOn, nil
 }
 
 // GetOIDCTokenSourceAndSetInitialTokenAsPassword validates the OIDC private_key_jwt