diff --git a/common/component/redis/entraid_onconnect_test.go b/common/component/redis/entraid_onconnect_test.go new file mode 100644 index 0000000000..f60035d6c7 --- /dev/null +++ b/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) + }) +} diff --git a/common/component/redis/redis.go b/common/component/redis/redis.go index 255a60bded..158bea59b5 100644 --- a/common/component/redis/redis.go +++ b/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,10 +220,13 @@ 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) @@ -232,8 +234,8 @@ func ParseClientFromProperties(properties map[string]string, componentType metad 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 diff --git a/common/component/redis/settings.go b/common/component/redis/settings.go index 50080c2ac9..8fffcbc4ae 100644 --- a/common/component/redis/settings.go +++ b/common/component/redis/settings.go @@ -14,18 +14,28 @@ limitations under the License. package redis import ( + "context" "crypto/tls" + "errors" "fmt" "net" "strconv" "strings" "time" + "github.com/Azure/azure-sdk-for-go/sdk/azcore" + "github.com/Azure/azure-sdk-for-go/sdk/azcore/policy" + "github.com/dapr/kit/config" ) const defaultRedisPort = "6379" +// entraIDRedisScope is the OAuth scope used to acquire Entra ID access tokens for +// Azure Cache/Managed Redis. Kept as a single constant so the initial-token fetch, +// the per-connection OnConnect fetch, and the background refresh loop never drift. +const entraIDRedisScope = "https://redis.azure.com/.default" + type Settings struct { // The Redis host Host string `mapstructure:"redisHost"` @@ -149,6 +159,48 @@ type Settings struct { OidcCACert string `mapstructure:"oidcCACert"` internalOidcScopes []string `mapstructure:"-"` + + // == Entra ID runtime state (populated by InitEntraIDCredential when UseEntraID is true; not configurable via metadata) == + + // entraIDUsername is the OID parsed from the initial Entra access token's "oid" claim. + // Used as the Redis ACL username by both the per-new-connection AUTH (OnConnect) and the + // periodic AUTH ACL refresh goroutine. + entraIDUsername string + + // entraIDTokenCredential is the Azure SDK credential used to acquire fresh Entra access + // tokens on demand. The credential implementation caches tokens until close to expiry, + // so calling GetToken on every new pool connection is inexpensive in steady state. + entraIDTokenCredential azcore.TokenCredential +} + +// entraIDToken acquires a fresh Entra access token for the Redis scope using the cached +// credential. The credential caches tokens internally until shortly before expiry, so in +// steady state this is a cheap in-memory lookup rather than a network round-trip. +func (s *Settings) entraIDToken(ctx context.Context) (azcore.AccessToken, error) { + if s.entraIDTokenCredential == nil { + return azcore.AccessToken{}, errors.New("redis client: EntraID credential not initialized") + } + return s.entraIDTokenCredential.GetToken(ctx, policy.TokenRequestOptions{ + Scopes: []string{entraIDRedisScope}, + }) +} + +// entraIDFetchAuthArgs returns the Redis ACL username and a freshly-acquired Entra access +// token suitable for use as the AUTH password. It must only be called when UseEntraID is +// true and after InitEntraIDCredential has succeeded; otherwise it returns an error. +// +// This is invoked from the OnConnect callback installed on the underlying go-redis client +// so that every new pool connection authenticates with a current token, rather than the +// stale snapshot Password that would otherwise be sent during initial AUTH. +func (s *Settings) entraIDFetchAuthArgs(ctx context.Context) (username, password string, err error) { + if s.entraIDUsername == "" { + return "", "", errors.New("redis client: EntraID username (OID) not initialized; AUTH ACL requires a non-empty username") + } + tok, err := s.entraIDToken(ctx) + if err != nil { + return "", "", fmt.Errorf("failed to acquire EntraID token for redis AUTH: %w", err) + } + return s.entraIDUsername, tok.Token, nil } func (s *Settings) Decode(in interface{}) error { diff --git a/common/component/redis/v8client.go b/common/component/redis/v8client.go index 866d4bc520..9db43393b3 100644 --- a/common/component/redis/v8client.go +++ b/common/component/redis/v8client.go @@ -326,6 +326,28 @@ func (c v8Client) AuthACL(ctx context.Context, username, password string) error return err } +// entraIDOnConnectV8 returns an OnConnect callback for go-redis v8 client options. +// On every newly-dialed pool connection, the callback fetches a fresh Entra access +// token from the cached credential and runs AUTH ACL on that connection. This is the +// fix for dapr/components-contrib#3554 — without it, go-redis's initial AUTH on a new +// connection would replay the snapshot Password set at component init, which has +// expired by the time the connection is opened. +// +// ctx is supplied by go-redis and is bounded by the configured DialTimeout. The token +// fetch is a cheap in-memory lookup in steady state (azcore caches until near expiry). +// Returning an error here fails the dial: go-redis discards the connection and surfaces +// the error to the caller of the Redis command that triggered the dial, which is the +// intended behavior — a connection that cannot authenticate must not be handed out. +func entraIDOnConnectV8(s *Settings) func(ctx context.Context, cn *v8.Conn) error { + return func(ctx context.Context, cn *v8.Conn) error { + user, pass, err := s.entraIDFetchAuthArgs(ctx) + if err != nil { + return err + } + return cn.AuthACL(ctx, user, pass).Err() + } +} + func newV8FailoverClient(s *Settings) (RedisClient, error) { if s == nil { return nil, nil @@ -351,6 +373,9 @@ func newV8FailoverClient(s *Settings) (RedisClient, error) { IdleCheckFrequency: time.Duration(s.IdleCheckFrequency), IdleTimeout: time.Duration(s.IdleTimeout), } + if s.UseEntraID { + opts.OnConnect = entraIDOnConnectV8(s) + } if s.EnableTLS { opts.TLSConfig = &tls.Config{ @@ -408,6 +433,9 @@ func newV8Client(s *Settings) (RedisClient, error) { IdleCheckFrequency: time.Duration(s.IdleCheckFrequency), IdleTimeout: time.Duration(s.IdleTimeout), } + if s.UseEntraID { + options.OnConnect = entraIDOnConnectV8(s) + } /* #nosec */ if s.EnableTLS { options.TLSConfig = &tls.Config{ @@ -447,6 +475,9 @@ func newV8Client(s *Settings) (RedisClient, error) { IdleCheckFrequency: time.Duration(s.IdleCheckFrequency), IdleTimeout: time.Duration(s.IdleTimeout), } + if s.UseEntraID { + options.OnConnect = entraIDOnConnectV8(s) + } /* #nosec */ if s.EnableTLS { diff --git a/common/component/redis/v9client.go b/common/component/redis/v9client.go index 434eca06b5..9580169d2f 100644 --- a/common/component/redis/v9client.go +++ b/common/component/redis/v9client.go @@ -329,6 +329,28 @@ func (c v9Client) AuthACL(ctx context.Context, username, password string) error return err } +// entraIDOnConnectV9 returns an OnConnect callback for go-redis v9 client options. +// On every newly-dialed pool connection, the callback fetches a fresh Entra access +// token from the cached credential and runs AUTH ACL on that connection. This is the +// fix for dapr/components-contrib#3554 — without it, go-redis's initial AUTH on a new +// connection would replay the snapshot Password set at component init, which has +// expired by the time the connection is opened. +// +// ctx is supplied by go-redis and is bounded by the configured DialTimeout. The token +// fetch is a cheap in-memory lookup in steady state (azcore caches until near expiry). +// Returning an error here fails the dial: go-redis discards the connection and surfaces +// the error to the caller of the Redis command that triggered the dial, which is the +// intended behavior — a connection that cannot authenticate must not be handed out. +func entraIDOnConnectV9(s *Settings) func(ctx context.Context, cn *v9.Conn) error { + return func(ctx context.Context, cn *v9.Conn) error { + user, pass, err := s.entraIDFetchAuthArgs(ctx) + if err != nil { + return err + } + return cn.AuthACL(ctx, user, pass).Err() + } +} + func newV9FailoverClient(s *Settings) (RedisClient, error) { if s == nil { return nil, nil @@ -354,6 +376,9 @@ func newV9FailoverClient(s *Settings) (RedisClient, error) { ConnMaxIdleTime: time.Duration(s.IdleTimeout), ContextTimeoutEnabled: true, } + if s.UseEntraID { + opts.OnConnect = entraIDOnConnectV9(s) + } /* #nosec */ if s.EnableTLS { @@ -413,6 +438,9 @@ func newV9Client(s *Settings) (RedisClient, error) { ConnMaxIdleTime: time.Duration(s.IdleTimeout), ContextTimeoutEnabled: true, } + if s.UseEntraID { + options.OnConnect = entraIDOnConnectV9(s) + } /* #nosec */ if s.EnableTLS { /* #nosec */ @@ -453,6 +481,9 @@ func newV9Client(s *Settings) (RedisClient, error) { ConnMaxIdleTime: time.Duration(s.IdleTimeout), ContextTimeoutEnabled: true, } + if s.UseEntraID { + options.OnConnect = entraIDOnConnectV9(s) + } if s.EnableTLS { /* #nosec */