(dev/eng) Support transport feeds

Directory.Build.props

@@ -11,7 +11,7 @@
     <CodeAnalysisRuleset>$(MSBuildThisFileDirectory)Shared.ruleset</CodeAnalysisRuleset>
     <MSBuildWarningsAsMessages>NETSDK1069</MSBuildWarningsAsMessages>
     <!-- SER002/SER003/SER006 retired (Redis 8.4/8.6/8.8 features no longer experimental); IDs reserved, see Experiments.cs -->
-    <NoWarn>$(NoWarn);NU5105;NU1507;SER001;SER004;SER005;SER007;SER008</NoWarn>
+    <NoWarn>$(NoWarn);NU5105;NU1507;SER001;SER004;SER005;SER007;SER008;SER009</NoWarn>
     <PackageReleaseNotes>https://github.com/StackExchange/StackExchange.Redis/releases</PackageReleaseNotes>
     <PackageProjectUrl>https://stackexchange.github.io/StackExchange.Redis/</PackageProjectUrl>
     <PackageLicenseExpression>MIT</PackageLicenseExpression>

src/RESPite/PublicAPI/PublicAPI.Unshipped.txt

@@ -7,6 +7,7 @@
 [SER009]abstract RESPite.Transports.DuplexTransport.GetMemory(int sizeHint = 0) -> System.Memory<byte>
 [SER009]virtual RESPite.Transports.DuplexTransport.GetSpan(int sizeHint = 0) -> System.Span<byte>
 [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<byte> payload) -> bool

src/RESPite/Transports/DuplexTransport.cs

@@ -47,6 +47,15 @@ public abstract class DuplexTransport : IBufferWriter<byte>, IAsyncDisposable
     /// delivery runs on the transport's schedule and threads.</summary>
     public abstract void Start(TransportReceiver receiver);
 
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <remarks>Valid once the transport is connected; consumers check it before first use.</remarks>
+    public virtual bool IsEncrypted => false;
+
     public abstract ValueTask DisposeAsync();
 }
 

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
     /// </summary>
     protected abstract Stream Log(Stream stream, EndPoint endpoint, ConnectionType connectionType);
 
+    /// <inheritdoc/>
+    public override async ValueTask<DuplexTransport?> 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);
+    }
+
+    /// <summary>
+    /// Perform logging on the provided transport (push-mode transports, i.e. tunnels that supply an
+    /// entire <see cref="DuplexTransport"/> rather than a socket or <see cref="Stream"/>).
+    /// </summary>
+    /// <remarks>The default implementation returns the transport unchanged, i.e. connects but does not
+    /// log; override it (typically returning a <see cref="LoggingDuplexTransport"/>) to capture traffic
+    /// in transport mode.</remarks>
+    protected virtual DuplexTransport Log(DuplexTransport transport, EndPoint endpoint, ConnectionType connectionType) => transport;
+
     /// <inheritdoc/>
     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
+    /// <summary>
+    /// Captures the traffic of a push-mode <see cref="DuplexTransport"/> into a pair of streams, in the
+    /// same format <see cref="LoggingDuplexStream"/> produces, so the same replay/validate tooling reads
+    /// both. Outbound bytes are captured at <see cref="Advance"/> (the last point at which they are still
+    /// ours to read); inbound bytes are captured as they are delivered, before the consumer sees them.
+    /// </summary>
+    protected sealed class LoggingDuplexTransport : DuplexTransport
+    {
+        private readonly DuplexTransport _inner;
+        private readonly Stream _reads, _writes;
+        private Memory<byte> _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<byte> 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<byte> 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<byte> payload)
+        {
+#if NET
+            target.Write(payload);
+#else
+            var lease = ArrayPool<byte>.Shared.Rent(payload.Length);
+            payload.CopyTo(lease);
+            target.Write(lease, 0, payload.Length);
+            ArrayPool<byte>.Shared.Return(lease);
+#endif
+        }
+
+        private sealed class LoggingReceiver(LoggingDuplexTransport parent, TransportReceiver tail) : TransportReceiver
+        {
+            public override bool OnReceived(ReadOnlySpan<byte> 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;

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;
+
+/// <summary>
+/// A read-only view over just the TLS-relevant parts of a <see cref="ConfigurationOptions"/>, for
+/// transports that own TLS themselves (see <see cref="Tunnel.ConnectTransportAsync"/>): a transport
+/// cannot honour an intent it cannot see, but it has no business with the rest of the configuration.
+/// </summary>
+/// <remarks>
+/// 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 <c>default</c> instance reports "no TLS".
+/// </remarks>
+[Experimental(RESPite.Experiments.Transport, UrlFormat = RESPite.Experiments.UrlFormat)]
+public readonly struct TlsOptions
+{
+    private readonly ConfigurationOptions? _options;
+    private readonly bool _forceTls;
+
+    /// <summary>
+    /// Create a view over the TLS settings of the supplied configuration.
+    /// </summary>
+    public TlsOptions(ConfigurationOptions options)
+        : this(options ?? throw new ArgumentNullException(nameof(options)), forceTls: false)
+    {
+    }
+
+    private TlsOptions(ConfigurationOptions? options, bool forceTls)
+    {
+        _options = options;
+        _forceTls = forceTls;
+    }
+
+    /// <summary>
+    /// Whether TLS is required for this connection; corresponds to <see cref="ConfigurationOptions.Ssl"/>.
+    /// A transport that returns <c>false</c> from
+    /// <see cref="RESPite.Transports.DuplexTransport.IsEncrypted"/> when this is set will not be used.
+    /// </summary>
+    public bool IsEnabled => _forceTls || (_options?.Ssl ?? false);
+
+    /// <summary>
+    /// The same view, but with TLS required regardless of what the configuration says - for the intermediate
+    /// tunnel that cleared <see cref="ConfigurationOptions.Ssl"/> for its own reasons but is chaining to a
+    /// tail that owns the handshake. Everything else (host, protocols, callbacks) is unchanged.
+    /// </summary>
+    internal TlsOptions WithTls() => new(_options, forceTls: true);
+
+    /// <summary>
+    /// The configured TLS host name, used for SNI and certificate validation; corresponds to
+    /// <see cref="ConfigurationOptions.SslHost"/>. Usually prefer <see cref="ResolveHost"/>, which
+    /// applies the same endpoint fallback the library's own TLS path uses.
+    /// </summary>
+    public string? SslHost => _options?.SslHost;
+
+    /// <summary>
+    /// The permitted TLS protocol versions, or <c>null</c> for the platform default; corresponds to
+    /// <see cref="ConfigurationOptions.SslProtocols"/>.
+    /// </summary>
+    public SslProtocols? SslProtocols => _options?.SslProtocols;
+
+    /// <summary>
+    /// Whether the certificate revocation list should be checked during authentication; corresponds to
+    /// <see cref="ConfigurationOptions.CheckCertificateRevocation"/>.
+    /// </summary>
+    public bool CheckCertificateRevocation => _options?.CheckCertificateRevocation ?? false;
+
+    /// <summary>
+    /// The callback that validates the server certificate, or <c>null</c> for the platform default.
+    /// </summary>
+    /// <remarks>This includes the ambient environment fallback (<c>SERedis_IssuerCertPath</c>), so a
+    /// transport that uses this behaves as the library's own TLS path does.</remarks>
+    public RemoteCertificateValidationCallback? CertificateValidationCallback
+        => _options is null ? null : _options.CertificateValidationCallback ?? PhysicalConnection.GetAmbientIssuerCertificateCallback();
+
+    /// <summary>
+    /// The callback that selects the client certificate, or <c>null</c> for the platform default.
+    /// </summary>
+    /// <remarks>This includes the ambient environment fallback (<c>SERedis_ClientCertPfxPath</c> etc), so
+    /// a transport that uses this behaves as the library's own TLS path does.</remarks>
+    public LocalCertificateSelectionCallback? CertificateSelectionCallback
+        => _options is null ? null : _options.CertificateSelectionCallback ?? PhysicalConnection.GetAmbientClientCertificateCallback();
+
+#if NET
+    /// <summary>
+    /// The caller-supplied authentication options for the given host, if any; corresponds to
+    /// <see cref="ConfigurationOptions.SslClientAuthenticationOptions"/>. When this returns non-null it
+    /// supersedes the individual values here, exactly as it does on the library's own TLS path.
+    /// </summary>
+    public SslClientAuthenticationOptions? GetSslClientAuthenticationOptions(string host)
+        => _options?.SslClientAuthenticationOptions?.Invoke(host);
+#endif
+
+    /// <summary>
+    /// The TLS host name to use for the given endpoint: the configured <see cref="SslHost"/> if there is
+    /// one, otherwise the host portion of the endpoint - which is what the library's own TLS path does.
+    /// </summary>
+    public string ResolveHost(EndPoint endpoint)
+    {
+        var host = SslHost;
+        return host.IsNullOrWhiteSpace() ? Format.ToStringHostOnly(endpoint) : host!;
+    }
+}

src/StackExchange.Redis/Configuration/Tunnel.cs

@@ -37,14 +37,28 @@ public abstract class Tunnel
         public virtual ValueTask<Stream?> BeforeAuthenticateAsync(EndPoint endpoint, ConnectionType connectionType, Socket? socket, CancellationToken cancellationToken) => default;
 
         /// <summary>
-        /// Optionally supply the ENTIRE transport for this connection — the same hijack as
+        /// Optionally supply the ENTIRE transport for this connection - the same hijack as
         /// <see cref="BeforeAuthenticateAsync"/> one level deeper: instead of yielding a
         /// <see cref="Stream"/> over a socket the library owns, yield a
         /// <see cref="RESPite.Transports.DuplexTransport"/> the tunnel owns, and no socket is created at
         /// all. Return null (the default for every existing tunnel) for the standard socket path.
         /// </summary>
+        /// <param name="endpoint">The logical endpoint being connected to.</param>
+        /// <param name="connectionType">Whether this is the interactive or subscription connection.</param>
+        /// <param name="tls">
+        /// The TLS intent this transport must honour: since the transport owns connect and TLS
+        /// end-to-end, the library does not (and cannot) apply <see cref="ConfigurationOptions.Ssl"/>
+        /// and friends on its behalf.
+        /// </param>
+        /// <param name="cancellationToken">Cancellation for the connect attempt.</param>
+        /// <remarks>
+        /// A transport that establishes TLS must report that via
+        /// <see cref="RESPite.Transports.DuplexTransport.IsEncrypted"/>; if
+        /// <see cref="TlsOptions.IsEnabled"/> is set and the transport reports otherwise, the
+        /// connection is failed rather than used.
+        /// </remarks>
         [Experimental(RESPite.Experiments.Transport, UrlFormat = RESPite.Experiments.UrlFormat)]
-        public virtual ValueTask<RESPite.Transports.DuplexTransport?> ConnectTransportAsync(EndPoint endpoint, ConnectionType connectionType, CancellationToken cancellationToken) => default;
+        public virtual ValueTask<RESPite.Transports.DuplexTransport?> ConnectTransportAsync(EndPoint endpoint, ConnectionType connectionType, TlsOptions tls, CancellationToken cancellationToken) => default;
 
         private sealed class HttpProxyTunnel : Tunnel
         {