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,33 @@ 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. Empty options without additional
+  endpoint security settings encrypt the connection without verifying the
+  server certificate. Cluster-level ``ssl_options=None`` does not enable TLS by
+  itself; an ``ssl_context`` or endpoint SSL options still enable it. An
+  endpoint ``server_hostname`` used only for SNI routing is not treated as an
+  additional security setting for explicit empty cluster options.
+* 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. Serialized symbolic stdlib protocol names are supported;
+  plain integer protocol values are mapped only when unambiguous and otherwise
+  rejected rather than being silently interpreted as a different pyOpenSSL
+  method. 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

@@ -37,6 +37,7 @@
 import re
 import queue
 import socket
+import ssl
 import sys
 import time
 from threading import Lock, RLock, Thread, Event
@@ -883,12 +884,29 @@ def default_retry_policy(self, policy):
     when new sockets are created. This should be used when client encryption is enabled
     in Cassandra.
 
-    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.
+    The following documentation only applies when cluster-level ``ssl_options``
+    is used without ``ssl_context``. An :class:`~cassandra.connection.EndPoint`
+    may independently supply SSL options for its connections.
+
+    .. versionchanged:: 3.29.12
+
+        An explicit empty dict (``ssl_options={}``) enables TLS with default
+        options. Cluster-level ``ssl_options=None`` does not enable TLS by
+        itself; an ``ssl_context`` or SSL options supplied by the selected
+        :class:`~cassandra.connection.EndPoint` still enable it. Empty options
+        do not verify the server certificate unless the endpoint supplies
+        additional security settings. 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 +1316,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,12 +1527,25 @@ 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 '
                  'to prepare for that release.',
                  DeprecationWarning)
+            if not ssl_options:
+                log.warning(
+                    'Cluster-level ssl_options enable TLS without requesting '
+                    'server certificate verification; endpoint-level SSL '
+                    'options may enable verification for individual '
+                    'connections')
+            elif (ssl_options.get('check_hostname', False) and
+                  ssl_options.get('cert_reqs') == ssl.CERT_NONE):
+                log.warning(
+                    'Cluster-level check_hostname=True requires server '
+                    'certificate verification and overrides '
+                    'cert_reqs=CERT_NONE at cluster level; endpoint-level SSL '
+                    'options may override these settings')
 
         self.ssl_options = ssl_options
         self.ssl_context = ssl_context

cassandra/connection.py

@@ -20,11 +20,11 @@
 import io
 import logging
 import socket
+import ssl
 import struct
 import sys
 from threading import Thread, Event, RLock, Condition
 import time
-import ssl
 import uuid
 import weakref
 import random
@@ -50,6 +50,10 @@
                                 AuthSuccessMessage, ProtocolException,
                                 RegisterMessage, ReviseRequestMessage)
 from cassandra.segment import SegmentCodec, CrcException
+from cassandra.tls import (_build_ssl_context_from_options,
+                           _prepare_ssl_options,
+                           _ssl_options_requiring_new_context,
+                           _wrap_socket_from_context)
 from cassandra.util import OrderedDict
 from cassandra.shard_info import ShardingInfo  # noqa: F401  # re-exported for cassandra.connection.ShardingInfo
 
@@ -803,6 +807,8 @@ class Connection(object):
     endpoint = None
     ssl_options = None
     ssl_context = None
+    _ssl_options_explicit = False
+    _ssl_options_verify_by_default = None
     last_error = None
 
     # The current number of operations that are in flight. More precisely,
@@ -858,6 +864,7 @@ class Connection(object):
     _socket = None
 
     _socket_impl = socket
+    _connect_socket_error_types = (socket.error,)
 
     _check_hostname = False
     _product_type = None
@@ -883,9 +890,21 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None,
                  on_orphaned_stream_released=None, application_info: Optional[ApplicationInfoBase] = None):
         # TODO next major rename host to endpoint and remove port kwarg.
         self.endpoint = host if isinstance(host, EndPoint) else DefaultEndPoint(host, port)
+        endpoint_ssl_options = self.endpoint.ssl_options
+        self._endpoint_ssl_options = (
+            dict(endpoint_ssl_options)
+            if endpoint_ssl_options is not None
+            else None
+        )
 
         self.authenticator = authenticator
-        self.ssl_options = ssl_options.copy() if ssl_options else {}
+        (
+            self.ssl_options,
+            self._ssl_options_explicit,
+            self._ssl_options_verify_by_default,
+        ) = _prepare_ssl_options(
+            ssl_options, self._endpoint_ssl_options
+        )
         self.ssl_context = ssl_context
         self.sockopts = sockopts
         self.compression = compression
@@ -905,21 +924,28 @@ 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
-
-        # PYTHON-1331
-        #
-        # We always use SSLContext.wrap_socket() now but legacy configs may have other params that were passed to ssl.wrap_socket()...
-        # and either could have 'check_hostname'.  Remove these params into a separate map and use them to build an SSLContext if
-        # we need to do so.
-        #
-        # 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:
+        # An explicitly empty options mapping enables TLS with default options.
+        if self.ssl_context is None and self._ssl_options_explicit:
             self.ssl_context = self._build_ssl_context_from_options()
+        elif self.ssl_context is not None:
+            endpoint_context_options = _ssl_options_requiring_new_context(
+                self._endpoint_ssl_options)
+            if endpoint_context_options:
+                # SSLContext has no supported copy operation, and settings such
+                # as trust roots, client keys, and pyOpenSSL callbacks cannot be
+                # restored reliably. The cluster context is shared by concurrent
+                # connections, so mutating and restoring it would leak endpoint
+                # policy or race with another handshake. Require the existing
+                # context to remain immutable; callers needing endpoint-specific
+                # context settings can omit ssl_context and use ssl_options,
+                # which builds an independent context for every connection.
+                raise ValueError(
+                    "Endpoint SSL options %s require an independent SSL "
+                    "context and cannot be combined with a supplied "
+                    "ssl_context" % sorted(endpoint_context_options))
+
+        self._check_hostname = bool(self.ssl_options.get('check_hostname', False) or
+                                    getattr(self.ssl_context, 'check_hostname', False))
 
         self.max_request_id = min(self.max_in_flight - 1, (2 ** 15) - 1)
         # Don't fill the deque with 2**15 items right away. Start with some and add
@@ -942,6 +968,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):
         """
@@ -992,47 +1022,16 @@ def factory(cls, endpoint, timeout, host_conn = None, *args, **kwargs):
             return conn
 
     def _build_ssl_context_from_options(self):
-
-        # Extract a subset of names from self.ssl_options which apply to SSLContext creation
-        ssl_context_opt_names = ['ssl_version', 'cert_reqs', 'check_hostname', 'keyfile', 'certfile', 'ca_certs', 'ciphers']
-        opts = {k:self.ssl_options.get(k, None) for k in ssl_context_opt_names if k in self.ssl_options}
-
-        # 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
-        rv = ssl.SSLContext(protocol=int(ssl_version))
-        rv.check_hostname = bool(opts.get('check_hostname', False))
-        rv.options = int(cert_reqs)
-
-        certfile = opts.get('certfile', None)
-        keyfile = opts.get('keyfile', None)
-        if certfile:
-            rv.load_cert_chain(certfile, keyfile)
-        ca_certs = opts.get('ca_certs', None)
-        if ca_certs:
-            rv.load_verify_locations(ca_certs)
-        ciphers = opts.get('ciphers', None)
-        if ciphers:
-            rv.set_ciphers(ciphers)
-
-        return rv
+        return _build_ssl_context_from_options(
+            self.ssl_options,
+            verify_by_default=self._ssl_options_verify_by_default)
 
     def _wrap_socket_from_context(self):
-
-        # Extract a subset of names from self.ssl_options which apply to SSLContext.wrap_socket (or at least the parts
-        # of it that don't involve building an SSLContext under the covers)
-        wrap_socket_opt_names = ['server_side', 'do_handshake_on_connect', 'suppress_ragged_eofs', 'server_hostname']
-        opts = {k:self.ssl_options.get(k, None) for k in wrap_socket_opt_names if k in self.ssl_options}
-
-        # PYTHON-1186: set the server_hostname only if the SSLContext has
-        # check_hostname enabled and it is not already provided by the EndPoint ssl options
-        #opts['server_hostname'] = self.endpoint.address
-        if (self.ssl_context.check_hostname and 'server_hostname' not in opts):
-            server_hostname = self.endpoint.address
-            opts['server_hostname'] = server_hostname
-
-        return self.ssl_context.wrap_socket(self._socket, **opts)
+        return _wrap_socket_from_context(
+            self.ssl_context,
+            self._socket,
+            self.ssl_options,
+            self.endpoint.address)
 
     def _initiate_connection(self, sockaddr):
         if self.features.shard_id is not None:
@@ -1089,13 +1088,27 @@ def _connect_socket(self):
                     self._validate_hostname()
                 sockerr = None
                 break
-            except socket.error as err:
+            except self._connect_socket_error_types as err:
+                if self._socket:
+                    self._socket.close()
+                    self._socket = None
+                # Preserve a TLS or driver error over later socket failures so
+                # a final ECONNREFUSED cannot hide certificate diagnostics.
+                if (sockerr is None or
+                        (isinstance(sockerr, socket.error) and
+                         not isinstance(sockerr, ssl.SSLError))):
+                    sockerr = err
+            except Exception:
                 if self._socket:
                     self._socket.close()
                     self._socket = None
-                sockerr = err
+                raise
 
         if sockerr:
+            if isinstance(sockerr, ssl.SSLError):
+                raise sockerr
+            if not isinstance(sockerr, socket.error):
+                raise sockerr
             raise socket.error(sockerr.errno, "Tried connecting to %s. Last error: %s" %
                                ([a[4] for a in addresses], sockerr.strerror or sockerr))
 

cassandra/datastax/cloud/__init__.py

@@ -34,6 +34,7 @@
     from zipfile import BadZipfile as BadZipFile
 
 from cassandra import DriverException
+from cassandra.tls import _build_pyopenssl_context_from_options
 
 log = logging.getLogger(__name__)
 
@@ -181,12 +182,14 @@ def _pyopenssl_context_from_cert(ca_cert_location, cert_location, key_location):
         from OpenSSL import SSL
     except ImportError as e:
         raise ImportError(
-            "PyOpenSSL must be installed to connect to Astra with the Eventlet or Twisted event loops")\
-            .with_traceback(e.__traceback__)
-    ssl_context = SSL.Context(SSL.TLSv1_METHOD)
-    ssl_context.set_verify(SSL.VERIFY_PEER, callback=lambda _1, _2, _3, _4, ok: ok)
-    ssl_context.use_certificate_file(cert_location)
-    ssl_context.use_privatekey_file(key_location)
-    ssl_context.load_verify_locations(ca_cert_location)
-
-    return ssl_context
+            "PyOpenSSL must be installed to connect to Astra with the Eventlet or Twisted event loops"
+        ) from e
+    return _build_pyopenssl_context_from_options(
+        SSL,
+        {
+            'ca_certs': ca_cert_location,
+            'certfile': cert_location,
+            'keyfile': key_location,
+            'cert_reqs': CERT_REQUIRED,
+        }
+    )