diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 12ade2018f..c4402b7dcf 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -1298,7 +1298,8 @@ def __init__(self, if cloud is not None: self.cloud = cloud - if contact_points is not _NOT_SET or endpoint_factory or ssl_context or ssl_options: + if (contact_points is not _NOT_SET or endpoint_factory or + ssl_context is not None or ssl_options is not None): raise ValueError("contact_points, endpoint_factory, ssl_context, and ssl_options " "cannot be specified with a cloud configuration") @@ -1508,7 +1509,7 @@ def __init__(self, self.metrics_enabled = metrics_enabled - if ssl_options and not ssl_context: + if ssl_options is not None and ssl_context is None: warn('Using ssl_options without ssl_context is ' 'deprecated and will result in an error in ' 'the next major release. Please use ssl_context ' diff --git a/cassandra/connection.py b/cassandra/connection.py index f238416b29..23102f6417 100644 --- a/cassandra/connection.py +++ b/cassandra/connection.py @@ -17,6 +17,7 @@ import errno from functools import wraps, partial, total_ordering from heapq import heappush, heappop +import ipaddress import io import logging import socket @@ -55,6 +56,7 @@ log = logging.getLogger(__name__) + segment_codec_no_compression = SegmentCodec() segment_codec_lz4 = None @@ -130,6 +132,185 @@ def decompress(byts): frame_header_v3 = struct.Struct('>BhBi') +def _default_pyopenssl_ssl_method(ssl_module): + for method_name in ('TLS_CLIENT_METHOD', 'TLS_METHOD', 'TLSv1_2_METHOD'): + method = getattr(ssl_module, method_name, None) + if method is not None: + return method + raise ImportError('pyOpenSSL does not expose a secure TLS client method') + + +def _pyopenssl_ssl_method_from_stdlib(ssl_module, ssl_version): + if ssl_version is None: + return _default_pyopenssl_ssl_method(ssl_module) + + protocol_method_names = ( + ('PROTOCOL_TLS_CLIENT', ('TLS_CLIENT_METHOD', 'TLS_METHOD', 'TLSv1_2_METHOD')), + ('PROTOCOL_TLS', ('TLS_METHOD', 'TLS_CLIENT_METHOD', 'TLSv1_2_METHOD')), + ('PROTOCOL_SSLv23', ('TLS_METHOD', 'TLS_CLIENT_METHOD', 'TLSv1_2_METHOD')), + ('PROTOCOL_TLSv1_2', ('TLSv1_2_METHOD',)), + ('PROTOCOL_TLSv1_1', ('TLSv1_1_METHOD',)), + ('PROTOCOL_TLSv1', ('TLSv1_METHOD',)), + ) + for protocol_name, method_names in protocol_method_names: + protocol = getattr(ssl, protocol_name, None) + if (protocol is not None and + ssl_version.__class__ is protocol.__class__ and + ssl_version == protocol): + for method_name in method_names: + method = getattr(ssl_module, method_name, None) + if method is not None: + return method + raise ImportError('pyOpenSSL does not expose a method for %s' % (protocol_name,)) + + return ssl_version + + +def _pyopenssl_verify_mode_from_cert_reqs(ssl_module, cert_reqs): + if cert_reqs is None: + return None + if cert_reqs.__class__ is not type(ssl.CERT_REQUIRED): + return cert_reqs + if cert_reqs == ssl.CERT_NONE: + return ssl_module.VERIFY_NONE + if cert_reqs in (ssl.CERT_OPTIONAL, ssl.CERT_REQUIRED): + return ssl_module.VERIFY_PEER + return cert_reqs + + +def _build_pyopenssl_context_from_options(ssl_module, ssl_options): + ssl_options = ssl_options or {} + context = ssl_module.Context( + _pyopenssl_ssl_method_from_stdlib(ssl_module, ssl_options.get('ssl_version', None)) + ) + if 'certfile' in ssl_options: + context.use_certificate_file(ssl_options['certfile']) + if 'keyfile' in ssl_options: + context.use_privatekey_file(ssl_options['keyfile']) + if 'ca_certs' in ssl_options: + context.load_verify_locations(ssl_options['ca_certs']) + cert_reqs = _pyopenssl_verify_mode_from_cert_reqs( + ssl_module, ssl_options.get('cert_reqs', None)) + if cert_reqs is None: + cert_reqs = (ssl_module.VERIFY_PEER + if (ssl_options.get('ca_certs', None) or ssl_options.get('check_hostname', False)) + else ssl_module.VERIFY_NONE) + elif ssl_options.get('check_hostname', False) and cert_reqs == ssl_module.VERIFY_NONE: + cert_reqs = ssl_module.VERIFY_PEER + context.set_verify( + cert_reqs, + callback=lambda _connection, _x509, _errnum, _errdepth, ok: ok + ) + ciphers = ssl_options.get('ciphers', None) + if ciphers: + if isinstance(ciphers, str): + ciphers = ciphers.encode('ascii') + context.set_cipher_list(ciphers) + return context + + +def _normalized_hostname(hostname): + return hostname.rstrip('.').lower() + + +def _dnsname_match(dn, hostname): + dn = _normalized_hostname(dn) + hostname = _normalized_hostname(hostname) + + if '*' not in dn: + return dn == hostname + + dn_labels = dn.split('.') + hostname_labels = hostname.split('.') + if (len(dn_labels) != len(hostname_labels) or + not dn_labels or dn_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 _decode_x509_name(value): + if isinstance(value, bytes): + return value.decode('utf-8') + return value + + +def _pyopenssl_cert_subject_alt_names(cert): + dns_names = [] + ip_addresses = [] + + for i in range(cert.get_extension_count()): + extension = cert.get_extension(i) + if extension.get_short_name() != b'subjectAltName': + continue + for item in str(extension).split(','): + item = item.strip() + if item.startswith('DNS:'): + dns_names.append(item[4:]) + elif item.startswith('IP Address:'): + ip_addresses.append(item[11:]) + + return dns_names, ip_addresses + + +def _pyopenssl_cert_common_names(cert): + return [ + _decode_x509_name(value) + for key, value in cert.get_subject().get_components() + if key == b'CN' + ] + + +def _validate_pyopenssl_hostname(cert, hostname): + san_dns_names, san_ip_addresses = _pyopenssl_cert_subject_alt_names(cert) + san_names = san_dns_names + san_ip_addresses + hostname_is_ip = _is_ip_address(hostname) + + for cert_ip in san_ip_addresses: + if _ipaddress_match(cert_ip, hostname): + return + if hostname_is_ip and san_names: + raise ssl.CertificateError( + "hostname %r doesn't match certificate subjectAltName %r" % + (hostname, san_names)) + + for cert_hostname in san_dns_names: + if _dnsname_match(cert_hostname, hostname): + return + if san_names: + raise ssl.CertificateError( + "hostname %r doesn't match certificate subjectAltName %r" % + (hostname, san_names)) + + common_names = _pyopenssl_cert_common_names(cert) + for common_name in common_names: + if (_ipaddress_match(common_name, hostname) if hostname_is_ip + else _dnsname_match(common_name, hostname)): + return + + raise ssl.CertificateError( + "hostname %r doesn't match certificate commonName %r" % + (hostname, common_names)) + + class EndPoint(object): """ Represents the information to connect to a cassandra node. @@ -803,6 +984,7 @@ class Connection(object): endpoint = None ssl_options = None ssl_context = None + _ssl_options_explicit = False last_error = None # The current number of operations that are in flight. More precisely, @@ -885,7 +1067,11 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None, self.endpoint = host if isinstance(host, EndPoint) else DefaultEndPoint(host, port) self.authenticator = authenticator - self.ssl_options = ssl_options.copy() if ssl_options else {} + endpoint_ssl_options = self.endpoint.ssl_options + # Explicit ssl_options={} enables SSL with default options; omitted + # ssl_options=None leaves SSL disabled unless an endpoint supplies options. + self._ssl_options_explicit = ssl_options is not None + self.ssl_options = ssl_options.copy() if ssl_options is not None else {} self.ssl_context = ssl_context self.sockopts = sockopts self.compression = compression @@ -905,10 +1091,13 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None, self._on_orphaned_stream_released = on_orphaned_stream_released self._application_info = application_info - if ssl_options: - self.ssl_options.update(self.endpoint.ssl_options or {}) - elif self.endpoint.ssl_options: - self.ssl_options = self.endpoint.ssl_options + if ssl_options is not None: + self.ssl_options.update(endpoint_ssl_options or {}) + elif endpoint_ssl_options is not None: + self._ssl_options_explicit = True + self.ssl_options = endpoint_ssl_options + self._check_hostname = bool(self.ssl_options.get('check_hostname', False) or + getattr(self.ssl_context, 'check_hostname', False)) # PYTHON-1331 # @@ -918,7 +1107,7 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None, # # Note the use of pop() here; we are very deliberately removing these params from ssl_options if they're present. After this # operation ssl_options should contain only args needed for the ssl_context.wrap_socket() call. - if not self.ssl_context and self.ssl_options: + if self.ssl_context is None and self._ssl_options_explicit: self.ssl_context = self._build_ssl_context_from_options() self.max_request_id = min(self.max_in_flight - 1, (2 ** 15) - 1) @@ -942,6 +1131,10 @@ def host(self): def port(self): return self.endpoint.port + @property + def _ssl_enabled(self): + return self.ssl_context is not None or self._ssl_options_explicit + @classmethod def initialize_reactor(cls): """ @@ -1000,10 +1193,15 @@ def _build_ssl_context_from_options(self): # Python >= 3.10 requires either PROTOCOL_TLS_CLIENT or PROTOCOL_TLS_SERVER so we'll get ahead of things by always # being explicit ssl_version = opts.get('ssl_version', None) or ssl.PROTOCOL_TLS_CLIENT - cert_reqs = opts.get('cert_reqs', None) or ssl.CERT_REQUIRED + cert_reqs = opts.get('cert_reqs', None) + if cert_reqs is None: + cert_reqs = (ssl.CERT_REQUIRED + if (opts.get('ca_certs', None) or opts.get('check_hostname', False)) + else ssl.CERT_NONE) rv = ssl.SSLContext(protocol=int(ssl_version)) + rv.check_hostname = False + rv.verify_mode = cert_reqs rv.check_hostname = bool(opts.get('check_hostname', False)) - rv.options = int(cert_reqs) certfile = opts.get('certfile', None) keyfile = opts.get('keyfile', None) diff --git a/cassandra/datastax/cloud/__init__.py b/cassandra/datastax/cloud/__init__.py index be79d6db38..b74f62beed 100644 --- a/cassandra/datastax/cloud/__init__.py +++ b/cassandra/datastax/cloud/__init__.py @@ -34,6 +34,7 @@ from zipfile import BadZipfile as BadZipFile from cassandra import DriverException +from cassandra.connection import _default_pyopenssl_ssl_method log = logging.getLogger(__name__) @@ -181,12 +182,12 @@ 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) + "PyOpenSSL must be installed to connect to Astra with the Eventlet or Twisted event loops" + ) from e + ssl_context = SSL.Context(_default_pyopenssl_ssl_method(SSL)) ssl_context.set_verify(SSL.VERIFY_PEER, callback=lambda _1, _2, _3, _4, ok: ok) ssl_context.use_certificate_file(cert_location) ssl_context.use_privatekey_file(key_location) ssl_context.load_verify_locations(ca_cert_location) - return ssl_context \ No newline at end of file + return ssl_context diff --git a/cassandra/datastax/insights/reporter.py b/cassandra/datastax/insights/reporter.py index 83205fc458..1877af4dd8 100644 --- a/cassandra/datastax/insights/reporter.py +++ b/cassandra/datastax/insights/reporter.py @@ -33,6 +33,28 @@ log = logging.getLogger(__name__) +def _ssl_options_cert_validation_enabled(ssl_options): + cert_reqs = ssl_options.get('cert_reqs') + if cert_reqs is not None: + return cert_reqs != ssl.CERT_NONE + return bool(ssl_options.get('ca_certs') or ssl_options.get('check_hostname', False)) + + +def _safe_getattr(obj, name, default=None): + try: + return object.__getattribute__(obj, name) + except AttributeError: + return default + + +def _ssl_context_cert_validation_enabled(ssl_context): + if isinstance(ssl_context, ssl.SSLContext): + return ssl_context.verify_mode != ssl.CERT_NONE + + from OpenSSL import SSL + return ssl_context.get_verify_mode() != SSL.VERIFY_NONE + + class MonitorReporter(Thread): def __init__(self, interval_sec, session): @@ -139,16 +161,32 @@ def _get_startup_data(self): except AttributeError: compression_type = 'NONE' + connection = cc._connection + connection_ssl_context = _safe_getattr(connection, 'ssl_context', None) + connection_ssl_options = _safe_getattr(connection, 'ssl_options', None) + endpoint = _safe_getattr(connection, 'endpoint', None) + endpoint_ssl_options = _safe_getattr(endpoint, 'ssl_options', None) + + ssl_context = (connection_ssl_context + if connection_ssl_context is not None + else self._session.cluster.ssl_context) + ssl_options = (connection_ssl_options + if connection_ssl_options is not None + else self._session.cluster.ssl_options) + ssl_options_explicit = bool(_safe_getattr(connection, '_ssl_options_explicit', False)) + ssl_enabled = (ssl_context is not None or + ssl_options is not None or + endpoint_ssl_options is not None or + ssl_options_explicit) + 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) + elif endpoint_ssl_options is not None: + cert_validation = _ssl_options_cert_validation_enabled(endpoint_ssl_options) except Exception as e: log.debug('Unable to get the cert validation: {}'.format(e)) @@ -186,7 +224,7 @@ def _get_startup_data(self): 'compression': compression_type.upper() if compression_type else 'NONE', 'reconnectionPolicy': insights_registry.serialize(self._session.cluster.reconnection_policy), 'sslConfigured': { - 'enabled': bool(self._session.cluster.ssl_options or self._session.cluster.ssl_context), + 'enabled': ssl_enabled, 'certValidation': cert_validation }, 'authProvider': { diff --git a/cassandra/io/asyncioreactor.py b/cassandra/io/asyncioreactor.py index 92ab972e7d..7f2c7a208b 100644 --- a/cassandra/io/asyncioreactor.py +++ b/cassandra/io/asyncioreactor.py @@ -133,7 +133,7 @@ def __init__(self, *args, **kwargs): Connection.__init__(self, *args, **kwargs) self._background_tasks = set() self._transport = None - self._using_ssl = bool(self.ssl_context) + self._using_ssl = self._ssl_enabled self._connect_socket() self._socket.setblocking(0) diff --git a/cassandra/io/eventletreactor.py b/cassandra/io/eventletreactor.py index 234a4a574c..4613a9f9bc 100644 --- a/cassandra/io/eventletreactor.py +++ b/cassandra/io/eventletreactor.py @@ -23,7 +23,10 @@ from threading import Event import time -from cassandra.connection import Connection, ConnectionShutdown, Timer, TimerManager +from cassandra.connection import (Connection, ConnectionShutdown, Timer, TimerManager, + _build_pyopenssl_context_from_options as _build_pyopenssl_context, + _default_pyopenssl_ssl_method, + _validate_pyopenssl_hostname) try: from eventlet.green.OpenSSL import SSL _PYOPENSSL = True @@ -35,6 +38,10 @@ log = logging.getLogger(__name__) +def _default_ssl_method(): + return _default_pyopenssl_ssl_method(SSL) + + def _check_pyopenssl(): if not _PYOPENSSL: raise ImportError( @@ -43,6 +50,10 @@ def _check_pyopenssl(): ) +def _build_pyopenssl_context_from_options(ssl_options): + return _build_pyopenssl_context(SSL, ssl_options) + + class EventletConnection(Connection): """ An implementation of :class:`.Connection` that utilizes ``eventlet``. @@ -92,7 +103,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() @@ -108,23 +118,20 @@ def _wrap_socket_from_context(self): if self.ssl_options and 'server_hostname' in self.ssl_options: # This is necessary for SNI self._socket.set_tlsext_host_name(self.ssl_options['server_hostname'].encode('ascii')) + return self._socket def _initiate_connection(self, sockaddr): - if self.uses_legacy_ssl_options: - super(EventletConnection, self)._initiate_connection(sockaddr) - else: - self._socket.connect(sockaddr) - if self.ssl_context or self.ssl_options: - 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): + expected_name = (self.ssl_options or {}).get('server_hostname') or self.endpoint.address + _validate_pyopenssl_hostname(self._socket.get_peer_certificate(), expected_name) + + def _build_ssl_context_from_options(self): + _check_pyopenssl() + return _build_pyopenssl_context_from_options(self.ssl_options) def close(self): with self.lock: diff --git a/cassandra/io/twistedreactor.py b/cassandra/io/twistedreactor.py index 446200bf63..6914bc7f15 100644 --- a/cassandra/io/twistedreactor.py +++ b/cassandra/io/twistedreactor.py @@ -17,6 +17,7 @@ """ import atexit import logging +import ssl import time from functools import partial from threading import Thread, Lock @@ -28,7 +29,10 @@ from twisted.python.failure import Failure from zope.interface import implementer -from cassandra.connection import Connection, ConnectionShutdown, Timer, TimerManager, ConnectionException +from cassandra.connection import (Connection, ConnectionShutdown, Timer, TimerManager, ConnectionException, + _build_pyopenssl_context_from_options as _build_pyopenssl_context, + _default_pyopenssl_ssl_method, + _validate_pyopenssl_hostname) try: from OpenSSL import SSL @@ -39,6 +43,22 @@ log = logging.getLogger(__name__) +def _default_ssl_method(): + return _default_pyopenssl_ssl_method(SSL) + + +def _check_pyopenssl(): + if not _HAS_SSL: + raise ImportError( + str(import_exception) + + ', pyOpenSSL must be installed to enable SSL support with the Twisted event loop' + ) + + +def _build_pyopenssl_context_from_options(ssl_options): + return _build_pyopenssl_context(SSL, ssl_options) + + def _cleanup(cleanup_weakref): try: cleanup_weakref()._cleanup() @@ -141,25 +161,14 @@ def _on_loop_timer(self): class _SSLCreator(object): def __init__(self, endpoint, ssl_context, ssl_options, check_hostname, timeout): self.endpoint = endpoint - self.ssl_options = ssl_options + self.ssl_options = ssl_options or {} self.check_hostname = check_hostname self.timeout = timeout - if ssl_context: + if ssl_context is not None: self.context = ssl_context else: - self.context = SSL.Context(SSL.TLSv1_METHOD) - if "certfile" in self.ssl_options: - self.context.use_certificate_file(self.ssl_options["certfile"]) - if "keyfile" in self.ssl_options: - self.context.use_privatekey_file(self.ssl_options["keyfile"]) - if "ca_certs" in self.ssl_options: - self.context.load_verify_locations(self.ssl_options["ca_certs"]) - if "cert_reqs" in self.ssl_options: - self.context.set_verify( - self.ssl_options["cert_reqs"], - callback=self.verify_callback - ) + self.context = _build_pyopenssl_context_from_options(self.ssl_options) self.context.set_info_callback(self.info_callback) def verify_callback(self, connection, x509, errnum, errdepth, ok): @@ -167,9 +176,13 @@ def verify_callback(self, connection, x509, errnum, errdepth, 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 self.check_hostname: + expected_name = self.ssl_options.get('server_hostname') or self.endpoint.address + try: + _validate_pyopenssl_hostname(connection.get_peer_certificate(), expected_name) + except ssl.CertificateError as exc: + transport = connection.get_app_data() + transport.failVerification(Failure(ConnectionException(str(exc), self.endpoint))) def clientConnectionForTLS(self, tlsProtocol): connection = SSL.Connection(self.context, None) @@ -217,12 +230,12 @@ def __init__(self, *args, **kwargs): self._loop.maybe_start() def _check_pyopenssl(self): - if self.ssl_context or self.ssl_options: - if not _HAS_SSL: - raise ImportError( - str(import_exception) + - ', pyOpenSSL must be installed to enable SSL support with the Twisted event loop' - ) + if self._ssl_enabled: + _check_pyopenssl() + + def _build_ssl_context_from_options(self): + _check_pyopenssl() + return _build_pyopenssl_context_from_options(self.ssl_options) def add_connection(self): """ @@ -230,14 +243,14 @@ def add_connection(self): connector. """ host, port = self.endpoint.resolve() - if self.ssl_context or self.ssl_options: + if self._ssl_enabled: # Can't use optionsForClientTLS here because it *forces* hostname verification. # Cool they enforce strong security, but we have to be able to turn it off self._check_pyopenssl() ssl_connection_creator = _SSLCreator( self.endpoint, - self.ssl_context if self.ssl_context else None, + self.ssl_context if self.ssl_context is not None else None, self.ssl_options, self._check_hostname, self.connect_timeout, diff --git a/cassandra/pool.py b/cassandra/pool.py index 176751f60a..ad68de44f5 100644 --- a/cassandra/pool.py +++ b/cassandra/pool.py @@ -682,19 +682,23 @@ 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 + ssl_enabled = (cluster.ssl_context is not None or + cluster.ssl_options is not None or + self.host.endpoint.ssl_options is not None) endpoint = None - if ssl_enabled and self.host.sharding_info.shard_aware_port_ssl: - endpoint = copy.copy(self.host.endpoint) - endpoint._port = self.host.sharding_info.shard_aware_port_ssl - elif not ssl_enabled and self.host.sharding_info.shard_aware_port: + if ssl_enabled: + if self.host.sharding_info.shard_aware_port_ssl: + endpoint = copy.copy(self.host.endpoint) + endpoint._port = self.host.sharding_info.shard_aware_port_ssl + elif self.host.sharding_info.shard_aware_port: endpoint = copy.copy(self.host.endpoint) endpoint._port = self.host.sharding_info.shard_aware_port diff --git a/tests/unit/advanced/test_insights.py b/tests/unit/advanced/test_insights.py index 2050439804..4fe0005591 100644 --- a/tests/unit/advanced/test_insights.py +++ b/tests/unit/advanced/test_insights.py @@ -16,18 +16,22 @@ import unittest import logging +import ssl import sys -from unittest.mock import sentinel +from unittest.mock import Mock, sentinel from cassandra import ConsistencyLevel from cassandra.cluster import ( ExecutionProfile, GraphExecutionProfile, GraphAnalyticsExecutionProfile ) +from cassandra.connection import Connection from cassandra.datastax.graph.query import GraphOptions +from cassandra.datastax.insights.reporter import MonitorReporter from cassandra.datastax.insights.registry import insights_registry from cassandra.datastax.insights.serializers import initialize_registry from cassandra.policies import ( LoadBalancingPolicy, + HostDistance, DCAwareRoundRobinPolicy, TokenAwarePolicy, WhiteListRoundRobinPolicy, @@ -88,6 +92,98 @@ def superclass_sentinel_serializer(obj): # with default -- same behavior assert insights_registry.serialize(SubclassSentinel(), default=object()) is sentinel.serialized_superclass + +class TestMonitorReporterStartupData(unittest.TestCase): + + def _get_startup_data(self, ssl_options=None, ssl_context=None, + connection_ssl_options=None, connection_ssl_context=None, + connection_ssl_options_explicit=False): + reporter = MonitorReporter.__new__(MonitorReporter) + reporter._interval = 30 + + connection = Mock() + connection.host = '127.0.0.1' + connection._compression_type = 'NONE' + connection._socket.getsockname.return_value = ('127.0.0.1', 9042) + if connection_ssl_options is not None: + connection.ssl_options = connection_ssl_options + if connection_ssl_context is not None: + connection.ssl_context = connection_ssl_context + connection._ssl_options_explicit = connection_ssl_options_explicit + + cluster = Mock() + cluster.auth_provider = None + cluster.client_id = 'client-id' + cluster.connection_class = Connection + cluster.control_connection._connection = connection + cluster.application_name = None + cluster.application_version = None + cluster._endpoint_map_for_insights = {} + cluster.idle_heartbeat_interval = 30 + cluster.metadata.all_hosts.return_value = [] + cluster.profile_manager.distance.return_value = HostDistance.LOCAL + cluster.protocol_version = 4 + cluster.reconnection_policy = object() + cluster.ssl_context = ssl_context + cluster.ssl_options = ssl_options + + session = Mock() + session.cluster = cluster + session.hosts = [] + session.session_id = 'session-id' + + reporter._session = session + return reporter._get_startup_data() + + def test_empty_ssl_options_reported_as_enabled(self): + startup_data = self._get_startup_data(ssl_options={}) + + assert startup_data['data']['sslConfigured']['enabled'] is True + + def test_implicit_ssl_options_validation_reported(self): + startup_data = self._get_startup_data(ssl_options={'ca_certs': 'ca.pem'}) + + assert startup_data['data']['sslConfigured']['certValidation'] is True + + def test_check_hostname_validation_reported(self): + startup_data = self._get_startup_data(ssl_options={'check_hostname': True}) + + assert startup_data['data']['sslConfigured']['certValidation'] is True + + def test_empty_ssl_options_validation_reported_disabled(self): + startup_data = self._get_startup_data(ssl_options={}) + + assert startup_data['data']['sslConfigured']['certValidation'] is False + + def test_omitted_ssl_options_reported_as_disabled(self): + startup_data = self._get_startup_data() + + assert startup_data['data']['sslConfigured']['enabled'] is False + + def test_ssl_context_reported_as_enabled(self): + ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + + startup_data = self._get_startup_data(ssl_context=ssl_context) + + assert startup_data['data']['sslConfigured']['enabled'] is True + + 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 + + startup_data = self._get_startup_data(ssl_context=ssl_context) + + assert startup_data['data']['sslConfigured']['certValidation'] is True + + def test_connection_ssl_options_reported_as_enabled(self): + startup_data = self._get_startup_data( + connection_ssl_options={}, connection_ssl_options_explicit=True) + + assert startup_data['data']['sslConfigured']['enabled'] is True + assert startup_data['data']['sslConfigured']['certValidation'] is False + + class TestConfigAsDict(unittest.TestCase): # graph/query.py 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..6e8063d6b6 100644 --- a/tests/unit/io/test_eventletreactor.py +++ b/tests/unit/io/test_eventletreactor.py @@ -12,10 +12,12 @@ # See the License for the specific language governing permissions and # limitations under the License. - +import os +import ssl import unittest +from types import SimpleNamespace -from unittest.mock import patch +from unittest.mock import Mock, patch from tests.unit.io.utils import TimerTestMixin from tests import notpypy, EVENT_LOOP_MANAGER @@ -23,10 +25,169 @@ try: from eventlet import monkey_patch + from cassandra.io import eventletreactor from cassandra.io.eventletreactor import EventletConnection except (ImportError, AttributeError): + eventletreactor = None EventletConnection = None # noqa +from cassandra.connection import Connection, DefaultEndPoint + +CA_CERTS = os.path.abspath(os.path.join( + os.path.dirname(__file__), '..', '..', 'integration', 'long', 'ssl', 'rootCa.crt')) + + +class _FakeX509Name(object): + def __init__(self, common_name): + self._common_name = common_name + + def get_components(self): + return [(b'CN', self._common_name.encode('utf-8'))] + + +class _FakeX509Extension(object): + def __init__(self, short_name, value): + self._short_name = short_name + self._value = value + + def get_short_name(self): + return self._short_name + + def __str__(self): + return self._value + + +class _FakeX509Certificate(object): + def __init__(self, common_name, san_dns_names=None): + self._subject = _FakeX509Name(common_name) + self._extensions = [] + if san_dns_names: + self._extensions.append(_FakeX509Extension( + b'subjectAltName', + ', '.join('DNS:%s' % name for name in san_dns_names))) + + def get_subject(self): + return self._subject + + def get_extension_count(self): + return len(self._extensions) + + def get_extension(self, i): + return self._extensions[i] + + +def _make_certificate(common_name, san_dns_names=None): + return _FakeX509Certificate(common_name, san_dns_names) + + +@unittest.skipIf(EventletConnection is None, "eventlet is not available") +@unittest.skipIf(not getattr(eventletreactor, '_PYOPENSSL', False), "pyOpenSSL is not available") +class EventletSSLContextTest(unittest.TestCase): + + def test_empty_ssl_options_build_pyopenssl_context(self): + conn = EventletConnection.__new__(EventletConnection) + + Connection.__init__(conn, DefaultEndPoint('1.2.3.4'), ssl_options={}) + + assert conn._ssl_enabled + assert conn.ssl_context is not None + + def test_empty_ssl_options_default_to_negotiating_tls(self): + with patch.object(eventletreactor.SSL, 'Context') as context_mock: + context = eventletreactor._build_pyopenssl_context_from_options({}) + + context_mock.assert_called_once_with(eventletreactor._default_ssl_method()) + assert context is context_mock.return_value + + def test_default_ssl_method_falls_back_to_tls_method(self): + tls_method = object() + + with patch.object(eventletreactor, 'SSL', SimpleNamespace(TLS_METHOD=tls_method)): + assert eventletreactor._default_ssl_method() is tls_method + + def test_default_ssl_method_falls_back_to_tlsv1_2_method(self): + tlsv1_2_method = object() + + with patch.object(eventletreactor, 'SSL', SimpleNamespace(TLSv1_2_METHOD=tlsv1_2_method)): + assert eventletreactor._default_ssl_method() is tlsv1_2_method + + def test_ssl_version_option_is_preserved(self): + with patch.object(eventletreactor.SSL, 'Context') as context_mock: + eventletreactor._build_pyopenssl_context_from_options( + {'ssl_version': eventletreactor.SSL.TLSv1_2_METHOD}) + + context_mock.assert_called_once_with(eventletreactor.SSL.TLSv1_2_METHOD) + + def test_stdlib_ssl_version_option_is_translated(self): + with patch.object(eventletreactor.SSL, 'Context') as context_mock: + eventletreactor._build_pyopenssl_context_from_options( + {'ssl_version': ssl.PROTOCOL_TLS}) + + context_mock.assert_called_once_with(eventletreactor.SSL.TLS_METHOD) + + def test_ca_certs_default_to_required_validation(self): + conn = EventletConnection.__new__(EventletConnection) + + Connection.__init__(conn, DefaultEndPoint('1.2.3.4'), ssl_options={'ca_certs': CA_CERTS}) + + assert conn.ssl_context.get_verify_mode() == eventletreactor.SSL.VERIFY_PEER + + def test_stdlib_cert_reqs_option_is_translated(self): + context = eventletreactor._build_pyopenssl_context_from_options({'cert_reqs': ssl.CERT_REQUIRED}) + + assert context.get_verify_mode() == eventletreactor.SSL.VERIFY_PEER + + def test_ciphers_option_is_applied(self): + with patch.object(eventletreactor.SSL, 'Context') as context_mock: + eventletreactor._build_pyopenssl_context_from_options({'ciphers': 'ECDHE+AESGCM'}) + + context_mock.return_value.set_cipher_list.assert_called_once_with(b'ECDHE+AESGCM') + + 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_wrap_socket_from_context_returns_wrapped_socket(self): + conn = EventletConnection.__new__(EventletConnection) + conn.ssl_context = object() + conn.ssl_options = {} + original_socket = object() + conn._socket = original_socket + + with patch.object(eventletreactor.SSL, 'Connection') as mock_connection: + wrapped_socket = mock_connection.return_value + + assert conn._wrap_socket_from_context() is wrapped_socket + + mock_connection.assert_called_once_with(conn.ssl_context, original_socket) + wrapped_socket.set_connect_state.assert_called_once_with() + assert conn._socket is wrapped_socket + + 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_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_certificate( + 'node.example.com', san_dns_names=['other.example.com']) + + with self.assertRaises(ssl.CertificateError): + conn._validate_hostname() + + 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..767909386f 100644 --- a/tests/unit/io/test_twistedreactor.py +++ b/tests/unit/io/test_twistedreactor.py @@ -12,10 +12,13 @@ # See the License for the specific language governing permissions and # limitations under the License. +import os +import ssl import unittest +from types import SimpleNamespace from unittest.mock import Mock, patch -from cassandra.connection import DefaultEndPoint +from cassandra.connection import Connection, DefaultEndPoint try: from twisted.test import proto_helpers @@ -29,6 +32,71 @@ from tests.unit.io.utils import TimerTestMixin +CA_CERTS = os.path.abspath(os.path.join( + os.path.dirname(__file__), '..', '..', 'integration', 'long', 'ssl', 'rootCa.crt')) + + +@unittest.skipIf(TwistedConnection is None, "Twisted libraries are not available") +@unittest.skipIf(not getattr(twistedreactor, '_HAS_SSL', False), "pyOpenSSL is not available") +class TwistedSSLContextTest(unittest.TestCase): + + def test_empty_ssl_options_default_to_negotiating_tls(self): + with patch.object(twistedreactor.SSL, 'Context') as context_mock: + context = twistedreactor._build_pyopenssl_context_from_options({}) + + context_mock.assert_called_once_with(twistedreactor._default_ssl_method()) + assert context is context_mock.return_value + + def test_default_ssl_method_falls_back_to_tls_method(self): + tls_method = object() + + with patch.object(twistedreactor, 'SSL', SimpleNamespace(TLS_METHOD=tls_method)): + assert twistedreactor._default_ssl_method() is tls_method + + def test_default_ssl_method_falls_back_to_tlsv1_2_method(self): + tlsv1_2_method = object() + + with patch.object(twistedreactor, 'SSL', SimpleNamespace(TLSv1_2_METHOD=tlsv1_2_method)): + assert twistedreactor._default_ssl_method() is tlsv1_2_method + + def test_ssl_version_option_is_preserved(self): + with patch.object(twistedreactor.SSL, 'Context') as context_mock: + twistedreactor._build_pyopenssl_context_from_options( + {'ssl_version': twistedreactor.SSL.TLSv1_2_METHOD}) + + context_mock.assert_called_once_with(twistedreactor.SSL.TLSv1_2_METHOD) + + def test_stdlib_ssl_version_option_is_translated(self): + with patch.object(twistedreactor.SSL, 'Context') as context_mock: + twistedreactor._build_pyopenssl_context_from_options( + {'ssl_version': ssl.PROTOCOL_TLS}) + + context_mock.assert_called_once_with(twistedreactor.SSL.TLS_METHOD) + + def test_ca_certs_default_to_required_validation(self): + context = twistedreactor._build_pyopenssl_context_from_options({'ca_certs': CA_CERTS}) + + assert context.get_verify_mode() == twistedreactor.SSL.VERIFY_PEER + + def test_stdlib_cert_reqs_option_is_translated(self): + context = twistedreactor._build_pyopenssl_context_from_options({'cert_reqs': ssl.CERT_REQUIRED}) + + assert context.get_verify_mode() == twistedreactor.SSL.VERIFY_PEER + + def test_ciphers_option_is_applied(self): + with patch.object(twistedreactor.SSL, 'Context') as context_mock: + twistedreactor._build_pyopenssl_context_from_options({'ciphers': 'ECDHE+AESGCM'}) + + context_mock.return_value.set_cipher_list.assert_called_once_with(b'ECDHE+AESGCM') + + 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 + + class TestTwistedTimer(TimerTestMixin, unittest.TestCase): """ Simple test class that is used to validate that the TimerManager, and timer @@ -196,3 +264,19 @@ def test_push(self, mock_connectTCP): self.obj_ut.push('123 pickup') self.mock_reactor_cft.assert_called_with( transport_mock.write, '123 pickup') + + @unittest.skipIf(not getattr(twistedreactor, '_HAS_SSL', False), "pyOpenSSL is not available") + @patch('cassandra.io.twistedreactor.connectProtocol') + @patch('cassandra.io.twistedreactor.TCP4ClientEndpoint') + @patch('cassandra.io.twistedreactor.SSL4ClientEndpoint') + def test_empty_ssl_options_use_ssl_endpoint(self, mock_ssl_endpoint, mock_tcp_endpoint, mock_connect_protocol): + conn = twistedreactor.TwistedConnection( + DefaultEndPoint('1.2.3.4'), + cql_version='3.0.1', + ssl_options={}) + + conn.add_connection() + + mock_ssl_endpoint.assert_called_once() + mock_tcp_endpoint.assert_not_called() + mock_connect_protocol.assert_called_once() diff --git a/tests/unit/test_client_routes.py b/tests/unit/test_client_routes.py index 0aa82fc76a..8d38419a8e 100644 --- a/tests/unit/test_client_routes.py +++ b/tests/unit/test_client_routes.py @@ -449,6 +449,21 @@ def test_check_hostname_with_ssl_options_raises(self): ) self.assertIn("check_hostname", str(cm.exception)) + def test_empty_ssl_options_enable_tls_routes(self): + config = ClientRoutesConfig( + proxies=[ClientRouteProxy(str(uuid.uuid4()), "10.0.0.1")] + ) + with self.assertWarns(DeprecationWarning): + cluster = Cluster( + contact_points=["10.0.0.1"], + ssl_options={}, + client_routes_config=config, + ) + try: + self.assertTrue(cluster._client_routes_handler.ssl_enabled) + finally: + cluster.shutdown() + def test_disabled_check_hostname_with_client_routes_ok(self): """Cluster should allow check_hostname=False with client_routes_config.""" ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) diff --git a/tests/unit/test_cloud.py b/tests/unit/test_cloud.py new file mode 100644 index 0000000000..509228fd7c --- /dev/null +++ b/tests/unit/test_cloud.py @@ -0,0 +1,53 @@ +# 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. + +from types import SimpleNamespace + +import pytest + +from cassandra.datastax import cloud + + +def test_default_pyopenssl_ssl_method_prefers_tls_client_method(): + tls_client_method = object() + + ssl_module = SimpleNamespace( + TLS_CLIENT_METHOD=tls_client_method, + TLS_METHOD=object(), + TLSv1_2_METHOD=object()) + + assert cloud._default_pyopenssl_ssl_method(ssl_module) is tls_client_method + + +def test_default_pyopenssl_ssl_method_falls_back_to_tls_method(): + tls_method = object() + + ssl_module = SimpleNamespace( + TLS_METHOD=tls_method, + TLSv1_2_METHOD=object()) + + assert cloud._default_pyopenssl_ssl_method(ssl_module) is tls_method + + +def test_default_pyopenssl_ssl_method_falls_back_to_tlsv1_2_method(): + tlsv1_2_method = object() + + ssl_module = SimpleNamespace(TLSv1_2_METHOD=tlsv1_2_method) + + assert cloud._default_pyopenssl_ssl_method(ssl_module) is tlsv1_2_method + + +def test_default_pyopenssl_ssl_method_requires_secure_method(): + with pytest.raises(ImportError, match="secure TLS client method"): + cloud._default_pyopenssl_ssl_method(SimpleNamespace()) diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 3d55bc1860..0e1fb5a019 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. import unittest +import warnings from concurrent.futures import Future import logging @@ -278,6 +279,25 @@ def test_connection_factory_passes_compression_kwarg(self): assert factory.call_args.kwargs['compression'] == expected assert cluster.compression == expected + def test_empty_ssl_options_are_rejected_with_cloud_config(self): + with pytest.raises(ValueError) as exc: + Cluster(cloud={'secure_connect_bundle': 'unused'}, ssl_options={}) + + assert "cannot be specified with a cloud configuration" in str(exc.value) + + def test_empty_ssl_options_emit_deprecation_warning(self): + with pytest.warns(DeprecationWarning, match="Using ssl_options without ssl_context"): + cluster = Cluster(ssl_options={}) + + assert cluster.ssl_options == {} + + def test_omitted_ssl_options_do_not_emit_deprecation_warning(self): + with warnings.catch_warnings(record=True) as recorded_warnings: + warnings.simplefilter("always") + Cluster() + + assert not [warning for warning in recorded_warnings if warning.category is DeprecationWarning] + class SchedulerTest(unittest.TestCase): # TODO: this suite could be expanded; for now just adding a test covering a ticket diff --git a/tests/unit/test_connection.py b/tests/unit/test_connection.py index 1f9a3f682c..2cbac25c03 100644 --- a/tests/unit/test_connection.py +++ b/tests/unit/test_connection.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. import itertools +import ssl import unittest from io import BytesIO import time @@ -22,7 +23,8 @@ from cassandra.cluster import Cluster from cassandra.connection import (Connection, HEADER_DIRECTION_TO_CLIENT, ProtocolError, locally_supported_compressions, ConnectionHeartbeat, HeartbeatFuture, _Frame, Timer, TimerManager, - ConnectionException, ConnectionShutdown, DefaultEndPoint, ShardAwarePortGenerator) + ConnectionException, ConnectionShutdown, DefaultEndPoint, ShardAwarePortGenerator, + _build_pyopenssl_context_from_options, _validate_pyopenssl_hostname) from cassandra.marshal import uint8_pack, uint32_pack, int32_pack from cassandra.protocol import (write_stringmultimap, write_int, write_string, SupportedMessage, ProtocolHandler, ResultMessage, @@ -32,6 +34,88 @@ import pytest +class FakeX509Name(object): + def __init__(self, common_name): + self._common_name = common_name + + def get_components(self): + return [(b'CN', self._common_name.encode('utf-8'))] + + +class FakeX509Extension(object): + def __init__(self, value): + self._value = value + + def get_short_name(self): + return b'subjectAltName' + + def __str__(self): + return self._value + + +class FakeX509Certificate(object): + def __init__(self, common_name, subject_alt_name=None): + self._subject = FakeX509Name(common_name) + self._extensions = [] + if subject_alt_name is not None: + self._extensions.append(FakeX509Extension(subject_alt_name)) + + def get_subject(self): + return self._subject + + def get_extension_count(self): + return len(self._extensions) + + def get_extension(self, i): + return self._extensions[i] + + +class FakePyOpenSSLContext(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 PyOpenSSLHostnameValidationTest(unittest.TestCase): + + def test_ip_hostname_requires_ip_subject_alt_name(self): + cert = FakeX509Certificate('unused', subject_alt_name='DNS:*.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 = FakeX509Certificate('unused', subject_alt_name='IP Address:1.2.3.4') + + _validate_pyopenssl_hostname(cert, '1.2.3.4') + + def test_ip_hostname_accepts_exact_common_name_without_san(self): + cert = FakeX509Certificate('1.2.3.4') + + _validate_pyopenssl_hostname(cert, '1.2.3.4') + + +class PyOpenSSLContextTest(unittest.TestCase): + + 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 + + class ConnectionTest(unittest.TestCase): def make_connection(self, **kwargs): @@ -81,6 +165,34 @@ def test_connection_endpoint(self): assert c.endpoint == endpoint assert c.endpoint.address == endpoint.address + def test_omitted_ssl_options_do_not_enable_ssl(self): + c = Connection(DefaultEndPoint('1.2.3.4')) + + assert c.ssl_context is None + assert not c._ssl_enabled + + def test_empty_ssl_options_enable_ssl(self): + c = Connection(DefaultEndPoint('1.2.3.4'), ssl_options={}) + + assert isinstance(c.ssl_context, ssl.SSLContext) + assert c.ssl_context.verify_mode == ssl.CERT_NONE + assert c.ssl_options == {} + assert c._ssl_enabled + + def test_ssl_options_cert_reqs_applied_to_context(self): + c = Connection(DefaultEndPoint('1.2.3.4'), ssl_options={'cert_reqs': ssl.CERT_REQUIRED}) + + assert isinstance(c.ssl_context, ssl.SSLContext) + assert c.ssl_context.verify_mode == ssl.CERT_REQUIRED + + def test_ssl_options_check_hostname_requires_validation(self): + c = Connection(DefaultEndPoint('1.2.3.4'), ssl_options={'check_hostname': True}) + + assert isinstance(c.ssl_context, ssl.SSLContext) + assert c.ssl_context.verify_mode == ssl.CERT_REQUIRED + assert c.ssl_context.check_hostname + assert c._check_hostname + def 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)