connection: fix ssl_options TLS handling

Preserve the distinction between omitted and explicitly empty ssl_options across connection setup, reactors, shard-aware routing, cloud contexts, and Insights reporting. Strengthen pyOpenSSL protocol, verification, SNI, and hostname handling while retaining legacy option behavior.

Author: dkropachev

CHANGELOG.rst

@@ -7,6 +7,27 @@ Features
   statements skip re-sending result metadata on EXECUTE, and the driver automatically
   refreshes cached metadata when the server detects a schema change (DRIVER-153)
 
+Bug Fixes
+---------
+* Explicit ``ssl_options={}`` and endpoint SSL options now enable TLS instead
+  of being treated as omitted SSL options. An empty dict encrypts the
+  connection without verifying the server certificate; ``ssl_options=None``
+  remains plaintext.
+* Legacy ``ssl_options`` now work with the Twisted and Eventlet reactors by
+  using pyOpenSSL contexts with mapped protocol, verification, cipher, SNI, and
+  hostname-validation settings. Nonempty options without ``cert_reqs`` now
+  default to peer-certificate verification on these reactors instead of
+  pyOpenSSL's ``VERIFY_NONE``. When ``ca_certs`` is omitted, the system trust
+  roots are loaded. Existing configurations that connect to an untrusted
+  certificate must supply the appropriate CA or explicitly opt out with
+  ``cert_reqs=ssl.CERT_NONE`` and leave ``check_hostname`` disabled.
+* Hostname verification is now performed for Twisted and Eventlet connections
+  when ``check_hostname`` is enabled. The flag was previously not propagated
+  to the connection, so hostname mismatches were silently accepted; such
+  mismatches now fail the connection.
+* ``SSLContext`` setup now applies ``cert_reqs`` through ``verify_mode`` instead
+  of overwriting the SSL options bitmask, preserving default hardening flags.
+
 Others
 ------
 * ``PreparedStatement.result_metadata`` and ``PreparedStatement.result_metadata_id`` are

cassandra/cluster.py

@@ -885,10 +885,22 @@ def default_retry_policy(self, policy):
 
     The following documentation only applies when ssl_options is used without ssl_context.
 
-    By default, a ``ca_certs`` value should be supplied (the value should be
-    a string pointing to the location of the CA certs file), and you probably
-    want to specify ``ssl_version`` as ``ssl.PROTOCOL_TLS`` to match
-    Cassandra's default protocol.
+    .. versionchanged:: 3.29.12
+
+        An explicit empty dict (``ssl_options={}``) enables TLS with default
+        options, while ``ssl_options=None`` leaves the connection in plaintext.
+        Empty options do not verify the server certificate. A nonempty options
+        dict enables peer-certificate verification by default unless
+        ``cert_reqs`` explicitly disables it. Setting
+        ``'check_hostname': True`` always enables peer-certificate verification
+        because hostname checking requires an authenticated certificate chain.
+        When verification is enabled and ``ca_certs`` is omitted, the system
+        trust roots are loaded.
+
+    Supply ``ca_certs`` as a path to a CA certificate file when the server uses
+    a private CA that is not in the system trust store. You probably also want
+    to specify ``ssl_version`` as ``ssl.PROTOCOL_TLS`` to match Cassandra's
+    default protocol.
 
     .. versionchanged:: 3.3.0
 
@@ -1298,7 +1310,8 @@ def __init__(self,
 
         if cloud is not None:
             self.cloud = cloud
-            if contact_points is not _NOT_SET or endpoint_factory or ssl_context or ssl_options:
+            if (contact_points is not _NOT_SET or endpoint_factory or
+                    ssl_context is not None or ssl_options is not None):
                 raise ValueError("contact_points, endpoint_factory, ssl_context, and ssl_options "
                                  "cannot be specified with a cloud configuration")
 
@@ -1508,7 +1521,7 @@ def __init__(self,
 
         self.metrics_enabled = metrics_enabled
 
-        if ssl_options and not ssl_context:
+        if ssl_options is not None and ssl_context is None:
             warn('Using ssl_options without ssl_context is '
                  'deprecated and will result in an error in '
                  'the next major release. Please use ssl_context '

cassandra/connection.py

@@ -17,6 +17,7 @@
 import errno
 from functools import wraps, partial, total_ordering
 from heapq import heappush, heappop
+import ipaddress
 import io
 import logging
 import socket
@@ -130,6 +131,220 @@ def decompress(byts):
 frame_header_v3 = struct.Struct('>BhBi')
 
 
+def _default_pyopenssl_ssl_method(ssl_module):
+    for method_name in ('TLS_CLIENT_METHOD', 'TLS_METHOD', 'TLSv1_2_METHOD'):
+        method = getattr(ssl_module, method_name, None)
+        if method is not None:
+            return method
+    raise ImportError('pyOpenSSL does not expose a secure TLS client method')
+
+
+def _pyopenssl_ssl_method_from_stdlib(ssl_module, ssl_version):
+    if ssl_version is None:
+        return _default_pyopenssl_ssl_method(ssl_module)
+
+    protocol_method_names = (
+        ('PROTOCOL_TLS_CLIENT', ('TLS_CLIENT_METHOD', 'TLS_METHOD', 'TLSv1_2_METHOD')),
+        ('PROTOCOL_TLS', ('TLS_METHOD', 'TLS_CLIENT_METHOD', 'TLSv1_2_METHOD')),
+        ('PROTOCOL_SSLv23', ('TLS_METHOD', 'TLS_CLIENT_METHOD', 'TLSv1_2_METHOD')),
+        ('PROTOCOL_TLSv1_2', ('TLSv1_2_METHOD',)),
+        ('PROTOCOL_TLSv1_1', ('TLSv1_1_METHOD',)),
+        ('PROTOCOL_TLSv1', ('TLSv1_METHOD',)),
+    )
+    for protocol_name, method_names in protocol_method_names:
+        protocol = getattr(ssl, protocol_name, None)
+        if (protocol is not None and
+                ssl_version.__class__ is protocol.__class__ and
+                ssl_version == protocol):
+            for method_name in method_names:
+                method = getattr(ssl_module, method_name, None)
+                if method is not None:
+                    return method
+            raise ImportError('pyOpenSSL does not expose a method for %s' % (protocol_name,))
+
+    return ssl_version
+
+
+def _pyopenssl_verify_mode_from_cert_reqs(ssl_module, cert_reqs):
+    if cert_reqs is None:
+        return None
+    if cert_reqs == ssl.CERT_NONE:
+        return ssl_module.VERIFY_NONE
+    if cert_reqs in (ssl.CERT_OPTIONAL, ssl.CERT_REQUIRED):
+        return ssl_module.VERIFY_PEER
+    return cert_reqs
+
+
+def _ensure_pyopenssl_context_requires_verification(ssl_module, context, check_hostname):
+    """
+    Make hostname verification fail closed for a caller-supplied context.
+
+    When hostname checking is enabled, this may mutate the context's verify
+    mode and callback because pyOpenSSL does not expose a non-mutating way to
+    require peer verification.
+    """
+    if not check_hostname:
+        return
+
+    get_verify_mode = getattr(context, 'get_verify_mode', None)
+    verify_mode = get_verify_mode() if get_verify_mode is not None else None
+    if (verify_mode is not None and
+            verify_mode & ssl_module.VERIFY_PEER):
+        return
+
+    log.warning(
+        "check_hostname=True requires peer verification; mutating supplied "
+        "pyOpenSSL context to use VERIFY_PEER and replacing its verification "
+        "callback"
+    )
+    context.set_verify(
+        ssl_module.VERIFY_PEER,
+        callback=lambda _connection, _x509, _errnum, _errdepth, ok: ok
+    )
+
+
+def _build_pyopenssl_context_from_options(ssl_module, ssl_options):
+    ssl_options = ssl_options or {}
+    context = ssl_module.Context(
+        _pyopenssl_ssl_method_from_stdlib(ssl_module, ssl_options.get('ssl_version', None))
+    )
+    if 'certfile' in ssl_options:
+        context.use_certificate_file(ssl_options['certfile'])
+    if 'keyfile' in ssl_options:
+        context.use_privatekey_file(ssl_options['keyfile'])
+    if 'ca_certs' in ssl_options:
+        context.load_verify_locations(ssl_options['ca_certs'])
+    cert_reqs = _pyopenssl_verify_mode_from_cert_reqs(
+        ssl_module, ssl_options.get('cert_reqs', None))
+    if cert_reqs is None:
+        cert_reqs = (ssl_module.VERIFY_PEER
+                     if ssl_options
+                     else ssl_module.VERIFY_NONE)
+    elif ssl_options.get('check_hostname', False) and cert_reqs == ssl_module.VERIFY_NONE:
+        cert_reqs = ssl_module.VERIFY_PEER
+    context.set_verify(
+        cert_reqs,
+        callback=lambda _connection, _x509, _errnum, _errdepth, ok: ok
+    )
+    if (cert_reqs & ssl_module.VERIFY_PEER and
+            not ssl_options.get('ca_certs')):
+        set_default_verify_paths = getattr(context, 'set_default_verify_paths', None)
+        if set_default_verify_paths is not None:
+            set_default_verify_paths()
+    ciphers = ssl_options.get('ciphers', None)
+    if ciphers:
+        if isinstance(ciphers, str):
+            ciphers = ciphers.encode('ascii')
+        context.set_cipher_list(ciphers)
+    return context
+
+
+def _normalized_hostname(hostname):
+    return hostname.rstrip('.').lower()
+
+
+def _dnsname_match(dn, hostname):
+    dn = _normalized_hostname(dn)
+    hostname = _normalized_hostname(hostname)
+
+    if not dn or not hostname:
+        return False
+
+    if '*' not in dn:
+        return dn == hostname
+
+    dn_labels = dn.split('.')
+    hostname_labels = hostname.split('.')
+    if (len(dn_labels) != len(hostname_labels) or
+            len(dn_labels) < 2 or dn_labels[0] != '*' or
+            not hostname_labels[0] or
+            any('*' in label for label in dn_labels[1:])):
+        return False
+
+    return dn_labels[1:] == hostname_labels[1:]
+
+
+def _ipaddress_match(cert_ip, hostname):
+    try:
+        host_ip = ipaddress.ip_address(hostname)
+        cert_ip = ipaddress.ip_address(cert_ip)
+    except ValueError:
+        return False
+    return cert_ip == host_ip
+
+
+def _is_ip_address(hostname):
+    try:
+        ipaddress.ip_address(hostname)
+    except ValueError:
+        return False
+    return True
+
+
+def _pyopenssl_cert_subject_alt_names(cert):
+    # Imported lazily because cryptography is an optional driver dependency.
+    # pyOpenSSL supplies it whenever this code path is used.
+    from cryptography import x509
+
+    try:
+        san = cert.to_cryptography().extensions.get_extension_for_class(
+            x509.SubjectAlternativeName).value
+    except x509.ExtensionNotFound:
+        return [], []
+
+    return (
+        san.get_values_for_type(x509.DNSName),
+        [str(ip) for ip in san.get_values_for_type(x509.IPAddress)]
+    )
+
+
+def _pyopenssl_cert_common_names(cert):
+    # Imported lazily for the same reason as the SAN types above.
+    from cryptography.x509.oid import NameOID
+
+    return [
+        attribute.value
+        for attribute in cert.to_cryptography().subject.get_attributes_for_oid(
+            NameOID.COMMON_NAME)
+    ]
+
+
+def _validate_pyopenssl_hostname(cert, hostname):
+    if cert is None:
+        raise ssl.CertificateError(
+            "peer did not present a certificate; cannot verify hostname %r" % (hostname,))
+
+    san_dns_names, san_ip_addresses = _pyopenssl_cert_subject_alt_names(cert)
+    san_names = san_dns_names + san_ip_addresses
+    hostname_is_ip = _is_ip_address(hostname)
+
+    for cert_ip in san_ip_addresses:
+        if _ipaddress_match(cert_ip, hostname):
+            return
+    if hostname_is_ip and san_names:
+        raise ssl.CertificateError(
+            "hostname %r doesn't match certificate subjectAltName %r" %
+            (hostname, san_names))
+
+    for cert_hostname in san_dns_names:
+        if _dnsname_match(cert_hostname, hostname):
+            return
+    if san_names:
+        raise ssl.CertificateError(
+            "hostname %r doesn't match certificate subjectAltName %r" %
+            (hostname, san_names))
+
+    common_names = _pyopenssl_cert_common_names(cert)
+    for common_name in common_names:
+        if (_ipaddress_match(common_name, hostname) if hostname_is_ip
+                else _dnsname_match(common_name, hostname)):
+            return
+
+    raise ssl.CertificateError(
+        "hostname %r doesn't match certificate commonName %r" %
+        (hostname, common_names))
+
+
 class EndPoint(object):
     """
     Represents the information to connect to a cassandra node.
@@ -803,6 +1018,7 @@ class Connection(object):
     endpoint = None
     ssl_options = None
     ssl_context = None
+    _ssl_options_explicit = False
     last_error = None
 
     # The current number of operations that are in flight. More precisely,
@@ -885,7 +1101,11 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None,
         self.endpoint = host if isinstance(host, EndPoint) else DefaultEndPoint(host, port)
 
         self.authenticator = authenticator
-        self.ssl_options = ssl_options.copy() if ssl_options else {}
+        endpoint_ssl_options = self.endpoint.ssl_options
+        # Explicit ssl_options={} enables SSL with default options; omitted
+        # ssl_options=None leaves SSL disabled unless an endpoint supplies options.
+        self._ssl_options_explicit = ssl_options is not None
+        self.ssl_options = ssl_options.copy() if ssl_options is not None else {}
         self.ssl_context = ssl_context
         self.sockopts = sockopts
         self.compression = compression
@@ -905,10 +1125,13 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None,
         self._on_orphaned_stream_released = on_orphaned_stream_released
         self._application_info = application_info
 
-        if ssl_options:
-            self.ssl_options.update(self.endpoint.ssl_options or {})
-        elif self.endpoint.ssl_options:
-            self.ssl_options = self.endpoint.ssl_options
+        if ssl_options is not None:
+            self.ssl_options.update(endpoint_ssl_options or {})
+        elif endpoint_ssl_options is not None:
+            self._ssl_options_explicit = True
+            self.ssl_options = endpoint_ssl_options
+        self._check_hostname = bool(self.ssl_options.get('check_hostname', False) or
+                                    getattr(self.ssl_context, 'check_hostname', False))
 
         # PYTHON-1331
         #
@@ -918,7 +1141,7 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None,
         #
         # Note the use of pop() here; we are very deliberately removing these params from ssl_options if they're present.  After this
         # operation ssl_options should contain only args needed for the ssl_context.wrap_socket() call.
-        if not self.ssl_context and self.ssl_options:
+        if self.ssl_context is None and self._ssl_options_explicit:
             self.ssl_context = self._build_ssl_context_from_options()
 
         self.max_request_id = min(self.max_in_flight - 1, (2 ** 15) - 1)
@@ -942,6 +1165,10 @@ def host(self):
     def port(self):
         return self.endpoint.port
 
+    @property
+    def _ssl_enabled(self):
+        return self.ssl_context is not None or self._ssl_options_explicit
+
     @classmethod
     def initialize_reactor(cls):
         """
@@ -1000,10 +1227,18 @@ def _build_ssl_context_from_options(self):
         # Python >= 3.10 requires either PROTOCOL_TLS_CLIENT or PROTOCOL_TLS_SERVER so we'll get ahead of things by always
         # being explicit
         ssl_version = opts.get('ssl_version', None) or ssl.PROTOCOL_TLS_CLIENT
-        cert_reqs = opts.get('cert_reqs', None) or ssl.CERT_REQUIRED
+        cert_reqs = opts.get('cert_reqs', None)
+        check_hostname = bool(opts.get('check_hostname', False))
+        if cert_reqs is None:
+            cert_reqs = (ssl.CERT_REQUIRED
+                         if self.ssl_options
+                         else ssl.CERT_NONE)
+        elif check_hostname and cert_reqs == ssl.CERT_NONE:
+            cert_reqs = ssl.CERT_REQUIRED
         rv = ssl.SSLContext(protocol=int(ssl_version))
-        rv.check_hostname = bool(opts.get('check_hostname', False))
-        rv.options = int(cert_reqs)
+        rv.check_hostname = False
+        rv.verify_mode = cert_reqs
+        rv.check_hostname = check_hostname
 
         certfile = opts.get('certfile', None)
         keyfile = opts.get('keyfile', None)
@@ -1012,6 +1247,10 @@ def _build_ssl_context_from_options(self):
         ca_certs = opts.get('ca_certs', None)
         if ca_certs:
             rv.load_verify_locations(ca_certs)
+        elif cert_reqs != ssl.CERT_NONE:
+            load_default_certs = getattr(rv, 'load_default_certs', None)
+            if load_default_certs is not None:
+                load_default_certs()
         ciphers = opts.get('ciphers', None)
         if ciphers:
             rv.set_ciphers(ciphers)