diff --git a/Directory.Build.props b/Directory.Build.props index 62c89b137..86c8c6ccc 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -11,7 +11,7 @@ $(MSBuildThisFileDirectory)Shared.ruleset NETSDK1069 - $(NoWarn);NU5105;NU1507;SER001;SER004;SER005;SER007;SER008 + $(NoWarn);NU5105;NU1507;SER001;SER004;SER005;SER007;SER008;SER009 https://github.com/StackExchange/StackExchange.Redis/releases https://stackexchange.github.io/StackExchange.Redis/ MIT diff --git a/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt b/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt index 3b10f3da4..4c74fea9a 100644 --- a/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/RESPite/PublicAPI/PublicAPI.Unshipped.txt @@ -7,6 +7,7 @@ [SER009]abstract RESPite.Transports.DuplexTransport.GetMemory(int sizeHint = 0) -> System.Memory [SER009]virtual RESPite.Transports.DuplexTransport.GetSpan(int sizeHint = 0) -> System.Span [SER009]abstract RESPite.Transports.DuplexTransport.Start(RESPite.Transports.TransportReceiver! receiver) -> void +[SER009]virtual RESPite.Transports.DuplexTransport.IsEncrypted.get -> bool [SER009]RESPite.Transports.TransportReceiver [SER009]RESPite.Transports.TransportReceiver.TransportReceiver() -> void [SER009]abstract RESPite.Transports.TransportReceiver.OnReceived(System.ReadOnlySpan payload) -> bool diff --git a/src/RESPite/Transports/DuplexTransport.cs b/src/RESPite/Transports/DuplexTransport.cs index 2701d60b6..68e7996f2 100644 --- a/src/RESPite/Transports/DuplexTransport.cs +++ b/src/RESPite/Transports/DuplexTransport.cs @@ -47,6 +47,15 @@ public abstract class DuplexTransport : IBufferWriter, IAsyncDisposable /// delivery runs on the transport's schedule and threads. public abstract void Start(TransportReceiver receiver); + /// + /// Whether this transport's bytes are encrypted on the wire. Since the transport owns connect and + /// TLS end-to-end, this is the only thing a consumer can ask; it is the transport's assertion, and + /// the default is the safe answer (no). A transport that terminates TLS itself MUST override this, + /// otherwise a configuration that demands TLS will refuse to use it. + /// + /// Valid once the transport is connected; consumers check it before first use. + public virtual bool IsEncrypted => false; + public abstract ValueTask DisposeAsync(); } diff --git a/src/StackExchange.Redis/Configuration/LoggingTunnel.cs b/src/StackExchange.Redis/Configuration/LoggingTunnel.cs index 42a85b710..a76231cd7 100644 --- a/src/StackExchange.Redis/Configuration/LoggingTunnel.cs +++ b/src/StackExchange.Redis/Configuration/LoggingTunnel.cs @@ -12,6 +12,7 @@ using RESPite; using RESPite.Buffers; using RESPite.Messages; +using RESPite.Transports; using static StackExchange.Redis.PhysicalConnection; namespace StackExchange.Redis.Configuration; @@ -315,6 +316,18 @@ internal DirectoryLoggingTunnel(string path, ConfigurationOptions? options = nul } protected override Stream Log(Stream stream, EndPoint endpoint, ConnectionType connectionType) + { + CreateLogFiles(endpoint, connectionType, out var reads, out var writes); + return new LoggingDuplexStream(stream, reads, writes); + } + + protected override DuplexTransport Log(DuplexTransport transport, EndPoint endpoint, ConnectionType connectionType) + { + CreateLogFiles(endpoint, connectionType, out var reads, out var writes); + return new LoggingDuplexTransport(transport, reads, writes); + } + + private void CreateLogFiles(EndPoint endpoint, ConnectionType connectionType, out Stream reads, out Stream writes) { int index = Interlocked.Increment(ref _nextIndex); var name = $"{Format.ToString(endpoint)} {connectionType} {index}.tmp"; @@ -323,9 +336,8 @@ protected override Stream Log(Stream stream, EndPoint endpoint, ConnectionType c name = name.Replace(c, ' '); } name = Path.Combine(path, name); - var reads = File.Create(Path.ChangeExtension(name, ".in")); - var writes = File.Create(Path.ChangeExtension(name, ".out")); - return new LoggingDuplexStream(stream, reads, writes); + reads = File.Create(Path.ChangeExtension(name, ".in")); + writes = File.Create(Path.ChangeExtension(name, ".out")); } private static readonly char[] InvalidChars = Path.GetInvalidFileNameChars(); @@ -352,6 +364,30 @@ protected override Stream Log(Stream stream, EndPoint endpoint, ConnectionType c /// protected abstract Stream Log(Stream stream, EndPoint endpoint, ConnectionType connectionType); + /// + public override async ValueTask ConnectTransportAsync(EndPoint endpoint, ConnectionType connectionType, TlsOptions tls, CancellationToken cancellationToken) + { + if (_tail is null) return null; // nothing to wrap; the socket path applies + + // in transport mode the tail owns connect *and* TLS, so it needs the original TLS intent - which + // our constructor deliberately cleared from the live options, since when we do the handshake + // ourselves the library must not also wrap the connection. Note that what we log here is + // therefore the plaintext RESP either way: we sit above the tail's own encryption. + if (_ssl) tls = tls.WithTls(); + + var transport = await _tail.ConnectTransportAsync(endpoint, connectionType, tls, cancellationToken).ForAwait(); + return transport is null ? null : Log(transport, endpoint, connectionType); + } + + /// + /// Perform logging on the provided transport (push-mode transports, i.e. tunnels that supply an + /// entire rather than a socket or ). + /// + /// The default implementation returns the transport unchanged, i.e. connects but does not + /// log; override it (typically returning a ) to capture traffic + /// in transport mode. + protected virtual DuplexTransport Log(DuplexTransport transport, EndPoint endpoint, ConnectionType connectionType) => transport; + /// public override ValueTask BeforeSocketConnectAsync(EndPoint endPoint, ConnectionType connectionType, Socket? socket, CancellationToken cancellationToken) { @@ -486,6 +522,120 @@ public static string DefaultFormatResponse(RedisResult value) } #pragma warning disable CS1591 // Missing XML comment for publicly visible type or member + /// + /// Captures the traffic of a push-mode into a pair of streams, in the + /// same format produces, so the same replay/validate tooling reads + /// both. Outbound bytes are captured at (the last point at which they are still + /// ours to read); inbound bytes are captured as they are delivered, before the consumer sees them. + /// + protected sealed class LoggingDuplexTransport : DuplexTransport + { + private readonly DuplexTransport _inner; + private readonly Stream _reads, _writes; + private Memory _pending; // the buffer most recently handed out, so Advance can log from it + private volatile bool _logsClosed; + + internal LoggingDuplexTransport(DuplexTransport inner, Stream reads, Stream writes) + { + _inner = inner; + _reads = reads; + _writes = writes; + } + + // the wrapper adds capture, not encryption; whether the bytes are encrypted is the inner + // transport's assertion to make + public override bool IsEncrypted => _inner.IsEncrypted; + + public override Memory GetMemory(int sizeHint = 0) => _pending = _inner.GetMemory(sizeHint); + + // deliberately NOT delegating to _inner.GetSpan: we need the memory to still be reachable at + // Advance, which is the only point where we learn how much of it was actually written + public override Span GetSpan(int sizeHint = 0) => GetMemory(sizeHint).Span; + + public override void Advance(int count) + { + if (count > 0 && !_logsClosed) + { + // capture *before* advancing; from Advance onwards the bytes belong to the transport, + // which is free to send and recycle them + WriteLog(_writes, _pending.Span.Slice(0, count)); + } + _pending = default; + _inner.Advance(count); + } + + public override bool Flush() + { + // capture first, so the log is never behind the wire + if (!_logsClosed) _writes.Flush(); + return _inner.Flush(); + } + + public override void Start(TransportReceiver receiver) => _inner.Start(new LoggingReceiver(this, receiver)); + + public override async ValueTask DisposeAsync() + { + try + { + await _inner.DisposeAsync().ForAwait(); + } + finally + { + CloseLogs(); + } + } + + private void CloseLogs() + { + if (_logsClosed) return; + _logsClosed = true; + try { _reads.Flush(); } catch { } + try { _reads.Dispose(); } catch { } + try { _writes.Flush(); } catch { } + try { _writes.Dispose(); } catch { } + } + + private static void WriteLog(Stream target, ReadOnlySpan payload) + { +#if NET + target.Write(payload); +#else + var lease = ArrayPool.Shared.Rent(payload.Length); + payload.CopyTo(lease); + target.Write(lease, 0, payload.Length); + ArrayPool.Shared.Return(lease); +#endif + } + + private sealed class LoggingReceiver(LoggingDuplexTransport parent, TransportReceiver tail) : TransportReceiver + { + public override bool OnReceived(ReadOnlySpan payload) + { + // capture before handing on, so a capture exists even if the consumer faults on it + if (!parent._logsClosed) + { + WriteLog(parent._reads, payload); + parent._reads.Flush(); + } + return tail.OnReceived(payload); + } + + public override void OnBatchEnd() => tail.OnBatchEnd(); + + public override void OnClosed(Exception? fault) + { + try + { + tail.OnClosed(fault); + } + finally + { + parent.CloseLogs(); + } + } + } + } + protected sealed class LoggingDuplexStream : Stream { private readonly Stream _inner, _reads, _writes; diff --git a/src/StackExchange.Redis/Configuration/TlsOptions.cs b/src/StackExchange.Redis/Configuration/TlsOptions.cs new file mode 100644 index 000000000..2e0d9f533 --- /dev/null +++ b/src/StackExchange.Redis/Configuration/TlsOptions.cs @@ -0,0 +1,107 @@ +using System; +using System.Diagnostics.CodeAnalysis; +using System.Net; +using System.Net.Security; +using System.Security.Authentication; + +namespace StackExchange.Redis.Configuration; + +/// +/// A read-only view over just the TLS-relevant parts of a , for +/// transports that own TLS themselves (see ): a transport +/// cannot honour an intent it cannot see, but it has no business with the rest of the configuration. +/// +/// +/// This wraps the configuration rather than copying it: the accessors proxy through, so the cost of +/// passing this around is one reference, and nothing here can mutate the configuration or reach +/// anything beyond TLS. A default instance reports "no TLS". +/// +[Experimental(RESPite.Experiments.Transport, UrlFormat = RESPite.Experiments.UrlFormat)] +public readonly struct TlsOptions +{ + private readonly ConfigurationOptions? _options; + private readonly bool _forceTls; + + /// + /// Create a view over the TLS settings of the supplied configuration. + /// + public TlsOptions(ConfigurationOptions options) + : this(options ?? throw new ArgumentNullException(nameof(options)), forceTls: false) + { + } + + private TlsOptions(ConfigurationOptions? options, bool forceTls) + { + _options = options; + _forceTls = forceTls; + } + + /// + /// Whether TLS is required for this connection; corresponds to . + /// A transport that returns false from + /// when this is set will not be used. + /// + public bool IsEnabled => _forceTls || (_options?.Ssl ?? false); + + /// + /// The same view, but with TLS required regardless of what the configuration says - for the intermediate + /// tunnel that cleared for its own reasons but is chaining to a + /// tail that owns the handshake. Everything else (host, protocols, callbacks) is unchanged. + /// + internal TlsOptions WithTls() => new(_options, forceTls: true); + + /// + /// The configured TLS host name, used for SNI and certificate validation; corresponds to + /// . Usually prefer , which + /// applies the same endpoint fallback the library's own TLS path uses. + /// + public string? SslHost => _options?.SslHost; + + /// + /// The permitted TLS protocol versions, or null for the platform default; corresponds to + /// . + /// + public SslProtocols? SslProtocols => _options?.SslProtocols; + + /// + /// Whether the certificate revocation list should be checked during authentication; corresponds to + /// . + /// + public bool CheckCertificateRevocation => _options?.CheckCertificateRevocation ?? false; + + /// + /// The callback that validates the server certificate, or null for the platform default. + /// + /// This includes the ambient environment fallback (SERedis_IssuerCertPath), so a + /// transport that uses this behaves as the library's own TLS path does. + public RemoteCertificateValidationCallback? CertificateValidationCallback + => _options is null ? null : _options.CertificateValidationCallback ?? PhysicalConnection.GetAmbientIssuerCertificateCallback(); + + /// + /// The callback that selects the client certificate, or null for the platform default. + /// + /// This includes the ambient environment fallback (SERedis_ClientCertPfxPath etc), so + /// a transport that uses this behaves as the library's own TLS path does. + public LocalCertificateSelectionCallback? CertificateSelectionCallback + => _options is null ? null : _options.CertificateSelectionCallback ?? PhysicalConnection.GetAmbientClientCertificateCallback(); + +#if NET + /// + /// The caller-supplied authentication options for the given host, if any; corresponds to + /// . When this returns non-null it + /// supersedes the individual values here, exactly as it does on the library's own TLS path. + /// + public SslClientAuthenticationOptions? GetSslClientAuthenticationOptions(string host) + => _options?.SslClientAuthenticationOptions?.Invoke(host); +#endif + + /// + /// The TLS host name to use for the given endpoint: the configured if there is + /// one, otherwise the host portion of the endpoint - which is what the library's own TLS path does. + /// + public string ResolveHost(EndPoint endpoint) + { + var host = SslHost; + return host.IsNullOrWhiteSpace() ? Format.ToStringHostOnly(endpoint) : host!; + } +} diff --git a/src/StackExchange.Redis/Configuration/Tunnel.cs b/src/StackExchange.Redis/Configuration/Tunnel.cs index 72cdbc4d2..85438a54f 100644 --- a/src/StackExchange.Redis/Configuration/Tunnel.cs +++ b/src/StackExchange.Redis/Configuration/Tunnel.cs @@ -37,14 +37,28 @@ public abstract class Tunnel public virtual ValueTask BeforeAuthenticateAsync(EndPoint endpoint, ConnectionType connectionType, Socket? socket, CancellationToken cancellationToken) => default; /// - /// Optionally supply the ENTIRE transport for this connection — the same hijack as + /// Optionally supply the ENTIRE transport for this connection - the same hijack as /// one level deeper: instead of yielding a /// over a socket the library owns, yield a /// the tunnel owns, and no socket is created at /// all. Return null (the default for every existing tunnel) for the standard socket path. /// + /// The logical endpoint being connected to. + /// Whether this is the interactive or subscription connection. + /// + /// The TLS intent this transport must honour: since the transport owns connect and TLS + /// end-to-end, the library does not (and cannot) apply + /// and friends on its behalf. + /// + /// Cancellation for the connect attempt. + /// + /// A transport that establishes TLS must report that via + /// ; if + /// is set and the transport reports otherwise, the + /// connection is failed rather than used. + /// [Experimental(RESPite.Experiments.Transport, UrlFormat = RESPite.Experiments.UrlFormat)] - public virtual ValueTask ConnectTransportAsync(EndPoint endpoint, ConnectionType connectionType, CancellationToken cancellationToken) => default; + public virtual ValueTask ConnectTransportAsync(EndPoint endpoint, ConnectionType connectionType, TlsOptions tls, CancellationToken cancellationToken) => default; private sealed class HttpProxyTunnel : Tunnel { diff --git a/src/StackExchange.Redis/LoggerExtensions.cs b/src/StackExchange.Redis/LoggerExtensions.cs index 381ee4426..552a38115 100644 --- a/src/StackExchange.Redis/LoggerExtensions.cs +++ b/src/StackExchange.Redis/LoggerExtensions.cs @@ -61,7 +61,13 @@ internal static void LogWithThreadPoolStats(this ILogger? log, string message) log.LogInformationThreadPoolStats(composed); } - // Generated LoggerMessage methods + // Generated LoggerMessage methods. + // + // NOTE: renaming any method here is NOT a refactor. The generator emits the method name as the event + // name -- new EventId(, nameof()) -- so the name reaches consumers' telemetry (EventId.Name; + // "EventName" in most structured sinks), where it may be filtered or alerted on. At our download volume + // that is not a cost worth paying for tidiness: treat these names as fixed once shipped, however + // awkwardly they read. If a name genuinely must change, pin the old one via the attribute's EventName. [LoggerMessage( Level = LogLevel.Error, Message = "Connection failed: {EndPoint} ({ConnectionType}, {FailureType}): {ErrorMessage}")] @@ -648,6 +654,9 @@ internal static void LogWithThreadPoolStats(this ILogger? log, string message) Message = "{EndPoint}: (socket shutdown)")] internal static partial void LogErrorSocketShutdown(this ILogger logger, Exception exception, EndPointLogValue endPoint); + // "TLS" here breaks the usual TLA convention (compare SslStream, TlsCipherSuite, and TlsOptions in this + // repo), and is kept that way ON PURPOSE: these names shipped, so they are the event names our consumers + // already see. See the note at the top of the generated section. [LoggerMessage( Level = LogLevel.Information, EventId = 98, @@ -709,4 +718,21 @@ internal static void LogWithThreadPoolStats(this ILogger? log, string message) EventId = 109, Message = "Service name not defined.")] internal static partial void LogInformationServiceNameNotDefined(this ILogger logger); + +#if NET + // deliberately the SAME event as its sibling above (id 99, same name): one logical event that + // simply carries an extra property where the platform can report it. A new id would mean "TLS connection + // established" arrived under one id on netfx and another on modern TFMs, for no gain to anyone. + [LoggerMessage( + Level = LogLevel.Information, + EventId = 99, + Message = "TLS connection established successfully using protocol: {SslProtocol}, cipher suite: {CipherSuite}")] + internal static partial void LogInformationTLSConnectionEstablished(this ILogger logger, System.Security.Authentication.SslProtocols sslProtocol, System.Net.Security.TlsCipherSuite cipherSuite); +#endif + + [LoggerMessage( + Level = LogLevel.Information, + EventId = 110, + Message = "{BridgeName}: Transport connected (encrypted: {IsEncrypted})")] + internal static partial void LogInformationTransportConnected(this ILogger logger, string bridgeName, bool isEncrypted); } diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs index 049824536..051460dd5 100644 --- a/src/StackExchange.Redis/PhysicalBridge.cs +++ b/src/StackExchange.Redis/PhysicalBridge.cs @@ -1236,6 +1236,36 @@ private void ProcessBridgeBacklog() } } + /// + /// Flush anything earlier writes left staged, but only if that can be done without waiting. + /// + /// + /// Called from a push transport's batch-end callback, which runs on a transport thread that must + /// not block. The write lock is what makes staging exclusive - a flush that raced a stage could + /// hand a half-written frame to a transport mid-Advance - so this either gets the lock + /// instantly or does nothing, leaving the flush to whoever does hold it. + /// + internal void TryFlushStagedWrites(PhysicalConnection physical) + { + if (!_singleWriter.TryTakeInstant()) return; + try + { + if (physical is { HasOutputPipe: true }) + { + physical.Flush(); + } + } + catch + { + // opportunistic: a flush failure here means the transport is going away, and that is + // reported through the receiver's OnClosed + } + finally + { + _singleWriter.Release(); + } + } + public bool HasPendingCallerFacingItems() { if (BacklogHasItems) diff --git a/src/StackExchange.Redis/PhysicalConnection.Read.cs b/src/StackExchange.Redis/PhysicalConnection.Read.cs index 9052f6494..4349c60ad 100644 --- a/src/StackExchange.Redis/PhysicalConnection.Read.cs +++ b/src/StackExchange.Redis/PhysicalConnection.Read.cs @@ -29,6 +29,13 @@ internal static PhysicalConnection Dummy(Stream stream, BufferedStreamWriter.Wri internal void StartReading(CancellationToken cancellation = default) { + if (_transport is { } transport) + { + // push mode: no reader thread/Task -- the transport delivers into the parser directly + StartTransportReading(transport); + return; + } + if (cancellation.CanBeCanceled && cancellation != InputCancel) { cancellation.ThrowIfCancellationRequested(); diff --git a/src/StackExchange.Redis/PhysicalConnection.Transport.cs b/src/StackExchange.Redis/PhysicalConnection.Transport.cs new file mode 100644 index 000000000..a0258166a --- /dev/null +++ b/src/StackExchange.Redis/PhysicalConnection.Transport.cs @@ -0,0 +1,166 @@ +using System; +using System.Buffers; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using RESPite.Buffers; +using RESPite.Streams; +using RESPite.Transports; + +namespace StackExchange.Redis +{ + internal sealed partial class PhysicalConnection + { + // TRANSPORT MODE: when Tunnel.ConnectTransportAsync yields a DuplexTransport, there is no + // socket, no Stream and no SslStream — the tunnel owns connect and TLS end-to-end. Outbound + // rides the existing _output seam via an adapter (the transport IS an IBufferWriter with an + // explicit Flush, which is the BufferedStreamWriter contract minus the Stream); inbound is + // PUSH — the transport calls into the same CommitAndParseFrames the pull loops use, on the + // transport's own threads, so there is no reader thread, no read Task and no read syscall + // surface at all on this side. + private DuplexTransport? _transport; + + internal bool HasTransport => _transport is not null; + + private void InitTransportOutput(DuplexTransport transport) + { + _output = new TransportWriter(transport, OutputCancel); + + // Same reasoning as InitOutput: nothing awaits WriteComplete in production; observe it so + // a teardown-time write fault never becomes an UnobservedTaskException. + _output.WriteComplete.RedisFireAndForget(); +#if DEBUG + if (BridgeCouldBeNull?.Multiplexer?.RawConfig?.OutputLog is { } log) + { + _output.DebugSetLog(log); + } +#endif + } + + private void StartTransportReading(DuplexTransport transport) + { + _readStatus = ReadStatus.Init; + _readState = default; + _readBuffer = CycleBuffer.Create(pool: ReaderBufferPool); + transport.Start(new TransportFeed(this)); + } + + /// Outbound half: the transport is already the writer; this adapter exists only to + /// satisfy the seam every call site already uses. Staging + /// forwards straight through; hands the batch to the transport; the write + /// fault/completion surface is a simple once-only task. + private sealed class TransportWriter(DuplexTransport transport, CancellationToken cancellationToken) + : BufferedStreamWriter(Stream.Null, cancellationToken) + { + private readonly TaskCompletionSource _done = new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _staged; // bytes advanced since the last flush; see HasStagedBytes + + public override Task WriteComplete => _done.Task; + + /// Whether anything has been advanced but not yet flushed; writes that do not demand + /// a flush (fire-and-forget) leave bytes here for a later flush to pick up. Deliberately + /// approximate in the safe direction: it can over-report (a concurrent stage during a flush + /// stays counted), which costs a redundant flush attempt, never a missed one. + public bool HasStagedBytes => Volatile.Read(ref _staged) != 0; + + public override Memory GetMemory(int sizeHint = 0) => transport.GetMemory(sizeHint); + + public override Span GetSpan(int sizeHint = 0) => transport.GetSpan(sizeHint); + + public override void Advance(int count) + { + transport.Advance(count); + Interlocked.Add(ref _staged, count); + OnWritten(count); + } + + // A false return means the transport is closed; that failure surfaces via the receiver's + // OnClosed (which records the connection failure), so it is not duplicated here. + public override void Flush() + { + var flushing = Volatile.Read(ref _staged); + transport.Flush(); + Interlocked.Add(ref _staged, -flushing); + } + + public override void Complete(Exception? exception = null) + { + try { Flush(); } + catch { } + if (exception is null) + { + _done.TrySetResult(true); + } + else + { + _done.TrySetException(exception); + } + } + } + + /// Inbound half: transport-owned spans arrive on the transport's schedule and are + /// committed into the same + + /// machinery the pull loops use — one copy, zero thread hops, no reader loop. + private sealed class TransportFeed(PhysicalConnection connection) : TransportReceiver + { + public override bool OnReceived(ReadOnlySpan payload) + { + var conn = connection; + try + { + while (!payload.IsEmpty) + { + var space = conn._readBuffer.GetUncommittedMemory().Span; + var take = Math.Min(space.Length, payload.Length); + payload.Slice(0, take).CopyTo(space); + payload = payload.Slice(take); + + conn._readStatus = ReadStatus.TryParseResult; + if (!conn.CommitAndParseFrames(take)) + { + return false; + } + } + conn.UpdateLastReadTime(); + return !conn.ForceReconnect; + } + catch (Exception ex) + { + conn._readStatus = ReadStatus.Faulted; + conn.RecordConnectionFailed(ConnectionFailureType.InternalFailure, ex); + return false; + } + } + + public override void OnBatchEnd() + { + // The contract asks for ONE flush per delivery burst rather than per OnReceived. The only + // thing that can be staged-but-unflushed here is a write issued by inbound processing + // that did not demand a flush (typically a fire-and-forget command from a completion + // continuation running inline on this thread), so check before reaching for the lock - + // the common case is nothing staged and nothing to do. + var conn = connection; + if (conn._output is TransportWriter { HasStagedBytes: true } && conn.BridgeCouldBeNull is { } bridge) + { + bridge.TryFlushStagedWrites(conn); + } + } + + public override void OnClosed(Exception? fault) + { + var conn = connection; + conn._readStatus = ReadStatus.RanToCompletion; + conn._readBuffer.Release(); + conn._readBuffer = default; + if (fault is null) + { + conn.RecordConnectionFailed(ConnectionFailureType.SocketClosed); + } + else + { + conn.RecordConnectionFailed(ConnectionFailureType.InternalFailure, fault); + } + } + } + } +} diff --git a/src/StackExchange.Redis/PhysicalConnection.cs b/src/StackExchange.Redis/PhysicalConnection.cs index e44f48f04..0d93eefeb 100644 --- a/src/StackExchange.Redis/PhysicalConnection.cs +++ b/src/StackExchange.Redis/PhysicalConnection.cs @@ -180,7 +180,20 @@ internal async Task BeginConnectAsync(ILogger? log) var connectTo = endpoint; if (tunnel is not null) { - connectTo = await tunnel.GetSocketConnectEndpointAsync(endpoint, CancellationToken.None).ForAwait(); + // A transport tunnel replaces the socket outright (the widest form of the existing + // connectTo=null no-socket pattern): connect and TLS belong to the tunnel, and the + // stream/SslStream machinery below never runs. The TLS intent goes with it precisely + // because TLS is now the tunnel's job: it cannot honour an intent it cannot see. + var transport = await tunnel.ConnectTransportAsync(endpoint, bridge.ConnectionType, new(rawConfig), CancellationToken.None).ForAwait(); + if (transport is not null) + { + _transport = transport; + connectTo = null; + } + else + { + connectTo = await tunnel.GetSocketConnectEndpointAsync(endpoint, CancellationToken.None).ForAwait(); + } } if (connectTo is not null) { @@ -314,6 +327,7 @@ internal void Shutdown(ConnectionFailureType failureType = ConnectionFailureType { var output = Interlocked.Exchange(ref _output, null); // compare to the critical read var socket = Interlocked.Exchange(ref _socket, null); + var transport = Interlocked.Exchange(ref _transport, null); if (output != null) { @@ -328,11 +342,16 @@ internal void Shutdown(ConnectionFailureType failureType = ConnectionFailureType try { socket.Close(); } catch { } try { socket.Dispose(); } catch { } } + + if (transport != null) + { + try { transport.DisposeAsync().AsTask().RedisFireAndForget(); } catch { } + } } public void Dispose() { - bool markDisposed = VolatileSocket != null; + bool markDisposed = VolatileSocket != null || _transport != null; Shutdown(); if (markDisposed) { @@ -1014,6 +1033,31 @@ internal async ValueTask ConnectedAsync(Socket? socket, ILogger? log) // TLS: [Socket]<==[NetworkStream]<==[SslStream]<==[StreamConnection:IDuplexPipe] var config = bridge.Multiplexer.RawConfig; + if (_transport is { } transport) + { + // Transport mode: no socket, no stream, no SslStream. TLS is the transport's job + // (it saw the config on ConnectTransportAsync), so the only question here is whether + // the transport DISAGREES with the configured intent. Configured TLS plus a + // transport that reports plaintext is a hard failure: silently sending credentials + // in the clear is exactly what Ssl=true forbids. The converse - a transport that is + // encrypted when the config did not demand it - is fine, and merely noted. + if (config.Ssl && !transport.IsEncrypted) + { + var mismatch = new RedisConnectionException( + ConnectionFailureType.AuthenticationFailure, + CommandFlags.None, + "TLS was requested (Ssl=true), but the transport supplied by the tunnel reports an unencrypted connection."); + RecordConnectionFailed(ConnectionFailureType.AuthenticationFailure, mismatch, isInitialConnect: true); + bridge.Multiplexer.Trace("Transport is not encrypted"); + return false; + } + + InitTransportOutput(transport); + log?.LogInformationTransportConnected(bridge.Name, transport.IsEncrypted); + await bridge.OnConnectedAsync(this, log).ForAwait(); + return true; + } + var tunnel = config.Tunnel; if (tunnel is not null) { @@ -1064,7 +1108,20 @@ static Stream DemandSocketStream(Socket? socket) bridge.Multiplexer.Logger?.LogErrorConnectionIssue(ex, ex.Message); throw; } + // Note on the "assert ssl.IsEncrypted after the handshake" advice: on every TFM we + // target, SslStream.IsEncrypted (and IsSigned) is literally an alias for + // IsAuthenticated, i.e. "handshake completed, no exception" - which is exactly what + // the await above already proved, so asserting it would guarantee nothing about the + // bytes on the wire. What actually forbids plaintext here is the + // EncryptionPolicy.RequireEncryption passed to the ctor above; instead of a + // tautological assert we log what was negotiated, so a surprise is visible. + // .NET: https://github.com/dotnet/dotnet/blob/b0f34d51fccc69fd334253924abd8d6853fad7aa/src/runtime/src/libraries/System.Net.Security/src/System/Net/Security/SslStream.cs#L475 + // netfx: https://github.com/microsoft/referencesource/blob/main/System/net/System/Net/SecureProtocols/SslStream.cs#L312-L334 +#if NET + log?.LogInformationTLSConnectionEstablished(ssl.SslProtocol, ssl.NegotiatedCipherSuite); +#else log?.LogInformationTLSConnectionEstablished(ssl.SslProtocol); +#endif } catch (AuthenticationException authexception) { diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt index f05af89aa..80d94801e 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt @@ -3,5 +3,15 @@ StackExchange.Redis.ConfigurationOptions.SentinelPassword.get -> string? StackExchange.Redis.ConfigurationOptions.SentinelPassword.set -> void StackExchange.Redis.ConfigurationOptions.SentinelUser.get -> string? StackExchange.Redis.ConfigurationOptions.SentinelUser.set -> void +[SER009]StackExchange.Redis.Configuration.TlsOptions +[SER009]StackExchange.Redis.Configuration.TlsOptions.TlsOptions() -> void +[SER009]StackExchange.Redis.Configuration.TlsOptions.TlsOptions(StackExchange.Redis.ConfigurationOptions! options) -> void +[SER009]StackExchange.Redis.Configuration.TlsOptions.IsEnabled.get -> bool +[SER009]StackExchange.Redis.Configuration.TlsOptions.SslHost.get -> string? +[SER009]StackExchange.Redis.Configuration.TlsOptions.SslProtocols.get -> System.Security.Authentication.SslProtocols? +[SER009]StackExchange.Redis.Configuration.TlsOptions.CheckCertificateRevocation.get -> bool +[SER009]StackExchange.Redis.Configuration.TlsOptions.CertificateValidationCallback.get -> System.Net.Security.RemoteCertificateValidationCallback? +[SER009]StackExchange.Redis.Configuration.TlsOptions.CertificateSelectionCallback.get -> System.Net.Security.LocalCertificateSelectionCallback? +[SER009]StackExchange.Redis.Configuration.TlsOptions.ResolveHost(System.Net.EndPoint! endpoint) -> string! +[SER009]virtual StackExchange.Redis.Configuration.Tunnel.ConnectTransportAsync(System.Net.EndPoint! endpoint, StackExchange.Redis.ConnectionType connectionType, StackExchange.Redis.Configuration.TlsOptions tls, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.ValueTask StackExchange.Redis.RedisFeatures.Hello.get -> bool -[SER009]virtual StackExchange.Redis.Configuration.Tunnel.ConnectTransportAsync(System.Net.EndPoint! endpoint, StackExchange.Redis.ConnectionType connectionType, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.ValueTask diff --git a/src/StackExchange.Redis/PublicAPI/net6.0/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/net6.0/PublicAPI.Unshipped.txt index ab058de62..494bb1b62 100644 --- a/src/StackExchange.Redis/PublicAPI/net6.0/PublicAPI.Unshipped.txt +++ b/src/StackExchange.Redis/PublicAPI/net6.0/PublicAPI.Unshipped.txt @@ -1 +1,2 @@ #nullable enable +[SER009]StackExchange.Redis.Configuration.TlsOptions.GetSslClientAuthenticationOptions(string! host) -> System.Net.Security.SslClientAuthenticationOptions?