[PYTHON-1163] Twisted cloud support

Author: TheRealFalcon

cassandra/cluster.py

@@ -36,6 +36,12 @@
 
 import weakref
 from weakref import WeakValueDictionary
+
+try:
+    from cassandra.io.twistedreactor import TwistedConnection
+except ImportError:
+    TwistedConnection = None
+
 try:
     from weakref import WeakSet
 except ImportError:
@@ -906,13 +912,18 @@ def __init__(self,
 
         Any of the mutable Cluster attributes may be set as keyword arguments to the constructor.
         """
+        if connection_class is not None:
+            self.connection_class = connection_class
 
         if cloud is not None:
             if contact_points is not _NOT_SET or endpoint_factory or ssl_context or ssl_options:
                 raise ValueError("contact_points, endpoint_factory, ssl_context, and ssl_options "
                                  "cannot be specified with a cloud configuration")
 
-            cloud_config = dscloud.get_cloud_config(cloud)
+            cloud_config = dscloud.get_cloud_config(
+                cloud,
+                create_pyopenssl_context=self.connection_class is TwistedConnection
+            )
 
             ssl_context = cloud_config.ssl_context
             ssl_options = {'check_hostname': True}
@@ -994,9 +1005,6 @@ def __init__(self,
                 raise TypeError("address_translator should not be a class, it should be an instance of that class")
             self.address_translator = address_translator
 
-        if connection_class is not None:
-            self.connection_class = connection_class
-
         if timestamp_generator is not None:
             if not callable(timestamp_generator):
                 raise ValueError("timestamp_generator must be callable")

cassandra/datastax/cloud/__init__.py

@@ -75,34 +75,37 @@ def from_dict(cls, d):
         return c
 
 
-def get_cloud_config(cloud_config):
+def get_cloud_config(cloud_config, create_pyopenssl_context=False):
     if not _HAS_SSL:
         raise DriverException("A Python installation with SSL is required to connect to a cloud cluster.")
 
     if 'secure_connect_bundle' not in cloud_config:
         raise ValueError("The cloud config doesn't have a secure_connect_bundle specified.")
 
     try:
-        config = read_cloud_config_from_zip(cloud_config)
+        config = read_cloud_config_from_zip(cloud_config, create_pyopenssl_context)
     except BadZipFile:
         raise ValueError("Unable to open the zip file for the cloud config. Check your secure connect bundle.")
 
-    return read_metadata_info(config, cloud_config)
+    config = read_metadata_info(config, cloud_config)
+    if create_pyopenssl_context:
+        config.ssl_context = config.pyopenssl_context
+    return config
 
 
-def read_cloud_config_from_zip(cloud_config):
+def read_cloud_config_from_zip(cloud_config, create_pyopenssl_context):
     secure_bundle = cloud_config['secure_connect_bundle']
     with ZipFile(secure_bundle) as zipfile:
         base_dir = os.path.dirname(secure_bundle)
         tmp_dir = tempfile.mkdtemp(dir=base_dir)
         try:
             zipfile.extractall(path=tmp_dir)
-            return parse_cloud_config(os.path.join(tmp_dir, 'config.json'), cloud_config)
+            return parse_cloud_config(os.path.join(tmp_dir, 'config.json'), cloud_config, create_pyopenssl_context)
         finally:
             shutil.rmtree(tmp_dir)
 
 
-def parse_cloud_config(path, cloud_config):
+def parse_cloud_config(path, cloud_config, create_pyopenssl_context):
     with open(path, 'r') as stream:
         data = json.load(stream)
 
@@ -116,7 +119,11 @@ def parse_cloud_config(path, cloud_config):
         ca_cert_location = os.path.join(config_dir, 'ca.crt')
         cert_location = os.path.join(config_dir, 'cert')
         key_location = os.path.join(config_dir, 'key')
+        # Regardless of if we create a pyopenssl context, we still need the builtin one
+        # to connect to the metadata service
         config.ssl_context = _ssl_context_from_cert(ca_cert_location, cert_location, key_location)
+        if create_pyopenssl_context:
+            config.pyopenssl_context = _pyopenssl_context_from_cert(ca_cert_location, cert_location, key_location)
 
     return config
 
@@ -165,3 +172,17 @@ def _ssl_context_from_cert(ca_cert_location, cert_location, key_location):
     ssl_context.load_cert_chain(certfile=cert_location, keyfile=key_location)
 
     return ssl_context
+
+
+def _pyopenssl_context_from_cert(ca_cert_location, cert_location, key_location):
+    try:
+        from OpenSSL import SSL
+    except ImportError:
+        return None
+    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

cassandra/io/twistedreactor.py

@@ -28,7 +28,7 @@
 from twisted.python.failure import Failure
 from zope.interface import implementer
 
-from cassandra.connection import Connection, ConnectionShutdown, Timer, TimerManager
+from cassandra.connection import Connection, ConnectionShutdown, Timer, TimerManager, ConnectionException
 
 try:
     from OpenSSL import SSL
@@ -77,6 +77,7 @@ def connectionMade(self):
 
     def connectionLost(self, reason):
         # reason is a Failure instance
+        log.debug("Connect lost: %s", reason)
         self.connection.defunct(reason.value)
 
 
@@ -134,9 +135,9 @@ def _on_loop_timer(self):
 
 
 @implementer(IOpenSSLClientConnectionCreator)
-class SSLCreator(object):
-    def __init__(self, host, ssl_context, ssl_options, check_hostname, timeout):
-        self.host = host
+class _SSLCreator(object):
+    def __init__(self, endpoint, ssl_context, ssl_options, check_hostname, timeout):
+        self.endpoint = endpoint
         self.ssl_options = ssl_options
         self.check_hostname = check_hostname
         self.timeout = timeout
@@ -163,9 +164,9 @@ 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.host != connection.get_peer_certificate().get_subject().commonName:
+            if self.check_hostname and self.endpoint.address != connection.get_peer_certificate().get_subject().commonName:
                 transport = connection.get_app_data()
-                transport.failVerification(Failure(Exception("Hostname verification failed")))
+                transport.failVerification(Failure(ConnectionException("Hostname verification failed", self.endpoint)))
 
     def clientConnectionForTLS(self, tlsProtocol):
         connection = SSL.Connection(self.context, None)
@@ -213,7 +214,7 @@ def __init__(self, *args, **kwargs):
         self._loop.maybe_start()
 
     def _check_pyopenssl(self):
-        if self.ssl_options:
+        if self.ssl_context or self.ssl_options:
             if not _HAS_SSL:
                 raise ImportError(
                     str(import_exception) +
@@ -231,29 +232,29 @@ def add_connection(self):
             # Cool they enforce strong security, but we have to be able to turn it off
             self._check_pyopenssl()
 
-            ssl_options = SSLCreator(
-                self.endpoint.address,
+            ssl_connection_creator = _SSLCreator(
+                self.endpoint,
                 self.ssl_context if self.ssl_context else None,
                 self.ssl_options,
                 self._check_hostname,
                 self.connect_timeout,
             )
 
-            point = SSL4ClientEndpoint(
+            endpoint = SSL4ClientEndpoint(
                 reactor,
                 host,
                 port,
-                sslContextFactory=ssl_options,
+                sslContextFactory=ssl_connection_creator,
                 timeout=self.connect_timeout,
             )
         else:
-            point = TCP4ClientEndpoint(
+            endpoint = TCP4ClientEndpoint(
                 reactor,
                 host,
                 port,
                 timeout=self.connect_timeout
             )
-        connectProtocol(point, TwistedConnectionProtocol(self))
+        connectProtocol(endpoint, TwistedConnectionProtocol(self))
 
     def client_connection_made(self, transport):
         """

test-requirements.txt

@@ -7,12 +7,11 @@ unittest2
 pytz
 sure
 pure-sasl
-twisted; python_version >= '3.5'
+twisted[tls]; python_version >= '3.5'
 twisted[tls]==19.2.1; python_version < '3.5'
 gevent>=1.0
 eventlet
 cython>=0.20,<0.30
 packaging
 futurist; python_version >= '3.7'
 asynctest; python_version > '3.4'
-pyopenssl

tests/integration/advanced/cloud/test_cloud.py

@@ -107,7 +107,7 @@ def test_verify_hostname(self):
         with patch('cassandra.datastax.cloud.parse_metadata_info', wraps=self._bad_hostname_metadata):
             with self.assertRaises(NoHostAvailable) as e:
                 self.connect(self.creds)
-            self.assertIn("hostname", str(e.exception))
+            self.assertIn("hostname", str(e.exception).lower())
 
     def test_error_when_bundle_doesnt_exist(self):
         try: