ssl: avoid insecure pyopenssl fallback

Author: dkropachev

cassandra/connection.py

@@ -229,6 +229,81 @@ 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)
+    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
+
+
 class EndPoint(object):
     """
     Represents the information to connect to a cassandra node.

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__)
 
@@ -183,10 +184,10 @@ def _pyopenssl_context_from_cert(ca_cert_location, cert_location, key_location):
         raise ImportError(
             "PyOpenSSL must be installed to connect to Astra with the Eventlet or Twisted event loops")\
             .with_traceback(e.__traceback__)
-    ssl_context = SSL.Context(SSL.TLSv1_METHOD)
+    ssl_context = 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
+    return ssl_context

cassandra/io/eventletreactor.py

@@ -24,6 +24,8 @@
 import time
 
 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
@@ -37,8 +39,7 @@
 
 
 def _default_ssl_method():
-    return getattr(SSL, 'TLS_CLIENT_METHOD',
-                   getattr(SSL, 'TLS_METHOD', SSL.TLSv1_METHOD))
+    return _default_pyopenssl_ssl_method(SSL)
 
 
 def _check_pyopenssl():
@@ -50,25 +51,7 @@ def _check_pyopenssl():
 
 
 def _build_pyopenssl_context_from_options(ssl_options):
-    ssl_version = (ssl_options['ssl_version']
-                   if 'ssl_version' in ssl_options else _default_ssl_method())
-    context = SSL.Context(ssl_version)
-    if 'certfile' in ssl_options:
-        context.use_certificate_file(ssl_options['certfile'])
-    if 'keyfile' in ssl_options:
-        context.use_privatekey_file(ssl_options['keyfile'])
-    if 'ca_certs' in ssl_options:
-        context.load_verify_locations(ssl_options['ca_certs'])
-    cert_reqs = ssl_options.get('cert_reqs', None)
-    if cert_reqs is None:
-        cert_reqs = (SSL.VERIFY_PEER
-                     if (ssl_options.get('ca_certs', None) or ssl_options.get('check_hostname', False))
-                     else SSL.VERIFY_NONE)
-    context.set_verify(
-        cert_reqs,
-        callback=lambda _connection, _x509, _errnum, _errdepth, ok: ok
-    )
-    return context
+    return _build_pyopenssl_context(SSL, ssl_options)
 
 
 class EventletConnection(Connection):

cassandra/io/twistedreactor.py

@@ -29,6 +29,8 @@
 from zope.interface import implementer
 
 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:
@@ -41,8 +43,7 @@
 
 
 def _default_ssl_method():
-    return getattr(SSL, "TLS_CLIENT_METHOD",
-                   getattr(SSL, "TLS_METHOD", SSL.TLSv1_METHOD))
+    return _default_pyopenssl_ssl_method(SSL)
 
 
 def _check_pyopenssl():
@@ -54,25 +55,7 @@ def _check_pyopenssl():
 
 
 def _build_pyopenssl_context_from_options(ssl_options):
-    ssl_version = (ssl_options["ssl_version"]
-                   if "ssl_version" in ssl_options else _default_ssl_method())
-    context = SSL.Context(ssl_version)
-    if "certfile" in ssl_options:
-        context.use_certificate_file(ssl_options["certfile"])
-    if "keyfile" in ssl_options:
-        context.use_privatekey_file(ssl_options["keyfile"])
-    if "ca_certs" in ssl_options:
-        context.load_verify_locations(ssl_options["ca_certs"])
-    cert_reqs = ssl_options.get("cert_reqs", None)
-    if cert_reqs is None:
-        cert_reqs = (SSL.VERIFY_PEER
-                     if (ssl_options.get("ca_certs", None) or ssl_options.get("check_hostname", False))
-                     else SSL.VERIFY_NONE)
-    context.set_verify(
-        cert_reqs,
-        callback=lambda _connection, _x509, _errnum, _errdepth, ok: ok
-    )
-    return context
+    return _build_pyopenssl_context(SSL, ssl_options)
 
 
 def _cleanup(cleanup_weakref):

tests/unit/io/test_eventletreactor.py

@@ -13,8 +13,10 @@
 # limitations under the License.
 
 import os
+import ssl
 import unittest
 from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
 
 from unittest.mock import Mock, patch
 
@@ -85,12 +87,31 @@ def test_empty_ssl_options_default_to_negotiating_tls(self):
         context_mock.assert_called_once_with(eventletreactor.SSL.TLS_CLIENT_METHOD)
         assert context is context_mock.return_value
 
+    def test_default_ssl_method_falls_back_to_tls_method(self):
+        tls_method = object()
+
+        with patch.object(eventletreactor, 'SSL', SimpleNamespace(TLS_METHOD=tls_method)):
+            assert 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_METHOD})
+                {'ssl_version': eventletreactor.SSL.TLSv1_2_METHOD})
 
-        context_mock.assert_called_once_with(eventletreactor.SSL.TLSv1_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)
@@ -99,6 +120,17 @@ def test_ca_certs_default_to_required_validation(self):
 
         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)
 

tests/unit/io/test_twistedreactor.py

@@ -13,8 +13,10 @@
 # limitations under the License.
 
 import os
+import ssl
 import unittest
 from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
 from unittest.mock import Mock, patch
 
 try:
@@ -76,18 +78,48 @@ def test_empty_ssl_options_default_to_negotiating_tls(self):
         context_mock.assert_called_once_with(twistedreactor.SSL.TLS_CLIENT_METHOD)
         assert context is context_mock.return_value
 
+    def test_default_ssl_method_falls_back_to_tls_method(self):
+        tls_method = object()
+
+        with patch.object(twistedreactor, 'SSL', SimpleNamespace(TLS_METHOD=tls_method)):
+            assert 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_METHOD})
+                {'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.TLSv1_METHOD)
+        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)
 

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())