Fix two RESP3 connect failure modes: credential desync (#3172) and subscription poisoning on downgrade (#3154)

src/StackExchange.Redis/PhysicalBridge.cs

@@ -784,6 +784,19 @@ internal bool TryEnqueue(List<Message> messages, bool isReplica)
 
         private WriteResult WriteMessageInsideLock(PhysicalConnection physical, Message message)
         {
+            if (ConnectionType == ConnectionType.Interactive
+                && message.IsForSubscriptionBridge
+                && Protocol is RedisProtocol.Resp2
+                && ServerEndPoint.TryRerouteToSubscriptionBridge(message, this))
+            {
+                // this was queued while we expected RESP3 - where subscriptions share the interactive
+                // connection - but the handshake resolved to RESP2, which needs them on their own connection.
+                // Writing it here would put *this* connection into subscriber mode, which rejects every
+                // ordinary command on it from then on; see #3154
+                Trace($"Rerouting {message.CommandAndKey} to the subscription connection (RESP2)");
+                return WriteResult.Success;
+            }
+
             WriteResult result;
             var existingMessage = Interlocked.CompareExchange(ref _activeMessage, message, null);
             if (existingMessage != null)
@@ -911,6 +924,16 @@ private bool TryPushToBacklog(Message message, bool onlyIfExists, bool bypassBac
             return true;
         }
 
+        /// <summary>
+        /// Takes over a message that was queued against a different bridge of the same endpoint, because the
+        /// protocol resolved differently to what we expected when it was queued (see #3154).
+        /// </summary>
+        internal void AcceptRerouted(Message message)
+        {
+            BacklogEnqueue(message);
+            StartBacklogProcessor();
+        }
+
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         private void BacklogEnqueue(Message message)
         {

src/StackExchange.Redis/ServerEndPoint.cs

@@ -283,6 +283,23 @@ public void Dispose()
                 : interactive ??= CreateBridge(ConnectionType.Interactive, null);
         }
 
+        /// <summary>
+        /// Moves a subscription message that was queued against the interactive bridge over to the subscription
+        /// bridge, which is where it belongs now that we know the connection is RESP2 (see #3154).
+        /// </summary>
+        /// <returns><c>true</c> if the message is now the subscription bridge's responsibility.</returns>
+        internal bool TryRerouteToSubscriptionBridge(Message message, PhysicalBridge from)
+        {
+            if (isDisposed) return false;
+
+            // deliberately not via GetBridge: that consults the same expectation that got us here
+            var target = subscription ??= CreateBridge(ConnectionType.Subscription, null);
+            if (target is null || ReferenceEquals(target, from)) return false;
+
+            target.AcceptRerouted(message);
+            return true;
+        }
+
         public PhysicalBridge? GetBridge(RedisCommand command, bool create = true)
         {
             if (isDisposed) return null;
@@ -969,6 +986,19 @@ internal ValueTask WriteDirectOrQueueFireAndForgetAsync<T>(PhysicalConnection? c
             return bridge;
         }
 
+        /// <summary>
+        /// Issues <c>HELLO</c>, optionally carrying the credentials and client name.
+        /// </summary>
+        /// <remarks>The server can reject RESP3 either with an error (<c>HELLO</c> not understood, or an
+        /// unsupported protocol version) or by simply reporting RESP2, so we don't assign the protocol here:
+        /// that happens when the reply is processed (and as a last resort, when the tracer completes).</remarks>
+        private async Task WriteHelloAsync(PhysicalConnection connection, ILogger? log, int protocolVersion, string? user, string? password, string? clientName)
+        {
+            var hello = Message.CreateHello(protocolVersion, user, password, clientName, CommandFlags.FireAndForget | Message.NoFlushFlag);
+            hello.SetInternalCall();
+            await WriteDirectOrQueueFireAndForgetAsync(connection, hello, ResultProcessor.AutoConfigureProcessor.Create(log)).ForAwait();
+        }
+
         private async Task HandshakeAsync(PhysicalConnection connection, ILogger? log)
         {
             log?.LogInformationServerHandshake(new(this));
@@ -1028,45 +1058,59 @@ private async Task HandshakeAsync(PhysicalConnection connection, ILogger? log)
                 RoleKnownFromHello = false;
             }
 
-            // HELLO comes in two flavours: as the RESP3 negotiation (which has to be the *first* command, and has to
-            // carry the credentials), or - when we're staying on RESP2 - purely for discovery, in which case it is
-            // issued *after* AUTH, below; see #2968 for why we want it even on RESP2
+            // HELLO serves two purposes: negotiating RESP3, and reporting details we would otherwise need INFO or
+            // CONFIG for (see #2968) - so we issue it whenever the server should understand it, at 2 or 3.
             bool helloAvailable = Multiplexer.RawConfig.TryHello(out int helloProtocol); // includes an availability check on HELLO
             bool negotiateResp3 = helloAvailable && helloProtocol >= 3;
-            bool discoveryHello = helloAvailable && !negotiateResp3 && isInteractive;
-            if (negotiateResp3)
+
+            // HELLO can carry the credentials, but we only use that when we have no other way of authenticating,
+            // and *never* both HELLO AUTH and a standalone AUTH. This is defensive against a redis bug (7.4
+            // through at least 8.x; valkey is unaffected): inside a pipelined batch, a *failing* AUTH only gets
+            // a reply if it is the first command in that batch - otherwise the error is silently dropped, which
+            // desynchronizes every reply that follows it on the connection. When that happens during the
+            // handshake, the tracer never gets its answer and the connection never becomes usable: what should
+            // have been a clean authentication failure instead presents as a connection that times out
+            // everything. So AUTH goes first in the batch, and HELLO follows it (bare).
+            bool haveCredentials = !string.IsNullOrWhiteSpace(user) || !string.IsNullOrWhiteSpace(password);
+            bool canAuthDirectly = Multiplexer.CommandMap.IsAvailable(RedisCommand.AUTH);
+            bool helloCarriesCredentials = helloAvailable && haveCredentials && !canAuthDirectly;
+
+            // ...which also means HELLO doesn't have to come first: only the credential-carrying flavour does,
+            // as nothing else can authenticate the connection in that case
+            if (helloCarriesCredentials)
             {
                 log?.LogInformationAuthenticatingViaHello(new(this));
-                var hello = Message.CreateHello(helloProtocol, user, password, clientName, CommandFlags.FireAndForget | Message.NoFlushFlag);
-                hello.SetInternalCall();
-                await WriteDirectOrQueueFireAndForgetAsync(connection, hello, autoConfig ??= ResultProcessor.AutoConfigureProcessor.Create(log)).ForAwait();
-
-                // note that the server can reject RESP3 via either an -ERR response (HELLO not understood), or by simply saying "nope",
-                // so we don't set the actual .Protocol until we process the result of the HELLO request
+                await WriteHelloAsync(connection, log, helloProtocol, user, password, clientName).ForAwait();
             }
-            else
+            else if (!negotiateResp3)
             {
                 // whether or not we issue HELLO for discovery below, we're RESP2
                 connection.SetProtocol(RedisProtocol.Resp2);
             }
 
-            // note: we auth EVEN IF we have used HELLO to AUTH; because otherwise the fallback/detection path is pure hell,
-            // and: we're pipelined here, so... meh
-            if (!string.IsNullOrWhiteSpace(user) && Multiplexer.CommandMap.IsAvailable(RedisCommand.AUTH))
+            if (!string.IsNullOrWhiteSpace(user) && canAuthDirectly)
             {
                 log?.LogInformationAuthenticatingUserPassword(new(this));
                 msg = Message.Create(-1, CommandFlags.FireAndForget | Message.NoFlushFlag, RedisCommand.AUTH, user.AsRedisValue(), password.AsRedisValue());
                 msg.SetInternalCall();
                 await WriteDirectOrQueueFireAndForgetAsync(connection, msg, ResultProcessor.DemandOK).ForAwait();
             }
-            else if (!string.IsNullOrWhiteSpace(password) && Multiplexer.CommandMap.IsAvailable(RedisCommand.AUTH))
+            else if (!string.IsNullOrWhiteSpace(password) && canAuthDirectly)
             {
                 log?.LogInformationAuthenticatingPassword(new(this));
                 msg = Message.Create(-1, CommandFlags.FireAndForget | Message.NoFlushFlag, RedisCommand.AUTH, password.AsRedisValue());
                 msg.SetInternalCall();
                 await WriteDirectOrQueueFireAndForgetAsync(connection, msg, ResultProcessor.DemandOK).ForAwait();
             }
 
+            // a bare HELLO, now that the connection is authenticated; for RESP2 this is discovery only, so we
+            // limit it to the interactive connection (the subscription connection does no discovery)
+            bool bareHello = helloAvailable && !helloCarriesCredentials && (negotiateResp3 || isInteractive);
+            if (bareHello)
+            {
+                await WriteHelloAsync(connection, log, helloProtocol, user: null, password: null, clientName: null).ForAwait();
+            }
+
             if (Multiplexer.CommandMap.IsAvailable(RedisCommand.CLIENT))
             {
                 if (!string.IsNullOrWhiteSpace(clientName))
@@ -1114,18 +1158,7 @@ private async Task HandshakeAsync(PhysicalConnection connection, ILogger? log)
             var connType = bridge.ConnectionType;
             if (connType == ConnectionType.Interactive)
             {
-                if (discoveryHello)
-                {
-                    // a bare HELLO (no AUTH clause of its own): we're already authenticated by this point, and
-                    // folding the credentials in here would change how credential failures surface. We only want
-                    // the reply, which reports version/role/mode/connection-id without INFO or CONFIG (both
-                    // @dangerous, hence often ACL-restricted); see #2968
-                    var hello = Message.CreateHello(helloProtocol, null, null, null, CommandFlags.FireAndForget | Message.NoFlushFlag);
-                    hello.SetInternalCall();
-                    await WriteDirectOrQueueFireAndForgetAsync(connection, hello, autoConfig ??= ResultProcessor.AutoConfigureProcessor.Create(log)).ForAwait();
-                }
-
-                await AutoConfigureAsync(connection, log, extraFlags: Message.NoFlushFlag, helloPending: negotiateResp3 || discoveryHello).ForAwait();
+                await AutoConfigureAsync(connection, log, extraFlags: Message.NoFlushFlag, helloPending: helloAvailable).ForAwait();
             }
 
             // note that the final messages *are* flushed (no Message.NoFlushFlag)

tests/StackExchange.Redis.Tests/ConfigTests.cs

@@ -774,7 +774,6 @@ public async Task MutableOptions()
         var originalUser = options.User = "originalUser";
         var originalPassword = options.Password = "originalPassword";
         Assert.Equal("Details", options.ClientName);
-        Assert.SkipWhen(options.TryResp3(), "only validate RESP2");
         Log(options.ToString());
         await using var conn = await ConnectionMultiplexer.ConnectAsync(options, log: Writer);
         Assert.NotNull(conn.AuthException);

tests/StackExchange.Redis.Tests/HelloHandshakeTests.cs

@@ -102,6 +102,64 @@ public async Task ReplicaProbeStillUsedWhenHelloUnavailable()
         Assert.All(sets, args => Assert.Equal("replica-read-only", args.Skip(1).FirstOrDefault()));
     }
 
+    /// <summary>
+    /// Redis (7.4 through at least 8.x; valkey is unaffected) drops the error reply of a *failing* <c>AUTH</c>
+    /// unless that AUTH is the first command in the pipelined batch, which desynchronizes every reply after it.
+    /// During the handshake that means the tracer never gets its answer and the connection never becomes usable
+    /// (an authentication failure presenting as a connection that times out everything). So the handshake sends
+    /// AUTH first and never sends both a credential-carrying <c>HELLO</c> and a standalone <c>AUTH</c>.
+    /// </summary>
+    [Theory]
+    [InlineData(RedisProtocol.Resp2)]
+    [InlineData(RedisProtocol.Resp3)]
+    public async Task CredentialsGoViaAuthNotHello(RedisProtocol protocol)
+    {
+        using var server = new RecordingServer(log) { Password = "correcthorse" };
+        var config = server.GetClientConfig();
+        config.Protocol = protocol;
+        config.Password = "correcthorse";
+
+        await using var conn = await ConnectionMultiplexer.ConnectAsync(config);
+
+        // AUTH carries the credentials...
+        var auths = server.Recorded("AUTH");
+        Assert.NotEmpty(auths);
+        Assert.All(auths, args => Assert.Equal("correcthorse", args.FirstOrDefault()));
+
+        // ...and HELLO is bare: just the protocol version, no AUTH (or SETNAME) clause
+        var hellos = server.Recorded("HELLO");
+        Assert.NotEmpty(hellos);
+        Assert.All(hellos, args => Assert.Single(args));
+
+        // and AUTH is issued first, so the connection is authenticated before HELLO runs
+        Assert.Equal("AUTH", server.RecordedOrder().First(x => x is "AUTH" or "HELLO"));
+    }
+
+    /// <summary>
+    /// The converse: when <c>AUTH</c> is unavailable, <c>HELLO</c> is the only way to authenticate, so it does
+    /// carry the credentials - and is issued first, since nothing else can authenticate the connection.
+    /// </summary>
+    /// <remarks>Only RESP3 here: a password with <c>AUTH</c> disabled is rejected up-front on RESP2, see the
+    /// <c>AssertAvailable</c> check in the <see cref="ConnectionMultiplexer"/> constructor.</remarks>
+    [Fact]
+    public async Task HelloCarriesCredentialsWhenAuthUnavailable()
+    {
+        const RedisProtocol protocol = RedisProtocol.Resp3;
+        using var server = new RecordingServer(log) { Password = "correcthorse" };
+        var config = server.GetClientConfig();
+        config.Protocol = protocol;
+        config.Password = "correcthorse";
+        config.CommandMap = server.CreateCommandMap(except: "AUTH");
+
+        await using var conn = await ConnectionMultiplexer.ConnectAsync(config);
+
+        Assert.Empty(server.Recorded("AUTH"));
+        var hellos = server.Recorded("HELLO");
+        Assert.NotEmpty(hellos);
+        Assert.All(hellos, args => Assert.Equal(["AUTH", "default", "correcthorse"], args.Skip(1).Take(3)));
+        Assert.Equal("HELLO", server.RecordedOrder().First());
+    }
+
     private sealed class RecordingServer(ITestOutputHelper log) : InProcessTestServer(log)
     {
         private readonly ConcurrentQueue<(string Command, string[] Args)> _commands = new();
@@ -120,6 +178,8 @@ public override TypedRedisValue Execute(RedisClient client, in RedisRequest requ
         public List<string[]> Recorded(string command)
             => _commands.Where(x => x.Command == command).Select(x => x.Args).ToList();
 
+        public List<string> RecordedOrder() => _commands.Select(x => x.Command).ToList();
+
         public CommandMap CreateCommandMap(params string[] except)
         {
             var commands = GetCommands();