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

@@ -18,6 +18,7 @@
 from functools import wraps, partial, total_ordering
 from heapq import heappush, heappop
 import io
+import ipaddress
 import logging
 import socket
 import struct
@@ -55,6 +56,104 @@
 
 log = logging.getLogger(__name__)
 
+
+def _pyopenssl_cert_to_ssl_cert(peer_cert):
+    from cryptography import x509
+    from cryptography.x509.oid import NameOID
+
+    cert = peer_cert.to_cryptography()
+    ssl_cert = {}
+
+    subject = tuple((('commonName', attr.value),)
+                    for attr in cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME))
+    if subject:
+        ssl_cert['subject'] = subject
+
+    subject_alt_names = []
+    try:
+        san = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName).value
+    except x509.ExtensionNotFound:
+        pass
+    else:
+        subject_alt_names.extend(('DNS', name) for name in san.get_values_for_type(x509.DNSName))
+        subject_alt_names.extend(('IP Address', str(address))
+                                 for address in san.get_values_for_type(x509.IPAddress))
+
+    if subject_alt_names:
+        ssl_cert['subjectAltName'] = tuple(subject_alt_names)
+
+    return ssl_cert
+
+
+def _validate_pyopenssl_hostname(peer_cert, hostname):
+    cert = _pyopenssl_cert_to_ssl_cert(peer_cert)
+    try:
+        host_ip = ipaddress.ip_address(hostname)
+    except ValueError:
+        host_ip = None
+
+    names = []
+    for key, value in cert.get('subjectAltName', ()):
+        if key == 'DNS':
+            if host_ip is None and _dnsname_match(value, hostname):
+                return
+            names.append(value)
+        elif key == 'IP Address':
+            if host_ip is not None and _ipaddress_match(value, host_ip):
+                return
+            names.append(value)
+
+    if not names:
+        for subject in cert.get('subject', ()):
+            for key, value in subject:
+                if key == 'commonName':
+                    if _dnsname_match(value, hostname):
+                        return
+                    names.append(value)
+
+    if len(names) > 1:
+        raise ssl.CertificateError("hostname {!r} doesn't match either of {}".format(
+            hostname, ', '.join(map(repr, names))))
+    if len(names) == 1:
+        raise ssl.CertificateError("hostname {!r} doesn't match {!r}".format(hostname, names[0]))
+    raise ssl.CertificateError("no appropriate commonName or subjectAltName fields were found")
+
+
+def _dnsname_match(pattern, hostname):
+    if not pattern:
+        return False
+
+    wildcards = pattern.count('*')
+    if not wildcards:
+        return pattern.lower() == hostname.lower()
+    if wildcards > 1:
+        raise ssl.CertificateError(
+            "too many wildcards in certificate DNS name: {!r}.".format(pattern))
+
+    leftmost, sep, remainder = pattern.partition('.')
+    if '*' in remainder:
+        raise ssl.CertificateError(
+            "wildcard can only be present in the leftmost label: {!r}.".format(pattern))
+    if not sep:
+        raise ssl.CertificateError(
+            "sole wildcard without additional labels is not supported: {!r}.".format(pattern))
+    if leftmost != '*':
+        raise ssl.CertificateError(
+            "partial wildcards in leftmost label are not supported: {!r}.".format(pattern))
+
+    hostname_leftmost, sep, hostname_remainder = hostname.partition('.')
+    if not hostname_leftmost or not sep:
+        return False
+    return remainder.lower() == hostname_remainder.lower()
+
+
+def _ipaddress_match(pattern, host_ip):
+    try:
+        return ipaddress.ip_address(pattern) == host_ip
+    except ValueError:
+        return False
+
+
 segment_codec_no_compression = SegmentCodec()
 segment_codec_lz4 = None
 
@@ -803,6 +902,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 +985,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 +1009,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 +1025,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 +1049,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 +1111,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/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,8 @@
 from threading import Event
 import time
 
-from cassandra.connection import Connection, ConnectionShutdown, Timer, TimerManager
+from cassandra.connection import (Connection, ConnectionShutdown, Timer, TimerManager,
+                                  _validate_pyopenssl_hostname)
 try:
     from eventlet.green.OpenSSL import SSL
     _PYOPENSSL = True
@@ -35,6 +36,11 @@
 log = logging.getLogger(__name__)
 
 
+def _default_ssl_method():
+    return getattr(SSL, 'TLS_CLIENT_METHOD',
+                   getattr(SSL, 'TLS_METHOD', SSL.TLSv1_METHOD))
+
+
 def _check_pyopenssl():
     if not _PYOPENSSL:
         raise ImportError(
@@ -43,6 +49,28 @@ def _check_pyopenssl():
         )
 
 
+def _build_pyopenssl_context_from_options(ssl_options):
+    ssl_version = (ssl_options['ssl_version']
+                   if 'ssl_version' in ssl_options else _default_ssl_method())
+    context = SSL.Context(ssl_version)
+    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 = ssl_options.get('cert_reqs', None)
+    if cert_reqs is None:
+        cert_reqs = (SSL.VERIFY_PEER
+                     if (ssl_options.get('ca_certs', None) or ssl_options.get('check_hostname', False))
+                     else SSL.VERIFY_NONE)
+    context.set_verify(
+        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 +120,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,23 +136,26 @@ 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))
+            expected_name = (self.ssl_options or {}).get('server_hostname') or self.endpoint.address
+            _validate_pyopenssl_hostname(self._socket.get_peer_certificate(), expected_name)
+
+    def _build_ssl_context_from_options(self):
+        _check_pyopenssl()
+        return _build_pyopenssl_context_from_options(self.ssl_options)
 
     def close(self):
         with self.lock: