Support using ssl.VerifyMode enum for ssl_cert_reqs (#3346)

Author: lattwood

redis/asyncio/client.py

@@ -80,9 +80,10 @@
 )
 
 if TYPE_CHECKING and SSL_AVAILABLE:
-    from ssl import TLSVersion
+    from ssl import TLSVersion, VerifyMode
 else:
     TLSVersion = None
+    VerifyMode = None
 
 PubSubHandler = Callable[[Dict[str, str]], Awaitable[None]]
 _KeyT = TypeVar("_KeyT", bound=KeyT)
@@ -227,7 +228,7 @@ def __init__(
         ssl: bool = False,
         ssl_keyfile: Optional[str] = None,
         ssl_certfile: Optional[str] = None,
-        ssl_cert_reqs: str = "required",
+        ssl_cert_reqs: Union[str, VerifyMode] = "required",
         ssl_ca_certs: Optional[str] = None,
         ssl_ca_data: Optional[str] = None,
         ssl_check_hostname: bool = False,

redis/asyncio/cluster.py

@@ -74,9 +74,10 @@
 )
 
 if SSL_AVAILABLE:
-    from ssl import TLSVersion
+    from ssl import TLSVersion, VerifyMode
 else:
     TLSVersion = None
+    VerifyMode = None
 
 TargetNodesT = TypeVar(
     "TargetNodesT", str, "ClusterNode", List["ClusterNode"], Dict[Any, "ClusterNode"]
@@ -267,7 +268,7 @@ def __init__(
         ssl: bool = False,
         ssl_ca_certs: Optional[str] = None,
         ssl_ca_data: Optional[str] = None,
-        ssl_cert_reqs: str = "required",
+        ssl_cert_reqs: Union[str, VerifyMode] = "required",
         ssl_certfile: Optional[str] = None,
         ssl_check_hostname: bool = False,
         ssl_keyfile: Optional[str] = None,

redis/asyncio/connection.py

@@ -768,7 +768,7 @@ def __init__(
         self,
         ssl_keyfile: Optional[str] = None,
         ssl_certfile: Optional[str] = None,
-        ssl_cert_reqs: str = "required",
+        ssl_cert_reqs: Union[str, ssl.VerifyMode] = "required",
         ssl_ca_certs: Optional[str] = None,
         ssl_ca_data: Optional[str] = None,
         ssl_check_hostname: bool = False,
@@ -842,7 +842,7 @@ def __init__(
         self,
         keyfile: Optional[str] = None,
         certfile: Optional[str] = None,
-        cert_reqs: Optional[str] = None,
+        cert_reqs: Optional[Union[str, ssl.VerifyMode]] = None,
         ca_certs: Optional[str] = None,
         ca_data: Optional[str] = None,
         check_hostname: bool = False,
@@ -855,7 +855,7 @@ def __init__(
         self.keyfile = keyfile
         self.certfile = certfile
         if cert_reqs is None:
-            self.cert_reqs = ssl.CERT_NONE
+            cert_reqs = ssl.CERT_NONE
         elif isinstance(cert_reqs, str):
             CERT_REQS = {  # noqa: N806
                 "none": ssl.CERT_NONE,
@@ -866,7 +866,8 @@ def __init__(
                 raise RedisError(
                     f"Invalid SSL Certificate Requirements Flag: {cert_reqs}"
                 )
-            self.cert_reqs = CERT_REQS[cert_reqs]
+            cert_reqs = CERT_REQS[cert_reqs]
+        self.cert_reqs = cert_reqs
         self.ca_certs = ca_certs
         self.ca_data = ca_data
         self.check_hostname = check_hostname

redis/client.py

@@ -210,7 +210,7 @@ def __init__(
         ssl: bool = False,
         ssl_keyfile: Optional[str] = None,
         ssl_certfile: Optional[str] = None,
-        ssl_cert_reqs: str = "required",
+        ssl_cert_reqs: Union[str, "ssl.VerifyMode"] = "required",
         ssl_ca_certs: Optional[str] = None,
         ssl_ca_path: Optional[str] = None,
         ssl_ca_data: Optional[str] = None,

redis/connection.py

@@ -1017,7 +1017,7 @@ def __init__(
         Args:
             ssl_keyfile: Path to an ssl private key. Defaults to None.
             ssl_certfile: Path to an ssl certificate. Defaults to None.
-            ssl_cert_reqs: The string value for the SSLContext.verify_mode (none, optional, required). Defaults to "required".
+            ssl_cert_reqs: The string value for the SSLContext.verify_mode (none, optional, required), or an ssl.VerifyMode. Defaults to "required".
             ssl_ca_certs: The path to a file of concatenated CA certificates in PEM format. Defaults to None.
             ssl_ca_data: Either an ASCII string of one or more PEM-encoded certificates or a bytes-like object of DER-encoded certificates.
             ssl_check_hostname: If set, match the hostname during the SSL handshake. Defaults to False.