Fix SSL verification with ssl_cert_reqs=none and ssl_check_hostname=True

Author: omerfeyzioglu

CHANGES

@@ -70,6 +70,7 @@
     * Close Unix sockets if the connection attempt fails. This prevents `ResourceWarning`s. (#3314)
     * Close SSL sockets if the connection attempt fails, or if validations fail. (#3317)
     * Eliminate mutable default arguments in the `redis.commands.core.Script` class. (#3332)
+    * Fix SSL verification with `ssl_cert_reqs="none"` and `ssl_check_hostname=True` by automatically setting `check_hostname=False` when `verify_mode=ssl.CERT_NONE` (#3636) 
 
 * 4.1.3 (Feb 8, 2022)
   * Fix flushdb and flushall (#1926)

redis/asyncio/connection.py

@@ -901,7 +901,10 @@ def __init__(
     def get(self) -> SSLContext:
         if not self.context:
             context = ssl.create_default_context()
-            context.check_hostname = self.check_hostname
+            if self.cert_reqs == ssl.CERT_NONE:
+                context.check_hostname = False
+            else:
+                context.check_hostname = self.check_hostname
             context.verify_mode = self.cert_reqs
             if self.certfile and self.keyfile:
                 context.load_cert_chain(certfile=self.certfile, keyfile=self.keyfile)

redis/connection.py

@@ -1115,7 +1115,10 @@ def _wrap_socket_with_ssl(self, sock):
             An SSL wrapped socket.
         """
         context = ssl.create_default_context()
-        context.check_hostname = self.check_hostname
+        if self.cert_reqs == ssl.CERT_NONE:
+            context.check_hostname = False
+        else:
+            context.check_hostname = self.check_hostname
         context.verify_mode = self.cert_reqs
         if self.certfile or self.keyfile:
             context.load_cert_chain(

tests/test_asyncio/test_ssl.py

@@ -0,0 +1,56 @@
+import asyncio
+import ssl
+import socket
+from urllib.parse import urlparse
+import pytest
+import pytest_asyncio
+import redis.asyncio as redis
+from redis.exceptions import RedisError, ConnectionError
+import ssl
+
+# Skip test or not based on cryptography installation
+try:
+    import cryptography  # noqa
+    skip_if_cryptography = pytest.mark.skipif(False, reason="")
+    skip_if_nocryptography = pytest.mark.skipif(False, reason="")
+except ImportError:
+    skip_if_cryptography = pytest.mark.skipif(True, reason="cryptography not installed")
+    skip_if_nocryptography = pytest.mark.skipif(True, reason="cryptography not installed")
+
+@pytest.mark.ssl
+class TestSSL:
+    """Tests for SSL connections in asyncio."""
+
+    @pytest_asyncio.fixture()
+    async def _get_client(self, request):
+        ssl_url = request.config.option.redis_ssl_url
+        p = urlparse(ssl_url)[1].split(":")
+        client = redis.Redis(host=p[0], port=p[1], ssl=True)
+        yield client
+        await client.aclose()
+
+    async def test_ssl_with_invalid_cert(self, _get_client):
+        """Test SSL connection with invalid certificate."""
+        pass
+
+    async def test_cert_reqs_none_with_check_hostname(self, request):
+        """Test that when ssl_cert_reqs=none is used with ssl_check_hostname=True,
+        the connection is created successfully with check_hostname internally set to False"""
+        ssl_url = request.config.option.redis_ssl_url
+        p = urlparse(ssl_url)[1].split(":")
+        r = redis.Redis(
+            host=p[0],
+            port=p[1],
+            ssl=True,
+            ssl_cert_reqs="none",
+            ssl_check_hostname=True,  # This should work now because we handle the incompatibility
+        )
+        try:
+            # Connection should be successful
+            assert await r.ping()
+            # check_hostname should have been automatically set to False
+            assert r.connection_pool.connection_class == redis.SSLConnection
+            conn = r.connection_pool.make_connection()
+            assert conn.check_hostname is False
+        finally:
+            await r.aclose() 

tests/test_ssl.py

@@ -309,3 +309,25 @@ def test_mock_ocsp_staple(self, request):
             r.ping()
         assert "no ocsp response present" in str(e)
         r.close()
+
+    def test_cert_reqs_none_with_check_hostname(self, request):
+        """Test that when ssl_cert_reqs=none is used with ssl_check_hostname=True,
+        the connection is created successfully with check_hostname internally set to False"""
+        ssl_url = request.config.option.redis_ssl_url
+        p = urlparse(ssl_url)[1].split(":")
+        r = redis.Redis(
+            host=p[0],
+            port=p[1],
+            ssl=True,
+            ssl_cert_reqs="none",
+            ssl_check_hostname=True,  # This should work now because we handle the incompatibility
+        )
+        try:
+            # Connection should be successful
+            assert r.ping()
+            # check_hostname should have been automatically set to False
+            assert r.connection_pool.connection_kwargs["connection_class"] == redis.SSLConnection
+            conn = r.connection_pool.make_connection()
+            assert conn.check_hostname is False
+        finally:
+            r.close()