diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs index 049824536..38051161c 100644 --- a/src/StackExchange.Redis/PhysicalBridge.cs +++ b/src/StackExchange.Redis/PhysicalBridge.cs @@ -784,6 +784,19 @@ internal bool TryEnqueue(List 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; } + /// + /// 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). + /// + internal void AcceptRerouted(Message message) + { + BacklogEnqueue(message); + StartBacklogProcessor(); + } + [MethodImpl(MethodImplOptions.AggressiveInlining)] private void BacklogEnqueue(Message message) { diff --git a/src/StackExchange.Redis/ServerEndPoint.cs b/src/StackExchange.Redis/ServerEndPoint.cs index ee5cd3281..e48274ed8 100644 --- a/src/StackExchange.Redis/ServerEndPoint.cs +++ b/src/StackExchange.Redis/ServerEndPoint.cs @@ -283,6 +283,23 @@ public void Dispose() : interactive ??= CreateBridge(ConnectionType.Interactive, null); } + /// + /// 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). + /// + /// true if the message is now the subscription bridge's responsibility. + 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(PhysicalConnection? c return bridge; } + /// + /// Issues HELLO, optionally carrying the credentials and client name. + /// + /// The server can reject RESP3 either with an error (HELLO 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). + 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,38 +1058,44 @@ 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()); @@ -1067,6 +1103,14 @@ private async Task HandshakeAsync(PhysicalConnection connection, ILogger? log) 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) diff --git a/tests/StackExchange.Redis.Tests/ConfigTests.cs b/tests/StackExchange.Redis.Tests/ConfigTests.cs index 098430fa2..6c62e3bb2 100644 --- a/tests/StackExchange.Redis.Tests/ConfigTests.cs +++ b/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); diff --git a/tests/StackExchange.Redis.Tests/HelloHandshakeTests.cs b/tests/StackExchange.Redis.Tests/HelloHandshakeTests.cs index 1b929802c..687a55a62 100644 --- a/tests/StackExchange.Redis.Tests/HelloHandshakeTests.cs +++ b/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())); } + /// + /// Redis (7.4 through at least 8.x; valkey is unaffected) drops the error reply of a *failing* AUTH + /// 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 HELLO and a standalone AUTH. + /// + [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")); + } + + /// + /// The converse: when AUTH is unavailable, HELLO is the only way to authenticate, so it does + /// carry the credentials - and is issued first, since nothing else can authenticate the connection. + /// + /// Only RESP3 here: a password with AUTH disabled is rejected up-front on RESP2, see the + /// AssertAvailable check in the constructor. + [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 Recorded(string command) => _commands.Where(x => x.Command == command).Select(x => x.Args).ToList(); + public List RecordedOrder() => _commands.Select(x => x.Command).ToList(); + public CommandMap CreateCommandMap(params string[] except) { var commands = GetCommands(); diff --git a/tests/StackExchange.Redis.Tests/Resp3DowngradeTests.cs b/tests/StackExchange.Redis.Tests/Resp3DowngradeTests.cs new file mode 100644 index 000000000..2694cf430 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/Resp3DowngradeTests.cs @@ -0,0 +1,141 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Net; +using System.Threading.Tasks; +using StackExchange.Redis.Server; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// A connection can negotiate RESP3 and then, on a later reconnect, fail to (the HELLO times out, a token +/// expires, the endpoint moves to a down-level server, ...). We then downgrade to RESP2, which needs pub/sub on +/// its own connection - but subscriptions queued while we still expected RESP3 are sitting in the *interactive* +/// bridge's backlog. Writing those to the interactive connection puts it into subscriber mode, which breaks +/// every normal command on it; see issue #3154. +/// +public class Resp3DowngradeTests(ITestOutputHelper log) +{ + [Fact] + public async Task SubscribeQueuedBeforeDowngradeDoesNotPoisonInteractive() + { + using var server = new DowngradeServer(log); + var config = server.GetClientConfig(); + config.Protocol = RedisProtocol.Resp3; + config.AllowSimulateConnectionFailure = true; + config.AbortOnConnectFail = false; + config.ConnectTimeout = 5000; + + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + var endpoint = conn.GetServerSnapshot()[0]; + Assert.Equal(RedisProtocol.Resp3, endpoint.Protocol); + + var sub = conn.GetSubscriber(); + var db = conn.GetDatabase(); + RedisChannel channel = RedisChannel.Literal("resp3-downgrade"); + ConcurrentBag received = []; + await sub.SubscribeAsync(channel, (_, value) => received.Add(value!)); + + // from here on the server no longer understands HELLO, so the next handshake lands on RESP2; and hold + // the door shut so that we get a window where we are disconnected but still expecting RESP3 + server.HelloSupported = false; + server.BlockAccepts(); + Assert.SkipUnless(endpoint.CanSimulateConnectionFailure, "Cannot simulate connection failure"); + endpoint.SimulateConnectionFailure(SimulatedFailureType.All); + await server.WaitForBlockedAcceptAsync(); + + // queue a subscribe while disconnected: this goes to the backlog of whichever bridge we *expect* to + // need, and we still expect RESP3 - i.e. the interactive bridge + await sub.SubscribeAsync(channel, (_, value) => received.Add(value!), CommandFlags.FireAndForget); + + // now let it reconnect (as RESP2) and drain that backlog + server.ReleaseAccepts(); + await UntilConditionAsync(TimeSpan.FromSeconds(10), () => endpoint.Protocol == RedisProtocol.Resp2); + Assert.Equal(RedisProtocol.Resp2, endpoint.Protocol); + + // the interactive connection must still be usable for normal commands + RedisKey key = "resp3-downgrade-key"; + await db.StringSetAsync(key, "value"); + Assert.Equal("value", await db.StringGetAsync(key)); + + // ...and pub/sub must still work, on the subscription connection now + await sub.PublishAsync(channel, "payload"); + await UntilConditionAsync(TimeSpan.FromSeconds(10), () => !received.IsEmpty); + Assert.Contains("payload", received); + + // finally, the invariant that matters (and that the assertions above only observe indirectly): under + // RESP2 no single connection may carry both subscriptions and ordinary commands + var mixed = server.ConnectionsMixingSubscriptionsAndCommands(RedisProtocol.Resp2); + Assert.Empty(mixed); + } + + private static async Task UntilConditionAsync(TimeSpan timeout, Func condition) + { + var deadline = DateTime.UtcNow + timeout; + while (!condition() && DateTime.UtcNow < deadline) + { + await Task.Delay(50, TestContext.Current.CancellationToken); + } + } + + private sealed class DowngradeServer(ITestOutputHelper log) : InProcessTestServer(log) + { + private static readonly HashSet SubscriptionCommands = new(StringComparer.OrdinalIgnoreCase) + { + "SUBSCRIBE", "UNSUBSCRIBE", "PSUBSCRIBE", "PUNSUBSCRIBE", "SSUBSCRIBE", "SUNSUBSCRIBE", + }; + + // commands that only ever make sense on a connection serving ordinary traffic + private static readonly HashSet InteractiveCommands = new(StringComparer.OrdinalIgnoreCase) + { + "GET", "SET", "DEL", "EXISTS", "INFO", "CONFIG", "CLUSTER", "PUBLISH", + }; + + private readonly ConcurrentDictionary> _byClient = new(); + private TaskCompletionSource? _gate, _blocked; + + public bool HelloSupported { get; set; } = true; + + public void BlockAccepts() + { + // everything recorded from here on belongs to the post-downgrade connections; the RESP3 connection + // that preceded them legitimately mixed subscriptions and ordinary commands + _byClient.Clear(); + _blocked = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + } + + public Task WaitForBlockedAcceptAsync() => _blocked?.Task ?? Task.CompletedTask; + + public void ReleaseAccepts() => _gate?.TrySetResult(true); + + protected override async ValueTask OnAcceptClientAsync(EndPoint endpoint) + { + if (_gate is { } gate) + { + _blocked?.TrySetResult(true); + await gate.Task; + } + } + + public override TypedRedisValue Execute(RedisClient client, in RedisRequest request) + { + _byClient.GetOrAdd(client.Id, _ => new()).Enqueue(request.GetString(0).ToUpperInvariant()); + return base.Execute(client, in request); + } + + protected override TypedRedisValue Hello(RedisClient client, in RedisRequest request) + => HelloSupported ? base.Hello(client, in request) : request.CommandNotFound(); + + /// + /// Connections that carried both subscription and ordinary commands; legitimate under RESP3 (one + /// connection does everything), fatal under RESP2 (subscriber mode rejects ordinary commands). + /// + public List ConnectionsMixingSubscriptionsAndCommands(RedisProtocol protocol) => + [.. _byClient + .Where(pair => pair.Value.Any(SubscriptionCommands.Contains) && pair.Value.Any(InteractiveCommands.Contains)) + .Select(pair => $"[{protocol}] client {pair.Key}: {string.Join(",", pair.Value)}")]; + } +}