From e92678e9ec5978145a3f2dd489ae7adbb457b4f3 Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Wed, 29 Jul 2026 00:28:13 -0400 Subject: [PATCH] connection: fix ssl_options TLS handling Preserve the distinction between omitted and explicitly empty ssl_options across connection setup, reactors, shard-aware routing, cloud contexts, and Insights reporting. Strengthen pyOpenSSL protocol, verification, SNI, and hostname handling while retaining legacy option behavior. --- CHANGELOG.rst | 28 ++ cassandra/client_routes.py | 11 +- cassandra/cluster.py | 70 ++- cassandra/connection.py | 142 +++--- cassandra/datastax/cloud/__init__.py | 21 +- cassandra/datastax/insights/reporter.py | 79 ++-- cassandra/io/asyncioreactor.py | 2 +- cassandra/io/eventletreactor.py | 93 +++- cassandra/io/twistedreactor.py | 116 +++-- cassandra/pool.py | 23 +- cassandra/tls.py | 579 ++++++++++++++++++++++++ docs/security.rst | 63 ++- scripts/validate_astra_tls.py | 91 ++++ tests/unit/advanced/test_insights.py | 207 ++++++++- tests/unit/io/fixtures/root_ca.pem | 19 + tests/unit/io/test_asyncioreactor.py | 51 +++ tests/unit/io/test_eventletreactor.py | 339 +++++++++++++- tests/unit/io/test_twistedreactor.py | 308 ++++++++++++- tests/unit/io/utils.py | 45 ++ tests/unit/test_client_routes.py | 69 ++- tests/unit/test_cloud.py | 44 ++ tests/unit/test_cluster.py | 47 ++ tests/unit/test_connection.py | 208 ++++++++- tests/unit/test_shard_aware.py | 58 ++- tests/unit/test_tls.py | 568 +++++++++++++++++++++++ 25 files changed, 3074 insertions(+), 207 deletions(-) create mode 100644 cassandra/tls.py create mode 100644 scripts/validate_astra_tls.py create mode 100644 tests/unit/io/fixtures/root_ca.pem create mode 100644 tests/unit/test_cloud.py create mode 100644 tests/unit/test_tls.py diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 2a02f1ac54..05797ed122 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -7,6 +7,34 @@ Features statements skip re-sending result metadata on EXECUTE, and the driver automatically refreshes cached metadata when the server detects a schema change (DRIVER-153) +Bug Fixes +--------- +* Explicit ``ssl_options={}`` and endpoint SSL options now enable TLS instead + of being treated as omitted SSL options. Empty options without additional + endpoint security settings encrypt the connection without verifying the + server certificate. Cluster-level ``ssl_options=None`` does not enable TLS by + itself; an ``ssl_context`` or endpoint SSL options still enable it. An + endpoint ``server_hostname`` used only for SNI routing is not treated as an + additional security setting for explicit empty cluster options. +* Legacy ``ssl_options`` now work with the Twisted and Eventlet reactors by + using pyOpenSSL contexts with mapped protocol, verification, cipher, SNI, and + hostname-validation settings. Nonempty options without ``cert_reqs`` now + default to peer-certificate verification on these reactors instead of + pyOpenSSL's ``VERIFY_NONE``. When ``ca_certs`` is omitted, the system trust + roots are loaded. Serialized symbolic stdlib protocol names are supported; + plain integer protocol values are mapped only when unambiguous and otherwise + rejected rather than being silently interpreted as a different pyOpenSSL + method. Existing configurations that connect to an untrusted certificate + must supply the appropriate CA or explicitly opt out with + ``cert_reqs=ssl.CERT_NONE`` and leave ``check_hostname`` disabled. +* Hostname verification is now performed consistently when ``check_hostname`` + is enabled. The flag was previously not propagated to some connection paths, + so hostname mismatches could be silently accepted; such mismatches now fail + the connection. For a supplied standard-library context, enabling hostname + checks also enables peer-certificate verification when necessary. +* ``SSLContext`` setup now applies ``cert_reqs`` through ``verify_mode`` instead + of overwriting the SSL options bitmask, preserving default hardening flags. + Others ------ * ``PreparedStatement.result_metadata`` and ``PreparedStatement.result_metadata_id`` are diff --git a/cassandra/client_routes.py b/cassandra/client_routes.py index 80b2477a6d..f4729a5566 100644 --- a/cassandra/client_routes.py +++ b/cassandra/client_routes.py @@ -213,20 +213,29 @@ class _ClientRoutesHandler: config: 'ClientRoutesConfig' ssl_enabled: bool + endpoint_ssl_options: Optional[Dict] _routes: _RouteStore _connection_ids: Set[str] _proxy_addresses_override: Dict[str, str] - def __init__(self, config: 'ClientRoutesConfig', ssl_enabled: bool = False): + def __init__(self, config: 'ClientRoutesConfig', ssl_enabled: bool = False, + endpoint_ssl_options: Optional[Dict] = None): """ :param config: ClientRoutesConfig instance :param ssl_enabled: Whether TLS is enabled (determines port selection) + :param endpoint_ssl_options: SSL options inherited from endpoint + contact points for generated client-routes endpoints """ if not isinstance(config, ClientRoutesConfig): raise TypeError("config must be a ClientRoutesConfig instance") self.config = config self.ssl_enabled = ssl_enabled + self.endpoint_ssl_options = ( + dict(endpoint_ssl_options) + if endpoint_ssl_options is not None + else None + ) self._routes = _RouteStore() self._connection_ids = {dep.connection_id for dep in config.proxies} # Precalculate proxy address mappings for efficient lookup diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 88c8d2707a..65f934616b 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -37,6 +37,7 @@ import re import queue import socket +import ssl import sys import time from threading import Lock, RLock, Thread, Event @@ -883,12 +884,29 @@ def default_retry_policy(self, policy): when new sockets are created. This should be used when client encryption is enabled in Cassandra. - The following documentation only applies when ssl_options is used without ssl_context. - - By default, a ``ca_certs`` value should be supplied (the value should be - a string pointing to the location of the CA certs file), and you probably - want to specify ``ssl_version`` as ``ssl.PROTOCOL_TLS`` to match - Cassandra's default protocol. + The following documentation only applies when cluster-level ``ssl_options`` + is used without ``ssl_context``. An :class:`~cassandra.connection.EndPoint` + may independently supply SSL options for its connections. + + .. versionchanged:: 3.29.12 + + An explicit empty dict (``ssl_options={}``) enables TLS with default + options. Cluster-level ``ssl_options=None`` does not enable TLS by + itself; an ``ssl_context`` or SSL options supplied by the selected + :class:`~cassandra.connection.EndPoint` still enable it. Empty options + do not verify the server certificate unless the endpoint supplies + additional security settings. A nonempty options dict enables + peer-certificate verification by default unless ``cert_reqs`` explicitly + disables it. Setting + ``'check_hostname': True`` always enables peer-certificate verification + because hostname checking requires an authenticated certificate chain. + When verification is enabled and ``ca_certs`` is omitted, the system + trust roots are loaded. + + Supply ``ca_certs`` as a path to a CA certificate file when the server uses + a private CA that is not in the system trust store. You probably also want + to specify ``ssl_version`` as ``ssl.PROTOCOL_TLS`` to match Cassandra's + default protocol. .. versionchanged:: 3.3.0 @@ -1298,7 +1316,8 @@ def __init__(self, if cloud is not None: self.cloud = cloud - if contact_points is not _NOT_SET or endpoint_factory or ssl_context or ssl_options: + if (contact_points is not _NOT_SET or endpoint_factory or + ssl_context is not None or ssl_options is not None): raise ValueError("contact_points, endpoint_factory, ssl_context, and ssl_options " "cannot be specified with a cloud configuration") @@ -1345,6 +1364,13 @@ def __init__(self, if not isinstance(client_routes_config, ClientRoutesConfig): raise TypeError("client_routes_config must be a ClientRoutesConfig instance") + endpoint_ssl_options = next(( + cp.ssl_options + for cp in self.contact_points + if (isinstance(cp, EndPoint) and + cp.ssl_options is not None) + ), None) + # SSL hostname verification is incompatible with client routes: # connections go through NLB proxies whose addresses won't match # server certificates. @@ -1353,6 +1379,9 @@ def __init__(self, _check_hostname_enabled = True if ssl_options is not None and ssl_options.get('check_hostname', False): _check_hostname_enabled = True + if (endpoint_ssl_options is not None and + endpoint_ssl_options.get('check_hostname', False)): + _check_hostname_enabled = True if _check_hostname_enabled: raise ValueError( "SSL hostname verification (check_hostname=True) is currently incompatible " @@ -1362,8 +1391,16 @@ def __init__(self, "ssl_context.check_hostname = False." ) - ssl_enabled = ssl_context is not None or ssl_options is not None - self._client_routes_handler = _ClientRoutesHandler(client_routes_config, ssl_enabled=ssl_enabled) + ssl_enabled = ( + ssl_context is not None or + ssl_options is not None or + endpoint_ssl_options is not None + ) + self._client_routes_handler = _ClientRoutesHandler( + client_routes_config, + ssl_enabled=ssl_enabled, + endpoint_ssl_options=endpoint_ssl_options, + ) if contact_points is _NOT_SET or not self._contact_points_explicit: seed_addrs = [dep.connection_addr_override for dep in client_routes_config.proxies @@ -1508,12 +1545,25 @@ def __init__(self, self.metrics_enabled = metrics_enabled - if ssl_options and not ssl_context: + if ssl_options is not None and ssl_context is None: warn('Using ssl_options without ssl_context is ' 'deprecated and will result in an error in ' 'the next major release. Please use ssl_context ' 'to prepare for that release.', DeprecationWarning) + if not ssl_options: + log.warning( + 'Cluster-level ssl_options enable TLS without requesting ' + 'server certificate verification; endpoint-level SSL ' + 'options may enable verification for individual ' + 'connections') + elif (ssl_options.get('check_hostname', False) and + ssl_options.get('cert_reqs') == ssl.CERT_NONE): + log.warning( + 'Cluster-level check_hostname=True requires server ' + 'certificate verification and overrides ' + 'cert_reqs=CERT_NONE at cluster level; endpoint-level SSL ' + 'options may override these settings') self.ssl_options = ssl_options self.ssl_context = ssl_context diff --git a/cassandra/connection.py b/cassandra/connection.py index f238416b29..8c5d2cf582 100644 --- a/cassandra/connection.py +++ b/cassandra/connection.py @@ -20,11 +20,11 @@ import io import logging import socket +import ssl import struct import sys from threading import Thread, Event, RLock, Condition import time -import ssl import uuid import weakref import random @@ -50,6 +50,10 @@ AuthSuccessMessage, ProtocolException, RegisterMessage, ReviseRequestMessage) from cassandra.segment import SegmentCodec, CrcException +from cassandra.tls import (_build_ssl_context_from_options, + _prepare_ssl_options, + _ssl_options_requiring_new_context, + _wrap_socket_from_context) from cassandra.util import OrderedDict from cassandra.shard_info import ShardingInfo # noqa: F401 # re-exported for cassandra.connection.ShardingInfo @@ -455,6 +459,10 @@ def port(self) -> Optional[int]: def host_id(self) -> uuid.UUID: return self._host_id + @property + def ssl_options(self) -> Optional[Dict]: + return self._handler.endpoint_ssl_options + def resolve(self) -> Tuple[str, int]: """ Resolve endpoint by delegating to the handler. @@ -803,6 +811,8 @@ class Connection(object): endpoint = None ssl_options = None ssl_context = None + _ssl_options_explicit = False + _ssl_options_verify_by_default = None last_error = None # The current number of operations that are in flight. More precisely, @@ -858,6 +868,7 @@ class Connection(object): _socket = None _socket_impl = socket + _connect_socket_error_types = (socket.error,) _check_hostname = False _product_type = None @@ -883,9 +894,21 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None, on_orphaned_stream_released=None, application_info: Optional[ApplicationInfoBase] = None): # TODO next major rename host to endpoint and remove port kwarg. self.endpoint = host if isinstance(host, EndPoint) else DefaultEndPoint(host, port) + endpoint_ssl_options = self.endpoint.ssl_options + self._endpoint_ssl_options = ( + dict(endpoint_ssl_options) + if endpoint_ssl_options is not None + else None + ) self.authenticator = authenticator - self.ssl_options = ssl_options.copy() if ssl_options else {} + ( + self.ssl_options, + self._ssl_options_explicit, + self._ssl_options_verify_by_default, + ) = _prepare_ssl_options( + ssl_options, self._endpoint_ssl_options + ) self.ssl_context = ssl_context self.sockopts = sockopts self.compression = compression @@ -905,21 +928,39 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None, self._on_orphaned_stream_released = on_orphaned_stream_released self._application_info = application_info - if ssl_options: - self.ssl_options.update(self.endpoint.ssl_options or {}) - elif self.endpoint.ssl_options: - self.ssl_options = self.endpoint.ssl_options - - # PYTHON-1331 - # - # We always use SSLContext.wrap_socket() now but legacy configs may have other params that were passed to ssl.wrap_socket()... - # and either could have 'check_hostname'. Remove these params into a separate map and use them to build an SSLContext if - # we need to do so. - # - # Note the use of pop() here; we are very deliberately removing these params from ssl_options if they're present. After this - # operation ssl_options should contain only args needed for the ssl_context.wrap_socket() call. - if not self.ssl_context and self.ssl_options: + # An explicitly empty options mapping enables TLS with default options. + if self.ssl_context is None and self._ssl_options_explicit: self.ssl_context = self._build_ssl_context_from_options() + elif self.ssl_context is not None: + endpoint_context_options = _ssl_options_requiring_new_context( + self._endpoint_ssl_options) + if endpoint_context_options: + # SSLContext has no supported copy operation, and settings such + # as trust roots, client keys, and pyOpenSSL callbacks cannot be + # restored reliably. The cluster context is shared by concurrent + # connections, so mutating and restoring it would leak endpoint + # policy or race with another handshake. Require the existing + # context to remain immutable; callers needing endpoint-specific + # context settings can omit ssl_context and use ssl_options, + # which builds an independent context for every connection. + raise ValueError( + "Endpoint SSL options %s require an independent SSL " + "context and cannot be combined with a supplied " + "ssl_context" % sorted(endpoint_context_options)) + + if (self.ssl_options.get('check_hostname', False) and + isinstance(self.ssl_context, ssl.SSLContext) and + not self.ssl_context.check_hostname): + # ``check_hostname`` is a context setting, not a wrap_socket + # option. Apply the caller's request to a supplied stdlib context + # so the default and Asyncio reactors do not rely on their no-op + # ``_validate_hostname`` hook. + if self.ssl_context.verify_mode == ssl.CERT_NONE: + self.ssl_context.verify_mode = ssl.CERT_REQUIRED + self.ssl_context.check_hostname = True + + self._check_hostname = bool(self.ssl_options.get('check_hostname', False) or + getattr(self.ssl_context, 'check_hostname', False)) self.max_request_id = min(self.max_in_flight - 1, (2 ** 15) - 1) # Don't fill the deque with 2**15 items right away. Start with some and add @@ -942,6 +983,10 @@ def host(self): def port(self): return self.endpoint.port + @property + def _ssl_enabled(self): + return self.ssl_context is not None or self._ssl_options_explicit + @classmethod def initialize_reactor(cls): """ @@ -992,47 +1037,16 @@ def factory(cls, endpoint, timeout, host_conn = None, *args, **kwargs): return conn def _build_ssl_context_from_options(self): - - # Extract a subset of names from self.ssl_options which apply to SSLContext creation - ssl_context_opt_names = ['ssl_version', 'cert_reqs', 'check_hostname', 'keyfile', 'certfile', 'ca_certs', 'ciphers'] - opts = {k:self.ssl_options.get(k, None) for k in ssl_context_opt_names if k in self.ssl_options} - - # Python >= 3.10 requires either PROTOCOL_TLS_CLIENT or PROTOCOL_TLS_SERVER so we'll get ahead of things by always - # being explicit - ssl_version = opts.get('ssl_version', None) or ssl.PROTOCOL_TLS_CLIENT - cert_reqs = opts.get('cert_reqs', None) or ssl.CERT_REQUIRED - rv = ssl.SSLContext(protocol=int(ssl_version)) - rv.check_hostname = bool(opts.get('check_hostname', False)) - rv.options = int(cert_reqs) - - certfile = opts.get('certfile', None) - keyfile = opts.get('keyfile', None) - if certfile: - rv.load_cert_chain(certfile, keyfile) - ca_certs = opts.get('ca_certs', None) - if ca_certs: - rv.load_verify_locations(ca_certs) - ciphers = opts.get('ciphers', None) - if ciphers: - rv.set_ciphers(ciphers) - - return rv + return _build_ssl_context_from_options( + self.ssl_options, + verify_by_default=self._ssl_options_verify_by_default) def _wrap_socket_from_context(self): - - # Extract a subset of names from self.ssl_options which apply to SSLContext.wrap_socket (or at least the parts - # of it that don't involve building an SSLContext under the covers) - wrap_socket_opt_names = ['server_side', 'do_handshake_on_connect', 'suppress_ragged_eofs', 'server_hostname'] - opts = {k:self.ssl_options.get(k, None) for k in wrap_socket_opt_names if k in self.ssl_options} - - # PYTHON-1186: set the server_hostname only if the SSLContext has - # check_hostname enabled and it is not already provided by the EndPoint ssl options - #opts['server_hostname'] = self.endpoint.address - if (self.ssl_context.check_hostname and 'server_hostname' not in opts): - server_hostname = self.endpoint.address - opts['server_hostname'] = server_hostname - - return self.ssl_context.wrap_socket(self._socket, **opts) + return _wrap_socket_from_context( + self.ssl_context, + self._socket, + self.ssl_options, + self.endpoint.address) def _initiate_connection(self, sockaddr): if self.features.shard_id is not None: @@ -1089,13 +1103,27 @@ def _connect_socket(self): self._validate_hostname() sockerr = None break - except socket.error as err: + except self._connect_socket_error_types as err: + if self._socket: + self._socket.close() + self._socket = None + # Preserve a TLS or driver error over later socket failures so + # a final ECONNREFUSED cannot hide certificate diagnostics. + if (sockerr is None or + (isinstance(sockerr, socket.error) and + not isinstance(sockerr, ssl.SSLError))): + sockerr = err + except Exception: if self._socket: self._socket.close() self._socket = None - sockerr = err + raise if sockerr: + if isinstance(sockerr, ssl.SSLError): + raise sockerr + if not isinstance(sockerr, socket.error): + raise sockerr raise socket.error(sockerr.errno, "Tried connecting to %s. Last error: %s" % ([a[4] for a in addresses], sockerr.strerror or sockerr)) diff --git a/cassandra/datastax/cloud/__init__.py b/cassandra/datastax/cloud/__init__.py index be79d6db38..d69ae0ae69 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.tls import _build_pyopenssl_context_from_options log = logging.getLogger(__name__) @@ -181,12 +182,14 @@ def _pyopenssl_context_from_cert(ca_cert_location, cert_location, key_location): from OpenSSL import SSL except ImportError as e: raise ImportError( - "PyOpenSSL must be installed to connect to Astra with the Eventlet or Twisted event loops")\ - .with_traceback(e.__traceback__) - ssl_context = SSL.Context(SSL.TLSv1_METHOD) - ssl_context.set_verify(SSL.VERIFY_PEER, callback=lambda _1, _2, _3, _4, ok: ok) - ssl_context.use_certificate_file(cert_location) - ssl_context.use_privatekey_file(key_location) - ssl_context.load_verify_locations(ca_cert_location) - - return ssl_context \ No newline at end of file + "PyOpenSSL must be installed to connect to Astra with the Eventlet or Twisted event loops" + ) from e + return _build_pyopenssl_context_from_options( + SSL, + { + 'ca_certs': ca_cert_location, + 'certfile': cert_location, + 'keyfile': key_location, + 'cert_reqs': CERT_REQUIRED, + } + ) diff --git a/cassandra/datastax/insights/reporter.py b/cassandra/datastax/insights/reporter.py index 83205fc458..4c8c58eeaf 100644 --- a/cassandra/datastax/insights/reporter.py +++ b/cassandra/datastax/insights/reporter.py @@ -20,7 +20,6 @@ import random import platform import socket -import ssl import sys from threading import Event, Thread import time @@ -29,6 +28,8 @@ from cassandra.util import ms_timestamp_from_datetime from cassandra.datastax.insights.registry import insights_registry from cassandra.datastax.insights.serializers import initialize_registry +from cassandra.tls import (_ssl_context_cert_validation_enabled, + _ssl_options_cert_validation_enabled) log = logging.getLogger(__name__) @@ -81,6 +82,7 @@ def _send_via_rpc(self, data): def _get_status_data(self): cc = self._session.cluster.control_connection + connection = cc._connection connected_nodes = { host.address: { @@ -110,22 +112,24 @@ def _get_status_data(self): 'clientId': str(self._session.cluster.client_id), # // 'sessionId' must be the same as the one provided in the startup message 'sessionId': str(self._session.session_id), - 'controlConnection': cc._connection.host if cc._connection else None, + 'controlConnection': connection.host if connection else None, 'connectedNodes': connected_nodes } } def _get_startup_data(self): - cc = self._session.cluster.control_connection + cluster = self._session.cluster + cc = cluster.control_connection + connection = cc._connection try: - local_ipaddr = cc._connection._socket.getsockname()[0] + local_ipaddr = connection._socket.getsockname()[0] except Exception as e: local_ipaddr = None - log.debug('Unable to get local socket addr from {}: {}'.format(cc._connection, e)) + log.debug('Unable to get local socket addr from {}: {}'.format(connection, e)) hostname = socket.getfqdn() host_distances_counter = Counter( - self._session.cluster.profile_manager.distance(host) + cluster.profile_manager.distance(host) for host in self._session.hosts ) host_distances_dict = { @@ -135,20 +139,33 @@ def _get_startup_data(self): } try: - compression_type = cc._connection._compression_type + compression_type = connection._compression_type except AttributeError: compression_type = 'NONE' + if connection is not None: + ssl_enabled = connection._ssl_enabled + ssl_context = connection.ssl_context + ssl_options = None + else: + ssl_context = cluster.ssl_context + ssl_options = cluster.ssl_options + if ssl_options is None: + ssl_options = next(( + endpoint_options + for endpoint in getattr( + cluster, 'endpoints_resolved', ()) + for endpoint_options in (endpoint.ssl_options,) + if endpoint_options is not None + ), None) + ssl_enabled = ssl_context is not None or ssl_options is not None + cert_validation = None try: - if self._session.cluster.ssl_context: - 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 + if ssl_context is not None: + cert_validation = _ssl_context_cert_validation_enabled(ssl_context) + elif ssl_options is not None: + cert_validation = _ssl_options_cert_validation_enabled(ssl_options) except Exception as e: log.debug('Unable to get the cert validation: {}'.format(e)) @@ -167,31 +184,31 @@ def _get_startup_data(self): 'data': { 'driverName': 'DataStax Python Driver', 'driverVersion': sys.modules['cassandra'].__version__, - 'clientId': str(self._session.cluster.client_id), + 'clientId': str(cluster.client_id), 'sessionId': str(self._session.session_id), - 'applicationName': self._session.cluster.application_name or 'python', - 'applicationNameWasGenerated': not self._session.cluster.application_name, - 'applicationVersion': self._session.cluster.application_version, - 'contactPoints': self._session.cluster._endpoint_map_for_insights, - 'dataCenters': list(set(h.datacenter for h in self._session.cluster.metadata.all_hosts() + 'applicationName': cluster.application_name or 'python', + 'applicationNameWasGenerated': not cluster.application_name, + 'applicationVersion': cluster.application_version, + 'contactPoints': cluster._endpoint_map_for_insights, + 'dataCenters': list(set(h.datacenter for h in cluster.metadata.all_hosts() if (h.datacenter and - self._session.cluster.profile_manager.distance(h) == HostDistance.LOCAL))), - 'initialControlConnection': cc._connection.host if cc._connection else None, - 'protocolVersion': self._session.cluster.protocol_version, + cluster.profile_manager.distance(h) == HostDistance.LOCAL))), + 'initialControlConnection': connection.host if connection else None, + 'protocolVersion': cluster.protocol_version, 'localAddress': local_ipaddr, 'hostName': hostname, - 'executionProfiles': insights_registry.serialize(self._session.cluster.profile_manager), + 'executionProfiles': insights_registry.serialize(cluster.profile_manager), 'configuredConnectionLength': host_distances_dict, - 'heartbeatInterval': self._session.cluster.idle_heartbeat_interval, + 'heartbeatInterval': cluster.idle_heartbeat_interval, 'compression': compression_type.upper() if compression_type else 'NONE', - 'reconnectionPolicy': insights_registry.serialize(self._session.cluster.reconnection_policy), + 'reconnectionPolicy': insights_registry.serialize(cluster.reconnection_policy), 'sslConfigured': { - 'enabled': bool(self._session.cluster.ssl_options or self._session.cluster.ssl_context), + 'enabled': ssl_enabled, 'certValidation': cert_validation }, 'authProvider': { - 'type': (self._session.cluster.auth_provider.__class__.__name__ - if self._session.cluster.auth_provider else + 'type': (cluster.auth_provider.__class__.__name__ + if cluster.auth_provider else None) }, 'otherOptions': { @@ -208,7 +225,7 @@ def _get_startup_data(self): }, 'runtime': { 'python': sys.version, - 'event_loop': self._session.cluster.connection_class.__name__ + 'event_loop': cluster.connection_class.__name__ } }, 'periodicStatusInterval': self._interval 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..3d8a1c40d4 100644 --- a/cassandra/io/eventletreactor.py +++ b/cassandra/io/eventletreactor.py @@ -20,13 +20,27 @@ from eventlet.queue import Queue from greenlet import GreenletExit import logging +import ssl from threading import Event import time -from cassandra.connection import Connection, ConnectionShutdown, Timer, TimerManager +from cassandra.connection import (Connection, ConnectionException, + ConnectionShutdown, SniEndPoint, Timer, + TimerManager) +from cassandra.tls import ( + _build_pyopenssl_context_from_options as _build_pyopenssl_context, + _encode_server_hostname, + _ensure_pyopenssl_context_requires_verification, + _resolve_pyopenssl_server_names, + _validate_pyopenssl_hostname, +) +_PYOPENSSL_ERROR = () +_PYOPENSSL_ZERO_RETURN_ERROR = () try: from eventlet.green.OpenSSL import SSL _PYOPENSSL = True + _PYOPENSSL_ERROR = (SSL.Error,) + _PYOPENSSL_ZERO_RETURN_ERROR = (SSL.ZeroReturnError,) except ImportError as e: _PYOPENSSL = False no_pyopenssl_error = e @@ -55,6 +69,9 @@ class EventletConnection(Connection): _socket_impl = eventlet.green.socket _ssl_impl = eventlet.green.ssl + _connect_socket_error_types = ( + (socket.error, ConnectionException) + _PYOPENSSL_ERROR + ) _timers = None _timeout_watcher = None @@ -92,7 +109,6 @@ 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._write_queue = Queue() self._connect_socket() @@ -103,28 +119,41 @@ def __init__(self, *args, **kwargs): def _wrap_socket_from_context(self): _check_pyopenssl() + _ensure_pyopenssl_context_requires_verification(SSL, self.ssl_context, self._check_hostname) self._socket = SSL.Connection(self.ssl_context, self._socket) self._socket.set_connect_state() - if self.ssl_options and 'server_hostname' in self.ssl_options: + server_hostname, _expected_name = self._tls_server_names() + if server_hostname is not None: # This is necessary for SNI - self._socket.set_tlsext_host_name(self.ssl_options['server_hostname'].encode('ascii')) + self._socket.set_tlsext_host_name(_encode_server_hostname(server_hostname)) + return self._socket + + def _tls_server_names(self): + return _resolve_pyopenssl_server_names( + self.endpoint.address, + (self.ssl_options or {}).get('server_hostname'), + self._check_hostname, + verify_endpoint_address=isinstance(self.endpoint, SniEndPoint)) 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: - self._socket.do_handshake() - - def _match_hostname(self): - if self.uses_legacy_ssl_options: - super(EventletConnection, self)._match_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)) + super(EventletConnection, self)._initiate_connection(sockaddr) + if self._ssl_enabled: + self._socket.do_handshake() + + def _validate_hostname(self): + _server_hostname, expected_name = self._tls_server_names() + try: + _validate_pyopenssl_hostname(self._socket.get_peer_certificate(), expected_name) + except ssl.CertificateError as exc: + raise ConnectionException( + "Hostname verification failed: %s" % (exc,), self.endpoint) from exc + + def _build_ssl_context_from_options(self): + _check_pyopenssl() + return _build_pyopenssl_context( + SSL, + self.ssl_options, + verify_by_default=self._ssl_options_verify_by_default) def close(self): with self.lock: @@ -161,6 +190,14 @@ def handle_write(self): try: next_msg = self._write_queue.get() self._socket.sendall(next_msg) + except _PYOPENSSL_ZERO_RETURN_ERROR: + log.debug("Connection %s closed by server during socket send", self) + self.close() + return + except _PYOPENSSL_ERROR as err: + log.debug("Exception during TLS socket send for %s: %s", self, err) + self.defunct(err) + return except socket.error as err: log.debug("Exception during socket send for %s: %s", self, err) self.defunct(err) @@ -172,7 +209,19 @@ def handle_read(self): while True: try: buf = self._socket.recv(self.in_buffer_size) + if not buf: + log.debug("Connection %s closed by server", self) + self.close() + return self._iobuf.write(buf) + except _PYOPENSSL_ZERO_RETURN_ERROR: + log.debug("Connection %s closed by server during socket recv", self) + self.close() + return + except _PYOPENSSL_ERROR as err: + log.debug("Exception during TLS socket recv for %s: %s", self, err) + self.defunct(err) + return except socket.error as err: log.debug("Exception during socket recv for %s: %s", self, err) @@ -181,12 +230,8 @@ def handle_read(self): except GreenletExit: # graceful greenthread exit return - if buf and self._iobuf.tell(): + if self._iobuf.tell(): self.process_io_buffer() - else: - log.debug("Connection %s closed by server", self) - self.close() - return def push(self, data): chunk_size = self.out_buffer_size diff --git a/cassandra/io/twistedreactor.py b/cassandra/io/twistedreactor.py index 446200bf63..b4da9d7d69 100644 --- a/cassandra/io/twistedreactor.py +++ b/cassandra/io/twistedreactor.py @@ -28,7 +28,16 @@ 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, ConnectionException, + ConnectionShutdown, SniEndPoint, Timer, + TimerManager) +from cassandra.tls import ( + _build_pyopenssl_context_from_options as _build_pyopenssl_context, + _encode_server_hostname, + _ensure_pyopenssl_context_requires_verification, + _resolve_pyopenssl_server_names, + _validate_pyopenssl_hostname, +) try: from OpenSSL import SSL @@ -39,6 +48,16 @@ log = logging.getLogger(__name__) +_TLS_APP_DATA_ATTR = '_cassandra_tls_app_data' + + +class _TLSAppData(object): + def __init__(self, endpoint, expected_name, check_hostname): + self.endpoint = endpoint + self.expected_name = expected_name + self.check_hostname = check_hostname + + def _cleanup(cleanup_weakref): try: cleanup_weakref()._cleanup() @@ -139,43 +158,53 @@ def _on_loop_timer(self): @implementer(IOpenSSLClientConnectionCreator) class _SSLCreator(object): - def __init__(self, endpoint, ssl_context, ssl_options, check_hostname, timeout): + def __init__(self, endpoint, ssl_context, ssl_options, check_hostname): 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: - 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) - - def verify_callback(self, connection, x509, errnum, errdepth, ok): - return ok - - 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 ssl_context is None: + raise ValueError( + '_SSLCreator requires a prepared pyOpenSSL context') + self.context = ssl_context + _ensure_pyopenssl_context_requires_verification(SSL, self.context, self.check_hostname) + if self.check_hostname and not hasattr(SSL.Connection, 'set_info_callback'): + log.warning( + 'Installed pyOpenSSL does not support per-connection TLS info ' + 'callbacks; replacing any callback configured on the supplied ' + 'context') + self.context.set_info_callback(_SSLCreator.info_callback) + + @staticmethod + def info_callback(connection, where, ret): + if not where & SSL.SSL_CB_HANDSHAKE_DONE: + return + tls_protocol = connection.get_app_data() + app_data = getattr(tls_protocol, _TLS_APP_DATA_ATTR, None) + if not (app_data and app_data.check_hostname): + return + try: + _validate_pyopenssl_hostname( + connection.get_peer_certificate(), app_data.expected_name) + except Exception as exc: + tls_protocol.failVerification( + Failure(ConnectionException( + "Hostname verification failed: %s" % (exc,), app_data.endpoint))) def clientConnectionForTLS(self, tlsProtocol): connection = SSL.Connection(self.context, None) + if self.check_hostname and hasattr(connection, 'set_info_callback'): + connection.set_info_callback(_SSLCreator.info_callback) + server_hostname, expected_name = _resolve_pyopenssl_server_names( + self.endpoint.address, + self.ssl_options.get('server_hostname'), + self.check_hostname, + verify_endpoint_address=isinstance(self.endpoint, SniEndPoint)) + setattr(tlsProtocol, _TLS_APP_DATA_ATTR, + _TLSAppData(self.endpoint, expected_name, self.check_hostname)) connection.set_app_data(tlsProtocol) - if self.ssl_options and "server_hostname" in self.ssl_options: - connection.set_tlsext_host_name(self.ssl_options['server_hostname'].encode('ascii')) + if server_hostname is not None: + connection.set_tlsext_host_name( + _encode_server_hostname(server_hostname)) return connection @@ -217,12 +246,18 @@ 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 and not _HAS_SSL: + raise ImportError( + str(import_exception) + + ', pyOpenSSL must be installed to enable SSL support with the Twisted event loop' + ) + + def _build_ssl_context_from_options(self): + self._check_pyopenssl() + return _build_pyopenssl_context( + SSL, + self.ssl_options, + verify_by_default=self._ssl_options_verify_by_default) def add_connection(self): """ @@ -230,17 +265,16 @@ 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, self.ssl_options, self._check_hostname, - self.connect_timeout, ) endpoint = SSL4ClientEndpoint( diff --git a/cassandra/pool.py b/cassandra/pool.py index 176751f60a..c706812973 100644 --- a/cassandra/pool.py +++ b/cassandra/pool.py @@ -682,21 +682,26 @@ def _get_shard_aware_endpoint(self): shard_aware_port_ssl; if it is absent, return None so the pool opens a regular SSL connection instead of falling back to the plaintext port. Explicit ssl_options={}, like ssl_context, marks the cluster SSL-enabled. + Endpoint ssl_options also imply SSL for custom endpoint factories. """ if (self.advanced_shardaware_block_until and self.advanced_shardaware_block_until > time.time()) or \ self._session.cluster.shard_aware_options.disable_shardaware_port: return None cluster = self._session.cluster - 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: - endpoint = copy.copy(self.host.endpoint) - endpoint._port = self.host.sharding_info.shard_aware_port + host_endpoint = self.host.endpoint + sharding_info = self.host.sharding_info + ssl_enabled = (cluster.ssl_context is not None or + cluster.ssl_options is not None or + host_endpoint.ssl_options is not None) + + port = (sharding_info.shard_aware_port_ssl + if ssl_enabled else sharding_info.shard_aware_port) + if not port: + return None + + endpoint = copy.copy(host_endpoint) + endpoint._port = port return endpoint diff --git a/cassandra/tls.py b/cassandra/tls.py new file mode 100644 index 0000000000..867f62995a --- /dev/null +++ b/cassandra/tls.py @@ -0,0 +1,579 @@ +# 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. + +""" +Internal TLS configuration and certificate validation helpers. + +This module keeps reactor-independent TLS behavior in one place. The +connection and reactor modules remain responsible only for socket lifecycle +and event-loop integration. +""" + +import ipaddress +import logging +import ssl +import unicodedata + + +log = logging.getLogger(__name__) + + +_STDLIB_PROTOCOL_TO_PYOPENSSL_METHODS = { + 'PROTOCOL_TLS_CLIENT': ('TLS_CLIENT_METHOD', 'TLS_METHOD', 'TLSv1_2_METHOD'), + 'PROTOCOL_TLS': ('TLS_METHOD', 'TLS_CLIENT_METHOD', 'TLSv1_2_METHOD'), + 'PROTOCOL_SSLv23': ('TLS_METHOD', 'TLS_CLIENT_METHOD', 'TLSv1_2_METHOD'), + 'PROTOCOL_TLSv1_2': ('TLSv1_2_METHOD',), + 'PROTOCOL_TLSv1_1': ('TLSv1_1_METHOD',), + 'PROTOCOL_TLSv1': ('TLSv1_METHOD',), +} + + +_SSL_CONTEXT_OPTION_NAMES = frozenset(( + 'ssl_version', + 'cert_reqs', + 'check_hostname', + 'keyfile', + 'certfile', + 'ca_certs', + 'ciphers', +)) + + +def _ssl_options_requiring_new_context(ssl_options): + """Return endpoint options that cannot safely alter an existing context.""" + return set(ssl_options or {}) & _SSL_CONTEXT_OPTION_NAMES + + +def _prepare_ssl_options(ssl_options, endpoint_ssl_options): + """ + Merge caller and endpoint options and derive their default verification. + + Endpoint options override caller options. An explicitly empty caller + mapping enables TLS without verification, while endpoint-provided options + default to verification. A caller's empty mapping also remains + unverified when the endpoint contributes only SNI. + """ + endpoint_has_non_sni_options = bool( + endpoint_ssl_options and + any(name != 'server_hostname' for name in endpoint_ssl_options) + ) + if ssl_options is not None: + verify_by_default = bool(ssl_options) or endpoint_has_non_sni_options + else: + verify_by_default = bool(endpoint_ssl_options) + + merged_options = dict(ssl_options or {}) + merged_options.update(endpoint_ssl_options or {}) + enabled = ssl_options is not None or endpoint_ssl_options is not None + return merged_options, enabled, verify_by_default + + +def _stdlib_protocol_from_symbolic_name(protocol_name): + protocol = getattr(ssl, protocol_name, None) + if not protocol_name.startswith('PROTOCOL_') or protocol is None: + raise ValueError( + 'unknown symbolic stdlib SSL protocol %r' % (protocol_name,)) + return protocol + + +def _build_ssl_context_from_options(ssl_options, verify_by_default=None): + """Build a stdlib :class:`ssl.SSLContext` from legacy SSL options.""" + ssl_options = ssl_options or {} + ssl_version = ( + ssl_options.get('ssl_version') or ssl.PROTOCOL_TLS_CLIENT + ) + if isinstance(ssl_version, str): + ssl_version = _stdlib_protocol_from_symbolic_name(ssl_version) + + cert_reqs = ssl_options.get('cert_reqs') + check_hostname = bool(ssl_options.get('check_hostname', False)) + if cert_reqs is None: + if verify_by_default is None: + verify_by_default = bool(ssl_options) + cert_reqs = ( + ssl.CERT_REQUIRED if verify_by_default else ssl.CERT_NONE + ) + if check_hostname and cert_reqs == ssl.CERT_NONE: + cert_reqs = ssl.CERT_REQUIRED + + context = ssl.SSLContext(protocol=int(ssl_version)) + context.check_hostname = False + context.verify_mode = cert_reqs + context.check_hostname = check_hostname + + certfile = ssl_options.get('certfile') + if certfile: + context.load_cert_chain( + certfile, ssl_options.get('keyfile') or None) + + ca_certs = ssl_options.get('ca_certs') + if ca_certs: + context.load_verify_locations(ca_certs) + elif cert_reqs != ssl.CERT_NONE: + load_default_certs = getattr(context, 'load_default_certs', None) + if load_default_certs is not None: + load_default_certs() + + ciphers = ssl_options.get('ciphers') + if ciphers: + context.set_ciphers(ciphers) + return context + + +def _wrap_socket_from_context(context, sock, ssl_options, endpoint_address): + """Wrap a socket using only options accepted by ``SSLContext.wrap_socket``.""" + wrap_option_names = ( + 'server_side', + 'do_handshake_on_connect', + 'suppress_ragged_eofs', + 'server_hostname', + ) + options = { + name: ssl_options[name] + for name in wrap_option_names + if name in ssl_options + } + if context.check_hostname and 'server_hostname' not in options: + options['server_hostname'] = endpoint_address + return context.wrap_socket(sock, **options) + + +def _ssl_context_cert_validation_enabled(ssl_context): + """Return whether a stdlib or pyOpenSSL context verifies peer certificates.""" + if isinstance(ssl_context, ssl.SSLContext): + return ssl_context.verify_mode != ssl.CERT_NONE + + from OpenSSL import SSL + return bool(ssl_context.get_verify_mode() & SSL.VERIFY_PEER) + + +def _ssl_options_cert_validation_enabled(ssl_options): + """Return whether legacy SSL options request peer-certificate validation.""" + cert_reqs = ssl_options.get('cert_reqs') + if cert_reqs is None: + return bool(ssl_options) + if ssl_options.get('check_hostname', False): + return True + if cert_reqs == ssl.CERT_NONE: + return False + if cert_reqs in (ssl.CERT_OPTIONAL, ssl.CERT_REQUIRED): + return True + + try: + from OpenSSL import SSL + return bool(cert_reqs & SSL.VERIFY_PEER) + except (ImportError, TypeError): + 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 _numeric_constants(module, name_filter): + constants = {} + for name in dir(module): + if not name_filter(name): + continue + value = getattr(module, name, None) + try: + numeric_value = int(value) + except (TypeError, ValueError, OverflowError): + continue + constants.setdefault(numeric_value, []).append(name) + return constants + + +def _stdlib_protocol_names_by_value(): + return _numeric_constants( + ssl, lambda name: name.startswith('PROTOCOL_')) + + +def _pyopenssl_method_names_by_value(ssl_module): + return _numeric_constants( + ssl_module, lambda name: name.endswith('_METHOD')) + + +def _pyopenssl_method_from_stdlib_names(ssl_module, protocol_names): + resolved_methods = [] + unsupported_protocols = [] + for protocol_name in protocol_names: + method_names = _STDLIB_PROTOCOL_TO_PYOPENSSL_METHODS.get(protocol_name) + if method_names is None: + unsupported_protocols.append(protocol_name) + continue + for method_name in method_names: + method = getattr(ssl_module, method_name, None) + if method is not None: + resolved_methods.append((protocol_name, method_name, method)) + break + else: + raise ImportError( + 'pyOpenSSL does not expose a method for %s' % protocol_name) + + if unsupported_protocols: + raise ValueError( + 'stdlib SSL protocol %s is not supported for client connections' % + ', '.join(sorted(unsupported_protocols))) + + numeric_methods = {} + for protocol_name, method_name, method in resolved_methods: + try: + method_key = int(method) + except (TypeError, ValueError, OverflowError): + method_key = id(method) + numeric_methods.setdefault(method_key, []).append( + (protocol_name, method_name, method)) + if len(numeric_methods) != 1: + raise ValueError( + 'stdlib SSL protocol aliases %s resolve to different ' + 'pyOpenSSL methods' % ', '.join(sorted(protocol_names))) + return next(iter(numeric_methods.values()))[0][2] + + +def _pyopenssl_ssl_method_from_stdlib(ssl_module, ssl_version): + """ + Resolve stdlib protocols without guessing the meaning of ambiguous ints. + + pyOpenSSL exposes its protocol methods as plain integers, and several of + their values collide with different stdlib protocols. Symbolic stdlib + names provide an unambiguous serialized representation. + """ + if ssl_version is None: + return _default_pyopenssl_ssl_method(ssl_module) + + stdlib_protocols = _stdlib_protocol_names_by_value() + + if isinstance(ssl_version, str): + _stdlib_protocol_from_symbolic_name(ssl_version) + return _pyopenssl_method_from_stdlib_names( + ssl_module, (ssl_version,)) + + for protocol_names in stdlib_protocols.values(): + protocol = getattr(ssl, protocol_names[0]) + if (ssl_version.__class__ is protocol.__class__ and + ssl_version == protocol): + return _pyopenssl_method_from_stdlib_names( + ssl_module, protocol_names) + + if isinstance(ssl_version, int): + protocol_names = stdlib_protocols.get(ssl_version, ()) + pyopenssl_method_names = _pyopenssl_method_names_by_value( + ssl_module).get(ssl_version, ()) + + if protocol_names: + method = _pyopenssl_method_from_stdlib_names( + ssl_module, protocol_names) + if pyopenssl_method_names and int(method) != ssl_version: + raise ValueError( + 'plain integer ssl_version %r is ambiguous between ' + 'stdlib protocols %s and pyOpenSSL methods %s; pass an ' + 'ssl.PROTOCOL_* value directly or use its symbolic name' % + ( + ssl_version, + ', '.join(sorted(protocol_names)), + ', '.join(sorted(pyopenssl_method_names)), + ) + ) + return method + + if pyopenssl_method_names: + return ssl_version + + raise ValueError( + 'plain integer ssl_version %r does not identify a protocol ' + 'exposed by stdlib ssl or pyOpenSSL' % ssl_version) + + return ssl_version + + +def _pyopenssl_verify_mode_from_cert_reqs(ssl_module, cert_reqs): + # ``cert_reqs`` is a legacy stdlib-style option, so stdlib constants win + # when their integer values collide with pyOpenSSL verification flags. + if cert_reqs is None: + return None + 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 _ensure_pyopenssl_context_requires_verification( + ssl_module, context, check_hostname): + """ + Make hostname verification fail closed for a caller-supplied context. + + When hostname checking is enabled, this may mutate the context's verify + mode and callback because pyOpenSSL does not expose a non-mutating way to + require peer verification. Twisted may also replace a context-level info + callback when the installed pyOpenSSL version does not support callbacks + on individual connections. + """ + if not check_hostname: + return + + get_verify_mode = getattr(context, 'get_verify_mode', None) + verify_mode = get_verify_mode() if get_verify_mode is not None else None + if (verify_mode is not None and + verify_mode & ssl_module.VERIFY_PEER): + return + + promoted_verify_mode = ssl_module.VERIFY_PEER + if get_verify_mode is not None: + promoted_verify_mode = ( + (verify_mode or ssl_module.VERIFY_NONE) | + ssl_module.VERIFY_PEER + ) + + log.warning( + "check_hostname=True requires peer verification; mutating supplied " + "pyOpenSSL context to use VERIFY_PEER and replacing its verification " + "callback" + ) + context.set_verify( + promoted_verify_mode, + callback=lambda _connection, _x509, _errnum, _errdepth, ok: ok + ) + + +def _build_pyopenssl_context_from_options( + ssl_module, ssl_options, verify_by_default=None): + """Build a pyOpenSSL context from legacy stdlib-style SSL options.""" + ssl_options = ssl_options or {} + method = _pyopenssl_ssl_method_from_stdlib( + ssl_module, ssl_options.get('ssl_version')) + context = ssl_module.Context(method) + + certfile = ssl_options.get('certfile') + if certfile: + use_certificate_chain_file = getattr( + context, 'use_certificate_chain_file', None) + if use_certificate_chain_file is not None: + use_certificate_chain_file(certfile) + else: + context.use_certificate_file(certfile) + context.use_privatekey_file( + ssl_options.get('keyfile') or certfile) + + ca_certs = ssl_options.get('ca_certs') + if ca_certs: + context.load_verify_locations(ca_certs) + + cert_reqs = _pyopenssl_verify_mode_from_cert_reqs( + ssl_module, ssl_options.get('cert_reqs')) + if cert_reqs is None: + if verify_by_default is None: + verify_by_default = bool(ssl_options) + cert_reqs = ( + ssl_module.VERIFY_PEER + if verify_by_default + else ssl_module.VERIFY_NONE + ) + if (ssl_options.get('check_hostname', False) and + not cert_reqs & ssl_module.VERIFY_PEER): + cert_reqs |= ssl_module.VERIFY_PEER + + context.set_verify( + cert_reqs, + callback=lambda _connection, _x509, _errnum, _errdepth, ok: ok + ) + if (cert_reqs & ssl_module.VERIFY_PEER and + not ssl_options.get('ca_certs')): + set_default_verify_paths = getattr( + context, 'set_default_verify_paths', None) + if set_default_verify_paths is not None: + set_default_verify_paths() + + ciphers = ssl_options.get('ciphers') + if ciphers: + if isinstance(ciphers, str): + ciphers = ciphers.encode('ascii') + context.set_cipher_list(ciphers) + return context + + +def _idna_ascii_hostname(hostname): + """ + Convert a DNS name to ASCII without accepting lossy IDNA 2003 mappings. + + Python's built-in ``idna`` codec maps some distinct IDNA 2008 names to the + same ASCII name (for example, ``faß.de`` to ``fass.de``). Reject any + non-ASCII input that does not round-trip through the codec so callers can + supply the unambiguous IDNA 2008 A-label instead. + """ + try: + ascii_hostname = hostname.encode('idna').decode('ascii') + if any(ord(character) > 127 for character in hostname): + round_tripped = ascii_hostname.encode('ascii').decode('idna') + normalized_input = unicodedata.normalize('NFC', hostname).lower() + normalized_round_trip = unicodedata.normalize( + 'NFC', round_tripped).lower() + if normalized_input != normalized_round_trip: + return None + except (AttributeError, UnicodeError): + return None + return ascii_hostname.lower() + + +def _normalized_hostname(hostname, is_reference=False): + if is_reference: + if hostname.endswith('.'): + hostname = hostname[:-1] + if hostname.endswith('.'): + return None + elif hostname.endswith('.'): + return None + return _idna_ascii_hostname(hostname) + + +def _encode_server_hostname(hostname): + normalized = _normalized_hostname(hostname, is_reference=True) + if not normalized: + raise ValueError( + "server_hostname %r cannot be safely encoded as an IDNA name" % + hostname) + return normalized.encode('ascii') + + +def _dnsname_match(dn, hostname): + dn = _normalized_hostname(dn) + hostname = _normalized_hostname(hostname, is_reference=True) + + if not dn or not hostname: + return False + + if '*' not in dn: + return dn == hostname + + dn_labels = dn.split('.') + hostname_labels = hostname.split('.') + if (len(dn_labels) != len(hostname_labels) or + len(dn_labels) < 2 or dn_labels[0] != '*' or + not hostname_labels[0] or + any('*' in label for label in dn_labels[1:])): + return False + + return dn_labels[1:] == hostname_labels[1:] + + +def _ipaddress_match(cert_ip, hostname): + try: + host_ip = ipaddress.ip_address(hostname) + cert_ip = ipaddress.ip_address(cert_ip) + except ValueError: + return False + return cert_ip == host_ip + + +def _is_ip_address(hostname): + try: + ipaddress.ip_address(hostname) + except ValueError: + return False + return True + + +def _resolve_pyopenssl_server_names( + endpoint_address, server_hostname, check_hostname, + verify_endpoint_address=False): + """ + Resolve the SNI name and certificate identity for pyOpenSSL reactors. + + SNI proxy endpoints route using an explicit server name but authenticate + the proxy address. Ordinary endpoints authenticate an explicit + ``server_hostname`` when present, otherwise their endpoint address. + """ + expected_name = ( + endpoint_address + if verify_endpoint_address + else server_hostname or endpoint_address + ) + if (server_hostname is None and check_hostname and + not _is_ip_address(endpoint_address)): + server_hostname = endpoint_address + return server_hostname, expected_name + + +def _pyopenssl_cert_subject_alt_names(cert): + # Imported lazily because cryptography is an optional driver dependency. + # pyOpenSSL supplies it whenever this code path is used. + from cryptography import x509 + + try: + san = cert.to_cryptography().extensions.get_extension_for_class( + x509.SubjectAlternativeName).value + except x509.ExtensionNotFound: + return [], [] + + return ( + san.get_values_for_type(x509.DNSName), + [str(ip) for ip in san.get_values_for_type(x509.IPAddress)] + ) + + +def _pyopenssl_cert_common_names(cert): + # Imported lazily for the same reason as the SAN types above. + from cryptography.x509.oid import NameOID + + return [ + attribute.value + for attribute in cert.to_cryptography().subject.get_attributes_for_oid( + NameOID.COMMON_NAME) + ] + + +def _validate_pyopenssl_hostname(cert, hostname): + if cert is None: + raise ssl.CertificateError( + "peer did not present a certificate; cannot verify hostname %r" % + hostname) + + san_dns_names, san_ip_addresses = _pyopenssl_cert_subject_alt_names(cert) + hostname_is_ip = _is_ip_address(hostname) + + if hostname_is_ip: + for cert_ip in san_ip_addresses: + if _ipaddress_match(cert_ip, hostname): + return + raise ssl.CertificateError( + "IP address %r doesn't match certificate IP subjectAltName %r" % + (hostname, san_ip_addresses)) + + for cert_hostname in san_dns_names: + if _dnsname_match(cert_hostname, hostname): + return + if san_dns_names: + raise ssl.CertificateError( + "hostname %r doesn't match certificate DNS subjectAltName %r" % + (hostname, san_dns_names)) + + # Match SSLContext/OpenSSL X509_check_host semantics: for a DNS reference, + # only DNS subjectAltNames suppress commonName fallback. IP subjectAltNames + # are a different identity type and do not suppress it. Conversely, IP + # references are handled above and never fall back to commonName. The + # deprecated ssl.match_hostname() helper differs in the DNS-with-IP-SAN + # case and is not the compatibility target here. + common_names = _pyopenssl_cert_common_names(cert) + for common_name in common_names: + if _dnsname_match(common_name, hostname): + return + + raise ssl.CertificateError( + "hostname %r doesn't match certificate commonName %r" % + (hostname, common_names)) diff --git a/docs/security.rst b/docs/security.rst index 5c8645e685..1309ff3f96 100644 --- a/docs/security.rst +++ b/docs/security.rst @@ -47,10 +47,36 @@ These docs will include some examples for how to achieve common configurations, but the `ssl.SSLContext `_ documentation gives a more complete description of what is possible. -To enable SSL with version 3.17.0 and higher, you will need to set :attr:`.Cluster.ssl_context` to a -``ssl.SSLContext`` instance to enable SSL. Optionally, you can also set :attr:`.Cluster.ssl_options` -to a dict of options. These will be passed as kwargs to ``ssl.SSLContext.wrap_socket()`` -when new sockets are created. +With version 3.17.0 and higher, the preferred way to enable SSL is to set +:attr:`.Cluster.ssl_context` to an ``ssl.SSLContext`` instance. Optionally, you +can also set :attr:`.Cluster.ssl_options` to a dict of options passed as keyword +arguments to ``ssl.SSLContext.wrap_socket()`` when new sockets are created. + +For compatibility, :attr:`.Cluster.ssl_options` can also enable SSL without an +``ssl_context``. This use is deprecated and will be removed in the next major +release. An explicit empty dict (``ssl_options={}``) enables encryption without +verifying the server certificate when the endpoint supplies no additional +security settings. Cluster-level ``ssl_options=None`` does not enable SSL by +itself; an ``ssl_context`` or SSL options supplied by the selected +:class:`~.connection.EndPoint` still enable it. A nonempty options dict enables +peer-certificate verification by default unless ``cert_reqs`` explicitly +disables it. Setting +``check_hostname=True`` always enables peer-certificate verification because +hostname checking requires an authenticated certificate chain. When +verification is enabled and ``ca_certs`` is omitted, the driver loads the +system trust roots. + +When ``ssl_options={'check_hostname': True}`` is combined with a supplied +standard-library ``SSLContext`` whose hostname checking is disabled, the driver +enables ``SSLContext.check_hostname`` and promotes ``CERT_NONE`` to +``CERT_REQUIRED``. This mutates the supplied context because hostname checking +cannot be enabled per connection. + +An endpoint ``server_hostname`` used only for SNI routing is not an additional +security setting. Consequently, explicit ``ssl_options={}`` remains +unverified when an endpoint contributes only ``server_hostname``. Omitting +``ssl_options`` instead delegates TLS configuration to the endpoint, whose +nonempty options enable verification by default. If you create your SSLContext using `ssl.create_default_context `_, be aware that SSLContext.check_hostname is set to True by default, so the hostname validation will be done @@ -67,12 +93,31 @@ keystore files with these instructions: SSL with Twisted or Eventlet ^^^^^^^^^^^^^^^^^^^^^^^^^^^^ -Twisted and Eventlet both use an alternative SSL implementation called pyOpenSSL, so if your `Cluster`'s connection class is -:class:`~cassandra.io.twistedreactor.TwistedConnection` or :class:`~cassandra.io.eventletreactor.EventletConnection`, you must pass a -`pyOpenSSL context `_ instead. -An example is provided in these docs, and more details can be found in the -`documentation `_. +Twisted and Eventlet use pyOpenSSL instead of the standard-library SSL +implementation. When :attr:`.Cluster.ssl_options` is supplied without an +``ssl_context``, the driver automatically builds a compatible +`pyOpenSSL context `_ +and maps the supported protocol, verification, certificate, cipher, SNI, and +hostname-validation settings. + +Standard-library ``ssl.PROTOCOL_*`` values are translated to the corresponding +pyOpenSSL method. Serialized configurations may use the symbolic stdlib name, +such as ``'PROTOCOL_TLS'``. Avoid serializing protocol values as plain +integers: stdlib and pyOpenSSL assign different meanings to some of the same +integers, and ambiguous values are rejected. The ``cert_reqs`` option accepts +the corresponding stdlib certificate requirement constants. + +If you supply :attr:`.Cluster.ssl_context` directly with +:class:`~cassandra.io.twistedreactor.TwistedConnection` or +:class:`~cassandra.io.eventletreactor.EventletConnection`, it must be a +pyOpenSSL context rather than a standard-library ``ssl.SSLContext``. An example +is provided below, and more details can be found in the +`pyOpenSSL documentation `_. pyOpenSSL is not installed by the driver and must be installed separately. +Enabling hostname checks on an unverified supplied context promotes it to peer +verification and replaces its verification callback. On older pyOpenSSL +versions without per-connection info callbacks, Twisted also replaces the +context-level info callback. SSL Configuration Examples ^^^^^^^^^^^^^^^^^^^^^^^^^^ diff --git a/scripts/validate_astra_tls.py b/scripts/validate_astra_tls.py new file mode 100644 index 0000000000..bf080cdac8 --- /dev/null +++ b/scripts/validate_astra_tls.py @@ -0,0 +1,91 @@ +#!/usr/bin/env python +# 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. + +"""Validate Astra bundle TLS hostname verification on a pyOpenSSL reactor. + +Set ``ASTRA_SECURE_CONNECT_BUNDLE``, ``ASTRA_CLIENT_ID``, and +``ASTRA_CLIENT_SECRET``. Run once for each reactor: + + uv run python scripts/validate_astra_tls.py twisted + uv run python scripts/validate_astra_tls.py eventlet + +The script never prints credentials or bundle contents. +""" + +import argparse +import os + + +def _required_environment(): + names = ( + 'ASTRA_SECURE_CONNECT_BUNDLE', + 'ASTRA_CLIENT_ID', + 'ASTRA_CLIENT_SECRET', + ) + missing = [name for name in names if not os.environ.get(name)] + if missing: + raise SystemExit( + 'Missing required environment variables: %s' % + ', '.join(missing)) + return {name: os.environ[name] for name in names} + + +def _connection_class(reactor_name): + if reactor_name == 'eventlet': + import eventlet + eventlet.monkey_patch() + from cassandra.io.eventletreactor import EventletConnection + return EventletConnection + + from cassandra.io.twistedreactor import TwistedConnection + return TwistedConnection + + +def main(): + parser = argparse.ArgumentParser( + description='Validate Astra TLS with Twisted or Eventlet') + parser.add_argument('reactor', choices=('twisted', 'eventlet')) + args = parser.parse_args() + environment = _required_environment() + connection_class = _connection_class(args.reactor) + + from cassandra.auth import PlainTextAuthProvider + from cassandra.cluster import Cluster + + Cluster.connection_class = connection_class + cluster = Cluster( + cloud={ + 'secure_connect_bundle': + environment['ASTRA_SECURE_CONNECT_BUNDLE'], + }, + auth_provider=PlainTextAuthProvider( + environment['ASTRA_CLIENT_ID'], + environment['ASTRA_CLIENT_SECRET']), + ) + try: + session = cluster.connect(wait_for_all_pools=True) + row = session.execute( + "SELECT release_version FROM system.local WHERE key='local'" + ).one() + if row is None: + raise RuntimeError('Astra query returned no system.local row') + print( + '%s Astra TLS validation passed (release %s)' % + (args.reactor, row.release_version)) + finally: + cluster.shutdown() + + +if __name__ == '__main__': + main() diff --git a/tests/unit/advanced/test_insights.py b/tests/unit/advanced/test_insights.py index 2050439804..66d564af57 100644 --- a/tests/unit/advanced/test_insights.py +++ b/tests/unit/advanced/test_insights.py @@ -16,18 +16,23 @@ 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, DefaultEndPoint 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.tls import _ssl_options_cert_validation_enabled from cassandra.policies import ( LoadBalancingPolicy, + HostDistance, DCAwareRoundRobinPolicy, TokenAwarePolicy, WhiteListRoundRobinPolicy, @@ -40,12 +45,24 @@ WrapperPolicy ) +try: + from OpenSSL import SSL as pyopenssl_ssl +except (ImportError, AttributeError): + pyopenssl_ssl = None + log = logging.getLogger(__name__) initialize_registry(insights_registry) +class EndpointSSLOptionsEndPoint(DefaultEndPoint): + + @property + def ssl_options(self): + return {} + + class TestGetConfig(unittest.TestCase): def test_invalid_object(self): @@ -88,6 +105,194 @@ 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, connection=sentinel.default_connection, + ssl_options=None, ssl_context=None, + endpoints_resolved=()): + reporter = MonitorReporter.__new__(MonitorReporter) + reporter._interval = 30 + + if connection is sentinel.default_connection: + connection = Connection() + + cluster = Mock() + cluster.auth_provider = None + cluster.client_id = 'client-id' + cluster.connection_class = Connection + cluster.control_connection._connection = connection + cluster.ssl_options = ssl_options + cluster.ssl_context = ssl_context + cluster.endpoints_resolved = endpoints_resolved + 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() + + 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_without_validation(self): + connection = Connection(ssl_options={}) + + startup_data = self._get_startup_data(connection) + + assert connection._ssl_options_explicit is True + assert startup_data['data']['sslConfigured']['enabled'] is True + assert startup_data['data']['sslConfigured']['certValidation'] is False + + def test_missing_control_connection_uses_cluster_ssl_configuration(self): + startup_data = self._get_startup_data(None, ssl_options={}) + + assert startup_data['data']['initialControlConnection'] is None + assert startup_data['data']['localAddress'] is None + assert startup_data['data']['compression'] == 'NONE' + assert startup_data['data']['sslConfigured']['enabled'] is True + assert startup_data['data']['sslConfigured']['certValidation'] is False + + def test_missing_control_connection_uses_endpoint_ssl_configuration(self): + startup_data = self._get_startup_data( + None, + endpoints_resolved=[ + EndpointSSLOptionsEndPoint('127.0.0.1') + ]) + + assert startup_data['data']['sslConfigured']['enabled'] is True + assert startup_data['data']['sslConfigured']['certValidation'] is False + + def test_ssl_options_cert_required_validation_reported(self): + connection = Connection(ssl_options={'cert_reqs': ssl.CERT_REQUIRED}) + + startup_data = self._get_startup_data(connection) + + assert startup_data['data']['sslConfigured']['certValidation'] is True + + def test_check_hostname_validation_reported(self): + connection = Connection(ssl_options={'check_hostname': True}) + + startup_data = self._get_startup_data(connection) + + assert startup_data['data']['sslConfigured']['certValidation'] is True + + def test_omitted_ssl_options_reported_as_disabled(self): + connection = Connection() + + startup_data = self._get_startup_data(connection) + + assert connection.ssl_options == {} + assert connection._ssl_options_explicit is False + assert startup_data['data']['sslConfigured']['enabled'] is False + assert startup_data['data']['sslConfigured']['certValidation'] is None + + def test_ssl_context_reported_as_enabled(self): + ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + connection = Connection(ssl_context=ssl_context) + + startup_data = self._get_startup_data(connection) + + assert startup_data['data']['sslConfigured']['enabled'] is True + + def test_ssl_context_cert_optional_validation_reported_enabled(self): + ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ssl_context.check_hostname = False + ssl_context.verify_mode = ssl.CERT_OPTIONAL + connection = Connection(ssl_context=ssl_context) + + startup_data = self._get_startup_data(connection) + + assert startup_data['data']['sslConfigured']['certValidation'] is True + + @unittest.skipIf(pyopenssl_ssl is None, "pyOpenSSL is not available") + def test_pyopenssl_context_without_verify_peer_validation_reported_disabled(self): + ssl_method = getattr( + pyopenssl_ssl, 'TLS_METHOD', pyopenssl_ssl.SSLv23_METHOD) + ssl_context = pyopenssl_ssl.Context(ssl_method) + ssl_context.set_verify( + pyopenssl_ssl.VERIFY_CLIENT_ONCE, callback=lambda *args: True) + connection = Connection(ssl_context=ssl_context) + + startup_data = self._get_startup_data(connection) + + assert startup_data['data']['sslConfigured']['certValidation'] is False + + @unittest.skipIf(pyopenssl_ssl is None, "pyOpenSSL is not available") + def test_pyopenssl_context_with_verify_peer_validation_reported_enabled(self): + ssl_method = getattr( + pyopenssl_ssl, 'TLS_METHOD', pyopenssl_ssl.SSLv23_METHOD) + ssl_context = pyopenssl_ssl.Context(ssl_method) + ssl_context.set_verify( + pyopenssl_ssl.VERIFY_PEER | pyopenssl_ssl.VERIFY_CLIENT_ONCE, + callback=lambda *args: True) + connection = Connection(ssl_context=ssl_context) + + startup_data = self._get_startup_data(connection) + + assert startup_data['data']['sslConfigured']['certValidation'] is True + + @unittest.skipIf(pyopenssl_ssl is None, "pyOpenSSL is not available") + def test_pyopenssl_options_without_verify_peer_report_validation_disabled(self): + assert not _ssl_options_cert_validation_enabled({ + 'cert_reqs': pyopenssl_ssl.VERIFY_CLIENT_ONCE, + }) + + @unittest.skipIf(pyopenssl_ssl is None, "pyOpenSSL is not available") + def test_pyopenssl_options_with_verify_peer_report_validation_enabled(self): + assert _ssl_options_cert_validation_enabled({ + 'cert_reqs': ( + pyopenssl_ssl.VERIFY_PEER | + pyopenssl_ssl.VERIFY_CLIENT_ONCE + ), + }) + + def test_endpoint_ssl_options_reported_as_enabled(self): + connection = Connection(EndpointSSLOptionsEndPoint('127.0.0.1')) + + startup_data = self._get_startup_data(connection) + + assert connection._ssl_options_explicit is True + assert startup_data['data']['sslConfigured']['enabled'] is True + assert startup_data['data']['sslConfigured']['certValidation'] is False + + +class TestMonitorReporterStatusData(unittest.TestCase): + + def test_control_connection_is_snapshotted_during_shutdown(self): + connection = Mock(host='127.0.0.1') + + class ChangingControlConnection(object): + calls = 0 + + @property + def _connection(self): + self.calls += 1 + return connection if self.calls == 1 else None + + reporter = MonitorReporter.__new__(MonitorReporter) + cluster = Mock() + cluster.client_id = 'client-id' + cluster.control_connection = ChangingControlConnection() + session = Mock() + session.cluster = cluster + session.session_id = 'session-id' + session.get_pool_state.return_value = {} + reporter._session = session + + status_data = reporter._get_status_data() + + assert status_data['data']['controlConnection'] == '127.0.0.1' + assert cluster.control_connection.calls == 1 + + class TestConfigAsDict(unittest.TestCase): # graph/query.py diff --git a/tests/unit/io/fixtures/root_ca.pem b/tests/unit/io/fixtures/root_ca.pem new file mode 100644 index 0000000000..a0a0ec73cf --- /dev/null +++ b/tests/unit/io/fixtures/root_ca.pem @@ -0,0 +1,19 @@ +-----BEGIN CERTIFICATE----- +MIIDCzCCAfMCFCoTNYhIQpOXMBnAq8Bw72qfKwGLMA0GCSqGSIb3DQEBCwUAMEIx +CzAJBgNVBAYTAlVTMREwDwYDVQQKDAhkYXRhc3RheDEPMA0GA1UECwwGZmllbGRz +MQ8wDQYDVQQDDAZyb290Q2EwHhcNMjEwMzE3MTcwNTE2WhcNMzEwMzE1MTcwNTE2 +WjBCMQswCQYDVQQGEwJVUzERMA8GA1UECgwIZGF0YXN0YXgxDzANBgNVBAsMBmZp +ZWxkczEPMA0GA1UEAwwGcm9vdENhMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB +CgKCAQEApFoQtNu0+XQuMBPle4WAJYIMR74HL15uk9ToKBqMEXL7ah3r23xTTeGr +NyUXicM6Owiup7DK27F4vni+MYKAn7L4uZ99mW0ATYNXBDLFB+wwy1JBk4Dw5+eZ +q9lz1TGK7uBvTOXCllOA2qxRqtMTl2aPy5OuciWQe794abwFqs5+1l9GEuzJGsp1 +P9L4yljbmijC8RmvDFAeUZoKRdKXw2G5kUOHqK9Aej5gLxIK920PezpgLxm0V/PD +ZAlwlsW0vT79RgZCF/vtKcKSLtFTHgPBNPPbkZmOdE7s/6KoAkORBV/9CIsKeTC3 +Y/YeYQ2+G0gxiq1RcMavPw8f58POTQIDAQABMA0GCSqGSIb3DQEBCwUAA4IBAQA1 +MXBlk6u2oVBM+4SyYc2nsaHyerM+omUEysAUNFJq6S6i0pu32ULcusDfrnrIQoyR +xPJ/GSYqZkIDX0s9LvPVD6A6bnugR+Z6VfEniLkG1+TkFC+JMCblgJyaF/EbuayU +3iJX+uj7ikTySjMSDvXxOHik2i0aOh90B/351+sFnSPQrFDQ0XqxeG8s0d7EiLTV +wWJmsYglSeTo1vF3ilVRwjmHO9sX6cmQhRvRNmiQrdWaM3gLS5F6yoQ2UQQ3YdFp +quhYuNwy0Ip6ZpORHYtzkCKSanz/oUh17QWvi7aaJyqD5G5hWZgn3R4RCutoOHRS +TEJ+xzhY768rpsrrNUou +-----END CERTIFICATE----- diff --git a/tests/unit/io/test_asyncioreactor.py b/tests/unit/io/test_asyncioreactor.py index f3ed942090..60686aaf9c 100644 --- a/tests/unit/io/test_asyncioreactor.py +++ b/tests/unit/io/test_asyncioreactor.py @@ -9,6 +9,7 @@ from tests import is_monkey_patched, connection_class from tests.unit.io.utils import TimerCallback, TimerTestMixin +from cassandra.connection import DefaultEndPoint from unittest.mock import patch, MagicMock import selectors import unittest @@ -19,6 +20,56 @@ (connection_class is not AsyncioConnection)) +@unittest.skipIf(is_monkey_patched(), 'runtime is monkey patched for another reactor') +@unittest.skipIf(connection_class is not AsyncioConnection, + 'not running asyncio tests; current connection_class is {}'.format(connection_class)) +@unittest.skipUnless(ASYNCIO_AVAILABLE, "asyncio is not available for this runtime") +class AsyncioSSLContextTest(unittest.TestCase): + + def setUp(self): + if skip_me: + return + self.old_loop = AsyncioConnection._loop + AsyncioConnection._loop = object() + self.addCleanup(setattr, AsyncioConnection, '_loop', self.old_loop) + + def test_empty_ssl_options_use_tls_transport_path(self): + socket_mock = MagicMock() + scheduled_tasks = [] + + def set_socket(conn): + conn._socket = socket_mock + + def schedule_task(task, loop): + scheduled_tasks.append((task, loop)) + close = getattr(task, 'close', None) + if close: + close() + return MagicMock() + + with patch.object(AsyncioConnection, '_connect_socket', autospec=True, + side_effect=set_socket) as connect_socket: + with patch.object(AsyncioConnection, '_setup_ssl_and_run') as setup_ssl_and_run: + with patch.object(AsyncioConnection, 'handle_read') as handle_read: + with patch.object(AsyncioConnection, 'handle_write') as handle_write: + with patch('cassandra.io.asyncioreactor.asyncio.run_coroutine_threadsafe', + side_effect=schedule_task) as run_coro: + with patch.object(AsyncioConnection, '_send_options_message'): + conn = AsyncioConnection(DefaultEndPoint('1.2.3.4'), ssl_options={}) + + assert conn._using_ssl + assert conn._protocol is not None + assert conn._ssl_ready is not None + connect_socket.assert_called_once_with(conn) + socket_mock.setblocking.assert_called_once_with(0) + setup_ssl_and_run.assert_called_once_with() + handle_read.assert_not_called() + handle_write.assert_called_once_with() + assert run_coro.call_count == 2 + assert scheduled_tasks[0][1] is AsyncioConnection._loop + assert scheduled_tasks[1][1] is AsyncioConnection._loop + + @unittest.skipIf(is_monkey_patched(), 'runtime is monkey patched for another reactor') @unittest.skipIf(connection_class is not AsyncioConnection, 'not running asyncio tests; current connection_class is {}'.format(connection_class)) diff --git a/tests/unit/io/test_eventletreactor.py b/tests/unit/io/test_eventletreactor.py index d3962196a4..ac7a5bbf79 100644 --- a/tests/unit/io/test_eventletreactor.py +++ b/tests/unit/io/test_eventletreactor.py @@ -12,21 +12,354 @@ # See the License for the specific language governing permissions and # limitations under the License. - import unittest -from unittest.mock import patch +from unittest.mock import Mock, patch -from tests.unit.io.utils import TimerTestMixin +from tests.unit.io.utils import (TimerTestMixin, UNIT_CA_CERT, + make_pyopenssl_x509_certificate) from tests import notpypy, EVENT_LOOP_MANAGER 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, ConnectionException, + DefaultEndPoint, SniEndPoint) + + +@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_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_explicit_empty_options_with_endpoint_sni_remain_unverified(self): + conn = EventletConnection.__new__(EventletConnection) + + Connection.__init__( + conn, + SniEndPoint('1.2.3.4', 'node.example.com'), + ssl_options={}) + + assert conn.ssl_options['server_hostname'] == 'node.example.com' + assert conn.ssl_context.get_verify_mode() == eventletreactor.SSL.VERIFY_NONE + + def test_endpoint_context_options_rejected_for_supplied_context(self): + endpoint = SniEndPoint('1.2.3.4', 'node.example.com') + endpoint._ssl_options.update({ + 'ca_certs': UNIT_CA_CERT, + 'check_hostname': True, + }) + context = eventletreactor.SSL.Context( + eventletreactor.SSL.TLS_METHOD) + conn = EventletConnection.__new__(EventletConnection) + + with self.assertRaisesRegex(ValueError, "independent SSL context"): + Connection.__init__(conn, endpoint, ssl_context=context) + + assert context.get_verify_mode() == eventletreactor.SSL.VERIFY_NONE + + def test_explicit_empty_options_with_endpoint_ca_require_verification(self): + endpoint = SniEndPoint('1.2.3.4', 'node.example.com') + endpoint._ssl_options = {'ca_certs': UNIT_CA_CERT} + conn = EventletConnection.__new__(EventletConnection) + + Connection.__init__(conn, endpoint, ssl_options={}) + + assert conn.ssl_options == {'ca_certs': UNIT_CA_CERT} + assert conn.ssl_context.get_verify_mode() == eventletreactor.SSL.VERIFY_PEER + + def test_wrap_socket_from_context_returns_wrapped_socket(self): + conn = EventletConnection.__new__(EventletConnection) + conn.endpoint = DefaultEndPoint('1.2.3.4') + conn.ssl_context = Mock() + conn.ssl_context.get_verify_mode.return_value = eventletreactor.SSL.VERIFY_PEER + conn.ssl_options = {} + conn._check_hostname = False + 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() + wrapped_socket.set_tlsext_host_name.assert_not_called() + assert conn._socket is wrapped_socket + + def test_wrap_socket_uses_endpoint_address_for_sni_when_checking_hostname(self): + conn = EventletConnection.__new__(EventletConnection) + conn.endpoint = DefaultEndPoint('node.example.com') + conn.ssl_context = Mock() + conn.ssl_context.get_verify_mode.return_value = eventletreactor.SSL.VERIFY_PEER + conn.ssl_options = {} + conn._check_hostname = True + 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_tlsext_host_name.assert_called_once_with(b'node.example.com') + assert conn._socket is wrapped_socket + + def test_wrap_socket_idna_encodes_unicode_sni(self): + conn = EventletConnection.__new__(EventletConnection) + conn.endpoint = DefaultEndPoint('täst.example') + conn.ssl_context = Mock() + conn.ssl_context.get_verify_mode.return_value = eventletreactor.SSL.VERIFY_PEER + conn.ssl_options = {} + conn._check_hostname = True + conn._socket = object() + + with patch.object(eventletreactor.SSL, 'Connection') as mock_connection: + wrapped_socket = mock_connection.return_value + assert conn._wrap_socket_from_context() is wrapped_socket + + wrapped_socket.set_tlsext_host_name.assert_called_once_with( + b'xn--tst-qla.example') + + def test_wrap_socket_omits_implicit_sni_for_ip_addresses(self): + for address in ('1.2.3.4', '2001:db8::1'): + with self.subTest(address=address): + conn = EventletConnection.__new__(EventletConnection) + conn.endpoint = DefaultEndPoint(address) + conn.ssl_context = Mock() + conn.ssl_context.get_verify_mode.return_value = eventletreactor.SSL.VERIFY_PEER + conn.ssl_options = {} + conn._check_hostname = True + conn._socket = object() + + with patch.object(eventletreactor.SSL, 'Connection') as mock_connection: + wrapped_socket = mock_connection.return_value + assert conn._wrap_socket_from_context() is wrapped_socket + + wrapped_socket.set_tlsext_host_name.assert_not_called() + certificate = wrapped_socket.get_peer_certificate.return_value + with patch.object( + eventletreactor, '_validate_pyopenssl_hostname') as validate_hostname: + conn._validate_hostname() + + validate_hostname.assert_called_once_with(certificate, address) + + def test_wrap_socket_preserves_explicit_ip_sni(self): + conn = EventletConnection.__new__(EventletConnection) + conn.endpoint = DefaultEndPoint('node.example.com') + conn.ssl_context = Mock() + conn.ssl_context.get_verify_mode.return_value = eventletreactor.SSL.VERIFY_PEER + conn.ssl_options = {'server_hostname': '1.2.3.4'} + conn._check_hostname = True + conn._socket = object() + + with patch.object(eventletreactor.SSL, 'Connection') as mock_connection: + wrapped_socket = mock_connection.return_value + assert conn._wrap_socket_from_context() is wrapped_socket + + wrapped_socket.set_tlsext_host_name.assert_called_once_with(b'1.2.3.4') + + def test_sni_endpoint_routes_by_host_id_but_verifies_proxy_address(self): + host_id = '01234567-89ab-cdef-0123-456789abcdef' + conn = EventletConnection.__new__(EventletConnection) + conn.endpoint = SniEndPoint('proxy.example.com', host_id) + conn.ssl_context = Mock() + conn.ssl_context.get_verify_mode.return_value = ( + eventletreactor.SSL.VERIFY_PEER) + conn.ssl_options = conn.endpoint.ssl_options + conn._check_hostname = True + conn._socket = object() + + with patch.object(eventletreactor.SSL, 'Connection') as connection_mock: + wrapped_socket = connection_mock.return_value + assert conn._wrap_socket_from_context() is wrapped_socket + + wrapped_socket.set_tlsext_host_name.assert_called_once_with( + host_id.encode('ascii')) + certificate = wrapped_socket.get_peer_certificate.return_value + with patch.object( + eventletreactor, '_validate_pyopenssl_hostname') as validate_hostname: + conn._validate_hostname() + + validate_hostname.assert_called_once_with( + certificate, conn.endpoint.address) + + def test_validate_hostname_uses_server_hostname_and_san(self): + conn = EventletConnection.__new__(EventletConnection) + conn.endpoint = DefaultEndPoint('10.0.0.1') + conn.ssl_options = {'server_hostname': 'node.example.com'} + conn._socket = Mock() + conn._socket.get_peer_certificate.return_value = make_pyopenssl_x509_certificate( + 'wrong.example.com', san_dns_names=['node.example.com']) + + conn._validate_hostname() + + def test_validate_hostname_prefers_san_over_common_name(self): + conn = EventletConnection.__new__(EventletConnection) + conn.endpoint = DefaultEndPoint('node.example.com') + conn.ssl_options = {} + conn._socket = Mock() + conn._socket.get_peer_certificate.return_value = make_pyopenssl_x509_certificate( + 'node.example.com', san_dns_names=['other.example.com']) + + with self.assertRaises(ConnectionException): + conn._validate_hostname() + + @staticmethod + def _connection_for_address_attempts(sockets): + conn = EventletConnection.__new__(EventletConnection) + conn.endpoint = DefaultEndPoint('node.example.com') + conn._get_socket_addresses = Mock(return_value=[ + (eventletreactor.socket.AF_INET, eventletreactor.socket.SOCK_STREAM, + 6, '', ('192.0.2.1', 9042)), + (eventletreactor.socket.AF_INET, eventletreactor.socket.SOCK_STREAM, + 6, '', ('192.0.2.2', 9042)), + ]) + conn._socket_impl = Mock() + conn._socket_impl.socket.side_effect = sockets + conn.ssl_context = None + conn._ssl_options_explicit = True + conn.features = Mock(shard_id=None) + conn.connect_timeout = 5 + conn._check_hostname = False + conn.sockopts = None + return conn + + def test_handshake_error_closes_socket_and_tries_next_address(self): + first_socket = Mock() + second_socket = Mock() + handshake_error = eventletreactor.SSL.Error('handshake failed') + first_socket.do_handshake.side_effect = handshake_error + conn = self._connection_for_address_attempts( + [first_socket, second_socket]) + + conn._connect_socket() + + first_socket.connect.assert_called_once_with(('192.0.2.1', 9042)) + first_socket.close.assert_called_once_with() + second_socket.connect.assert_called_once_with(('192.0.2.2', 9042)) + second_socket.do_handshake.assert_called_once_with() + second_socket.close.assert_not_called() + assert conn._socket is second_socket + + def test_hostname_error_closes_socket_and_tries_next_address(self): + first_socket = Mock() + second_socket = Mock() + conn = self._connection_for_address_attempts( + [first_socket, second_socket]) + conn._check_hostname = True + hostname_error = ConnectionException( + 'Hostname verification failed', conn.endpoint) + conn._validate_hostname = Mock( + side_effect=[hostname_error, None]) + + conn._connect_socket() + + first_socket.connect.assert_called_once_with(('192.0.2.1', 9042)) + first_socket.close.assert_called_once_with() + second_socket.connect.assert_called_once_with(('192.0.2.2', 9042)) + second_socket.close.assert_not_called() + assert conn._socket is second_socket + assert conn._validate_hostname.call_count == 2 + + def test_hostname_error_is_not_masked_by_later_socket_error(self): + first_socket = Mock() + second_socket = Mock() + second_socket.connect.side_effect = eventletreactor.socket.error( + 111, 'Connection refused') + conn = self._connection_for_address_attempts( + [first_socket, second_socket]) + conn._check_hostname = True + hostname_error = ConnectionException( + 'Hostname verification failed', conn.endpoint) + conn._validate_hostname = Mock(side_effect=hostname_error) + + with self.assertRaises(ConnectionException) as raised: + conn._connect_socket() + + assert raised.exception is hostname_error + first_socket.close.assert_called_once_with() + second_socket.close.assert_called_once_with() + + @staticmethod + def _run_io_handler(handler_name, error): + conn = Mock(in_buffer_size=4096) + conn._write_queue.get.return_value = b'message' + operation = 'recv' if handler_name == 'handle_read' else 'sendall' + getattr(conn._socket, operation).side_effect = error + + getattr(EventletConnection, handler_name)(conn) + return conn + + def test_zero_return_during_io_closes_connection_cleanly(self): + for handler_name in ('handle_read', 'handle_write'): + with self.subTest(handler=handler_name): + conn = self._run_io_handler( + handler_name, eventletreactor.SSL.ZeroReturnError()) + + conn.close.assert_called_once_with() + conn.defunct.assert_not_called() + + def test_tls_error_during_io_defuncts_connection(self): + for handler_name in ('handle_read', 'handle_write'): + with self.subTest(handler=handler_name): + error = eventletreactor.SSL.Error('boom') + conn = self._run_io_handler(handler_name, error) + + conn.defunct.assert_called_once_with(error) + conn.close.assert_not_called() + + def test_socket_error_during_io_still_defuncts_connection(self): + for handler_name in ('handle_read', 'handle_write'): + with self.subTest(handler=handler_name): + error = eventletreactor.socket.error('boom') + conn = self._run_io_handler(handler_name, error) + + conn.defunct.assert_called_once_with(error) + conn.close.assert_not_called() + + def test_greenlet_exit_during_io_remains_graceful(self): + for handler_name in ('handle_read', 'handle_write'): + with self.subTest(handler=handler_name): + conn = self._run_io_handler( + handler_name, eventletreactor.GreenletExit()) + + conn.defunct.assert_not_called() + conn.close.assert_not_called() + + def test_falsey_tls_recv_closes_before_writing_to_buffer(self): + conn = Mock(in_buffer_size=4096) + conn._socket.recv.return_value = '' + + EventletConnection.handle_read(conn) + + conn.close.assert_called_once_with() + conn._iobuf.write.assert_not_called() + conn.defunct.assert_not_called() + + 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..26ac45149a 100644 --- a/tests/unit/io/test_twistedreactor.py +++ b/tests/unit/io/test_twistedreactor.py @@ -15,7 +15,8 @@ import unittest from unittest.mock import Mock, patch -from cassandra.connection import DefaultEndPoint +from cassandra.connection import (Connection, ConnectionException, + DefaultEndPoint, SniEndPoint) try: from twisted.test import proto_helpers @@ -27,7 +28,294 @@ from cassandra.connection import _Frame -from tests.unit.io.utils import TimerTestMixin +from tests.unit.io.utils import TimerTestMixin, UNIT_CA_CERT + + +@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_ssl_creator_requires_prepared_context(self): + with self.assertRaisesRegex(ValueError, 'prepared pyOpenSSL context'): + twistedreactor._SSLCreator( + DefaultEndPoint('node.example.com'), None, {}, False) + + 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_explicit_empty_options_with_endpoint_sni_remain_unverified(self): + conn = TwistedConnection.__new__(TwistedConnection) + + Connection.__init__( + conn, + SniEndPoint('1.2.3.4', 'node.example.com'), + ssl_options={}) + + assert conn.ssl_options['server_hostname'] == 'node.example.com' + assert conn.ssl_context.get_verify_mode() == twistedreactor.SSL.VERIFY_NONE + + def test_endpoint_context_options_rejected_for_supplied_context(self): + endpoint = SniEndPoint('1.2.3.4', 'node.example.com') + endpoint._ssl_options.update({ + 'ca_certs': UNIT_CA_CERT, + 'check_hostname': True, + }) + context = twistedreactor.SSL.Context( + twistedreactor.SSL.TLS_METHOD) + conn = TwistedConnection.__new__(TwistedConnection) + + with self.assertRaisesRegex(ValueError, "independent SSL context"): + Connection.__init__(conn, endpoint, ssl_context=context) + + assert context.get_verify_mode() == twistedreactor.SSL.VERIFY_NONE + + def test_supplied_context_check_hostname_requires_peer_verification(self): + context = twistedreactor.SSL.Context(twistedreactor.SSL.TLS_METHOD) + + assert context.get_verify_mode() == twistedreactor.SSL.VERIFY_NONE + + twistedreactor._SSLCreator( + DefaultEndPoint('node.example.com'), + context, + {'check_hostname': True}, + True) + + assert context.get_verify_mode() == twistedreactor.SSL.VERIFY_PEER + + def test_ssl_creator_sets_hostname_callback_on_connection(self): + context = Mock() + context.get_verify_mode.return_value = twistedreactor.SSL.VERIFY_PEER + + with patch.object(twistedreactor.SSL, 'Connection') as connection_mock: + creator = twistedreactor._SSLCreator( + DefaultEndPoint('node.example.com'), + context, + {'server_hostname': 'node.example.com'}, + True) + connection = connection_mock.return_value + result = creator.clientConnectionForTLS(Mock()) + + context.set_info_callback.assert_not_called() + assert result is connection + connection.set_info_callback.assert_called_once_with(twistedreactor._SSLCreator.info_callback) + connection.set_tlsext_host_name.assert_called_once_with(b'node.example.com') + + def test_sni_endpoint_routes_by_host_id_but_verifies_proxy_address(self): + host_id = '01234567-89ab-cdef-0123-456789abcdef' + endpoint = SniEndPoint('proxy.example.com', host_id) + context = Mock() + context.get_verify_mode.return_value = twistedreactor.SSL.VERIFY_PEER + + with patch.object(twistedreactor.SSL, 'Connection') as connection_mock: + tls_protocol = Mock() + creator = twistedreactor._SSLCreator( + endpoint, context, endpoint.ssl_options, True) + connection = connection_mock.return_value + creator.clientConnectionForTLS(tls_protocol) + + connection.set_tlsext_host_name.assert_called_once_with( + host_id.encode('ascii')) + tls_app_data = getattr( + tls_protocol, twistedreactor._TLS_APP_DATA_ATTR) + assert tls_app_data.expected_name == endpoint.address + + connection.get_app_data.return_value = tls_protocol + certificate = connection.get_peer_certificate.return_value + with patch.object( + twistedreactor, '_validate_pyopenssl_hostname') as validate_hostname: + twistedreactor._SSLCreator.info_callback( + connection, twistedreactor.SSL.SSL_CB_HANDSHAKE_DONE, 1) + + validate_hostname.assert_called_once_with( + certificate, endpoint.address) + + def test_ssl_creator_uses_endpoint_address_for_sni_when_checking_hostname(self): + context = Mock() + context.get_verify_mode.return_value = twistedreactor.SSL.VERIFY_PEER + + with patch.object(twistedreactor.SSL, 'Connection') as connection_mock: + creator = twistedreactor._SSLCreator( + DefaultEndPoint('node.example.com'), + context, + {}, + True) + connection = connection_mock.return_value + result = creator.clientConnectionForTLS(Mock()) + + assert result is connection + connection.set_tlsext_host_name.assert_called_once_with(b'node.example.com') + + def test_ssl_creator_idna_encodes_unicode_sni(self): + context = Mock() + context.get_verify_mode.return_value = twistedreactor.SSL.VERIFY_PEER + + with patch.object(twistedreactor.SSL, 'Connection') as connection_mock: + creator = twistedreactor._SSLCreator( + DefaultEndPoint('täst.example'), + context, + {}, + True) + connection = connection_mock.return_value + result = creator.clientConnectionForTLS(Mock()) + + assert result is connection + connection.set_tlsext_host_name.assert_called_once_with( + b'xn--tst-qla.example') + + def test_ssl_creator_omits_implicit_sni_for_ip_addresses(self): + context = Mock() + context.get_verify_mode.return_value = twistedreactor.SSL.VERIFY_PEER + + for address in ('1.2.3.4', '2001:db8::1'): + with self.subTest(address=address): + with patch.object(twistedreactor.SSL, 'Connection') as connection_mock: + creator = twistedreactor._SSLCreator( + DefaultEndPoint(address), + context, + {}, + True) + tls_protocol = Mock() + connection = connection_mock.return_value + result = creator.clientConnectionForTLS(tls_protocol) + + assert result is connection + connection.set_tlsext_host_name.assert_not_called() + tls_app_data = getattr( + tls_protocol, twistedreactor._TLS_APP_DATA_ATTR) + assert tls_app_data.expected_name == address + + def test_ssl_creator_preserves_explicit_ip_sni(self): + context = Mock() + context.get_verify_mode.return_value = twistedreactor.SSL.VERIFY_PEER + + with patch.object(twistedreactor.SSL, 'Connection') as connection_mock: + creator = twistedreactor._SSLCreator( + DefaultEndPoint('node.example.com'), + context, + {'server_hostname': '1.2.3.4'}, + True) + connection = connection_mock.return_value + tls_protocol = Mock() + result = creator.clientConnectionForTLS(tls_protocol) + + assert result is connection + connection.set_tlsext_host_name.assert_called_once_with(b'1.2.3.4') + tls_app_data = getattr(tls_protocol, twistedreactor._TLS_APP_DATA_ATTR) + assert tls_app_data.expected_name == '1.2.3.4' + + def test_ssl_creator_uses_context_callback_when_connection_callback_is_unavailable(self): + class ConnectionWithoutInfoCallback(object): + def __init__(self, context, socket): + self.context = context + self.socket = socket + self.app_data = None + self.peer_cert = Mock() + + def set_app_data(self, app_data): + self.app_data = app_data + + def get_app_data(self): + return self.app_data + + def get_peer_certificate(self): + return self.peer_cert + + def set_tlsext_host_name(self, server_hostname): + self.server_hostname = server_hostname + + context = Mock() + context.get_verify_mode.return_value = twistedreactor.SSL.VERIFY_PEER + + with self.assertLogs( + 'cassandra.io.twistedreactor', level='WARNING') as logs: + with patch.object( + twistedreactor.SSL, 'Connection', + ConnectionWithoutInfoCallback): + creator = twistedreactor._SSLCreator( + DefaultEndPoint('node.example.com'), + context, + {'server_hostname': 'node.example.com'}, + True) + tls_protocol = Mock() + result = creator.clientConnectionForTLS(tls_protocol) + + assert "replacing any callback" in logs.output[0] + context.set_info_callback.assert_called_once_with(twistedreactor._SSLCreator.info_callback) + assert result.context is context + assert result.socket is None + assert result.app_data is tls_protocol + tls_app_data = getattr(tls_protocol, twistedreactor._TLS_APP_DATA_ATTR) + assert tls_app_data.endpoint == creator.endpoint + assert tls_app_data.expected_name == 'node.example.com' + assert result.server_hostname == b'node.example.com' + + def test_context_callback_uses_connection_app_data(self): + class ConnectionWithoutInfoCallback(object): + def __init__(self, context, socket): + self.context = context + self.socket = socket + self.app_data = None + self.peer_cert = Mock() + + def set_app_data(self, app_data): + self.app_data = app_data + + def get_app_data(self): + return self.app_data + + def get_peer_certificate(self): + return self.peer_cert + + def set_tlsext_host_name(self, server_hostname): + self.server_hostname = server_hostname + + context = Mock() + context.get_verify_mode.return_value = twistedreactor.SSL.VERIFY_PEER + + with patch.object(twistedreactor.SSL, 'Connection', ConnectionWithoutInfoCallback): + first_creator = twistedreactor._SSLCreator( + DefaultEndPoint('node1.example.com'), + context, + {}, + True) + first_protocol = Mock() + first_connection = first_creator.clientConnectionForTLS(first_protocol) + twistedreactor._SSLCreator( + DefaultEndPoint('node2.example.com'), + context, + {}, + True).clientConnectionForTLS(Mock()) + + assert first_connection.get_app_data() is first_protocol + callback = context.set_info_callback.call_args[0][0] + with patch.object(twistedreactor, '_validate_pyopenssl_hostname') as validate_hostname: + callback(first_connection, twistedreactor.SSL.SSL_CB_HANDSHAKE_DONE, 1) + + validate_hostname.assert_called_once_with( + first_connection.peer_cert, 'node1.example.com') + + def test_info_callback_fails_closed_on_unexpected_exception(self): + connection = Mock() + tls_protocol = Mock() + setattr(tls_protocol, twistedreactor._TLS_APP_DATA_ATTR, + twistedreactor._TLSAppData( + DefaultEndPoint('node.example.com'), 'node.example.com', True)) + connection.get_app_data.return_value = tls_protocol + connection.get_peer_certificate.return_value = object() + + with patch.object(twistedreactor, '_validate_pyopenssl_hostname', + side_effect=RuntimeError('boom')): + twistedreactor._SSLCreator.info_callback( + connection, twistedreactor.SSL.SSL_CB_HANDSHAKE_DONE, 1) + + failure = tls_protocol.failVerification.call_args[0][0] + assert isinstance(failure.value, ConnectionException) + assert "Hostname verification failed" in str(failure.value) + class TestTwistedTimer(TimerTestMixin, unittest.TestCase): """ @@ -196,3 +484,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/io/utils.py b/tests/unit/io/utils.py index f43224058c..2f46c0e744 100644 --- a/tests/unit/io/utils.py +++ b/tests/unit/io/utils.py @@ -41,6 +41,51 @@ log = logging.getLogger(__name__) +UNIT_CA_CERT = os.path.join( + os.path.dirname(__file__), 'fixtures', 'root_ca.pem') + + +def make_pyopenssl_x509_certificate(common_name, san_dns_names=None, + san_ip_addresses=None): + """Build a real pyOpenSSL X509 certificate entirely in process.""" + from datetime import datetime, timedelta, timezone + import ipaddress + + from cryptography import x509 + from cryptography.hazmat.primitives import hashes + from cryptography.hazmat.primitives.asymmetric import ec + from cryptography.x509.oid import NameOID + from OpenSSL import crypto + + key = ec.generate_private_key(ec.SECP256R1()) + subject = x509.Name([ + x509.NameAttribute(NameOID.COMMON_NAME, common_name) + ]) + now = datetime.now(timezone.utc) + builder = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(subject) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - timedelta(minutes=1)) + .not_valid_after(now + timedelta(days=1)) + ) + + subject_alt_names = [ + x509.DNSName(name) for name in (san_dns_names or []) + ] + subject_alt_names.extend( + x509.IPAddress(ipaddress.ip_address(address)) + for address in (san_ip_addresses or []) + ) + if subject_alt_names: + builder = builder.add_extension( + x509.SubjectAlternativeName(subject_alt_names), critical=False) + + certificate = builder.sign(key, hashes.SHA256()) + return crypto.X509.from_cryptography(certificate) + class TimerCallback(object): diff --git a/tests/unit/test_client_routes.py b/tests/unit/test_client_routes.py index 0aa82fc76a..8fbe708bb0 100644 --- a/tests/unit/test_client_routes.py +++ b/tests/unit/test_client_routes.py @@ -26,10 +26,20 @@ _Route, _ClientRoutesHandler ) -from cassandra.connection import ClientRoutesEndPoint, ClientRoutesEndPointFactory +from cassandra.connection import ( + ClientRoutesEndPoint, ClientRoutesEndPointFactory, Connection, + DefaultEndPoint, +) from cassandra.cluster import Cluster +class EndpointSSLOptionsEndPoint(DefaultEndPoint): + + @property + def ssl_options(self): + return {} + + class TestClientRouteProxy(unittest.TestCase): def test_endpoint_none_connection_id(self): @@ -449,6 +459,63 @@ 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_endpoint_ssl_options_enable_tls_routes(self): + config = ClientRoutesConfig( + proxies=[ClientRouteProxy(str(uuid.uuid4()), "10.0.0.1")] + ) + cluster = Cluster( + contact_points=[ + EndpointSSLOptionsEndPoint("10.0.0.1") + ], + client_routes_config=config, + ) + try: + handler = cluster._client_routes_handler + self.assertTrue(handler.ssl_enabled) + self.assertEqual(handler.endpoint_ssl_options, {}) + + host_id = uuid.uuid4() + response = Mock() + response.column_names = [ + "connection_id", "host_id", "address", "port", "tls_port" + ] + response.parsed_rows = [[ + str(config.proxies[0].connection_id), + host_id, + "route.example.com", + 9042, + 9142, + ]] + connection = Mock() + connection.wait_for_response.return_value = response + handler.initialize(connection, timeout=5.0) + self.assertEqual(handler._routes.get_by_host_id(host_id).port, 9142) + + endpoint = cluster.endpoint_factory.create({ + "host_id": host_id, + "rpc_address": "10.0.0.5", + "native_transport_port": 9042, + }) + self.assertEqual(endpoint.ssl_options, {}) + self.assertTrue(Connection(endpoint)._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) diff --git a/tests/unit/test_cloud.py b/tests/unit/test_cloud.py new file mode 100644 index 0000000000..192dba1c9e --- /dev/null +++ b/tests/unit/test_cloud.py @@ -0,0 +1,44 @@ +# Copyright 2026 ScyllaDB, 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 ssl +import sys +from types import SimpleNamespace +from unittest.mock import Mock, patch + +from cassandra.datastax import cloud + + +def test_pyopenssl_context_from_cert_configures_context(): + context = Mock() + ssl_module = SimpleNamespace() + + with patch.dict(sys.modules, {'OpenSSL': SimpleNamespace(SSL=ssl_module)}), \ + patch.object( + cloud, + '_build_pyopenssl_context_from_options', + return_value=context) as build_context: + result = cloud._pyopenssl_context_from_cert( + 'ca.crt', 'client.crt', 'client.key') + + assert result is context + build_context.assert_called_once_with( + ssl_module, + { + 'ca_certs': 'ca.crt', + 'certfile': 'client.crt', + 'keyfile': 'client.key', + 'cert_reqs': ssl.CERT_REQUIRED, + } + ) diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 3d55bc1860..8c1e75099b 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -12,10 +12,12 @@ # 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 import socket +import ssl from types import SimpleNamespace from unittest.mock import patch, Mock @@ -278,6 +280,51 @@ 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 self.assertLogs('cassandra.cluster', level='WARNING') as logs: + with pytest.warns(DeprecationWarning, match="Using ssl_options without ssl_context"): + cluster = Cluster(ssl_options={}, compression=False) + try: + assert cluster.ssl_options == {} + assert "Cluster-level ssl_options" in logs.output[0] + assert "endpoint-level SSL options may enable verification" in logs.output[0] + finally: + cluster.shutdown() + + def test_omitted_ssl_options_do_not_emit_deprecation_warning(self): + with warnings.catch_warnings(record=True) as recorded_warnings: + warnings.simplefilter("always") + cluster = Cluster(compression=False) + try: + assert not [ + warning for warning in recorded_warnings + if (warning.category is DeprecationWarning and + "Using ssl_options without ssl_context" in + str(warning.message)) + ] + finally: + cluster.shutdown() + + def test_check_hostname_warns_when_overriding_cert_none(self): + ssl_options = { + 'check_hostname': True, + 'cert_reqs': ssl.CERT_NONE, + } + with self.assertLogs('cassandra.cluster', level='WARNING') as logs: + with pytest.warns(DeprecationWarning): + cluster = Cluster( + ssl_options=ssl_options, compression=False) + try: + assert "overrides cert_reqs=CERT_NONE" in logs.output[0] + finally: + cluster.shutdown() + 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..cc4ac18e54 100644 --- a/tests/unit/test_connection.py +++ b/tests/unit/test_connection.py @@ -12,6 +12,8 @@ # See the License for the specific language governing permissions and # limitations under the License. import itertools +import socket +import ssl import unittest from io import BytesIO import time @@ -22,12 +24,11 @@ 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, SniEndPoint, ShardAwarePortGenerator) from cassandra.marshal import uint8_pack, uint32_pack, int32_pack from cassandra.protocol import (write_stringmultimap, write_int, write_string, SupportedMessage, ProtocolHandler, ResultMessage, RESULT_KIND_SET_KEYSPACE) - from tests.util import wait_until, assertRegex import pytest @@ -81,6 +82,209 @@ 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_tls_error_is_not_masked_by_later_socket_error(self): + first_socket = Mock() + second_socket = Mock() + tls_error = ssl.SSLError(1, 'certificate verify failed') + first_socket.connect.side_effect = tls_error + second_socket.connect.side_effect = socket.error( + 111, 'Connection refused') + c = Connection.__new__(Connection) + c.endpoint = DefaultEndPoint('node.example.com') + c._get_socket_addresses = Mock(return_value=[ + (socket.AF_INET, socket.SOCK_STREAM, 6, '', + ('192.0.2.1', 9042)), + (socket.AF_INET, socket.SOCK_STREAM, 6, '', + ('192.0.2.2', 9042)), + ]) + c._socket_impl = Mock() + c._socket_impl.socket.side_effect = [first_socket, second_socket] + c.ssl_context = None + c.features = Mock(shard_id=None) + c.connect_timeout = 5 + c._check_hostname = False + c.sockopts = None + + with self.assertRaises(ssl.SSLError) as raised: + c._connect_socket() + + assert raised.exception is tls_error + first_socket.close.assert_called_once_with() + second_socket.close.assert_called_once_with() + + 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_symbolic_stdlib_ssl_protocol_builds_stdlib_context(self): + c = Connection( + DefaultEndPoint('1.2.3.4'), + ssl_options={'ssl_version': 'PROTOCOL_TLS'}) + + assert c.ssl_context.protocol == ssl.PROTOCOL_TLS + + def test_check_hostname_secures_supplied_stdlib_context(self): + context = ssl.SSLContext(ssl.PROTOCOL_TLS) + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + + c = Connection( + DefaultEndPoint('node.example.com'), + ssl_context=context, + ssl_options={'check_hostname': True}) + + assert c._check_hostname + assert context.check_hostname + assert context.verify_mode == ssl.CERT_REQUIRED + + def test_empty_ssl_options_with_endpoint_sni_preserve_cert_none(self): + c = Connection( + SniEndPoint('1.2.3.4', 'node.example.com'), + ssl_options={}) + + assert c.ssl_options == {'server_hostname': 'node.example.com'} + assert c.ssl_context.verify_mode == ssl.CERT_NONE + assert not c._ssl_options_verify_by_default + + def test_empty_ssl_options_with_endpoint_ca_default_to_cert_required(self): + endpoint = SniEndPoint('1.2.3.4', 'node.example.com') + endpoint._ssl_options = {'ca_certs': 'endpoint-ca.pem'} + context = Mock() + + with patch('cassandra.tls.ssl.SSLContext', return_value=context): + c = Connection(endpoint, ssl_options={}) + + assert c.ssl_options == {'ca_certs': 'endpoint-ca.pem'} + assert c._ssl_options_verify_by_default + assert context.verify_mode == ssl.CERT_REQUIRED + context.load_verify_locations.assert_called_once_with( + 'endpoint-ca.pem') + + def test_omitted_ssl_options_with_endpoint_sni_default_to_cert_required(self): + c = Connection(SniEndPoint('1.2.3.4', 'node.example.com')) + + assert c.ssl_options == {'server_hostname': 'node.example.com'} + assert c.ssl_context.verify_mode == ssl.CERT_REQUIRED + assert c._ssl_options_verify_by_default + + def test_explicit_verification_options_override_origin_default(self): + endpoint = SniEndPoint('1.2.3.4', 'endpoint.example.com') + + c = Connection( + endpoint, + ssl_options={ + 'server_hostname': 'caller.example.com', + 'cert_reqs': ssl.CERT_NONE, + 'check_hostname': False, + }) + + assert c.ssl_options['server_hostname'] == 'endpoint.example.com' + assert c.ssl_context.verify_mode == ssl.CERT_NONE + assert not c.ssl_context.check_hostname + + def test_endpoint_check_hostname_overrides_disabled_origin_default(self): + endpoint = SniEndPoint('1.2.3.4', 'endpoint.example.com') + endpoint._ssl_options['check_hostname'] = True + + c = Connection(endpoint, ssl_options={}) + + assert c.ssl_context.verify_mode == ssl.CERT_REQUIRED + assert c.ssl_context.check_hostname + + def test_endpoint_context_options_rejected_for_supplied_context(self): + endpoint = SniEndPoint('1.2.3.4', 'endpoint.example.com') + endpoint._ssl_options.update({ + 'ca_certs': 'endpoint-ca.pem', + 'cert_reqs': ssl.CERT_REQUIRED, + 'check_hostname': True, + }) + context = Mock(spec=ssl.SSLContext) + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + + with pytest.raises(ValueError, match="independent SSL context"): + Connection(endpoint, ssl_context=context) + + assert context.verify_mode == ssl.CERT_NONE + assert not context.check_hostname + context.load_verify_locations.assert_not_called() + + def test_endpoint_sni_does_not_mutate_supplied_context(self): + endpoint = SniEndPoint('1.2.3.4', 'endpoint.example.com') + context = Mock(spec=ssl.SSLContext) + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + + c = Connection(endpoint, ssl_context=context) + + assert c.ssl_context is context + assert c.ssl_options == { + 'server_hostname': 'endpoint.example.com'} + assert context.verify_mode == ssl.CERT_NONE + assert not context.check_hostname + context.load_verify_locations.assert_not_called() + context.load_cert_chain.assert_not_called() + context.set_ciphers.assert_not_called() + + def test_endpoint_option_does_not_promote_explicit_cert_none(self): + endpoint = SniEndPoint('1.2.3.4', 'endpoint.example.com') + endpoint._ssl_options = {'ciphers': 'DEFAULT'} + context = Mock(spec=ssl.SSLContext) + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + + with pytest.raises(ValueError, match="independent SSL context"): + Connection( + endpoint, + ssl_context=context, + ssl_options={'cert_reqs': ssl.CERT_NONE}) + + assert context.verify_mode == ssl.CERT_NONE + assert not context.check_hostname + context.set_ciphers.assert_not_called() + + def test_non_empty_ssl_options_default_to_cert_required(self): + for ssl_options in ({'server_hostname': 'node.example.com'}, {'ciphers': 'DEFAULT'}): + with self.subTest(ssl_options=ssl_options): + c = Connection(DefaultEndPoint('1.2.3.4'), ssl_options=ssl_options) + + assert isinstance(c.ssl_context, ssl.SSLContext) + assert c.ssl_context.verify_mode == ssl.CERT_REQUIRED + + 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 test_ssl_options_check_hostname_promotes_cert_none_to_cert_required(self): + c = Connection( + DefaultEndPoint('1.2.3.4'), + ssl_options={'cert_reqs': ssl.CERT_NONE, '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 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..c9e3e02516 100644 --- a/tests/unit/test_shard_aware.py +++ b/tests/unit/test_shard_aware.py @@ -72,7 +72,22 @@ def mock_connection_factory(self, *args, **kwargs): return connection +class EndpointSSLOptionsEndPoint(DefaultEndPoint): + @property + def ssl_options(self): + return {} + + 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 +115,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 +142,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, @@ -153,6 +166,40 @@ def test_ssl_advanced_shard_aware_port_requires_ssl_port(self): finally: session.cluster.executor.shutdown(wait=True) + def test_endpoint_ssl_options_use_ssl_advanced_shard_aware_port(self): + host = self.make_host() + host.endpoint = EndpointSSLOptionsEndPoint("1.2.3.4") + session = MockSession() + pool = HostConnection(host=host, host_distance=HostDistance.REMOTE, session=session) + + try: + for f in session.futures: + f.result() + + endpoint = pool._get_shard_aware_endpoint() + assert endpoint is not None + assert endpoint.port == 19045 + finally: + session.cluster.executor.shutdown(wait=True) + + def test_endpoint_ssl_options_do_not_use_plaintext_advanced_shard_aware_port(self): + host = self.make_host(shard_aware_port=19042, shard_aware_port_ssl=None) + host.endpoint = EndpointSSLOptionsEndPoint("1.2.3.4") + sharding_info = ShardingInfo( + shard_id=1, shards_count=4, partitioner="", sharding_algorithm="", + sharding_ignore_msb=0, shard_aware_port=19042, + shard_aware_port_ssl=None) + session = MockSession(sharding_info=sharding_info) + pool = HostConnection(host=host, host_distance=HostDistance.REMOTE, session=session) + + try: + for f in session.futures: + f.result() + + assert pool._get_shard_aware_endpoint() is None + finally: + session.cluster.executor.shutdown(wait=True) + def test_advanced_shard_aware_cooldown(self): """ `disable_advanced_shard_aware` must suppress the shard-aware endpoint for @@ -160,8 +207,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) diff --git a/tests/unit/test_tls.py b/tests/unit/test_tls.py new file mode 100644 index 0000000000..569b9685db --- /dev/null +++ b/tests/unit/test_tls.py @@ -0,0 +1,568 @@ +# 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 ssl +from types import SimpleNamespace +import unittest +from unittest.mock import Mock, patch + +import pytest + +from cassandra.tls import ( + _build_ssl_context_from_options, + _build_pyopenssl_context_from_options, + _default_pyopenssl_ssl_method, + _dnsname_match, + _encode_server_hostname, + _ensure_pyopenssl_context_requires_verification, + _pyopenssl_ssl_method_from_stdlib, + _resolve_pyopenssl_server_names, + _validate_pyopenssl_hostname, +) +from tests.unit.io.utils import make_pyopenssl_x509_certificate + + +try: + from OpenSSL import crypto as openssl_crypto +except (ImportError, AttributeError): + openssl_crypto = None + + +class FakePyOpenSSLContext(object): + def __init__(self, method): + self.method = method + self.verify_mode = None + self.default_verify_paths_loaded = False + self.verify_locations = [] + self.set_verify_calls = [] + self.certificate_chain_files = [] + self.certificate_files = [] + self.privatekey_files = [] + self.cipher_lists = [] + + def set_verify(self, verify_mode, callback): + self.verify_mode = verify_mode + self.verify_callback = callback + self.set_verify_calls.append((verify_mode, callback)) + + def get_verify_mode(self): + return self.verify_mode + + def set_default_verify_paths(self): + self.default_verify_paths_loaded = True + + def load_verify_locations(self, path): + self.verify_locations.append(path) + + def use_certificate_chain_file(self, path): + self.certificate_chain_files.append(path) + + def use_certificate_file(self, path): + self.certificate_files.append(path) + + def use_privatekey_file(self, path): + self.privatekey_files.append(path) + + def set_cipher_list(self, ciphers): + self.cipher_lists.append(ciphers) + + +class FakePyOpenSSLContextWithoutChainFile(FakePyOpenSSLContext): + use_certificate_chain_file = None + + +class FakePyOpenSSLContextWithoutGetVerifyMode(object): + def __init__(self, method): + self.method = method + self.verify_mode = None + + def set_verify(self, verify_mode, callback): + self.verify_mode = verify_mode + self.verify_callback = callback + + +class FakePyOpenSSLModule(object): + TLS_CLIENT_METHOD = object() + VERIFY_NONE = 0 + VERIFY_PEER = 1 + Context = FakePyOpenSSLContext + + +class PyOpenSSLMethodTest(unittest.TestCase): + + @staticmethod + def numeric_ssl_module(): + return SimpleNamespace( + SSLv23_METHOD=3, + TLSv1_METHOD=4, + TLSv1_1_METHOD=5, + TLSv1_2_METHOD=6, + TLS_METHOD=7, + TLS_CLIENT_METHOD=9, + ) + + def test_prefers_tls_client_method(self): + 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_falls_back_to_tls_method(self): + 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_falls_back_to_tlsv1_2_method(self): + 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_requires_secure_method(self): + with pytest.raises(ImportError, match="secure TLS client method"): + _default_pyopenssl_ssl_method(SimpleNamespace()) + + def test_maps_stdlib_protocol_enum(self): + ssl_module = self.numeric_ssl_module() + + method = _pyopenssl_ssl_method_from_stdlib( + ssl_module, ssl.PROTOCOL_TLSv1_2) + + assert method == ssl_module.TLSv1_2_METHOD + + def test_maps_unambiguous_serialized_stdlib_protocol(self): + ssl_module = self.numeric_ssl_module() + + method = _pyopenssl_ssl_method_from_stdlib( + ssl_module, int(ssl.PROTOCOL_TLS)) + + assert method == ssl_module.TLS_METHOD + + def test_rejects_ambiguous_serialized_protocol(self): + ssl_module = self.numeric_ssl_module() + + with pytest.raises( + ValueError, + match=r"ambiguous.*PROTOCOL_TLSv1_2.*TLSv1_1_METHOD"): + _pyopenssl_ssl_method_from_stdlib( + ssl_module, int(ssl.PROTOCOL_TLSv1_2)) + + def test_preserves_unambiguous_pyopenssl_method_integer(self): + ssl_module = self.numeric_ssl_module() + + method = _pyopenssl_ssl_method_from_stdlib( + ssl_module, ssl_module.TLSv1_2_METHOD) + + assert method == ssl_module.TLSv1_2_METHOD + + def test_maps_symbolic_stdlib_protocol(self): + ssl_module = self.numeric_ssl_module() + + method = _pyopenssl_ssl_method_from_stdlib( + ssl_module, 'PROTOCOL_TLSv1_2') + + assert method == ssl_module.TLSv1_2_METHOD + + def test_rejects_unknown_symbolic_protocol(self): + with pytest.raises( + ValueError, match="unknown symbolic stdlib SSL protocol"): + _pyopenssl_ssl_method_from_stdlib( + self.numeric_ssl_module(), 'PROTOCOL_TLSv9') + + def test_rejects_server_protocol(self): + with pytest.raises( + ValueError, match="not supported for client connections"): + _pyopenssl_ssl_method_from_stdlib( + self.numeric_ssl_module(), 'PROTOCOL_TLS_SERVER') + + def test_rejects_unknown_protocol_integer(self): + with pytest.raises( + ValueError, match="does not identify a protocol"): + _pyopenssl_ssl_method_from_stdlib( + self.numeric_ssl_module(), 999) + + +class PyOpenSSLServerNamesTest(unittest.TestCase): + + def test_explicit_server_name_is_used_for_sni_and_verification(self): + assert _resolve_pyopenssl_server_names( + '10.0.0.1', 'node.example.com', True) == ( + 'node.example.com', 'node.example.com') + + def test_sni_proxy_authenticates_endpoint_address(self): + assert _resolve_pyopenssl_server_names( + 'proxy.example.com', + 'host-id', + True, + verify_endpoint_address=True, + ) == ('host-id', 'proxy.example.com') + + def test_dns_endpoint_uses_implicit_sni_when_verifying(self): + assert _resolve_pyopenssl_server_names( + 'node.example.com', None, True) == ( + 'node.example.com', 'node.example.com') + + def test_ip_endpoint_omits_implicit_sni(self): + assert _resolve_pyopenssl_server_names( + '1.2.3.4', None, True) == (None, '1.2.3.4') + + +@unittest.skipIf(openssl_crypto is None, "pyOpenSSL is not available") +class PyOpenSSLHostnameValidationTest(unittest.TestCase): + + def test_dnsname_match_normalizes_unicode_hostname_to_idna(self): + assert _dnsname_match( + 'xn--tst-qla.example', 'täst.example') + + def test_dnsname_match_does_not_apply_idna2003_aliases(self): + assert not _dnsname_match('fass.de', 'faß.de') + + def test_server_hostname_encoding_rejects_idna2003_aliases(self): + with self.assertRaises(ValueError): + _encode_server_hostname('faß.de') + + def test_server_hostname_encoding_preserves_safe_idna(self): + assert (_encode_server_hostname('täst.example') == + b'xn--tst-qla.example') + + def test_dnsname_match_rejects_malformed_certificate_trailing_dots(self): + assert not _dnsname_match( + 'node.example.com..', 'node.example.com') + + def test_rejects_sole_wildcard_subject_alt_name(self): + cert = make_pyopenssl_x509_certificate( + 'unused', san_dns_names=['*']) + + with self.assertRaises(ssl.CertificateError): + _validate_pyopenssl_hostname(cert, 'node1') + + def test_ip_hostname_requires_ip_subject_alt_name(self): + cert = make_pyopenssl_x509_certificate( + 'unused', san_dns_names=['*.2.3.4']) + + with self.assertRaises(ssl.CertificateError): + _validate_pyopenssl_hostname(cert, '1.2.3.4') + + def test_ip_hostname_accepts_exact_ip_subject_alt_name(self): + cert = make_pyopenssl_x509_certificate( + 'unused', san_ip_addresses=['1.2.3.4']) + + _validate_pyopenssl_hostname(cert, '1.2.3.4') + + def test_ip_hostname_does_not_fall_back_to_common_name(self): + cert = make_pyopenssl_x509_certificate('1.2.3.4') + + with self.assertRaises(ssl.CertificateError): + _validate_pyopenssl_hostname(cert, '1.2.3.4') + + def test_dns_hostname_falls_back_to_common_name_with_only_ip_san(self): + cert = make_pyopenssl_x509_certificate( + 'node.example.com', san_ip_addresses=['1.2.3.4']) + + _validate_pyopenssl_hostname(cert, 'node.example.com') + + def test_dns_hostname_rejects_malformed_san_trailing_dots(self): + cert = make_pyopenssl_x509_certificate( + 'unused', san_dns_names=['node.example.com..']) + + with self.assertRaises(ssl.CertificateError): + _validate_pyopenssl_hostname(cert, 'node.example.com') + + def test_dns_hostname_accepts_subject_alt_name_on_real_x509(self): + cert = make_pyopenssl_x509_certificate( + 'wrong.example.com', san_dns_names=['node.example.com']) + + assert isinstance(cert, openssl_crypto.X509) + _validate_pyopenssl_hostname(cert, 'node.example.com') + + def test_rejects_missing_certificate(self): + with self.assertRaises(ssl.CertificateError): + _validate_pyopenssl_hostname(None, 'node.example.com') + + +class PyOpenSSLContextTest(unittest.TestCase): + + def test_empty_ssl_options_default_to_verify_none(self): + context = _build_pyopenssl_context_from_options( + FakePyOpenSSLModule, {}) + + assert context.method is FakePyOpenSSLModule.TLS_CLIENT_METHOD + assert context.verify_mode == FakePyOpenSSLModule.VERIFY_NONE + assert not context.default_verify_paths_loaded + + def test_stdlib_ssl_version_is_translated(self): + ssl_module = PyOpenSSLMethodTest.numeric_ssl_module() + ssl_module.VERIFY_NONE = 0 + ssl_module.VERIFY_PEER = 1 + ssl_module.Context = FakePyOpenSSLContext + + context = _build_pyopenssl_context_from_options( + ssl_module, {'ssl_version': ssl.PROTOCOL_TLS}) + + assert context.method == ssl_module.TLS_METHOD + + def test_ciphers_option_is_applied(self): + context = _build_pyopenssl_context_from_options( + FakePyOpenSSLModule, {'ciphers': 'ECDHE+AESGCM'}) + + assert context.cipher_lists == [b'ECDHE+AESGCM'] + + def test_non_empty_ssl_options_default_to_verify_peer(self): + context = _build_pyopenssl_context_from_options( + FakePyOpenSSLModule, + {'server_hostname': 'node.example.com'}) + + assert context.verify_mode == FakePyOpenSSLModule.VERIFY_PEER + assert context.default_verify_paths_loaded + + def test_default_verification_can_ignore_merged_endpoint_options(self): + context = _build_pyopenssl_context_from_options( + FakePyOpenSSLModule, + {'server_hostname': 'node.example.com'}, + verify_by_default=False) + + assert context.verify_mode == FakePyOpenSSLModule.VERIFY_NONE + assert not context.default_verify_paths_loaded + + def test_check_hostname_overrides_disabled_default_verification(self): + context = _build_pyopenssl_context_from_options( + FakePyOpenSSLModule, + {'check_hostname': True}, + verify_by_default=False) + + assert context.verify_mode == FakePyOpenSSLModule.VERIFY_PEER + assert context.default_verify_paths_loaded + + def test_explicit_ca_certs_do_not_load_default_verify_paths(self): + context = _build_pyopenssl_context_from_options( + FakePyOpenSSLModule, + {'ca_certs': 'ca.pem'}) + + assert context.verify_locations == ['ca.pem'] + assert not context.default_verify_paths_loaded + + def test_verify_peer_without_default_paths_api_is_supported(self): + ssl_module = SimpleNamespace( + TLS_CLIENT_METHOD=object(), + VERIFY_NONE=0, + VERIFY_PEER=1, + Context=FakePyOpenSSLContextWithoutGetVerifyMode) + + context = _build_pyopenssl_context_from_options( + ssl_module, {'server_hostname': 'node.example.com'}) + + assert context.verify_mode == ssl_module.VERIFY_PEER + + def test_verify_peer_bitmask_loads_default_verify_paths(self): + verify_peer_with_flags = FakePyOpenSSLModule.VERIFY_PEER | 4 + context = _build_pyopenssl_context_from_options( + FakePyOpenSSLModule, + {'cert_reqs': verify_peer_with_flags}) + + assert context.verify_mode == verify_peer_with_flags + assert context.default_verify_paths_loaded + + def test_check_hostname_promotes_verify_none_to_verify_peer(self): + context = _build_pyopenssl_context_from_options( + FakePyOpenSSLModule, + {'cert_reqs': ssl.CERT_NONE, 'check_hostname': True}) + + assert context.verify_mode == FakePyOpenSSLModule.VERIFY_PEER + + def test_check_hostname_preserves_flags_while_promoting_verify_peer(self): + context = _build_pyopenssl_context_from_options( + FakePyOpenSSLModule, + {'cert_reqs': 4, 'check_hostname': True}) + + assert context.verify_mode == (4 | FakePyOpenSSLModule.VERIFY_PEER) + + def test_plain_int_cert_reqs_option_is_translated(self): + context = _build_pyopenssl_context_from_options( + FakePyOpenSSLModule, + {'cert_reqs': int(ssl.CERT_REQUIRED)}) + + assert context.verify_mode == FakePyOpenSSLModule.VERIFY_PEER + + def test_loads_client_certificate_chain_and_key(self): + context = _build_pyopenssl_context_from_options( + FakePyOpenSSLModule, + { + 'certfile': 'client-chain.pem', + 'keyfile': 'client-key.pem', + 'cert_reqs': ssl.CERT_NONE, + }) + + assert context.certificate_chain_files == ['client-chain.pem'] + assert context.certificate_files == [] + assert context.privatekey_files == ['client-key.pem'] + + def test_combined_client_certificate_and_key_file(self): + context = _build_pyopenssl_context_from_options( + FakePyOpenSSLModule, + { + 'certfile': 'combined.pem', + 'cert_reqs': ssl.CERT_NONE, + }) + + assert context.certificate_chain_files == ['combined.pem'] + assert context.privatekey_files == ['combined.pem'] + + def test_falls_back_when_certificate_chain_api_is_unavailable(self): + ssl_module = SimpleNamespace( + TLS_CLIENT_METHOD=object(), + VERIFY_NONE=0, + VERIFY_PEER=1, + Context=FakePyOpenSSLContextWithoutChainFile) + + context = _build_pyopenssl_context_from_options( + ssl_module, + { + 'certfile': 'combined.pem', + 'cert_reqs': ssl.CERT_NONE, + }) + + assert context.certificate_chain_files == [] + assert context.certificate_files == ['combined.pem'] + assert context.privatekey_files == ['combined.pem'] + + def test_ignores_falsey_legacy_file_options(self): + for falsey in (None, ''): + with self.subTest(falsey=falsey): + context = _build_pyopenssl_context_from_options( + FakePyOpenSSLModule, + { + 'certfile': falsey, + 'keyfile': 'unused-key.pem', + 'ca_certs': falsey, + }) + + assert context.certificate_chain_files == [] + assert context.certificate_files == [] + assert context.privatekey_files == [] + assert context.verify_locations == [] + + def test_supplied_context_check_hostname_promotes_verify_none_to_verify_peer(self): + context = FakePyOpenSSLContext( + FakePyOpenSSLModule.TLS_CLIENT_METHOD) + context.verify_mode = FakePyOpenSSLModule.VERIFY_NONE + + _ensure_pyopenssl_context_requires_verification( + FakePyOpenSSLModule, context, True) + + assert context.verify_mode == FakePyOpenSSLModule.VERIFY_PEER + + def test_supplied_context_without_check_hostname_preserves_verify_none(self): + context = FakePyOpenSSLContext( + FakePyOpenSSLModule.TLS_CLIENT_METHOD) + context.verify_mode = FakePyOpenSSLModule.VERIFY_NONE + + _ensure_pyopenssl_context_requires_verification( + FakePyOpenSSLModule, context, False) + + assert context.verify_mode == FakePyOpenSSLModule.VERIFY_NONE + + def test_supplied_context_with_verify_peer_preserves_custom_callback(self): + context = FakePyOpenSSLContext( + FakePyOpenSSLModule.TLS_CLIENT_METHOD) + custom_callback = Mock() + secure_mode = FakePyOpenSSLModule.VERIFY_PEER | 4 + context.set_verify(secure_mode, custom_callback) + + _ensure_pyopenssl_context_requires_verification( + FakePyOpenSSLModule, context, True) + + assert context.verify_callback is custom_callback + assert context.set_verify_calls == [ + (secure_mode, custom_callback)] + + def test_supplied_context_without_verify_peer_flag_is_promoted(self): + context = FakePyOpenSSLContext( + FakePyOpenSSLModule.TLS_CLIENT_METHOD) + context.set_verify(4, Mock()) + + with self.assertLogs('cassandra.tls', level='WARNING'): + _ensure_pyopenssl_context_requires_verification( + FakePyOpenSSLModule, context, True) + + assert context.verify_mode == (4 | FakePyOpenSSLModule.VERIFY_PEER) + + def test_supplied_context_without_get_verify_mode_promotes_to_verify_peer(self): + context = FakePyOpenSSLContextWithoutGetVerifyMode( + FakePyOpenSSLModule.TLS_CLIENT_METHOD) + + with self.assertLogs('cassandra.tls', level='WARNING') as logs: + _ensure_pyopenssl_context_requires_verification( + FakePyOpenSSLModule, context, True) + + assert context.verify_mode == FakePyOpenSSLModule.VERIFY_PEER + assert "replacing its verification callback" in logs.output[0] + + +class StdlibSSLContextTest(unittest.TestCase): + + def test_non_empty_options_load_system_default_ca_certs(self): + context = Mock() + + with patch('cassandra.tls.ssl.SSLContext', return_value=context): + result = _build_ssl_context_from_options( + {'server_hostname': 'node.example.com'}) + + assert result is context + context.load_default_certs.assert_called_once_with() + context.load_verify_locations.assert_not_called() + + def test_explicit_ca_certs_do_not_load_system_defaults(self): + context = Mock() + + with patch('cassandra.tls.ssl.SSLContext', return_value=context): + result = _build_ssl_context_from_options( + {'ca_certs': 'ca.pem'}) + + assert result is context + context.load_verify_locations.assert_called_once_with('ca.pem') + context.load_default_certs.assert_not_called() + + def test_falsey_legacy_file_options_are_ignored(self): + context = Mock() + + with patch('cassandra.tls.ssl.SSLContext', return_value=context): + result = _build_ssl_context_from_options({ + 'certfile': '', + 'keyfile': '', + 'ca_certs': '', + 'cert_reqs': ssl.CERT_NONE, + }) + + assert result is context + context.load_cert_chain.assert_not_called() + context.load_verify_locations.assert_not_called() + + def test_falsey_keyfile_treats_certfile_as_combined_pem(self): + context = Mock() + + with patch('cassandra.tls.ssl.SSLContext', return_value=context): + result = _build_ssl_context_from_options({ + 'certfile': 'combined.pem', + 'keyfile': '', + 'cert_reqs': ssl.CERT_NONE, + }) + + assert result is context + context.load_cert_chain.assert_called_once_with('combined.pem', None)