connection: address SSL review feedback

Author: dkropachev

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
@@ -206,6 +207,93 @@ def _build_pyopenssl_context_from_options(ssl_module, ssl_options):
     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 _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
+
+    for cert_ip in san_ip_addresses:
+        if _ipaddress_match(cert_ip, hostname):
+            return
+    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 _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.

cassandra/datastax/cloud/__init__.py

@@ -182,8 +182,8 @@ 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__)
+            "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)

cassandra/datastax/insights/reporter.py

@@ -40,6 +40,21 @@ def _ssl_options_cert_validation_enabled(ssl_options):
     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_REQUIRED
+
+    from OpenSSL import SSL
+    return ssl_context.get_verify_mode() != SSL.VERIFY_NONE
+
+
 class MonitorReporter(Thread):
 
     def __init__(self, interval_sec, session):
@@ -146,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 is not None:
-                if isinstance(self._session.cluster.ssl_context, ssl.SSLContext):
-                    cert_validation = self._session.cluster.ssl_context.verify_mode == ssl.CERT_REQUIRED
-                else:  # pyopenssl
-                    from OpenSSL import SSL
-                    cert_validation = self._session.cluster.ssl_context.get_verify_mode() != SSL.VERIFY_NONE
-            elif self._session.cluster.ssl_options is not None:
-                cert_validation = _ssl_options_cert_validation_enabled(self._session.cluster.ssl_options)
+            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))
 
@@ -193,8 +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': (self._session.cluster.ssl_context is not None or
-                                self._session.cluster.ssl_options is not None),
+                    'enabled': ssl_enabled,
                     'certValidation': cert_validation
                 },
                 'authProvider': {

cassandra/io/eventletreactor.py

@@ -25,7 +25,8 @@
 
 from cassandra.connection import (Connection, ConnectionShutdown, Timer, TimerManager,
                                   _build_pyopenssl_context_from_options as _build_pyopenssl_context,
-                                  _default_pyopenssl_ssl_method)
+                                  _default_pyopenssl_ssl_method,
+                                  _validate_pyopenssl_hostname)
 try:
     from eventlet.green.OpenSSL import SSL
     _PYOPENSSL = True
@@ -102,7 +103,6 @@ def service_timeouts(cls):
 
     def __init__(self, *args, **kwargs):
         Connection.__init__(self, *args, **kwargs)
-        self.uses_legacy_ssl_options = False
         self._write_queue = Queue()
 
         self._connect_socket()
@@ -121,21 +121,13 @@ def _wrap_socket_from_context(self):
         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_enabled:
-                self._socket.do_handshake()
+        self._socket.connect(sockaddr)
+        if self._ssl_enabled:
+            self._socket.do_handshake()
 
     def _validate_hostname(self):
-        if self.uses_legacy_ssl_options:
-            super(EventletConnection, self)._validate_hostname()
-        else:
-            cert_name = self._socket.get_peer_certificate().get_subject().commonName
-            if cert_name != self.endpoint.address:
-                raise Exception("Hostname verification failed! Certificate name '{}' "
-                                "doesn't endpoint '{}'".format(cert_name, self.endpoint.address))
+        expected_name = (self.ssl_options or {}).get('server_hostname') or self.endpoint.address
+        _validate_pyopenssl_hostname(self._socket.get_peer_certificate(), expected_name)
 
     def _build_ssl_context_from_options(self):
         _check_pyopenssl()

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
@@ -30,7 +31,8 @@
 
 from cassandra.connection import (Connection, ConnectionShutdown, Timer, TimerManager, ConnectionException,
                                   _build_pyopenssl_context_from_options as _build_pyopenssl_context,
-                                  _default_pyopenssl_ssl_method)
+                                  _default_pyopenssl_ssl_method,
+                                  _validate_pyopenssl_hostname)
 
 try:
     from OpenSSL import SSL
@@ -174,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)

tests/unit/advanced/test_insights.py

@@ -95,14 +95,21 @@ def superclass_sentinel_serializer(obj):
 
 class TestMonitorReporterStartupData(unittest.TestCase):
 
-    def _get_startup_data(self, ssl_options=None, ssl_context=None):
+    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
@@ -160,6 +167,13 @@ def test_ssl_context_reported_as_enabled(self):
 
         assert startup_data['data']['sslConfigured']['enabled'] 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):
 

tests/unit/io/test_eventletreactor.py

@@ -17,7 +17,7 @@
 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
@@ -27,7 +27,9 @@
     from eventlet import monkey_patch
     from cassandra.io import eventletreactor
     from cassandra.io.eventletreactor import EventletConnection
+    from OpenSSL import crypto
 except (ImportError, AttributeError):
+    crypto = None
     eventletreactor = None
     EventletConnection = None  # noqa
 
@@ -37,6 +39,28 @@
     os.path.dirname(__file__), '..', '..', 'integration', 'long', 'ssl', 'rootCa.crt'))
 
 
+def _make_certificate(common_name, san_dns_names=None):
+    key = crypto.PKey()
+    key.generate_key(crypto.TYPE_RSA, 2048)
+
+    cert = crypto.X509()
+    cert.get_subject().CN = common_name
+    cert.set_serial_number(1)
+    cert.gmtime_adj_notBefore(-3600)
+    cert.gmtime_adj_notAfter(3600)
+    cert.set_issuer(cert.get_subject())
+    cert.set_pubkey(key)
+    if san_dns_names:
+        cert.add_extensions([
+            crypto.X509Extension(
+                b'subjectAltName',
+                False,
+                ', '.join('DNS:%s' % name for name in san_dns_names).encode('ascii'))
+        ])
+    cert.sign(key, 'sha256')
+    return cert
+
+
 @unittest.skipIf(EventletConnection is None, "eventlet is not available")
 @unittest.skipIf(not getattr(eventletreactor, '_PYOPENSSL', False), "pyOpenSSL is not available")
 class EventletSSLContextTest(unittest.TestCase):
@@ -53,7 +77,7 @@ 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.SSL.TLS_CLIENT_METHOD)
+        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):
@@ -123,6 +147,27 @@ def test_wrap_socket_from_context_returns_wrapped_socket(self):
         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

tests/unit/io/test_twistedreactor.py

@@ -44,7 +44,7 @@ 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.SSL.TLS_CLIENT_METHOD)
+        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):