Add ssl_ca_path support to async Redis client (#3879)

* Add ssl_ca_path support to async Redis client (#3414)

Add ssl_ca_path parameter to the async Redis client and SSLConnection
to enable specifying a directory of CA certificates, matching the sync
client's functionality.

Changes:
- Add ssl_ca_path parameter to Redis.__init__
- Add ssl_ca_path parameter to SSLConnection.__init__
- Add ca_path parameter to RedisSSLContext
- Pass capath to ssl.SSLContext.load_verify_locations()

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

* Add tests for ssl_ca_path and add to REDIS_ALLOWED_KEYS

- Add test_ssl_ca_path_parameter test to verify ssl_ca_path is properly
  passed through to SSLConnection
- Add ssl_ca_path to REDIS_ALLOWED_KEYS tuple in cluster.py for sync
  cluster client support

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: petyaslavova <petya.slavova@redis.com>

Author: 3 people

redis/asyncio/client.py

@@ -243,6 +243,7 @@ def __init__(
         ssl_exclude_verify_flags: Optional[List[VerifyFlags]] = None,
         ssl_ca_certs: Optional[str] = None,
         ssl_ca_data: Optional[str] = None,
+        ssl_ca_path: Optional[str] = None,
         ssl_check_hostname: bool = True,
         ssl_min_version: Optional[TLSVersion] = None,
         ssl_ciphers: Optional[str] = None,
@@ -354,6 +355,7 @@ def __init__(
                             "ssl_exclude_verify_flags": ssl_exclude_verify_flags,
                             "ssl_ca_certs": ssl_ca_certs,
                             "ssl_ca_data": ssl_ca_data,
+                            "ssl_ca_path": ssl_ca_path,
                             "ssl_check_hostname": ssl_check_hostname,
                             "ssl_min_version": ssl_min_version,
                             "ssl_ciphers": ssl_ciphers,

redis/asyncio/connection.py

@@ -813,6 +813,7 @@ def __init__(
         ssl_exclude_verify_flags: Optional[List["ssl.VerifyFlags"]] = None,
         ssl_ca_certs: Optional[str] = None,
         ssl_ca_data: Optional[str] = None,
+        ssl_ca_path: Optional[str] = None,
         ssl_check_hostname: bool = True,
         ssl_min_version: Optional[TLSVersion] = None,
         ssl_ciphers: Optional[str] = None,
@@ -829,6 +830,7 @@ def __init__(
             exclude_verify_flags=ssl_exclude_verify_flags,
             ca_certs=ssl_ca_certs,
             ca_data=ssl_ca_data,
+            ca_path=ssl_ca_path,
             check_hostname=ssl_check_hostname,
             min_version=ssl_min_version,
             ciphers=ssl_ciphers,
@@ -886,6 +888,7 @@ class RedisSSLContext:
         "exclude_verify_flags",
         "ca_certs",
         "ca_data",
+        "ca_path",
         "context",
         "check_hostname",
         "min_version",
@@ -901,6 +904,7 @@ def __init__(
         exclude_verify_flags: Optional[List["ssl.VerifyFlags"]] = None,
         ca_certs: Optional[str] = None,
         ca_data: Optional[str] = None,
+        ca_path: Optional[str] = None,
         check_hostname: bool = False,
         min_version: Optional[TLSVersion] = None,
         ciphers: Optional[str] = None,
@@ -928,6 +932,7 @@ def __init__(
         self.exclude_verify_flags = exclude_verify_flags
         self.ca_certs = ca_certs
         self.ca_data = ca_data
+        self.ca_path = ca_path
         self.check_hostname = (
             check_hostname if self.cert_reqs != ssl.CERT_NONE else False
         )
@@ -948,8 +953,10 @@ def get(self) -> SSLContext:
                     context.verify_flags &= ~flag
             if self.certfile and self.keyfile:
                 context.load_cert_chain(certfile=self.certfile, keyfile=self.keyfile)
-            if self.ca_certs or self.ca_data:
-                context.load_verify_locations(cafile=self.ca_certs, cadata=self.ca_data)
+            if self.ca_certs or self.ca_data or self.ca_path:
+                context.load_verify_locations(
+                    cafile=self.ca_certs, capath=self.ca_path, cadata=self.ca_data
+                )
             if self.min_version is not None:
                 context.minimum_version = self.min_version
             if self.ciphers is not None:

redis/cluster.py

@@ -185,6 +185,7 @@ def parse_cluster_myshardid(resp, **options):
     "ssl",
     "ssl_ca_certs",
     "ssl_ca_data",
+    "ssl_ca_path",
     "ssl_certfile",
     "ssl_cert_reqs",
     "ssl_include_verify_flags",

tests/test_asyncio/test_ssl.py

@@ -141,3 +141,29 @@ def capture_context_create_default():
 
         finally:
             await r.aclose()
+
+    async def test_ssl_ca_path_parameter(self, request):
+        """Test that ssl_ca_path parameter is properly passed to SSLConnection"""
+        ssl_url = request.config.option.redis_ssl_url
+        parsed_url = urlparse(ssl_url)
+
+        # Test with a mock ca_path directory
+        test_ca_path = "/tmp/test_ca_certs"
+
+        r = redis.Redis(
+            host=parsed_url.hostname,
+            port=parsed_url.port,
+            ssl=True,
+            ssl_cert_reqs="none",
+            ssl_ca_path=test_ca_path,
+        )
+
+        try:
+            # Get the connection to verify ssl_ca_path is passed through
+            conn = r.connection_pool.make_connection()
+            assert isinstance(conn, redis.SSLConnection)
+
+            # Verify the ca_path is stored in the SSL context
+            assert conn.ssl_context.ca_path == test_ca_path
+        finally:
+            await r.aclose()