Adding ssl_verify_flags_config argument for ssl connection configuration

Author: petyaslavova

redis/asyncio/client.py

@@ -81,10 +81,11 @@
 )
 
 if TYPE_CHECKING and SSL_AVAILABLE:
-    from ssl import TLSVersion, VerifyMode
+    from ssl import TLSVersion, VerifyFlags, VerifyMode
 else:
     TLSVersion = None
     VerifyMode = None
+    VerifyFlags = None
 
 PubSubHandler = Callable[[Dict[str, str]], Awaitable[None]]
 _KeyT = TypeVar("_KeyT", bound=KeyT)
@@ -238,6 +239,7 @@ def __init__(
         ssl_keyfile: Optional[str] = None,
         ssl_certfile: Optional[str] = None,
         ssl_cert_reqs: Union[str, VerifyMode] = "required",
+        ssl_verify_flags_config: Optional[List[Tuple[VerifyFlags, bool]]] = None,
         ssl_ca_certs: Optional[str] = None,
         ssl_ca_data: Optional[str] = None,
         ssl_check_hostname: bool = True,
@@ -347,6 +349,7 @@ def __init__(
                             "ssl_keyfile": ssl_keyfile,
                             "ssl_certfile": ssl_certfile,
                             "ssl_cert_reqs": ssl_cert_reqs,
+                            "ssl_verify_flags_config": ssl_verify_flags_config,
                             "ssl_ca_certs": ssl_ca_certs,
                             "ssl_ca_data": ssl_ca_data,
                             "ssl_check_hostname": ssl_check_hostname,

redis/asyncio/cluster.py

@@ -86,10 +86,11 @@
 )
 
 if SSL_AVAILABLE:
-    from ssl import TLSVersion, VerifyMode
+    from ssl import TLSVersion, VerifyFlags, VerifyMode
 else:
     TLSVersion = None
     VerifyMode = None
+    VerifyFlags = None
 
 TargetNodesT = TypeVar(
     "TargetNodesT", str, "ClusterNode", List["ClusterNode"], Dict[Any, "ClusterNode"]
@@ -299,6 +300,7 @@ def __init__(
         ssl_ca_certs: Optional[str] = None,
         ssl_ca_data: Optional[str] = None,
         ssl_cert_reqs: Union[str, VerifyMode] = "required",
+        ssl_verify_flags_config: Optional[List[Tuple[VerifyFlags, bool]]] = None,
         ssl_certfile: Optional[str] = None,
         ssl_check_hostname: bool = True,
         ssl_keyfile: Optional[str] = None,
@@ -358,6 +360,7 @@ def __init__(
                     "ssl_ca_certs": ssl_ca_certs,
                     "ssl_ca_data": ssl_ca_data,
                     "ssl_cert_reqs": ssl_cert_reqs,
+                    "ssl_verify_flags_config": ssl_verify_flags_config,
                     "ssl_certfile": ssl_certfile,
                     "ssl_check_hostname": ssl_check_hostname,
                     "ssl_keyfile": ssl_keyfile,

redis/asyncio/connection.py

@@ -1,7 +1,9 @@
+import ast
 import asyncio
 import copy
 import enum
 import inspect
+import re
 import socket
 import sys
 import warnings
@@ -30,11 +32,12 @@
 
 if SSL_AVAILABLE:
     import ssl
-    from ssl import SSLContext, TLSVersion
+    from ssl import SSLContext, TLSVersion, VerifyFlags
 else:
     ssl = None
     TLSVersion = None
     SSLContext = None
+    VerifyFlags = None
 
 from ..auth.token import TokenInterface
 from ..event import AsyncAfterConnectionReleasedEvent, EventDispatcher
@@ -793,6 +796,7 @@ def __init__(
         ssl_keyfile: Optional[str] = None,
         ssl_certfile: Optional[str] = None,
         ssl_cert_reqs: Union[str, ssl.VerifyMode] = "required",
+        ssl_verify_flags_config: Optional[List[Tuple["ssl.VerifyFlags", bool]]] = None,
         ssl_ca_certs: Optional[str] = None,
         ssl_ca_data: Optional[str] = None,
         ssl_check_hostname: bool = True,
@@ -807,6 +811,7 @@ def __init__(
             keyfile=ssl_keyfile,
             certfile=ssl_certfile,
             cert_reqs=ssl_cert_reqs,
+            verify_flags_config=ssl_verify_flags_config,
             ca_certs=ssl_ca_certs,
             ca_data=ssl_ca_data,
             check_hostname=ssl_check_hostname,
@@ -832,6 +837,10 @@ def certfile(self):
     def cert_reqs(self):
         return self.ssl_context.cert_reqs
 
+    @property
+    def verify_flags_config(self):
+        return self.ssl_context.verify_flags_config
+
     @property
     def ca_certs(self):
         return self.ssl_context.ca_certs
@@ -854,6 +863,7 @@ class RedisSSLContext:
         "keyfile",
         "certfile",
         "cert_reqs",
+        "verify_flags_config",
         "ca_certs",
         "ca_data",
         "context",
@@ -867,6 +877,7 @@ def __init__(
         keyfile: Optional[str] = None,
         certfile: Optional[str] = None,
         cert_reqs: Optional[Union[str, ssl.VerifyMode]] = None,
+        verify_flags_config: Optional[List[Tuple[ssl.VerifyFlags, bool]]] = None,
         ca_certs: Optional[str] = None,
         ca_data: Optional[str] = None,
         check_hostname: bool = False,
@@ -892,6 +903,7 @@ def __init__(
                 )
             cert_reqs = CERT_REQS[cert_reqs]
         self.cert_reqs = cert_reqs
+        self.verify_flags_config = verify_flags_config
         self.ca_certs = ca_certs
         self.ca_data = ca_data
         self.check_hostname = (
@@ -906,6 +918,12 @@ def get(self) -> SSLContext:
             context = ssl.create_default_context()
             context.check_hostname = self.check_hostname
             context.verify_mode = self.cert_reqs
+            if self.verify_flags_config:
+                for flag, enabled in self.verify_flags_config:
+                    if enabled:
+                        context.options |= flag
+                    else:
+                        context.options &= ~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:
@@ -1021,6 +1039,34 @@ def parse_url(url: str) -> ConnectKwargs:
 
         if parsed.scheme == "rediss":
             kwargs["connection_class"] = SSLConnection
+
+            if "ssl_verify_flags_config" in kwargs:
+                # flags are passed in as a string representation of a list,
+                # e.g. [(VERIFY_X509_STRICT, False), (VERIFY_X509_PARTIAL_CHAIN, True)]
+                # To parse it sucessfully, we need transform the flags to strings with quotes.
+                verify_flags_config_str = kwargs.pop("ssl_verify_flags_config")
+                # First wrap any VERIFY_* name in quotes
+                verify_flags_config_str = re.sub(
+                    r"\b(VERIFY_[A-Z0-9_]+)\b", r'"\1"', verify_flags_config_str
+                )
+
+                # transform the string to a list of tuples - the first element of each tuple is a string containing the name of the flag,
+                # and the second is a boolean that indicates if the flad should be enabled or disabled
+                verify_flags_config = ast.literal_eval(verify_flags_config_str)
+
+                verify_flags_config_config_parsed = []
+                for flag, enabled in verify_flags_config:
+                    if not hasattr(VerifyFlags, flag):
+                        raise ValueError(f"Invalid verify flag: {flag}")
+                    if not isinstance(enabled, bool):
+                        raise ValueError(
+                            f"Invalid verify flag enabled/disabled value: {enabled}"
+                        )
+                    verify_flags_config_config_parsed.append(
+                        (getattr(VerifyFlags, flag), enabled)
+                    )
+
+                kwargs["ssl_verify_flags_config"] = verify_flags_config_config_parsed
     else:
         valid_schemes = "redis://, rediss://, unix://"
         raise ValueError(

redis/client.py

@@ -12,6 +12,7 @@
     Mapping,
     Optional,
     Set,
+    Tuple,
     Type,
     Union,
 )
@@ -224,6 +225,7 @@ def __init__(
         ssl_keyfile: Optional[str] = None,
         ssl_certfile: Optional[str] = None,
         ssl_cert_reqs: Union[str, "ssl.VerifyMode"] = "required",
+        ssl_verify_flags_config: Optional[List[Tuple["ssl.VerifyFlags", bool]]] = None,
         ssl_ca_certs: Optional[str] = None,
         ssl_ca_path: Optional[str] = None,
         ssl_ca_data: Optional[str] = None,
@@ -330,6 +332,7 @@ def __init__(
                             "ssl_keyfile": ssl_keyfile,
                             "ssl_certfile": ssl_certfile,
                             "ssl_cert_reqs": ssl_cert_reqs,
+                            "ssl_verify_flags_config": ssl_verify_flags_config,
                             "ssl_ca_certs": ssl_ca_certs,
                             "ssl_ca_data": ssl_ca_data,
                             "ssl_check_hostname": ssl_check_hostname,

redis/cluster.py

@@ -184,6 +184,7 @@ def parse_cluster_myshardid(resp, **options):
     "ssl_ca_data",
     "ssl_certfile",
     "ssl_cert_reqs",
+    "ssl_verify_flags_config",
     "ssl_keyfile",
     "ssl_password",
     "ssl_check_hostname",

redis/connection.py

@@ -1,5 +1,7 @@
+import ast
 import copy
 import os
+import re
 import socket
 import sys
 import threading
@@ -16,6 +18,7 @@
     List,
     Literal,
     Optional,
+    Tuple,
     Type,
     TypeVar,
     Union,
@@ -68,8 +71,10 @@
 
 if SSL_AVAILABLE:
     import ssl
+    from ssl import VerifyFlags
 else:
     ssl = None
+    VerifyFlags = None
 
 if HIREDIS_AVAILABLE:
     import hiredis
@@ -1358,6 +1363,7 @@ def __init__(
         ssl_keyfile=None,
         ssl_certfile=None,
         ssl_cert_reqs="required",
+        ssl_verify_flags_config: Optional[List[Tuple["VerifyFlags", bool]]] = None,
         ssl_ca_certs=None,
         ssl_ca_data=None,
         ssl_check_hostname=True,
@@ -1376,7 +1382,19 @@ 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), or an ssl.VerifyMode. 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_verify_flags_config: A list with flags configuration to be set on the SSLContext. Defaults to None.
+                              Valid format is as follows:
+                                [
+                                    (config_flag, enabled/disabled),
+                                    ...
+                                ]
+                              Example:
+                                [
+                                    (ssl.VERIFY_X509_STRICT, False),       # disable strict
+                                    (ssl.VERIFY_X509_PARTIAL_CHAIN, True), # ensure partial chain is enabled
+                                ]
             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 True.
@@ -1412,6 +1430,7 @@ def __init__(
                 )
             ssl_cert_reqs = CERT_REQS[ssl_cert_reqs]
         self.cert_reqs = ssl_cert_reqs
+        self.ssl_verify_flags_config = ssl_verify_flags_config
         self.ca_certs = ssl_ca_certs
         self.ca_data = ssl_ca_data
         self.ca_path = ssl_ca_path
@@ -1451,6 +1470,12 @@ def _wrap_socket_with_ssl(self, sock):
         context = ssl.create_default_context()
         context.check_hostname = self.check_hostname
         context.verify_mode = self.cert_reqs
+        if self.ssl_verify_flags_config:
+            for flag, enabled in self.ssl_verify_flags_config:
+                if enabled:
+                    context.options |= flag
+                else:
+                    context.options &= ~flag
         if self.certfile or self.keyfile:
             context.load_cert_chain(
                 certfile=self.certfile,
@@ -1632,6 +1657,34 @@ def parse_url(url):
         if url.scheme == "rediss":
             kwargs["connection_class"] = SSLConnection
 
+            if "ssl_verify_flags_config" in kwargs:
+                # flags are passed in as a string representation of a list,
+                # e.g. [(VERIFY_X509_STRICT, False), (VERIFY_X509_PARTIAL_CHAIN, True)]
+                # To parse it sucessfully, we need transform the flags to strings with quotes.
+                verify_flags_config_str = kwargs.pop("ssl_verify_flags_config")
+                # First wrap any VERIFY_* name in quotes
+                verify_flags_config_str = re.sub(
+                    r"\b(VERIFY_[A-Z0-9_]+)\b", r'"\1"', verify_flags_config_str
+                )
+
+                # transform the string to a list of tuples - the first element of each tuple is a string containing the name of the flag,
+                # and the second is a boolean that indicates if the flad should be enabled or disabled
+                verify_flags_config = ast.literal_eval(verify_flags_config_str)
+
+                ssl_verify_flags_config_parsed = []
+                for flag, enabled in verify_flags_config:
+                    if not hasattr(VerifyFlags, flag):
+                        raise ValueError(f"Invalid ssl verify flag: {flag}")
+                    if not isinstance(enabled, bool):
+                        raise ValueError(
+                            f"Invalid ssl verify flag enabled/disabled value: {enabled}"
+                        )
+                    ssl_verify_flags_config_parsed.append(
+                        (getattr(VerifyFlags, flag), enabled)
+                    )
+
+                kwargs["ssl_verify_flags_config"] = ssl_verify_flags_config_parsed
+
     return kwargs