Implement high integrity mode for commands

src/StackExchange.Redis/Configuration/DefaultOptionsProvider.cs

@@ -104,6 +104,14 @@ public static DefaultOptionsProvider GetProvider(EndPoint endpoint)
         /// </summary>
         public virtual bool CheckCertificateRevocation => true;
 
+        /// <summary>
+        /// A Boolean value that specifies whether to use per-command validation of strict protocol validity
+        /// </summary>
+        /// <remarks>The regular RESP protocol does not include correlation identifiers between requests and responses; in exceptional
+        /// scenarios, protocol desynchronization can occur, which may not be noticed immediately; this option adds additional data
+        /// to ensure that this cannot occur, at the cost of some (small) additional bandwidth usage.</remarks>
+        public virtual bool HighIntegrity => true;
+
         /// <summary>
         /// The number of times to repeat the initial connect cycle if no servers respond promptly.
         /// </summary>

src/StackExchange.Redis/ConfigurationOptions.cs

@@ -110,7 +110,8 @@ internal const string
                 CheckCertificateRevocation = "checkCertificateRevocation",
                 Tunnel = "tunnel",
                 SetClientLibrary = "setlib",
-                Protocol = "protocol";
+                Protocol = "protocol",
+                HighIntegrity = "highIntegrity";
 
             private static readonly Dictionary<string, string> normalizedOptions = new[]
             {
@@ -141,6 +142,7 @@ internal const string
                 WriteBuffer,
                 CheckCertificateRevocation,
                 Protocol,
+                HighIntegrity,
             }.ToDictionary(x => x, StringComparer.OrdinalIgnoreCase);
 
             public static string TryNormalize(string value)
@@ -156,7 +158,7 @@ public static string TryNormalize(string value)
         private DefaultOptionsProvider? defaultOptions;
 
         private bool? allowAdmin, abortOnConnectFail, resolveDns, ssl, checkCertificateRevocation, heartbeatConsistencyChecks,
-                      includeDetailInExceptions, includePerformanceCountersInExceptions, setClientLibrary;
+                      includeDetailInExceptions, includePerformanceCountersInExceptions, setClientLibrary, highIntegrity;
 
         private string? tieBreaker, sslHost, configChannel, user, password;
 
@@ -279,6 +281,18 @@ public bool CheckCertificateRevocation
             set => checkCertificateRevocation = value;
         }
 
+        /// <summary>
+        /// A Boolean value that specifies whether to use per-command validation of strict protocol validity
+        /// </summary>
+        /// <remarks>The regular RESP protocol does not include correlation identifiers between requests and responses; in exceptional
+        /// scenarios, protocol desynchronization can occur, which may not be noticed immediately; this option adds additional data
+        /// to ensure that this cannot occur, at the cost of some (small) additional bandwidth usage.</remarks>
+        public bool HighIntegrity
+        {
+            get => highIntegrity ?? Defaults.HighIntegrity;
+            set => highIntegrity = value;
+        }
+
         /// <summary>
         /// Create a certificate validation check that checks against the supplied issuer even when not known by the machine.
         /// </summary>
@@ -769,6 +783,7 @@ public int ConfigCheckSeconds
             Protocol = Protocol,
             heartbeatInterval = heartbeatInterval,
             heartbeatConsistencyChecks = heartbeatConsistencyChecks,
+            highIntegrity = highIntegrity,
         };
 
         /// <summary>
@@ -849,6 +864,7 @@ public string ToString(bool includePassword)
             Append(sb, OptionKeys.ResponseTimeout, responseTimeout);
             Append(sb, OptionKeys.DefaultDatabase, DefaultDatabase);
             Append(sb, OptionKeys.SetClientLibrary, setClientLibrary);
+            Append(sb, OptionKeys.HighIntegrity, highIntegrity);
             Append(sb, OptionKeys.Protocol, FormatProtocol(Protocol));
             if (Tunnel is { IsInbuilt: true } tunnel)
             {
@@ -894,7 +910,7 @@ private void Clear()
         {
             ClientName = ServiceName = user = password = tieBreaker = sslHost = configChannel = null;
             keepAlive = syncTimeout = asyncTimeout = connectTimeout = connectRetry = configCheckSeconds = DefaultDatabase = null;
-            allowAdmin = abortOnConnectFail = resolveDns = ssl = setClientLibrary = null;
+            allowAdmin = abortOnConnectFail = resolveDns = ssl = setClientLibrary = highIntegrity = null;
             SslProtocols = null;
             defaultVersion = null;
             EndPoints.Clear();
@@ -1013,6 +1029,9 @@ private ConfigurationOptions DoParse(string configuration, bool ignoreUnknown)
                         case OptionKeys.SetClientLibrary:
                             SetClientLibrary = OptionKeys.ParseBoolean(key, value);
                             break;
+                        case OptionKeys.HighIntegrity:
+                            HighIntegrity = OptionKeys.ParseBoolean(key, value);
+                            break;
                         case OptionKeys.Tunnel:
                             if (value.IsNullOrWhiteSpace())
                             {

src/StackExchange.Redis/Message.cs

@@ -1,6 +1,7 @@
 using Microsoft.Extensions.Logging;
 using StackExchange.Redis.Profiling;
 using System;
+using System.Buffers.Binary;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
@@ -52,6 +53,8 @@ internal abstract class Message : ICompletable
     {
         public readonly int Db;
 
+        private int _highIntegrityChecksum;
+
         internal const CommandFlags InternalCallFlag = (CommandFlags)128;
 
         protected RedisCommand command;
@@ -198,6 +201,18 @@ public bool IsAdmin
 
         public bool IsAsking => (Flags & AskingFlag) != 0;
 
+        public bool IsHighIntegrity => _highIntegrityChecksum != 0;
+        public int HighIntegrityChecksum => _highIntegrityChecksum;
+
+        internal void WithHighIntegrity(Random rng)
+        {
+            // create a required value if needed; avoid sentinel zero
+            while (_highIntegrityChecksum is 0)
+            {
+                _highIntegrityChecksum = rng.Next();
+            }
+        }
+
         internal bool IsScriptUnavailable => (Flags & ScriptUnavailableFlag) != 0;
 
         internal void SetScriptUnavailable() => Flags |= ScriptUnavailableFlag;
@@ -710,6 +725,28 @@ internal void WriteTo(PhysicalConnection physical)
             }
         }
 
+        private static ReadOnlySpan<byte> ChecksumTemplate => "$4\r\nXXXX\r\n"u8;
+
+        internal void WriteHighIntegrityChecksumRequest(PhysicalConnection physical)
+        {
+            Debug.Assert(IsHighIntegrity, "should only be used for high-integrity");
+            try
+            {
+                physical.WriteHeader(RedisCommand.ECHO, 1); // use WriteHeader to allow command-rewrite
+
+                Span<byte> chk = stackalloc byte[10];
+                Debug.Assert(ChecksumTemplate.Length == chk.Length, "checksum template length error");
+                ChecksumTemplate.CopyTo(chk);
+                BinaryPrimitives.WriteInt32LittleEndian(chk.Slice(4, 4), _highIntegrityChecksum);
+                physical.WriteRaw(chk);
+            }
+            catch (Exception ex)
+            {
+                physical?.OnInternalError(ex);
+                Fail(ConnectionFailureType.InternalFailure, ex, null, physical?.BridgeCouldBeNull?.Multiplexer);
+            }
+        }
+
         internal static Message CreateHello(int protocolVersion, string? username, string? password, string? clientName, CommandFlags flags)
             => new HelloMessage(protocolVersion, username, password, clientName, flags);
 

src/StackExchange.Redis/PhysicalBridge.cs

@@ -70,6 +70,8 @@ internal sealed class PhysicalBridge : IDisposable
 
         internal string? PhysicalName => physical?.ToString();
 
+        private readonly Random? _highIntegrityEntropy = null;
+
         public DateTime? ConnectedAt { get; private set; }
 
         public PhysicalBridge(ServerEndPoint serverEndPoint, ConnectionType type, int timeoutMilliseconds)
@@ -82,6 +84,10 @@ public PhysicalBridge(ServerEndPoint serverEndPoint, ConnectionType type, int ti
 #if !NETCOREAPP
             _singleWriterMutex = new MutexSlim(timeoutMilliseconds: timeoutMilliseconds);
 #endif
+            if (type == ConnectionType.Interactive && Multiplexer.RawConfig.HighIntegrity)
+            {
+                _highIntegrityEntropy = new Random();
+            }
         }
 
         private readonly int TimeoutMilliseconds;
@@ -1546,10 +1552,27 @@ private WriteResult WriteMessageToServerInsideWriteLock(PhysicalConnection conne
                         break;
                 }
 
+                if (_highIntegrityEntropy is not null && !connection.TransactionActive)
+                {
+                    // make sure this value exists early to avoid a race condition
+                    // if the response comes back super quickly
+                    message.WithHighIntegrity(_highIntegrityEntropy);
+                    Debug.Assert(message.IsHighIntegrity, "message should be high integrity");
+                }
+                else
+                {
+                    Debug.Assert(!message.IsHighIntegrity, "prior high integrity message found during transaction?");
+                }
                 connection.EnqueueInsideWriteLock(message);
                 isQueued = true;
                 message.WriteTo(connection);
 
+                if (message.IsHighIntegrity)
+                {
+                    message.WriteHighIntegrityChecksumRequest(connection);
+                    IncrementOpCount();
+                }
+
                 message.SetRequestSent();
                 IncrementOpCount();