MOD-14850 Fix crash in race between MR_ClusterAsyncDisconnect and MR…

src/cluster.c

@@ -464,7 +464,10 @@ static void MR_ClusterReconnect(void* ctx){
 
 static void MR_ClusterAsyncDisconnect(void* ctx){
     Node* n = ctx;
-    redisAsyncFree(n->c);
+    if (n->c) {
+        redisAsyncFree(n->c);
+        n->c = NULL;
+    }
 }
 
 static void MR_ClusterOnDisconnectCallback(const struct redisAsyncContext* c, int status){
@@ -644,16 +647,11 @@ static void MR_OnConnectCallback(const struct redisAsyncContext* c, int status){
         }
         SSL *ssl = SSL_new(ssl_context);
         SSL_CTX_free(ssl_context);
-        const redisContextFuncs *old_callbacks = c->c.funcs;
         if (redisInitiateSSL((redisContext *)(&c->c), ssl) != REDIS_OK) {
             const char *err = "Unknown error";
             if (c->c.err != 0) {
                 err = c->c.errstr;
             }
-            // This is a temporary fix to the bug describe on https://github.com/redis/hiredis/issues/1233.
-            // In case of SSL initialization failure. We need to reset the callbacks value, as the `redisInitiateSSL`
-            // function will not do it for us.
-            ((struct redisAsyncContext*)c)->c.funcs = old_callbacks;
             RedisModule_Log(mr_staticCtx, "warning", "SSL auth to %s:%d failed, will initiate retry. %s.", c->c.tcp.host, c->c.tcp.port, err);
             // disconnect async, its not possible to free redisAsyncContext here
             MR_EventLoopAddTask(MR_ClusterAsyncDisconnect, n);

tests/mr_test_module/Makefile

@@ -54,6 +54,7 @@ test: build
 
 test_ssl: build
 	bash ./generate_tests_cert.sh
+	DEBUG=$(DEBUG_RUN) $(HERE)/pytests/run_tests.sh -t ./pytests/ --tls --tls-cert-file ./tests/tls/redis.crt --tls-key-file ./tests/tls/redis.key --tls-ca-cert-file ./tests/tls/ca.crt --tls-passphrase foobar
 	DEBUG=$(DEBUG_RUN) $(HERE)/pytests/run_tests.sh -t ./pytests/ --env oss-cluster --shards-count 2 --tls --tls-cert-file ./tests/tls/redis.crt --tls-key-file ./tests/tls/redis.key --tls-ca-cert-file ./tests/tls/ca.crt --tls-passphrase foobar
 	DEBUG=$(DEBUG_RUN) $(HERE)/pytests/run_tests.sh -t ./pytests/ --env oss-cluster --shards-count 3 --tls --tls-cert-file ./tests/tls/redis.crt --tls-key-file ./tests/tls/redis.key --tls-ca-cert-file ./tests/tls/ca.crt --tls-passphrase foobar
 

tests/mr_test_module/pytests/test_disconnect_race.py

@@ -0,0 +1,155 @@
+"""
+Race summary inside the libmr event-loop thread:
+  1. MR_OnConnectCallback is fired by hiredis once the TCP connect completes.
+     For a TLS-enabled cluster it then calls redisInitiateSSL on the brand-new
+     redisAsyncContext.  When the peer is plain TCP (or otherwise terminates
+     the handshake abruptly), redisInitiateSSL returns REDIS_ERR.
+     MR_OnConnectCallback logs "SSL auth ... failed", schedules a deferred
+     task via MR_EventLoopAddTask(MR_ClusterAsyncDisconnect, n) and returns.
+     The deferred task fires on the next event-loop iteration.
+  2. Hiredis sees the now-broken context and immediately drives its own
+     disconnect path which calls back into MR_ClusterOnDisconnectCallback;
+     that callback sets `n->c = NULL` (and hiredis frees the context shortly
+     after).
+  3. The deferred MR_ClusterAsyncDisconnect task runs and executes
+     redisAsyncFree(n->c).  Because n->c is now NULL, the very first
+     instruction of redisAsyncFree dereferences offset 0x90 of a NULL
+     pointer and the process is killed by SIGSEGV.
+
+Reproduction strategy:
+  - Run under a TLS-enabled Redis (--tls).  This forces MR_OnConnectCallback
+    down the SSL branch.
+  - Stand up a mock peer that accepts TCP but does NOT speak TLS.
+    SSL_connect from libmr will then fail, scheduling the deferred
+    MR_ClusterAsyncDisconnect.
+  - Force a connection attempt and wait long enough for the deferred task
+    to run.  Repeat several times to make the test robust under loaded
+    machines.
+  - Assert the Redis process is still alive after each cycle.
+"""
+
+import gevent.server
+import gevent.queue
+import socket
+import time
+import unittest
+
+from common import MRTestDecorator, promote_internal_client_if_supported
+from test_network import _get_hosts
+
+
+class _RejectingShard(object):
+    """Mock shard that accepts TCP and immediately closes the socket so any
+    TLS handshake from libmr fails with an EOF / handshake error."""
+
+    def __init__(self, host):
+        self.host = host
+        self.connections_accepted = 0
+        self.queue = gevent.queue.Queue()
+
+    def _handle_conn(self, sock, _addr):
+        self.connections_accepted += 1
+        self.queue.put(True)
+        try:
+            sock.close()
+        except Exception:
+            pass
+
+    def __enter__(self):
+        family = socket.AF_INET6 if ':' in self.host else socket.AF_INET
+        tmp = socket.socket(family, socket.SOCK_STREAM)
+        tmp.bind((self.host, 0))
+        self.port = tmp.getsockname()[1]
+        tmp.close()
+        self.server = gevent.server.StreamServer((self.host, self.port), self._handle_conn)
+        self.server.start()
+        return self
+
+    def __exit__(self, *_args):
+        try:
+            self.server.stop()
+        except Exception:
+            pass
+
+    def wait_for_connection(self, timeout=2):
+        try:
+            self.queue.get(block=True, timeout=timeout)
+            return True
+        except Exception:
+            return False
+
+
+def _tls_enabled(env):
+    """Return True when the running Redis was started with TLS enabled.
+    The bug only manifests on the TLS code path."""
+    try:
+        res = env.cmd('CONFIG', 'GET', 'tls-port')
+    except Exception:
+        return False
+    if not res or len(res) < 2:
+        return False
+    try:
+        return int(res[1]) != 0
+    except (TypeError, ValueError):
+        return False
+
+
+def _alive(env):
+    """Return True iff the underlying Redis process answered a PING."""
+    try:
+        env.cmd('PING')
+    except Exception:
+        return False
+    return True
+
+
+def _push_topology(env, host, mock_port):
+    """Tell libmr that there is a second shard listening on the mock peer."""
+    promote_internal_client_if_supported(env=env)
+    endpoint_host = '[%s]' % host if ':' in host else host
+    args = [
+        'NO-USED', 'NO-USED', 'NO-USED', 'NO-USED', 'NO-USED',
+        '1',
+        'RANGES', '2',
+        'SHARD', '1',
+        'SLOTRANGE', '0', '8192',
+        'ADDR', 'password@%s:6379' % endpoint_host,
+        'MASTER',
+        'SHARD', '2',
+        'SLOTRANGE', '8193', '16383',
+        'ADDR', 'password@%s:%d' % (endpoint_host, mock_port),
+        'MASTER',
+    ]
+    env.cmd('MRTESTS.CLUSTERSET', *args)
+    env.cmd('MRTESTS.FORCESHARDSCONNECTION')
+
+
+@MRTestDecorator(skipOnCluster=True)
+def testNoCrashOnTLSHandshakeFailure(env, conn):
+    if not _tls_enabled(env):
+        raise unittest.SkipTest('test requires --tls')
+
+    for host in _get_hosts():
+        with _RejectingShard(host) as mock:
+            # Drive several connection cycles. Each cycle exercises the path:
+            #   connect -> TLS handshake -> SSL_connect EOF -> schedule
+            #   MR_ClusterAsyncDisconnect -> hiredis disconnect callback nulls
+            #   n->c -> deferred task runs redisAsyncFree(NULL) -> CRASH.
+            # The first failed cycle is usually enough to crash an unfixed
+            # build; we loop to be robust under timing jitter.
+            for _ in range(20):
+                _push_topology(env, host, mock.port)
+                # Wait for libmr to actually attempt the connection so the
+                # event-loop thread has time to run the SSL handshake and the
+                # deferred MR_ClusterAsyncDisconnect task.
+                mock.wait_for_connection(timeout=2)
+                time.sleep(0.05)
+                # If the bug fired, this PING raises (connection refused or
+                # broken pipe) because the Redis process is dead. We only care
+                # that the server is alive; some redis-py versions return
+                # True for PING, others 'PONG' / b'PONG'.
+                env.assertTrue(_alive(env),
+                               message='Redis crashed after TLS handshake failure')
+
+        # Final liveness check after the mock has been torn down.
+        env.assertTrue(_alive(env))

tests/mr_test_module/pytests/test_network.py

@@ -169,6 +169,37 @@ def read_response(self):
             raise Exception('Invalid response: %s' % line)
 
 
+def _maybe_make_ssl_context(env):
+    """Return a gevent SSLContext configured as a TLS server using the same
+    cert bundle Redis is using, or None if the env is not TLS-enabled.
+
+    This lets ShardMock terminate TLS so that single-shard --tls tests can
+    actually exchange RESP traffic with the mock peer instead of dying at the
+    handshake.
+    """
+    if not getattr(env, 'useTLS', False):
+        return None
+    import gevent.ssl as gssl
+    cert = getattr(env, 'tlsCertFile', None)
+    key = getattr(env, 'tlsKeyFile', None)
+    ca = getattr(env, 'tlsCaCertFile', None)
+    passphrase = getattr(env, 'tlsPassphrase', None) or None
+    ctx = gssl.SSLContext(gssl.PROTOCOL_TLS_SERVER)
+    if cert and key:
+        ctx.load_cert_chain(certfile=cert, keyfile=key, password=passphrase)
+    if ca:
+        ctx.load_verify_locations(cafile=ca)
+    ctx.verify_mode = gssl.CERT_NONE
+    return ctx
+
+
+def _make_stream_server(host, port, handler, env):
+    ssl_context = _maybe_make_ssl_context(env)
+    if ssl_context is not None:
+        return gevent.server.StreamServer((host, port), handler, ssl_context=ssl_context)
+    return gevent.server.StreamServer((host, port), handler)
+
+
 class ShardMock():
     def __init__(self, env, host='localhost'):
         self.env = env
@@ -216,7 +247,7 @@ def __enter__(self):
         tmp_sock.bind((self.host, 0))
         self.port = tmp_sock.getsockname()[1]
         tmp_sock.close()
-        self.stream_server = gevent.server.StreamServer((self.host, self.port), self._handle_conn)
+        self.stream_server = _make_stream_server(self.host, self.port, self._handle_conn, self.env)
         self.stream_server.start()
         self._send_cluster_set()
         self.runId = self.env.cmd('MRTESTS.INFOCLUSTER')[3]
@@ -243,7 +274,7 @@ def StopListening(self):
         self.stream_server.stop()
 
     def StartListening(self):
-        self.stream_server = gevent.server.StreamServer((self.host, self.port), self._handle_conn)
+        self.stream_server = _make_stream_server(self.host, self.port, self._handle_conn, self.env)
         self.stream_server.start()
 
 def _is_ipv6_enabled():