diff --git a/cassandra/cluster.py b/cassandra/cluster.py
index 12ade2018f..47f06348c7 100644
--- a/cassandra/cluster.py
+++ b/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")
 
@@ -1349,7 +1350,7 @@ def __init__(self,
             # connections go through NLB proxies whose addresses won't match
             # server certificates.
             _check_hostname_enabled = False
-            if ssl_context is not None and ssl_context.check_hostname:
+            if ssl_context is not None and getattr(ssl_context, 'check_hostname', False):
                 _check_hostname_enabled = True
             if ssl_options is not None and ssl_options.get('check_hostname', False):
                 _check_hostname_enabled = True
@@ -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 '
diff --git a/cassandra/connection.py b/cassandra/connection.py
index f238416b29..019dd58df9 100644
--- a/cassandra/connection.py
+++ b/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,184 @@
 
 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
+
+
+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_ssl_version(ssl_version, ssl_module):
+    if not isinstance(ssl_version, type(ssl.PROTOCOL_TLS)):
+        return ssl_version
+
+    protocol_method_names = (
+        ('PROTOCOL_TLS', None),
+        ('PROTOCOL_SSLv23', None),
+        ('PROTOCOL_TLS_CLIENT', None),
+        ('PROTOCOL_TLS_SERVER', ('TLS_SERVER_METHOD', 'TLS_METHOD')),
+        ('PROTOCOL_TLSv1', ('TLSv1_METHOD',)),
+        ('PROTOCOL_TLSv1_1', ('TLSv1_1_METHOD',)),
+        ('PROTOCOL_TLSv1_2', ('TLSv1_2_METHOD',)),
+    )
+    for protocol_name, method_names in protocol_method_names:
+        if getattr(ssl, protocol_name, None) is ssl_version:
+            if method_names is None:
+                return _default_pyopenssl_ssl_method(ssl_module)
+            for method_name in method_names:
+                method = getattr(ssl_module, method_name, None)
+                if method is not None:
+                    return method
+            raise ValueError('pyOpenSSL does not expose a method for {}'.format(protocol_name))
+
+    return ssl_version
+
+
+def _pyopenssl_verify_mode_from_cert_reqs(cert_reqs, ssl_module):
+    if not isinstance(cert_reqs, type(ssl.CERT_NONE)):
+        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_options, ssl_module):
+    ssl_version = (ssl_options['ssl_version']
+                   if 'ssl_version' in ssl_options else _default_pyopenssl_ssl_method(ssl_module))
+    ssl_version = _pyopenssl_ssl_method_from_ssl_version(ssl_version, ssl_module)
+    context = ssl_module.Context(ssl_version)
+    certfile = ssl_options.get('certfile', None)
+    keyfile = ssl_options.get('keyfile', None)
+    if certfile:
+        use_certificate_chain_file = getattr(context, 'use_certificate_chain_file', None)
+        if use_certificate_chain_file:
+            use_certificate_chain_file(certfile)
+        else:
+            context.use_certificate_file(certfile)
+        context.use_privatekey_file(keyfile or certfile)
+        context.check_privatekey()
+    elif keyfile:
+        context.use_privatekey_file(keyfile)
+    ca_certs = ssl_options.get('ca_certs', None)
+    if ca_certs:
+        context.load_verify_locations(ca_certs)
+    ciphers = ssl_options.get('ciphers', None)
+    if ciphers:
+        context.set_cipher_list(ciphers)
+    cert_reqs = 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)
+    cert_reqs = _pyopenssl_verify_mode_from_cert_reqs(cert_reqs, ssl_module)
+    context.set_verify(
+        cert_reqs,
+        callback=lambda _connection, _x509, _errnum, _errdepth, ok: ok
+    )
+    return context
+
+
 segment_codec_no_compression = SegmentCodec()
 segment_codec_lz4 = None
 
@@ -803,6 +982,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 +1065,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 +1089,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 +1105,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 +1129,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 +1191,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)
diff --git a/cassandra/datastax/cloud/__init__.py b/cassandra/datastax/cloud/__init__.py
index be79d6db38..b9311ad7be 100644
--- a/cassandra/datastax/cloud/__init__.py
+++ b/cassandra/datastax/cloud/__init__.py
@@ -34,6 +34,7 @@
     from zipfile import BadZipfile as BadZipFile
 
 from cassandra import DriverException
+from cassandra.connection import _build_pyopenssl_context_from_options
 
 log = logging.getLogger(__name__)
 
@@ -89,7 +90,7 @@ def get_cloud_config(cloud_config, create_pyopenssl_context=False):
 
     config = read_metadata_info(config, cloud_config)
     if create_pyopenssl_context:
-        config.ssl_context = config.pyopenssl_context
+        config.ssl_context = getattr(config, 'pyopenssl_context', config.ssl_context)
     return config
 
 
@@ -112,19 +113,19 @@ def parse_cloud_config(path, cloud_config, create_pyopenssl_context):
 
     config = CloudConfig.from_dict(data)
     config_dir = os.path.dirname(path)
+    ca_cert_location = os.path.join(config_dir, 'ca.crt')
+    cert_location = os.path.join(config_dir, 'cert')
+    key_location = os.path.join(config_dir, 'key')
 
     if 'ssl_context' in cloud_config:
         config.ssl_context = cloud_config['ssl_context']
     else:
         # Load the ssl_context before we delete the temporary directory
-        ca_cert_location = os.path.join(config_dir, 'ca.crt')
-        cert_location = os.path.join(config_dir, 'cert')
-        key_location = os.path.join(config_dir, 'key')
         # Regardless of if we create a pyopenssl context, we still need the builtin one
         # to connect to the metadata service
         config.ssl_context = _ssl_context_from_cert(ca_cert_location, cert_location, key_location)
-        if create_pyopenssl_context:
-            config.pyopenssl_context = _pyopenssl_context_from_cert(ca_cert_location, cert_location, key_location)
+    if create_pyopenssl_context:
+        config.pyopenssl_context = _pyopenssl_context_from_cert(ca_cert_location, cert_location, key_location)
 
     return config
 
@@ -183,10 +184,12 @@ 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.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
\ No newline at end of file
+    return _build_pyopenssl_context_from_options(
+        {
+            'ca_certs': ca_cert_location,
+            'certfile': cert_location,
+            'keyfile': key_location,
+            'cert_reqs': SSL.VERIFY_PEER,
+        },
+        SSL
+    )
diff --git a/cassandra/datastax/insights/reporter.py b/cassandra/datastax/insights/reporter.py
index 83205fc458..1ca037f17b 100644
--- a/cassandra/datastax/insights/reporter.py
+++ b/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': {
diff --git a/cassandra/io/asyncioreactor.py b/cassandra/io/asyncioreactor.py
index 92ab972e7d..7f2c7a208b 100644
--- a/cassandra/io/asyncioreactor.py
+++ b/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)
diff --git a/cassandra/io/eventletreactor.py b/cassandra/io/eventletreactor.py
index 234a4a574c..716697d52a 100644
--- a/cassandra/io/eventletreactor.py
+++ b/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,
+                                  _validate_pyopenssl_hostname)
 try:
     from eventlet.green.OpenSSL import SSL
     _PYOPENSSL = True
@@ -92,7 +94,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 +110,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, SSL)
 
     def close(self):
         with self.lock:
diff --git a/cassandra/io/twistedreactor.py b/cassandra/io/twistedreactor.py
index 446200bf63..43adcec346 100644
--- a/cassandra/io/twistedreactor.py
+++ b/cassandra/io/twistedreactor.py
@@ -28,7 +28,9 @@
 from twisted.python.failure import Failure
 from zope.interface import implementer
 
-from cassandra.connection import Connection, ConnectionShutdown, Timer, TimerManager, ConnectionException
+from cassandra.connection import (Connection, ConnectionShutdown, Timer, TimerManager, ConnectionException,
+                                  _build_pyopenssl_context_from_options,
+                                  _validate_pyopenssl_hostname)
 
 try:
     from OpenSSL import SSL
@@ -39,6 +41,14 @@
 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 _cleanup(cleanup_weakref):
     try:
         cleanup_weakref()._cleanup()
@@ -141,39 +151,35 @@ 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.set_info_callback(self.info_callback)
+            self.context = _build_pyopenssl_context_from_options(self.ssl_options, SSL)
 
     def verify_callback(self, connection, x509, errnum, errdepth, ok):
         return ok
 
+    def _hostname_to_verify(self):
+        return self.ssl_options.get("server_hostname") or self.endpoint.address
+
     def info_callback(self, connection, where, ret):
         if where & SSL.SSL_CB_HANDSHAKE_DONE:
-            if self.check_hostname and self.endpoint.address != connection.get_peer_certificate().get_subject().commonName:
-                transport = connection.get_app_data()
-                transport.failVerification(Failure(ConnectionException("Hostname verification failed", self.endpoint)))
+            if self.check_hostname:
+                try:
+                    _validate_pyopenssl_hostname(connection.get_peer_certificate(), self._hostname_to_verify())
+                except Exception as exc:
+                    transport = connection.get_app_data()
+                    transport.failVerification(Failure(ConnectionException(str(exc), self.endpoint)))
 
     def clientConnectionForTLS(self, tlsProtocol):
         connection = SSL.Connection(self.context, None)
         connection.set_app_data(tlsProtocol)
+        if self.check_hostname:
+            connection.set_info_callback(self.info_callback)
         if self.ssl_options and "server_hostname" in self.ssl_options:
             connection.set_tlsext_host_name(self.ssl_options['server_hostname'].encode('ascii'))
         return connection
@@ -217,12 +223,12 @@ 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, SSL)
 
     def add_connection(self):
         """
@@ -230,14 +236,14 @@ def add_connection(self):
         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,
diff --git a/cassandra/pool.py b/cassandra/pool.py
index 176751f60a..cd489b1044 100644
--- a/cassandra/pool.py
+++ b/cassandra/pool.py
@@ -691,10 +691,11 @@ def _get_shard_aware_endpoint(self):
         ssl_enabled = cluster.ssl_context is not None or cluster.ssl_options is not None
 
         endpoint = None
-        if ssl_enabled and self.host.sharding_info.shard_aware_port_ssl:
-            endpoint = copy.copy(self.host.endpoint)
-            endpoint._port = self.host.sharding_info.shard_aware_port_ssl
-        elif not ssl_enabled and self.host.sharding_info.shard_aware_port:
+        if ssl_enabled:
+            if self.host.sharding_info.shard_aware_port_ssl:
+                endpoint = copy.copy(self.host.endpoint)
+                endpoint._port = self.host.sharding_info.shard_aware_port_ssl
+        elif self.host.sharding_info.shard_aware_port:
             endpoint = copy.copy(self.host.endpoint)
             endpoint._port = self.host.sharding_info.shard_aware_port
 
diff --git a/tests/unit/advanced/test_insights.py b/tests/unit/advanced/test_insights.py
index 2050439804..92f2bed51d 100644
--- a/tests/unit/advanced/test_insights.py
+++ b/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,75 @@ 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_implicit_ssl_options_validation_reported(self):
+        startup_data = self._get_startup_data(ssl_options={'ca_certs': 'ca.pem'})
+
+        assert startup_data['data']['sslConfigured']['certValidation'] is True
+
+    def test_check_hostname_validation_reported(self):
+        startup_data = self._get_startup_data(ssl_options={'check_hostname': True})
+
+        assert startup_data['data']['sslConfigured']['certValidation'] is True
+
+    def test_empty_ssl_options_validation_reported_disabled(self):
+        startup_data = self._get_startup_data(ssl_options={})
+
+        assert startup_data['data']['sslConfigured']['certValidation'] is False
+
+    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
diff --git a/tests/unit/io/test_eventletreactor.py b/tests/unit/io/test_eventletreactor.py
index d3962196a4..ebf6a346be 100644
--- a/tests/unit/io/test_eventletreactor.py
+++ b/tests/unit/io/test_eventletreactor.py
@@ -12,10 +12,22 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
+import os
+import ssl
 import unittest
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+
+from unittest.mock import Mock, patch
 
-from unittest.mock import patch
+try:
+    from cryptography import x509
+    from cryptography.hazmat.primitives import hashes
+    from cryptography.hazmat.primitives.asymmetric import rsa
+    from cryptography.x509.oid import NameOID
+    from OpenSSL import crypto
+except ImportError:
+    crypto = None
 
 from tests.unit.io.utils import TimerTestMixin
 from tests import notpypy, EVENT_LOOP_MANAGER
@@ -23,10 +35,157 @@
 
 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, _default_pyopenssl_ssl_method
+
+CA_CERTS = os.path.abspath(os.path.join(
+    os.path.dirname(__file__), '..', '..', 'integration', 'long', 'ssl', 'rootCa.crt'))
+
+
+def _make_certificate(common_name, san_dns_names=None):
+    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    subject = issuer = x509.Name([
+        x509.NameAttribute(NameOID.COMMON_NAME, common_name),
+    ])
+    now = datetime.now(timezone.utc)
+    builder = (x509.CertificateBuilder()
+               .subject_name(subject)
+               .issuer_name(issuer)
+               .public_key(key.public_key())
+               .serial_number(x509.random_serial_number())
+               .not_valid_before(now - timedelta(days=1))
+               .not_valid_after(now + timedelta(days=1)))
+    if san_dns_names:
+        builder = builder.add_extension(
+            x509.SubjectAlternativeName([x509.DNSName(name) for name in san_dns_names]),
+            critical=False)
+
+    return crypto.X509.from_cryptography(builder.sign(key, hashes.SHA256()))
+
+
+@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
+
+    def test_empty_ssl_options_default_to_negotiating_tls(self):
+        with patch.object(eventletreactor.SSL, 'Context') as context_mock:
+            context = eventletreactor._build_pyopenssl_context_from_options({}, eventletreactor.SSL)
+
+        context_mock.assert_called_once_with(eventletreactor.SSL.TLS_CLIENT_METHOD)
+        assert context is context_mock.return_value
+
+    def test_default_ssl_method_falls_back_to_tls_method(self):
+        tls_method = object()
+
+        with patch.object(eventletreactor, 'SSL', SimpleNamespace(TLS_METHOD=tls_method)):
+            assert _default_pyopenssl_ssl_method(eventletreactor.SSL) is tls_method
+
+    def test_default_ssl_method_falls_back_to_tlsv1_2_method(self):
+        tlsv1_2_method = object()
+
+        with patch.object(eventletreactor, 'SSL', SimpleNamespace(TLSv1_2_METHOD=tlsv1_2_method)):
+            assert _default_pyopenssl_ssl_method(eventletreactor.SSL) is tlsv1_2_method
+
+    def test_ssl_version_option_is_preserved(self):
+        with patch.object(eventletreactor.SSL, 'Context') as context_mock:
+            eventletreactor._build_pyopenssl_context_from_options(
+                {'ssl_version': eventletreactor.SSL.TLSv1_2_METHOD}, eventletreactor.SSL)
+
+        context_mock.assert_called_once_with(eventletreactor.SSL.TLSv1_2_METHOD)
+
+    def test_stdlib_ssl_version_option_is_translated(self):
+        with patch.object(eventletreactor.SSL, 'Context') as context_mock:
+            eventletreactor._build_pyopenssl_context_from_options(
+                {'ssl_version': ssl.PROTOCOL_TLS}, eventletreactor.SSL)
+
+        context_mock.assert_called_once_with(eventletreactor.SSL.TLS_CLIENT_METHOD)
+
+    def test_ca_certs_default_to_required_validation(self):
+        conn = EventletConnection.__new__(EventletConnection)
+
+        Connection.__init__(conn, DefaultEndPoint('1.2.3.4'), ssl_options={'ca_certs': CA_CERTS})
+
+        assert conn.ssl_context.get_verify_mode() == eventletreactor.SSL.VERIFY_PEER
+
+    def test_check_hostname_option_enables_hostname_validation(self):
+        conn = EventletConnection.__new__(EventletConnection)
+
+        Connection.__init__(conn, DefaultEndPoint('1.2.3.4'), ssl_options={'check_hostname': True})
+
+        assert conn._check_hostname
+
+    def test_validate_hostname_rejects_mismatch(self):
+        conn = EventletConnection.__new__(EventletConnection)
+        conn.uses_legacy_ssl_options = False
+        conn.endpoint = DefaultEndPoint('1.2.3.4')
+        conn.ssl_options = {}
+        conn._socket = Mock()
+        conn._socket.get_peer_certificate.return_value = _make_certificate('wrong.host')
+
+        with self.assertRaises(ssl.CertificateError):
+            conn._validate_hostname()
+
+    def test_validate_hostname_uses_server_hostname(self):
+        conn = EventletConnection.__new__(EventletConnection)
+        conn.uses_legacy_ssl_options = False
+        conn.endpoint = DefaultEndPoint('proxy.host')
+        conn.ssl_options = {'server_hostname': 'sni.host'}
+        conn._socket = Mock()
+        conn._socket.get_peer_certificate.return_value = _make_certificate('proxy.host', ['sni.host'])
+
+        conn._validate_hostname()
+
+    def test_validate_hostname_prefers_san_over_common_name(self):
+        conn = EventletConnection.__new__(EventletConnection)
+        conn.uses_legacy_ssl_options = False
+        conn.endpoint = DefaultEndPoint('sni.host')
+        conn.ssl_options = {}
+        conn._socket = Mock()
+        conn._socket.get_peer_certificate.return_value = _make_certificate('sni.host', ['other.host'])
+
+        with self.assertRaises(ssl.CertificateError):
+            conn._validate_hostname()
+
+    def test_validate_hostname_matches_wildcard_san(self):
+        conn = EventletConnection.__new__(EventletConnection)
+        conn.uses_legacy_ssl_options = False
+        conn.endpoint = DefaultEndPoint('node.example.com')
+        conn.ssl_options = {}
+        conn._socket = Mock()
+        conn._socket.get_peer_certificate.return_value = _make_certificate('other.host', ['*.example.com'])
+
+        conn._validate_hostname()
+
+    def test_wrap_socket_from_context_returns_wrapped_socket(self):
+        conn = EventletConnection.__new__(EventletConnection)
+        conn.ssl_context = object()
+        conn.ssl_options = {}
+        original_socket = object()
+        conn._socket = original_socket
+
+        with patch.object(eventletreactor.SSL, 'Connection') as mock_connection:
+            wrapped_socket = mock_connection.return_value
+
+            assert conn._wrap_socket_from_context() is wrapped_socket
+
+        mock_connection.assert_called_once_with(conn.ssl_context, original_socket)
+        wrapped_socket.set_connect_state.assert_called_once_with()
+        assert conn._socket is wrapped_socket
+
+
 skip_condition = EventletConnection is None or EVENT_LOOP_MANAGER != "eventlet"
 # There are some issues with some versions of pypy and eventlet
 @notpypy
diff --git a/tests/unit/io/test_twistedreactor.py b/tests/unit/io/test_twistedreactor.py
index 02bac10d8e..8b41adfaac 100644
--- a/tests/unit/io/test_twistedreactor.py
+++ b/tests/unit/io/test_twistedreactor.py
@@ -12,10 +12,23 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import os
+import ssl
 import unittest
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
 from unittest.mock import Mock, patch
 
-from cassandra.connection import DefaultEndPoint
+try:
+    from cryptography import x509
+    from cryptography.hazmat.primitives import hashes
+    from cryptography.hazmat.primitives.asymmetric import rsa
+    from cryptography.x509.oid import NameOID
+    from OpenSSL import crypto
+except ImportError:
+    crypto = None
+
+from cassandra.connection import Connection, DefaultEndPoint, _default_pyopenssl_ssl_method
 
 try:
     from twisted.test import proto_helpers
@@ -29,6 +42,131 @@
 
 from tests.unit.io.utils import TimerTestMixin
 
+CA_CERTS = os.path.abspath(os.path.join(
+    os.path.dirname(__file__), '..', '..', 'integration', 'long', 'ssl', 'rootCa.crt'))
+
+
+def _make_certificate(common_name, san_dns_names=None):
+    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    subject = issuer = x509.Name([
+        x509.NameAttribute(NameOID.COMMON_NAME, common_name),
+    ])
+    now = datetime.now(timezone.utc)
+    builder = (x509.CertificateBuilder()
+               .subject_name(subject)
+               .issuer_name(issuer)
+               .public_key(key.public_key())
+               .serial_number(x509.random_serial_number())
+               .not_valid_before(now - timedelta(days=1))
+               .not_valid_after(now + timedelta(days=1)))
+    if san_dns_names:
+        builder = builder.add_extension(
+            x509.SubjectAlternativeName([x509.DNSName(name) for name in san_dns_names]),
+            critical=False)
+
+    return crypto.X509.from_cryptography(builder.sign(key, hashes.SHA256()))
+
+
+@unittest.skipIf(TwistedConnection is None, "Twisted libraries are not available")
+@unittest.skipIf(not getattr(twistedreactor, '_HAS_SSL', False), "pyOpenSSL is not available")
+class TwistedSSLContextTest(unittest.TestCase):
+
+    def test_empty_ssl_options_default_to_negotiating_tls(self):
+        with patch.object(twistedreactor.SSL, 'Context') as context_mock:
+            context = twistedreactor._build_pyopenssl_context_from_options({}, twistedreactor.SSL)
+
+        context_mock.assert_called_once_with(twistedreactor.SSL.TLS_CLIENT_METHOD)
+        assert context is context_mock.return_value
+
+    def test_default_ssl_method_falls_back_to_tls_method(self):
+        tls_method = object()
+
+        with patch.object(twistedreactor, 'SSL', SimpleNamespace(TLS_METHOD=tls_method)):
+            assert _default_pyopenssl_ssl_method(twistedreactor.SSL) is tls_method
+
+    def test_default_ssl_method_falls_back_to_tlsv1_2_method(self):
+        tlsv1_2_method = object()
+
+        with patch.object(twistedreactor, 'SSL', SimpleNamespace(TLSv1_2_METHOD=tlsv1_2_method)):
+            assert _default_pyopenssl_ssl_method(twistedreactor.SSL) is tlsv1_2_method
+
+    def test_ssl_version_option_is_preserved(self):
+        with patch.object(twistedreactor.SSL, 'Context') as context_mock:
+            twistedreactor._build_pyopenssl_context_from_options(
+                {'ssl_version': twistedreactor.SSL.TLSv1_2_METHOD}, twistedreactor.SSL)
+
+        context_mock.assert_called_once_with(twistedreactor.SSL.TLSv1_2_METHOD)
+
+    def test_stdlib_ssl_version_option_is_translated(self):
+        with patch.object(twistedreactor.SSL, 'Context') as context_mock:
+            twistedreactor._build_pyopenssl_context_from_options(
+                {'ssl_version': ssl.PROTOCOL_TLS}, twistedreactor.SSL)
+
+        context_mock.assert_called_once_with(twistedreactor.SSL.TLS_CLIENT_METHOD)
+
+    def test_ca_certs_default_to_required_validation(self):
+        context = twistedreactor._build_pyopenssl_context_from_options({'ca_certs': CA_CERTS}, twistedreactor.SSL)
+
+        assert context.get_verify_mode() == twistedreactor.SSL.VERIFY_PEER
+
+    def test_check_hostname_option_enables_hostname_validation(self):
+        conn = TwistedConnection.__new__(TwistedConnection)
+
+        Connection.__init__(conn, DefaultEndPoint('1.2.3.4'), ssl_options={'check_hostname': True})
+
+        assert conn._check_hostname
+
+    def test_hostname_verification_uses_server_hostname(self):
+        context = Mock()
+        creator = twistedreactor._SSLCreator(DefaultEndPoint('proxy.host'), context,
+                                             {'server_hostname': 'sni.host'}, True, None)
+        connection = Mock()
+        connection.get_peer_certificate.return_value = _make_certificate('proxy.host', ['sni.host'])
+
+        creator.info_callback(connection, twistedreactor.SSL.SSL_CB_HANDSHAKE_DONE, None)
+
+        connection.get_app_data.assert_not_called()
+
+    def test_hostname_verification_prefers_san_over_common_name(self):
+        context = Mock()
+        creator = twistedreactor._SSLCreator(DefaultEndPoint('sni.host'), context, {}, True, None)
+        connection = Mock()
+        transport = Mock()
+        connection.get_app_data.return_value = transport
+        connection.get_peer_certificate.return_value = _make_certificate('sni.host', ['other.host'])
+
+        creator.info_callback(connection, twistedreactor.SSL.SSL_CB_HANDSHAKE_DONE, None)
+
+        transport.failVerification.assert_called_once()
+
+    def test_hostname_verification_matches_wildcard_san(self):
+        context = Mock()
+        creator = twistedreactor._SSLCreator(DefaultEndPoint('node.example.com'), context, {}, True, None)
+        connection = Mock()
+        connection.get_peer_certificate.return_value = _make_certificate('other.host', ['*.example.com'])
+
+        creator.info_callback(connection, twistedreactor.SSL.SSL_CB_HANDSHAKE_DONE, None)
+
+        connection.get_app_data.assert_not_called()
+
+    def test_hostname_callback_is_set_on_each_connection(self):
+        context = Mock()
+        first_connection = Mock()
+        second_connection = Mock()
+        first_creator = twistedreactor._SSLCreator(DefaultEndPoint('first.host'), context, {}, True, None)
+        second_creator = twistedreactor._SSLCreator(DefaultEndPoint('second.host'), context, {}, True, None)
+
+        with patch.object(twistedreactor.SSL, 'Connection', side_effect=[first_connection, second_connection]):
+            assert first_creator.clientConnectionForTLS(Mock()) is first_connection
+            assert second_creator.clientConnectionForTLS(Mock()) is second_connection
+
+        context.set_info_callback.assert_not_called()
+        first_callback = first_connection.set_info_callback.call_args[0][0]
+        second_callback = second_connection.set_info_callback.call_args[0][0]
+        assert first_callback.__self__ is first_creator
+        assert second_callback.__self__ is second_creator
+
+
 class TestTwistedTimer(TimerTestMixin, unittest.TestCase):
     """
     Simple test class that is used to validate that the TimerManager, and timer
@@ -196,3 +334,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()
diff --git a/tests/unit/test_client_routes.py b/tests/unit/test_client_routes.py
index 0aa82fc76a..aacd889bd8 100644
--- a/tests/unit/test_client_routes.py
+++ b/tests/unit/test_client_routes.py
@@ -449,6 +449,21 @@ def test_check_hostname_with_ssl_options_raises(self):
             )
         self.assertIn("check_hostname", str(cm.exception))
 
+    def test_empty_ssl_options_enable_tls_routes(self):
+        config = ClientRoutesConfig(
+            proxies=[ClientRouteProxy(str(uuid.uuid4()), "10.0.0.1")]
+        )
+        with self.assertWarns(DeprecationWarning):
+            cluster = Cluster(
+                contact_points=["10.0.0.1"],
+                ssl_options={},
+                client_routes_config=config,
+            )
+        try:
+            self.assertTrue(cluster._client_routes_handler.ssl_enabled)
+        finally:
+            cluster.shutdown()
+
     def test_disabled_check_hostname_with_client_routes_ok(self):
         """Cluster should allow check_hostname=False with client_routes_config."""
         ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
@@ -465,6 +480,21 @@ def test_disabled_check_hostname_with_client_routes_ok(self):
         )
         cluster.shutdown()
 
+    def test_pyopenssl_ssl_context_with_client_routes_ok(self):
+        """Cluster should allow pyOpenSSL-style contexts without check_hostname."""
+        ssl_ctx = Mock()
+        del ssl_ctx.check_hostname
+
+        config = ClientRoutesConfig(
+            proxies=[ClientRouteProxy(str(uuid.uuid4()), "10.0.0.1")]
+        )
+        cluster = Cluster(
+            contact_points=["10.0.0.1"],
+            ssl_context=ssl_ctx,
+            client_routes_config=config,
+        )
+        cluster.shutdown()
+
     def test_no_ssl_with_client_routes_ok(self):
         """Cluster should allow client_routes_config without SSL."""
         config = ClientRoutesConfig(
diff --git a/tests/unit/test_cloud.py b/tests/unit/test_cloud.py
new file mode 100644
index 0000000000..b90df3e27b
--- /dev/null
+++ b/tests/unit/test_cloud.py
@@ -0,0 +1,109 @@
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from cassandra.datastax import cloud
+from cassandra.connection import _default_pyopenssl_ssl_method
+
+
+def test_default_pyopenssl_ssl_method_prefers_tls_client_method():
+    tls_client_method = object()
+
+    ssl_module = SimpleNamespace(
+        TLS_CLIENT_METHOD=tls_client_method,
+        TLS_METHOD=object(),
+        TLSv1_2_METHOD=object())
+
+    assert _default_pyopenssl_ssl_method(ssl_module) is tls_client_method
+
+
+def test_default_pyopenssl_ssl_method_falls_back_to_tls_method():
+    tls_method = object()
+
+    ssl_module = SimpleNamespace(
+        TLS_METHOD=tls_method,
+        TLSv1_2_METHOD=object())
+
+    assert _default_pyopenssl_ssl_method(ssl_module) is tls_method
+
+
+def test_default_pyopenssl_ssl_method_falls_back_to_tlsv1_2_method():
+    tlsv1_2_method = object()
+
+    ssl_module = SimpleNamespace(TLSv1_2_METHOD=tlsv1_2_method)
+
+    assert _default_pyopenssl_ssl_method(ssl_module) is tlsv1_2_method
+
+
+def test_default_pyopenssl_ssl_method_requires_secure_method():
+    with pytest.raises(ImportError, match="secure TLS client method"):
+        _default_pyopenssl_ssl_method(SimpleNamespace())
+
+
+def test_pyopenssl_context_from_cert_uses_shared_options_builder():
+    ssl_module = SimpleNamespace(VERIFY_PEER=object())
+    openssl_module = SimpleNamespace(SSL=ssl_module)
+
+    with patch.dict('sys.modules', {'OpenSSL': openssl_module}):
+        with patch.object(cloud, '_build_pyopenssl_context_from_options') as build_context:
+            context = cloud._pyopenssl_context_from_cert('ca.crt', 'cert', 'key')
+
+    assert context is build_context.return_value
+    build_context.assert_called_once_with(
+        {
+            'ca_certs': 'ca.crt',
+            'certfile': 'cert',
+            'keyfile': 'key',
+            'cert_reqs': ssl_module.VERIFY_PEER,
+        },
+        ssl_module
+    )
+
+
+def test_parse_cloud_config_builds_pyopenssl_context_with_custom_ssl_context(tmp_path):
+    path = tmp_path / 'config.json'
+    path.write_text(json.dumps({'host': 'metadata.host', 'port': 443}))
+    ssl_context = object()
+    pyopenssl_context = object()
+
+    with patch.object(cloud, '_pyopenssl_context_from_cert', return_value=pyopenssl_context) as build_context:
+        config = cloud.parse_cloud_config(str(path), {'ssl_context': ssl_context}, create_pyopenssl_context=True)
+
+    assert config.ssl_context is ssl_context
+    assert config.pyopenssl_context is pyopenssl_context
+    build_context.assert_called_once_with(
+        str(tmp_path / 'ca.crt'),
+        str(tmp_path / 'cert'),
+        str(tmp_path / 'key'))
+
+
+def test_get_cloud_config_uses_pyopenssl_context_when_requested():
+    ssl_context = object()
+    pyopenssl_context = object()
+    config = cloud.CloudConfig()
+    config.ssl_context = ssl_context
+    config.pyopenssl_context = pyopenssl_context
+
+    with patch.object(cloud, 'read_cloud_config_from_zip', return_value=config):
+        with patch.object(cloud, 'read_metadata_info', return_value=config):
+            result = cloud.get_cloud_config(
+                {'secure_connect_bundle': 'bundle.zip'},
+                create_pyopenssl_context=True)
+
+    assert result.ssl_context is pyopenssl_context
diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py
index 3d55bc1860..0e1fb5a019 100644
--- a/tests/unit/test_cluster.py
+++ b/tests/unit/test_cluster.py
@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import unittest
+import warnings
 
 from concurrent.futures import Future
 import logging
@@ -278,6 +279,25 @@ def test_connection_factory_passes_compression_kwarg(self):
                 assert factory.call_args.kwargs['compression'] == expected
                 assert cluster.compression == expected
 
+    def test_empty_ssl_options_are_rejected_with_cloud_config(self):
+        with pytest.raises(ValueError) as exc:
+            Cluster(cloud={'secure_connect_bundle': 'unused'}, ssl_options={})
+
+        assert "cannot be specified with a cloud configuration" in str(exc.value)
+
+    def test_empty_ssl_options_emit_deprecation_warning(self):
+        with pytest.warns(DeprecationWarning, match="Using ssl_options without ssl_context"):
+            cluster = Cluster(ssl_options={})
+
+        assert cluster.ssl_options == {}
+
+    def test_omitted_ssl_options_do_not_emit_deprecation_warning(self):
+        with warnings.catch_warnings(record=True) as recorded_warnings:
+            warnings.simplefilter("always")
+            Cluster()
+
+        assert not [warning for warning in recorded_warnings if warning.category is DeprecationWarning]
+
 
 class SchedulerTest(unittest.TestCase):
     # TODO: this suite could be expanded; for now just adding a test covering a ticket
diff --git a/tests/unit/test_connection.py b/tests/unit/test_connection.py
index 1f9a3f682c..50b4382cb1 100644
--- a/tests/unit/test_connection.py
+++ b/tests/unit/test_connection.py
@@ -12,8 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import itertools
+import ssl
 import unittest
 from io import BytesIO
+from types import SimpleNamespace
 import time
 from threading import Lock
 from unittest.mock import Mock, ANY, call, patch
@@ -22,7 +24,8 @@
 from cassandra.cluster import Cluster
 from cassandra.connection import (Connection, HEADER_DIRECTION_TO_CLIENT, ProtocolError,
                                   locally_supported_compressions, ConnectionHeartbeat, HeartbeatFuture, _Frame, Timer, TimerManager,
-                                  ConnectionException, ConnectionShutdown, DefaultEndPoint, ShardAwarePortGenerator)
+                                  ConnectionException, ConnectionShutdown, DefaultEndPoint, ShardAwarePortGenerator,
+                                  _build_pyopenssl_context_from_options)
 from cassandra.marshal import uint8_pack, uint32_pack, int32_pack
 from cassandra.protocol import (write_stringmultimap, write_int, write_string,
                                 SupportedMessage, ProtocolHandler, ResultMessage,
@@ -81,6 +84,112 @@ def test_connection_endpoint(self):
         assert c.endpoint == endpoint
         assert c.endpoint.address == endpoint.address
 
+    def test_omitted_ssl_options_do_not_enable_ssl(self):
+        c = Connection(DefaultEndPoint('1.2.3.4'))
+
+        assert c.ssl_context is None
+        assert not c._ssl_enabled
+
+    def test_empty_ssl_options_enable_ssl(self):
+        c = Connection(DefaultEndPoint('1.2.3.4'), ssl_options={})
+
+        assert isinstance(c.ssl_context, ssl.SSLContext)
+        assert c.ssl_context.verify_mode == ssl.CERT_NONE
+        assert c.ssl_options == {}
+        assert c._ssl_enabled
+
+    def test_ssl_options_cert_reqs_applied_to_context(self):
+        c = Connection(DefaultEndPoint('1.2.3.4'), ssl_options={'cert_reqs': ssl.CERT_REQUIRED})
+
+        assert isinstance(c.ssl_context, ssl.SSLContext)
+        assert c.ssl_context.verify_mode == ssl.CERT_REQUIRED
+
+    def test_ssl_options_check_hostname_requires_validation(self):
+        c = Connection(DefaultEndPoint('1.2.3.4'), ssl_options={'check_hostname': True})
+
+        assert isinstance(c.ssl_context, ssl.SSLContext)
+        assert c.ssl_context.verify_mode == ssl.CERT_REQUIRED
+        assert c.ssl_context.check_hostname
+        assert c._check_hostname
+
+    def make_pyopenssl_module(self, **methods):
+        context = Mock()
+        module_attrs = {
+            'Context': Mock(return_value=context),
+            'VERIFY_NONE': object(),
+            'VERIFY_PEER': object(),
+        }
+        module_attrs.update(methods)
+        return SimpleNamespace(**module_attrs)
+
+    def test_pyopenssl_context_translates_stdlib_protocol_tls(self):
+        tls_client_method = object()
+        ssl_module = self.make_pyopenssl_module(TLS_CLIENT_METHOD=tls_client_method)
+
+        _build_pyopenssl_context_from_options({'ssl_version': ssl.PROTOCOL_TLS}, ssl_module)
+
+        ssl_module.Context.assert_called_once_with(tls_client_method)
+
+    def test_pyopenssl_context_translates_stdlib_tlsv1_2_protocol(self):
+        tlsv1_2_method = object()
+        ssl_module = self.make_pyopenssl_module(TLSv1_2_METHOD=tlsv1_2_method)
+
+        _build_pyopenssl_context_from_options({'ssl_version': ssl.PROTOCOL_TLSv1_2}, ssl_module)
+
+        ssl_module.Context.assert_called_once_with(tlsv1_2_method)
+
+    def test_pyopenssl_context_preserves_pyopenssl_method(self):
+        pyopenssl_method = object()
+        ssl_module = self.make_pyopenssl_module(TLS_CLIENT_METHOD=object())
+
+        _build_pyopenssl_context_from_options({'ssl_version': pyopenssl_method}, ssl_module)
+
+        ssl_module.Context.assert_called_once_with(pyopenssl_method)
+
+    def test_pyopenssl_context_applies_ciphers(self):
+        ssl_module = self.make_pyopenssl_module(TLS_CLIENT_METHOD=object())
+
+        _build_pyopenssl_context_from_options({'ciphers': 'HIGH:!aNULL'}, ssl_module)
+
+        ssl_module.Context.return_value.set_cipher_list.assert_called_once_with('HIGH:!aNULL')
+
+    def test_pyopenssl_context_loads_cert_chain_and_keyfile(self):
+        ssl_module = self.make_pyopenssl_module(TLS_CLIENT_METHOD=object())
+
+        _build_pyopenssl_context_from_options(
+            {'certfile': 'client.pem', 'keyfile': 'client.key'}, ssl_module)
+
+        context = ssl_module.Context.return_value
+        context.use_certificate_chain_file.assert_called_once_with('client.pem')
+        context.use_privatekey_file.assert_called_once_with('client.key')
+        context.check_privatekey.assert_called_once_with()
+
+    def test_pyopenssl_context_loads_private_key_from_certfile_by_default(self):
+        ssl_module = self.make_pyopenssl_module(TLS_CLIENT_METHOD=object())
+
+        _build_pyopenssl_context_from_options({'certfile': 'client-combined.pem'}, ssl_module)
+
+        context = ssl_module.Context.return_value
+        context.use_certificate_chain_file.assert_called_once_with('client-combined.pem')
+        context.use_privatekey_file.assert_called_once_with('client-combined.pem')
+        context.check_privatekey.assert_called_once_with()
+
+    def test_pyopenssl_context_translates_stdlib_cert_reqs(self):
+        verify_peer = object()
+        ssl_module = self.make_pyopenssl_module(TLS_CLIENT_METHOD=object(), VERIFY_PEER=verify_peer)
+
+        _build_pyopenssl_context_from_options({'cert_reqs': ssl.CERT_REQUIRED}, ssl_module)
+
+        ssl_module.Context.return_value.set_verify.assert_called_once_with(verify_peer, callback=ANY)
+
+    def test_pyopenssl_context_preserves_pyopenssl_verify_mode(self):
+        verify_mode = object()
+        ssl_module = self.make_pyopenssl_module(TLS_CLIENT_METHOD=object())
+
+        _build_pyopenssl_context_from_options({'cert_reqs': verify_mode}, ssl_module)
+
+        ssl_module.Context.return_value.set_verify.assert_called_once_with(verify_mode, callback=ANY)
+
     def test_bad_protocol_version(self, *args):
         c = self.make_connection()
         c._requests = Mock()
diff --git a/tests/unit/test_shard_aware.py b/tests/unit/test_shard_aware.py
index 902b48a276..bbf776cbf4 100644
--- a/tests/unit/test_shard_aware.py
+++ b/tests/unit/test_shard_aware.py
@@ -73,6 +73,15 @@ def mock_connection_factory(self, *args, **kwargs):
 
 
 class TestShardAware(unittest.TestCase):
+    def make_host(self, shard_aware_port=19042, shard_aware_port_ssl=19045):
+        host = MagicMock()
+        host.endpoint = DefaultEndPoint("1.2.3.4")
+        host.sharding_info = ShardingInfo(shard_id=1, shards_count=4, partitioner="",
+                                          sharding_algorithm="", sharding_ignore_msb=0,
+                                          shard_aware_port=shard_aware_port,
+                                          shard_aware_port_ssl=shard_aware_port_ssl)
+        return host
+
     def test_parsing_and_calculating_shard_id(self):
         """
         Testing the parsing of the options command
@@ -100,8 +109,7 @@ def test_advanced_shard_aware_port(self):
         Test that on given a `shard_aware_port` on the OPTIONS message (ShardInfo class)
         the next connections would be open using this port
         """
-        host = MagicMock()
-        host.endpoint = DefaultEndPoint("1.2.3.4")
+        host = self.make_host()
 
         for port, ssl_options, ssl_context in [
                 (19042, None, None),
@@ -128,8 +136,7 @@ def test_ssl_advanced_shard_aware_port_requires_ssl_port(self):
         Test that SSL connections do not fall back to the plaintext
         shard-aware port when the SSL shard-aware port is unavailable.
         """
-        host = MagicMock()
-        host.endpoint = DefaultEndPoint("1.2.3.4")
+        host = self.make_host(shard_aware_port=19042, shard_aware_port_ssl=None)
         sharding_info = ShardingInfo(
             shard_id=1, shards_count=4, partitioner="", sharding_algorithm="",
             sharding_ignore_msb=0, shard_aware_port=19042,
@@ -160,8 +167,7 @@ def test_advanced_shard_aware_cooldown(self):
         the deadline has passed. The hard-disable flag must suppress the endpoint
         unconditionally.
         """
-        host = MagicMock()
-        host.endpoint = DefaultEndPoint("1.2.3.4")
+        host = self.make_host()
         session = MockSession()
 
         pool = HostConnection(host=host, host_distance=HostDistance.REMOTE, session=session)