connection: preserve explicit empty ssl_options

Problem: several SSL paths used ssl_options truthiness, so explicit ssl_options={} was treated like omitted ssl_options=None. That disabled SSL setup in connection/reactor paths, skipped the ssl_options deprecation warning, and made Insights report SSL as disabled.

Fix: preserve whether ssl_options was supplied and expose a shared SSL-enabled predicate. Build default SSL contexts for explicit ssl_options on the standard path, use pyOpenSSL contexts for Twisted/Eventlet, and update Cluster and Insights checks to use explicit None comparisons. Add focused unit coverage for connection setup, client routes, Twisted/Eventlet, and Insights reporting.

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

@@ -803,6 +803,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 +886,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 +910,11 @@ 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
 
         # PYTHON-1331
         #
@@ -918,7 +924,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 +948,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):
         """

cassandra/datastax/insights/reporter.py

@@ -141,13 +141,13 @@ 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:
+            elif self._session.cluster.ssl_options is not None:
                 cert_validation = self._session.cluster.ssl_options.get('cert_reqs') == ssl.CERT_REQUIRED
         except Exception as e:
             log.debug('Unable to get the cert validation: {}'.format(e))
@@ -186,7 +186,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

@@ -43,6 +43,22 @@ def _check_pyopenssl():
         )
 
 
+def _build_pyopenssl_context_from_options(ssl_options):
+    context = SSL.Context(ssl_options.get('ssl_version', SSL.TLSv1_METHOD))
+    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'])
+    if 'cert_reqs' in ssl_options:
+        context.set_verify(
+            ssl_options['cert_reqs'],
+            callback=lambda _connection, _x509, _errnum, _errdepth, ok: ok
+        )
+    return context
+
+
 class EventletConnection(Connection):
     """
     An implementation of :class:`.Connection` that utilizes ``eventlet``.
@@ -92,7 +108,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()
@@ -114,7 +130,7 @@ def _initiate_connection(self, sockaddr):
             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):
@@ -126,6 +142,10 @@ def _match_hostname(self):
                 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:

cassandra/io/twistedreactor.py

@@ -39,6 +39,30 @@
 log = logging.getLogger(__name__)
 
 
+def _check_pyopenssl():
+    if not _HAS_SSL:
+        raise ImportError(
+            str(import_exception) +
+            ', pyOpenSSL must be installed to enable SSL support with the Twisted event loop'
+        )
+
+
+def _build_pyopenssl_context_from_options(ssl_options):
+    context = SSL.Context(ssl_options.get("ssl_version", SSL.TLSv1_METHOD))
+    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"])
+    if "cert_reqs" in ssl_options:
+        context.set_verify(
+            ssl_options["cert_reqs"],
+            callback=lambda _connection, _x509, _errnum, _errdepth, ok: ok
+        )
+    return context
+
+
 def _cleanup(cleanup_weakref):
     try:
         cleanup_weakref()._cleanup()
@@ -141,25 +165,14 @@ def _on_loop_timer(self):
 class _SSLCreator(object):
     def __init__(self, endpoint, ssl_context, ssl_options, check_hostname, timeout):
         self.endpoint = endpoint
-        self.ssl_options = ssl_options
+        self.ssl_options = ssl_options or {}
         self.check_hostname = check_hostname
         self.timeout = timeout
 
-        if ssl_context:
+        if ssl_context is not None:
             self.context = ssl_context
         else:
-            self.context = SSL.Context(SSL.TLSv1_METHOD)
-            if "certfile" in self.ssl_options:
-                self.context.use_certificate_file(self.ssl_options["certfile"])
-            if "keyfile" in self.ssl_options:
-                self.context.use_privatekey_file(self.ssl_options["keyfile"])
-            if "ca_certs" in self.ssl_options:
-                self.context.load_verify_locations(self.ssl_options["ca_certs"])
-            if "cert_reqs" in self.ssl_options:
-                self.context.set_verify(
-                    self.ssl_options["cert_reqs"],
-                    callback=self.verify_callback
-                )
+            self.context = _build_pyopenssl_context_from_options(self.ssl_options)
         self.context.set_info_callback(self.info_callback)
 
     def verify_callback(self, connection, x509, errnum, errdepth, ok):
@@ -217,27 +230,27 @@ def __init__(self, *args, **kwargs):
         self._loop.maybe_start()
 
     def _check_pyopenssl(self):
-        if self.ssl_context or self.ssl_options:
-            if not _HAS_SSL:
-                raise ImportError(
-                    str(import_exception) +
-                    ', pyOpenSSL must be installed to enable SSL support with the Twisted event loop'
-                )
+        if self._ssl_enabled:
+            _check_pyopenssl()
+
+    def _build_ssl_context_from_options(self):
+        _check_pyopenssl()
+        return _build_pyopenssl_context_from_options(self.ssl_options)
 
     def add_connection(self):
         """
         Convenience function to connect and store the resulting
         connector.
         """
         host, port = self.endpoint.resolve()
-        if self.ssl_context or self.ssl_options:
+        if self._ssl_enabled:
             # Can't use optionsForClientTLS here because it *forces* hostname verification.
             # Cool they enforce strong security, but we have to be able to turn it off
             self._check_pyopenssl()
 
             ssl_connection_creator = _SSLCreator(
                 self.endpoint,
-                self.ssl_context if self.ssl_context else None,
+                self.ssl_context if self.ssl_context is not None else None,
                 self.ssl_options,
                 self._check_hostname,
                 self.connect_timeout,

tests/unit/advanced/test_insights.py

@@ -16,18 +16,22 @@
 import unittest
 
 import logging
+import ssl
 import sys
-from unittest.mock import sentinel
+from unittest.mock import Mock, sentinel
 
 from cassandra import ConsistencyLevel
 from cassandra.cluster import (
     ExecutionProfile, GraphExecutionProfile, GraphAnalyticsExecutionProfile
 )
+from cassandra.connection import Connection
 from cassandra.datastax.graph.query import GraphOptions
+from cassandra.datastax.insights.reporter import MonitorReporter
 from cassandra.datastax.insights.registry import insights_registry
 from cassandra.datastax.insights.serializers import initialize_registry
 from cassandra.policies import (
     LoadBalancingPolicy,
+    HostDistance,
     DCAwareRoundRobinPolicy,
     TokenAwarePolicy,
     WhiteListRoundRobinPolicy,
@@ -88,6 +92,60 @@ def superclass_sentinel_serializer(obj):
         # with default -- same behavior
         assert insights_registry.serialize(SubclassSentinel(), default=object()) is sentinel.serialized_superclass
 
+
+class TestMonitorReporterStartupData(unittest.TestCase):
+
+    def _get_startup_data(self, ssl_options=None, ssl_context=None):
+        reporter = MonitorReporter.__new__(MonitorReporter)
+        reporter._interval = 30
+
+        connection = Mock()
+        connection.host = '127.0.0.1'
+        connection._compression_type = 'NONE'
+        connection._socket.getsockname.return_value = ('127.0.0.1', 9042)
+
+        cluster = Mock()
+        cluster.auth_provider = None
+        cluster.client_id = 'client-id'
+        cluster.connection_class = Connection
+        cluster.control_connection._connection = connection
+        cluster.application_name = None
+        cluster.application_version = None
+        cluster._endpoint_map_for_insights = {}
+        cluster.idle_heartbeat_interval = 30
+        cluster.metadata.all_hosts.return_value = []
+        cluster.profile_manager.distance.return_value = HostDistance.LOCAL
+        cluster.protocol_version = 4
+        cluster.reconnection_policy = object()
+        cluster.ssl_context = ssl_context
+        cluster.ssl_options = ssl_options
+
+        session = Mock()
+        session.cluster = cluster
+        session.hosts = []
+        session.session_id = 'session-id'
+
+        reporter._session = session
+        return reporter._get_startup_data()
+
+    def test_empty_ssl_options_reported_as_enabled(self):
+        startup_data = self._get_startup_data(ssl_options={})
+
+        assert startup_data['data']['sslConfigured']['enabled'] is True
+
+    def test_omitted_ssl_options_reported_as_disabled(self):
+        startup_data = self._get_startup_data()
+
+        assert startup_data['data']['sslConfigured']['enabled'] is False
+
+    def test_ssl_context_reported_as_enabled(self):
+        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+
+        startup_data = self._get_startup_data(ssl_context=ssl_context)
+
+        assert startup_data['data']['sslConfigured']['enabled'] is True
+
+
 class TestConfigAsDict(unittest.TestCase):
 
     #  graph/query.py

tests/unit/io/test_eventletreactor.py

@@ -23,10 +23,28 @@
 
 try:
     from eventlet import monkey_patch
+    from cassandra.io import eventletreactor
     from cassandra.io.eventletreactor import EventletConnection
 except (ImportError, AttributeError):
+    eventletreactor = None
     EventletConnection = None  # noqa
 
+from cassandra.connection import Connection, DefaultEndPoint
+
+
+@unittest.skipIf(EventletConnection is None, "eventlet is not available")
+@unittest.skipIf(not getattr(eventletreactor, '_PYOPENSSL', False), "pyOpenSSL is not available")
+class EventletSSLContextTest(unittest.TestCase):
+
+    def test_empty_ssl_options_build_pyopenssl_context(self):
+        conn = EventletConnection.__new__(EventletConnection)
+
+        Connection.__init__(conn, DefaultEndPoint('1.2.3.4'), ssl_options={})
+
+        assert conn._ssl_enabled
+        assert conn.ssl_context is not None
+
+
 skip_condition = EventletConnection is None or EVENT_LOOP_MANAGER != "eventlet"
 # There are some issues with some versions of pypy and eventlet
 @notpypy

tests/unit/io/test_twistedreactor.py

@@ -196,3 +196,19 @@ def test_push(self, mock_connectTCP):
         self.obj_ut.push('123 pickup')
         self.mock_reactor_cft.assert_called_with(
             transport_mock.write, '123 pickup')
+
+    @unittest.skipIf(not getattr(twistedreactor, '_HAS_SSL', False), "pyOpenSSL is not available")
+    @patch('cassandra.io.twistedreactor.connectProtocol')
+    @patch('cassandra.io.twistedreactor.TCP4ClientEndpoint')
+    @patch('cassandra.io.twistedreactor.SSL4ClientEndpoint')
+    def test_empty_ssl_options_use_ssl_endpoint(self, mock_ssl_endpoint, mock_tcp_endpoint, mock_connect_protocol):
+        conn = twistedreactor.TwistedConnection(
+            DefaultEndPoint('1.2.3.4'),
+            cql_version='3.0.1',
+            ssl_options={})
+
+        conn.add_connection()
+
+        mock_ssl_endpoint.assert_called_once()
+        mock_tcp_endpoint.assert_not_called()
+        mock_connect_protocol.assert_called_once()