connection: preserve explicit empty ssl_options

Author: dkropachev

cassandra/cluster.py

@@ -1298,7 +1298,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 +1509,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

@@ -55,6 +55,7 @@
 
 log = logging.getLogger(__name__)
 
+
 segment_codec_no_compression = SegmentCodec()
 segment_codec_lz4 = None
 
@@ -130,6 +131,81 @@ 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.__class__ is not type(ssl.CERT_REQUIRED):
+        return cert_reqs
+    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 _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.get('ca_certs', None) or ssl_options.get('check_hostname', False))
+                     else ssl_module.VERIFY_NONE)
+    context.set_verify(
+        cert_reqs,
+        callback=lambda _connection, _x509, _errnum, _errdepth, ok: ok
+    )
+    ciphers = ssl_options.get('ciphers', None)
+    if ciphers:
+        if isinstance(ciphers, str):
+            ciphers = ciphers.encode('ascii')
+        context.set_cipher_list(ciphers)
+    return context
+
+
 class EndPoint(object):
     """
     Represents the information to connect to a cassandra node.
@@ -803,6 +879,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 +962,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 +986,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 +1002,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 +1026,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 +1088,15 @@ 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)
+        if cert_reqs is None:
+            cert_reqs = (ssl.CERT_REQUIRED
+                         if (opts.get('ca_certs', None) or opts.get('check_hostname', False))
+                         else ssl.CERT_NONE)
         rv = ssl.SSLContext(protocol=int(ssl_version))
+        rv.check_hostname = False
+        rv.verify_mode = cert_reqs
         rv.check_hostname = bool(opts.get('check_hostname', False))
-        rv.options = int(cert_reqs)
 
         certfile = opts.get('certfile', None)
         keyfile = opts.get('keyfile', None)

cassandra/datastax/cloud/__init__.py

@@ -34,6 +34,7 @@
     from zipfile import BadZipfile as BadZipFile
 
 from cassandra import DriverException
+from cassandra.connection import _default_pyopenssl_ssl_method
 
 log = logging.getLogger(__name__)
 
@@ -183,10 +184,10 @@ def _pyopenssl_context_from_cert(ca_cert_location, cert_location, key_location):
         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 = SSL.Context(_default_pyopenssl_ssl_method(SSL))
     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
+    return ssl_context

cassandra/datastax/insights/reporter.py

@@ -33,6 +33,13 @@
 log = logging.getLogger(__name__)
 
 
+def _ssl_options_cert_validation_enabled(ssl_options):
+    cert_reqs = ssl_options.get('cert_reqs')
+    if cert_reqs is not None:
+        return cert_reqs != ssl.CERT_NONE
+    return bool(ssl_options.get('ca_certs') or ssl_options.get('check_hostname', False))
+
+
 class MonitorReporter(Thread):
 
     def __init__(self, interval_sec, session):
@@ -141,14 +148,14 @@ def _get_startup_data(self):
 
         cert_validation = None
         try:
-            if self._session.cluster.ssl_context:
+            if self._session.cluster.ssl_context is not None:
                 if isinstance(self._session.cluster.ssl_context, ssl.SSLContext):
                     cert_validation = self._session.cluster.ssl_context.verify_mode == ssl.CERT_REQUIRED
                 else:  # pyopenssl
                     from OpenSSL import SSL
                     cert_validation = self._session.cluster.ssl_context.get_verify_mode() != SSL.VERIFY_NONE
-            elif self._session.cluster.ssl_options:
-                cert_validation = self._session.cluster.ssl_options.get('cert_reqs') == ssl.CERT_REQUIRED
+            elif self._session.cluster.ssl_options is not None:
+                cert_validation = _ssl_options_cert_validation_enabled(self._session.cluster.ssl_options)
         except Exception as e:
             log.debug('Unable to get the cert validation: {}'.format(e))
 
@@ -186,7 +193,8 @@ def _get_startup_data(self):
                 'compression': compression_type.upper() if compression_type else 'NONE',
                 'reconnectionPolicy': insights_registry.serialize(self._session.cluster.reconnection_policy),
                 'sslConfigured': {
-                    'enabled': bool(self._session.cluster.ssl_options or self._session.cluster.ssl_context),
+                    'enabled': (self._session.cluster.ssl_context is not None or
+                                self._session.cluster.ssl_options is not None),
                     'certValidation': cert_validation
                 },
                 'authProvider': {

cassandra/io/asyncioreactor.py

@@ -133,7 +133,7 @@ def __init__(self, *args, **kwargs):
         Connection.__init__(self, *args, **kwargs)
         self._background_tasks = set()
         self._transport = None
-        self._using_ssl = bool(self.ssl_context)
+        self._using_ssl = self._ssl_enabled
 
         self._connect_socket()
         self._socket.setblocking(0)

cassandra/io/eventletreactor.py

@@ -23,7 +23,9 @@
 from threading import Event
 import time
 
-from cassandra.connection import Connection, ConnectionShutdown, Timer, TimerManager
+from cassandra.connection import (Connection, ConnectionShutdown, Timer, TimerManager,
+                                  _build_pyopenssl_context_from_options as _build_pyopenssl_context,
+                                  _default_pyopenssl_ssl_method)
 try:
     from eventlet.green.OpenSSL import SSL
     _PYOPENSSL = True
@@ -35,6 +37,10 @@
 log = logging.getLogger(__name__)
 
 
+def _default_ssl_method():
+    return _default_pyopenssl_ssl_method(SSL)
+
+
 def _check_pyopenssl():
     if not _PYOPENSSL:
         raise ImportError(
@@ -43,6 +49,10 @@ def _check_pyopenssl():
         )
 
 
+def _build_pyopenssl_context_from_options(ssl_options):
+    return _build_pyopenssl_context(SSL, ssl_options)
+
+
 class EventletConnection(Connection):
     """
     An implementation of :class:`.Connection` that utilizes ``eventlet``.
@@ -92,7 +102,7 @@ def service_timeouts(cls):
 
     def __init__(self, *args, **kwargs):
         Connection.__init__(self, *args, **kwargs)
-        self.uses_legacy_ssl_options = self.ssl_options and not self.ssl_context
+        self.uses_legacy_ssl_options = False
         self._write_queue = Queue()
 
         self._connect_socket()
@@ -108,24 +118,29 @@ def _wrap_socket_from_context(self):
         if self.ssl_options and 'server_hostname' in self.ssl_options:
             # This is necessary for SNI
             self._socket.set_tlsext_host_name(self.ssl_options['server_hostname'].encode('ascii'))
+        return self._socket
 
     def _initiate_connection(self, sockaddr):
         if self.uses_legacy_ssl_options:
             super(EventletConnection, self)._initiate_connection(sockaddr)
         else:
             self._socket.connect(sockaddr)
-            if self.ssl_context or self.ssl_options:
+            if self._ssl_enabled:
                 self._socket.do_handshake()
 
-    def _match_hostname(self):
+    def _validate_hostname(self):
         if self.uses_legacy_ssl_options:
-            super(EventletConnection, self)._match_hostname()
+            super(EventletConnection, self)._validate_hostname()
         else:
             cert_name = self._socket.get_peer_certificate().get_subject().commonName
             if cert_name != self.endpoint.address:
                 raise Exception("Hostname verification failed! Certificate name '{}' "
                                 "doesn't endpoint '{}'".format(cert_name, self.endpoint.address))
 
+    def _build_ssl_context_from_options(self):
+        _check_pyopenssl()
+        return _build_pyopenssl_context_from_options(self.ssl_options)
+
     def close(self):
         with self.lock:
             if self.is_closed: