Redis driver fork update v9.6.1

go.mod

@@ -559,7 +559,7 @@ replace (
 	github.com/julienschmidt/httprouter => github.com/gravitational/httprouter v1.3.1-0.20220408074523-c876c5e705a5
 	github.com/keys-pub/go-libfido2 => github.com/gravitational/go-libfido2 v1.5.3-teleport.1
 	github.com/microsoft/go-mssqldb => github.com/gravitational/go-mssqldb v0.11.1-0.20230331180905-0f76f1751cd3
-	github.com/redis/go-redis/v9 => github.com/gravitational/redis/v9 v9.0.2-teleport.2
+	github.com/redis/go-redis/v9 => github.com/gravitational/redis/v9 v9.6.1-teleport.1
 	github.com/sijms/go-ora/v2 => github.com/gravitational/go-ora/v2 v2.0.0-20230821114616-e2a9f1131a46
 	github.com/vulcand/predicate => github.com/gravitational/predicate v1.3.1
 )

go.sum

@@ -953,10 +953,10 @@ github.com/boombuler/barcode v1.0.1 h1:NDBbPmhS+EqABEs5Kg3n/5ZNjy73Pz7SIV+KCeqyX
 github.com/boombuler/barcode v1.0.1/go.mod h1:paBWMcWSl3LHKBqUq+rly7CNSldXjb2rDl3JlRe0mD8=
 github.com/bshuster-repo/logrus-logstash-hook v1.0.0 h1:e+C0SB5R1pu//O4MQ3f9cFuPGoOVeF2fE4Og9otCc70=
 github.com/bshuster-repo/logrus-logstash-hook v1.0.0/go.mod h1:zsTqEiSzDgAa/8GZR7E1qaXrhYNDKBYy5/dWPTIflbk=
-github.com/bsm/ginkgo/v2 v2.5.0 h1:aOAnND1T40wEdAtkGSkvSICWeQ8L3UASX7YVCqQx+eQ=
-github.com/bsm/ginkgo/v2 v2.5.0/go.mod h1:AiKlXPm7ItEHNc/2+OkrNG4E0ITzojb9/xWzvQ9XZ9w=
-github.com/bsm/gomega v1.20.0 h1:JhAwLmtRzXFTx2AkALSLa8ijZafntmhSoU63Ok18Uq8=
-github.com/bsm/gomega v1.20.0/go.mod h1:JifAceMQ4crZIWYUKrlGcmbN3bqHogVTADMD2ATsbwk=
+github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
+github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
+github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
+github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/bugsnag/bugsnag-go v0.0.0-20141110184014-b1d153021fcd h1:rFt+Y/IK1aEZkEHchZRSq9OQbsSzIT/OrI8YFFmRIng=
 github.com/bugsnag/bugsnag-go v0.0.0-20141110184014-b1d153021fcd/go.mod h1:2oa8nejYd4cQ/b0hMIopN0lCRxU0bueqREvZLWFrtK8=
 github.com/bugsnag/osext v0.0.0-20130617224835-0dd3f918b21b h1:otBG+dV+YK+Soembjv71DPz3uX/V/6MMlSyD9JBQ6kQ=
@@ -1534,8 +1534,8 @@ github.com/gravitational/predicate v1.3.1 h1:f1uGg2FF6z5wZbcafYpLZJ1gl+82I0MlSd0
 github.com/gravitational/predicate v1.3.1/go.mod h1:H5e9dUW7zb/cuKkkhfnyT9SsI/WHWJ8Ra011La16DTY=
 github.com/gravitational/protobuf v1.3.2-teleport.1 h1:h5mh+UOKPurqDxn1hRVcr1WzSkmBi+D9qkXpaXA9PFM=
 github.com/gravitational/protobuf v1.3.2-teleport.1/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
-github.com/gravitational/redis/v9 v9.0.2-teleport.2 h1:br9gSGJLpfrIoKsF5N99zWgjPpQpsQiCeN5HmAbYpio=
-github.com/gravitational/redis/v9 v9.0.2-teleport.2/go.mod h1:/xDTe9EF1LM61hek62Poq2nzQSGj0xSrEtEHbBQevps=
+github.com/gravitational/redis/v9 v9.6.1-teleport.1 h1:gPirfPKArN2nPhTKR3h9fnEg5YuYU933+CjlDJMo4H0=
+github.com/gravitational/redis/v9 v9.6.1-teleport.1/go.mod h1:0C0c6ycQsdpVNQpxb1njEQIqkx5UcsM8FJCQLgE9+RA=
 github.com/gravitational/roundtrip v1.0.2 h1:eOCY0NEKKaB0ksJmvhO6lPMFz1pIIef+vyPBTBROQ5c=
 github.com/gravitational/roundtrip v1.0.2/go.mod h1:fuI1booM2hLRA/B/m5MRAPOU6mBZNYcNycono2UuTw0=
 github.com/gravitational/saml v0.4.15-teleport.1 h1:kYSLpxEBEc7JLJJ+VjsZU8PbWI4gWxdCgll5cq1/rGU=

integrations/event-handler/go.mod

@@ -343,7 +343,7 @@ replace (
 	// replace module github.com/moby/spdystream until https://github.com/moby/spdystream/pull/91 merges and deps are updated
 	// otherwise tests fail with a data race detection.
 	github.com/moby/spdystream => github.com/gravitational/spdystream v0.0.0-20230512133543-4e46862ca9bf
-	github.com/redis/go-redis/v9 => github.com/gravitational/redis/v9 v9.0.2-teleport.2
+	github.com/redis/go-redis/v9 => github.com/gravitational/redis/v9 v9.6.1-teleport.1
 	github.com/sijms/go-ora/v2 => github.com/gravitational/go-ora/v2 v2.0.0-20230821114616-e2a9f1131a46
 	github.com/vulcand/predicate => github.com/gravitational/predicate v1.3.1
 	// replace module sigs.k8s.io/kustomize/api until https://github.com/kubernetes-sigs/kustomize/issues/5524 is resolved,

integrations/terraform/go.mod

@@ -400,7 +400,7 @@ replace (
 	github.com/keys-pub/go-libfido2 => github.com/gravitational/go-libfido2 v1.5.3-teleport.1
 	github.com/microsoft/go-mssqldb => github.com/gravitational/go-mssqldb v0.11.1-0.20230331180905-0f76f1751cd3
 	github.com/moby/spdystream => github.com/gravitational/spdystream v0.0.0-20230512133543-4e46862ca9bf
-	github.com/redis/go-redis/v9 => github.com/gravitational/redis/v9 v9.0.2-teleport.2
+	github.com/redis/go-redis/v9 => github.com/gravitational/redis/v9 v9.6.1-teleport.1
 	github.com/sijms/go-ora/v2 => github.com/gravitational/go-ora/v2 v2.0.0-20230821114616-e2a9f1131a46
 	github.com/vulcand/predicate => github.com/gravitational/predicate v1.3.1
 	sigs.k8s.io/kustomize/api => github.com/gravitational/kustomize/api v0.16.0-teleport.1

integrations/terraform/go.sum

@@ -1320,8 +1320,8 @@ github.com/gravitational/predicate v1.3.1 h1:f1uGg2FF6z5wZbcafYpLZJ1gl+82I0MlSd0
 github.com/gravitational/predicate v1.3.1/go.mod h1:H5e9dUW7zb/cuKkkhfnyT9SsI/WHWJ8Ra011La16DTY=
 github.com/gravitational/protobuf v1.3.2-teleport.1 h1:h5mh+UOKPurqDxn1hRVcr1WzSkmBi+D9qkXpaXA9PFM=
 github.com/gravitational/protobuf v1.3.2-teleport.1/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
-github.com/gravitational/redis/v9 v9.0.2-teleport.2 h1:br9gSGJLpfrIoKsF5N99zWgjPpQpsQiCeN5HmAbYpio=
-github.com/gravitational/redis/v9 v9.0.2-teleport.2/go.mod h1:/xDTe9EF1LM61hek62Poq2nzQSGj0xSrEtEHbBQevps=
+github.com/gravitational/redis/v9 v9.6.1-teleport.1 h1:gPirfPKArN2nPhTKR3h9fnEg5YuYU933+CjlDJMo4H0=
+github.com/gravitational/redis/v9 v9.6.1-teleport.1/go.mod h1:0C0c6ycQsdpVNQpxb1njEQIqkx5UcsM8FJCQLgE9+RA=
 github.com/gravitational/roundtrip v1.0.2 h1:eOCY0NEKKaB0ksJmvhO6lPMFz1pIIef+vyPBTBROQ5c=
 github.com/gravitational/roundtrip v1.0.2/go.mod h1:fuI1booM2hLRA/B/m5MRAPOU6mBZNYcNycono2UuTw0=
 github.com/gravitational/spdystream v0.0.0-20230512133543-4e46862ca9bf h1:aXnqDSit8L1qhI0+QdbJh+MTUFKXG7qbkZXnfr7L96A=

lib/srv/db/redis/client.go

@@ -92,6 +92,8 @@ const (
 const (
 	// aclWhoami is a subcommand of "acl" that requires special handling.
 	aclWhoami = "whoami"
+	// protocolV2 defines the RESP protocol v2 that Teleport uses.
+	protocolV2 = 2
 )
 
 // clusterClient is a wrapper around redis.ClusterClient
@@ -101,30 +103,26 @@ type clusterClient struct {
 
 // newClient creates a new Redis client based on given ConnectionMode. If connection mode is not supported
 // an error is returned.
-func newClient(ctx context.Context, connectionOptions *connection.Options, tlsConfig *tls.Config, onConnect onClientConnectFunc) (redis.UniversalClient, error) {
+func newClient(ctx context.Context, connectionOptions *connection.Options, tlsConfig *tls.Config, credentialsProvider fetchCredentialsFunc) (redis.UniversalClient, error) {
 	connectionAddr := net.JoinHostPort(connectionOptions.Address, connectionOptions.Port)
 	// TODO(jakub): Investigate Redis Sentinel.
 	switch connectionOptions.Mode {
 	case connection.Standalone:
 		return redis.NewClient(&redis.Options{
-			Addr:      connectionAddr,
-			TLSConfig: tlsConfig,
-			OnConnect: onConnect,
-
-			// Auth should be done by the `OnConnect` callback here. So disable
-			// "automatic" auth by the client.
-			DisableAuthOnConnect: true,
+			Addr:                       connectionAddr,
+			TLSConfig:                  tlsConfig,
+			CredentialsProviderContext: credentialsProvider,
+			Protocol:                   protocolV2,
+			DisableIndentity:           true,
 		}), nil
 	case connection.Cluster:
 		client := &clusterClient{
 			ClusterClient: *redis.NewClusterClient(&redis.ClusterOptions{
-				Addrs:     []string{connectionAddr},
-				TLSConfig: tlsConfig,
-				OnConnect: onConnect,
-				NewClient: func(opt *redis.Options) *redis.Client {
-					opt.DisableAuthOnConnect = true
-					return redis.NewClient(opt)
-				},
+				Addrs:                      []string{connectionAddr},
+				TLSConfig:                  tlsConfig,
+				CredentialsProviderContext: credentialsProvider,
+				Protocol:                   protocolV2,
+				DisableIndentity:           true,
 			}),
 		}
 		// Load cluster information.
@@ -137,33 +135,25 @@ func newClient(ctx context.Context, connectionOptions *connection.Options, tlsCo
 	}
 }
 
-// onClientConnectFunc is a callback function that performs setups after Redis
-// client makes a new connection.
-type onClientConnectFunc func(context.Context, *redis.Conn) error
-
 // fetchCredentialsFunc fetches credentials for a new connection.
 type fetchCredentialsFunc func(ctx context.Context) (username, password string, err error)
 
-func noopOnConnect(context.Context, *redis.Conn) error {
-	return nil
-}
-
-// authWithPasswordOnConnect returns an onClientConnectFunc that sends "auth"
+// authWithPasswordOnConnect returns an fetchCredentialsFunc that sends "auth"
 // with provided username and password.
-func authWithPasswordOnConnect(username, password string) onClientConnectFunc {
-	return func(ctx context.Context, conn *redis.Conn) error {
-		return authConnection(ctx, conn, username, password)
+func authWithPasswordOnConnect(username, password string) fetchCredentialsFunc {
+	return func(ctx context.Context) (string, string, error) {
+		return username, password, nil
 	}
 }
 
-// fetchCredentialsOnConnect returns an onClientConnectFunc that does an
+// fetchCredentialsOnConnect returns an fetchCredentialsFunc that does an
 // authorization check, calls a provided credential fetcher callback func,
 // then logs an AUTH query to the audit log once and and uses the credentials to
 // auth a new connection.
-func fetchCredentialsOnConnect(closeCtx context.Context, sessionCtx *common.Session, audit common.Audit, fetchCreds fetchCredentialsFunc) onClientConnectFunc {
+func fetchCredentialsOnConnect(closeCtx context.Context, sessionCtx *common.Session, audit common.Audit, fetchCreds fetchCredentialsFunc) fetchCredentialsFunc {
 	// audit log one time, to avoid excessive audit logs from reconnects.
 	var auditOnce sync.Once
-	return func(ctx context.Context, conn *redis.Conn) error {
+	return func(ctx context.Context) (string, string, error) {
 		err := sessionCtx.Checker.CheckAccess(sessionCtx.Database,
 			services.AccessState{MFAVerified: true},
 			role.GetDatabaseRoleMatchers(role.RoleMatchersConfig{
@@ -172,11 +162,11 @@ func fetchCredentialsOnConnect(closeCtx context.Context, sessionCtx *common.Sess
 				DatabaseName: sessionCtx.DatabaseName,
 			})...)
 		if err != nil {
-			return trace.Wrap(err)
+			return "", "", trace.Wrap(err)
 		}
 		username, password, err := fetchCreds(ctx)
 		if err != nil {
-			return trace.Wrap(err)
+			return "", "", trace.Wrap(err)
 		}
 		auditOnce.Do(func() {
 			var query string
@@ -187,7 +177,7 @@ func fetchCredentialsOnConnect(closeCtx context.Context, sessionCtx *common.Sess
 			}
 			audit.OnQuery(closeCtx, sessionCtx, common.Query{Query: query})
 		})
-		return authConnection(ctx, conn, username, password)
+		return username, password, nil
 	}
 }
 
@@ -256,23 +246,6 @@ func awsIAMTokenFetchFunc(sessionCtx *common.Session, auth common.Auth) (fetchCr
 	}
 }
 
-// authConnection is a helper function that sends "auth" command to provided
-// Redis connection with provided username and password.
-func authConnection(ctx context.Context, conn *redis.Conn, username, password string) error {
-	// Copied from redis.baseClient.initConn.
-	_, err := conn.Pipelined(ctx, func(pipe redis.Pipeliner) error {
-		if password != "" {
-			if username != "" {
-				pipe.AuthACL(ctx, username, password)
-			} else {
-				pipe.Auth(ctx, password)
-			}
-		}
-		return nil
-	})
-	return trace.Wrap(err)
-}
-
 // Process add supports for additional cluster commands. Our Redis implementation passes most commands to
 // go-redis `Process()` function which doesn't handel all Cluster commands like for ex. DBSIZE, FLUSHDB, etc.
 // This function provides additional processing for those commands enabling more Redis commands in Cluster mode.

lib/srv/db/redis/engine.go

@@ -25,6 +25,7 @@ import (
 	"net"
 	"slices"
 	"strings"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/elasticache"
@@ -75,6 +76,8 @@ type Engine struct {
 	redisClient redis.UniversalClient
 	// awsIAMAuthSupported is the saved result of isAWSIAMAuthSupported.
 	awsIAMAuthSupported *bool
+	// clientMessageRead indicates processing client messages has started.
+	clientMessageRead bool
 }
 
 // InitializeConnection initializes the database connection.
@@ -145,12 +148,52 @@ func (e *Engine) SendError(redisErr error) {
 		return
 	}
 
+	// If the first message is a HELLO test, do not return authentication
+	// errors to the HELLO command as it can be swallowed by the client as part
+	// of its fallback mechanism. First return the unknown command error then
+	// send the authentication errors to the next incoming command (usually
+	// AUTH).
+	//
+	// Background: The HELLO test is used for establishing the RESP3 protocol
+	// but Teleport currently only supports RESP2. The client generally
+	// fallbacks to RESP2 when they receive an unknown command error for the
+	// HELLO message.
+	e.maybeHandleFirstHello()
+
 	if err := e.sendToClient(redisErr); err != nil {
 		e.Log.ErrorContext(e.Context, "Failed to send message to the client.", "error", err)
 		return
 	}
 }
 
+// maybeHandleFirstHello replies an unknown command error to the client if the
+// first message is a HELLO test.
+func (e *Engine) maybeHandleFirstHello() {
+	// Return if not the first message.
+	if e.clientMessageRead {
+		return
+	}
+
+	// Let's not wait forever for the HELLO message.
+	ctx, cancel := context.WithTimeout(e.Context, 10*time.Second)
+	defer cancel()
+
+	cmd, err := e.readClientCmd(ctx)
+	if err != nil {
+		e.Log.ErrorContext(e.Context, "Failed to read first client message.", "error", err)
+		return
+	}
+
+	// Return if not a HELLO.
+	if strings.ToLower(cmd.Name()) != helloCmd {
+		return
+	}
+	response := protocol.MakeUnknownCommandErrorForCmd(cmd)
+	if err := e.sendToClient(response); err != nil {
+		e.Log.ErrorContext(e.Context, "Failed to send message to the client.", "error", err)
+	}
+}
+
 // sendToClient sends a command to connected Redis client.
 func (e *Engine) sendToClient(vals interface{}) error {
 	if vals == nil {
@@ -249,12 +292,12 @@ func (e *Engine) getNewClientFn(ctx context.Context, sessionCtx *common.Session)
 	}
 
 	return func(username, password string) (redis.UniversalClient, error) {
-		onConnect, err := e.createOnClientConnectFunc(ctx, sessionCtx, username, password)
+		credenialsProvider, err := e.createCredentialsProvider(ctx, sessionCtx, username, password)
 		if err != nil {
 			return nil, trace.Wrap(err)
 		}
 
-		redisClient, err := newClient(ctx, connectionOptions, tlsConfig, onConnect)
+		redisClient, err := newClient(ctx, connectionOptions, tlsConfig, credenialsProvider)
 		if err != nil {
 			return nil, trace.Wrap(err)
 		}
@@ -263,9 +306,10 @@ func (e *Engine) getNewClientFn(ctx context.Context, sessionCtx *common.Session)
 	}, nil
 }
 
-// createOnClientConnectFunc creates a callback function that is called after a
-// successful client connection with the Redis server.
-func (e *Engine) createOnClientConnectFunc(ctx context.Context, sessionCtx *common.Session, username, password string) (onClientConnectFunc, error) {
+// createCredentialsProvider creates a callback function that provides username
+// and password.
+// This function may return nil, nil as nil credenialsProvider is valid.
+func (e *Engine) createCredentialsProvider(ctx context.Context, sessionCtx *common.Session, username, password string) (fetchCredentialsFunc, error) {
 	switch {
 	// If password is provided by client.
 	case password != "":
@@ -299,7 +343,7 @@ func (e *Engine) createOnClientConnectFunc(ctx context.Context, sessionCtx *comm
 		return fetchCredentialsOnConnect(e.Context, sessionCtx, e.Audit, credFetchFn), nil
 
 	default:
-		return noopOnConnect, nil
+		return nil, nil
 	}
 }
 
@@ -454,6 +498,8 @@ func (e *Engine) process(ctx context.Context, sessionCtx *common.Session) error
 
 // readClientCmd reads commands from connected Redis client.
 func (e *Engine) readClientCmd(ctx context.Context) (*redis.Cmd, error) {
+	e.clientMessageRead = true
+
 	cmd := &redis.Cmd{}
 	if err := cmd.ReadReply(e.clientReader); err != nil {
 		return nil, trace.Wrap(err)

lib/srv/db/redis/test.go

@@ -78,15 +78,13 @@ func MakeTestClient(ctx context.Context, config common.TestClientConfig, opts ..
 	}
 
 	client := redis.NewClient(&redis.Options{
-		Addr:         config.Address,
-		TLSConfig:    tlsConfig,
-		DialTimeout:  clientOptions.timeout,
-		ReadTimeout:  clientOptions.timeout,
-		WriteTimeout: clientOptions.timeout,
-		// Set DisableAuthOnConnect to true to avoid automatically sending
-		// HELLO to the server to speed up the tests. Let the caller decide to
-		// send HELLO or not.
-		DisableAuthOnConnect: true,
+		Addr:             config.Address,
+		TLSConfig:        tlsConfig,
+		DialTimeout:      clientOptions.timeout,
+		ReadTimeout:      clientOptions.timeout,
+		WriteTimeout:     clientOptions.timeout,
+		Protocol:         protocolV2,
+		DisableIndentity: true,
 	})
 
 	if !clientOptions.skipPing {