Merge pull request datastax#933 from datastax/python-343

PYTHON-343 SSL support for Twisted

Author: beltran

CHANGELOG.rst

@@ -6,6 +6,7 @@ Features
 * Add one() function to the ResultSet API (PYTHON-947)
 * Create an utility function to fetch concurrently many keys from the same replica (PYTHON-647)
 * Allow filter queries with fields that have an index managed outside of cqlengine (PYTHON-966)
+* Twisted SSL Support (PYTHON-343)
 * Support IS NOT NULL operator in cqlengine (PYTHON-968)
 
 Other

build.yaml

@@ -118,13 +118,19 @@ build:
       export PATH=$JAVA_HOME/bin:$PATH
       export PYTHONPATH=""
 
+      # Install latest setuptools
+      pip install --upgrade pip
+      pip install -U setuptools
+
       pip install git+https://github.com/pcmanus/ccm.git
       # Install dependencies
-      sudo apt-get install -y libev4 libev-dev
+      sudo apt-get install -y libev4 libev-dev libssl-dev
 
       pip install -r test-requirements.txt
       pip install nose-ignore-docstring
       pip install nose-exclude
+      pip install service_identity
+
       FORCE_CYTHON=False
       if [[ $CYTHON == 'CYTHON' ]]; then
         FORCE_CYTHON=True

cassandra/io/twistedreactor.py

@@ -42,6 +42,9 @@ class TwistedConnectionProtocol(protocol.Protocol):
     made events.
     """
 
+    def __init__(self):
+        self.connection = None
+
     def dataReceived(self, data):
         """
         Callback function that is called when data has been received
@@ -50,21 +53,27 @@ def dataReceived(self, data):
         Reaches back to the Connection object and queues the data for
         processing.
         """
-        self.transport.connector.factory.conn._iobuf.write(data)
-        self.transport.connector.factory.conn.handle_read()
-
+        self.connection._iobuf.write(data)
+        self.connection.handle_read()
     def connectionMade(self):
         """
         Callback function that is called when a connection has succeeded.
 
         Reaches back to the Connection object and confirms that the connection
         is ready.
         """
-        self.transport.connector.factory.conn.client_connection_made()
+        try:
+            # Non SSL connection
+            self.connection = self.transport.connector.factory.conn
+        except AttributeError:
+            # SSL connection
+            self.connection = self.transport.connector.factory.wrappedFactory.conn
+
+        self.connection.client_connection_made(self.transport)
 
     def connectionLost(self, reason):
         # reason is a Failure instance
-        self.transport.connector.factory.conn.defunct(reason.value)
+        self.connection.defunct(reason.value)
 
 
 class TwistedConnectionClientFactory(protocol.ClientFactory):
@@ -157,6 +166,49 @@ def _on_loop_timer(self):
         self._schedule_timeout(self._timers.next_timeout)
 
 
+try:
+    from twisted.internet import ssl
+    import OpenSSL.crypto
+    from OpenSSL.crypto import load_certificate, FILETYPE_PEM
+
+    class _SSLContextFactory(ssl.ClientContextFactory):
+        def __init__(self, ssl_options, check_hostname, host):
+            self.ssl_options = ssl_options
+            self.check_hostname = check_hostname
+            self.host = host
+
+        def getContext(self):
+            # This version has to be OpenSSL.SSL.DESIRED_VERSION
+            # instead of ssl.DESIRED_VERSION as in other loops
+            self.method = self.ssl_options["ssl_version"]
+            context = ssl.ClientContextFactory.getContext(self)
+            if "certfile" in self.ssl_options:
+                context.use_certificate_file(self.ssl_options["certfile"])
+            if "keyfile" in self.ssl_options:
+                context.use_privatekey_file(self.ssl_options["keyfile"])
+            if "ca_certs" in self.ssl_options:
+                x509 = load_certificate(FILETYPE_PEM, open(self.ssl_options["ca_certs"]).read())
+                store = context.get_cert_store()
+                store.add_cert(x509)
+            if "cert_reqs" in self.ssl_options:
+                # This expects OpenSSL.SSL.VERIFY_NONE/OpenSSL.SSL.VERIFY_PEER
+                # or OpenSSL.SSL.VERIFY_FAIL_IF_NO_PEER_CERT
+                context.set_verify(self.ssl_options["cert_reqs"],
+                                   callback=self.verify_callback)
+            return context
+
+        def verify_callback(self, connection, x509, errnum, errdepth, ok):
+            if ok:
+                if self.check_hostname and self.host != x509.get_subject().commonName:
+                    return False
+            return ok
+
+    _HAS_SSL = True
+
+except ImportError as e:
+    _HAS_SSL = False
+
+
 class TwistedConnection(Connection):
     """
     An implementation of :class:`.Connection` that utilizes the
@@ -189,6 +241,7 @@ def __init__(self, *args, **kwargs):
 
         self.is_closed = True
         self.connector = None
+        self.transport = None
 
         reactor.callFromThread(self.add_connection)
         self._loop.maybe_start()
@@ -198,18 +251,33 @@ def add_connection(self):
         Convenience function to connect and store the resulting
         connector.
         """
-        self.connector = reactor.connectTCP(
-            host=self.host, port=self.port,
-            factory=TwistedConnectionClientFactory(self),
-            timeout=self.connect_timeout)
-
-    def client_connection_made(self):
+        if self.ssl_options:
+
+            if not _HAS_SSL:
+                raise ImportError(
+                    str(e) +
+                    ', pyOpenSSL must be installed to enable SSL support with the Twisted event loop'
+                )
+
+            self.connector = reactor.connectSSL(
+                host=self.host, port=self.port,
+                factory=TwistedConnectionClientFactory(self),
+                contextFactory=_SSLContextFactory(self.ssl_options, self._check_hostname, self.host),
+                timeout=self.connect_timeout)
+        else:
+            self.connector = reactor.connectTCP(
+                host=self.host, port=self.port,
+                factory=TwistedConnectionClientFactory(self),
+                timeout=self.connect_timeout)
+
+    def client_connection_made(self, transport):
         """
         Called by twisted protocol when a connection attempt has
         succeeded.
         """
         with self.lock:
             self.is_closed = False
+        self.transport = transport
         self._send_options_message()
 
     def close(self):
@@ -222,7 +290,7 @@ def close(self):
             self.is_closed = True
 
         log.debug("Closing connection (%s) to %s", id(self), self.host)
-        self.connector.disconnect()
+        reactor.callFromThread(self.connector.disconnect)
         log.debug("Closed socket to %s", self.host)
 
         if not self.is_defunct:
@@ -246,4 +314,4 @@ def push(self, data):
         it is not thread-safe, so we schedule it to run from within
         the event loop when it gets the chance.
         """
-        reactor.callFromThread(self.connector.transport.write, data)
+        reactor.callFromThread(self.transport.write, data)

docs/security.rst

@@ -87,3 +87,6 @@ This is only an example to show how to pass the ssl parameters. Consider reading
 the `python ssl documentation <https://docs.python.org/2/library/ssl.html#ssl.wrap_socket>`_ for
 your configuration. For further reading, Andrew Mussey has published a thorough guide on
 `Using SSL with the DataStax Python driver <http://blog.amussey.com/post/64036730812/cassandra-2-0-client-server-ssl-with-datastax-python>`_.
+
+*Note*: In case the twisted event loop is used pyOpenSSL must be installed or an exception will be risen. Also
+to set the ``ssl_version`` and ``cert_reqs`` in ``ssl_opts`` the appropriate constants from pyOpenSSL are expected.

test-requirements.txt

@@ -8,7 +8,7 @@ PyYAML
 pytz
 sure
 pure-sasl
-twisted
+twisted[tls]
 gevent>=1.0
 eventlet
 cython>=0.20,<0.28

tests/integration/long/test_ssl.py

@@ -21,7 +21,7 @@
 from cassandra.cluster import Cluster, NoHostAvailable
 from cassandra import ConsistencyLevel
 from cassandra.query import SimpleStatement
-from tests.integration import PROTOCOL_VERSION, get_cluster, remove_cluster, use_single_node
+from tests.integration import PROTOCOL_VERSION, get_cluster, remove_cluster, use_single_node, EVENT_LOOP_MANAGER
 
 log = logging.getLogger(__name__)
 
@@ -37,6 +37,16 @@
 DRIVER_CERTFILE = "tests/integration/long/ssl/driver.pem"
 DRIVER_CERTFILE_BAD = "tests/integration/long/ssl/python_driver_bad.pem"
 
+if "twisted" in EVENT_LOOP_MANAGER:
+    import OpenSSL
+    ssl_version = OpenSSL.SSL.TLSv1_METHOD
+    verify_certs = {'cert_reqs': OpenSSL.SSL.VERIFY_PEER,
+                    'check_hostname': True}
+
+else:
+    ssl_version = ssl.PROTOCOL_TLSv1
+    verify_certs = {'cert_reqs': ssl.CERT_REQUIRED,
+                    'check_hostname': True}
 
 def setup_cluster_ssl(client_auth=False):
     """
@@ -130,7 +140,7 @@ def test_can_connect_with_ssl_ca(self):
 
         # find absolute path to client CA_CERTS
         abs_path_ca_cert_path = os.path.abspath(CLIENT_CA_CERTS)
-        ssl_options = {'ca_certs': abs_path_ca_cert_path,'ssl_version': ssl.PROTOCOL_TLSv1}
+        ssl_options = {'ca_certs': abs_path_ca_cert_path,'ssl_version': ssl_version}
         validate_ssl_options(ssl_options=ssl_options)
 
     def test_can_connect_with_ssl_long_running(self):
@@ -147,7 +157,7 @@ def test_can_connect_with_ssl_long_running(self):
         # find absolute path to client CA_CERTS
         abs_path_ca_cert_path = os.path.abspath(CLIENT_CA_CERTS)
         ssl_options = {'ca_certs': abs_path_ca_cert_path,
-                       'ssl_version': ssl.PROTOCOL_TLSv1}
+                       'ssl_version': ssl_version}
         tries = 0
         while True:
             if tries > 5:
@@ -187,9 +197,8 @@ def test_can_connect_with_ssl_ca_host_match(self):
         # find absolute path to client CA_CERTS
         abs_path_ca_cert_path = os.path.abspath(CLIENT_CA_CERTS)
         ssl_options = {'ca_certs': abs_path_ca_cert_path,
-                       'ssl_version': ssl.PROTOCOL_TLSv1,
-                       'cert_reqs': ssl.CERT_REQUIRED,
-                       'check_hostname': True}
+                       'ssl_version': ssl_version}
+        ssl_options.update(verify_certs)
 
         validate_ssl_options(ssl_options=ssl_options)
 
@@ -225,7 +234,7 @@ def test_can_connect_with_ssl_client_auth(self):
         abs_driver_keyfile = os.path.abspath(DRIVER_KEYFILE)
         abs_driver_certfile = os.path.abspath(DRIVER_CERTFILE)
         ssl_options = {'ca_certs': abs_path_ca_cert_path,
-                       'ssl_version': ssl.PROTOCOL_TLSv1,
+                       'ssl_version': ssl_version,
                        'keyfile': abs_driver_keyfile,
                        'certfile': abs_driver_certfile}
         validate_ssl_options(ssl_options)
@@ -251,11 +260,11 @@ def test_can_connect_with_ssl_client_auth_host_name(self):
         abs_driver_certfile = os.path.abspath(DRIVER_CERTFILE)
 
         ssl_options = {'ca_certs': abs_path_ca_cert_path,
-                       'ssl_version': ssl.PROTOCOL_TLSv1,
+                       'ssl_version': ssl_version,
                        'keyfile': abs_driver_keyfile,
-                       'certfile': abs_driver_certfile,
-                       'cert_reqs': ssl.CERT_REQUIRED,
-                       'check_hostname': True}
+                       'certfile': abs_driver_certfile}
+        ssl_options.update(verify_certs)
+
         validate_ssl_options(ssl_options)
 
     def test_cannot_connect_without_client_auth(self):
@@ -273,7 +282,7 @@ def test_cannot_connect_without_client_auth(self):
 
         abs_path_ca_cert_path = os.path.abspath(CLIENT_CA_CERTS)
         cluster = Cluster(protocol_version=PROTOCOL_VERSION, ssl_options={'ca_certs': abs_path_ca_cert_path,
-                                                                          'ssl_version': ssl.PROTOCOL_TLSv1})
+                                                                          'ssl_version': ssl_version})
         # attempt to connect and expect an exception
 
         with self.assertRaises(NoHostAvailable) as context:
@@ -300,7 +309,7 @@ def test_cannot_connect_with_bad_client_auth(self):
         abs_driver_certfile = os.path.abspath(DRIVER_CERTFILE_BAD)
 
         cluster = Cluster(protocol_version=PROTOCOL_VERSION, ssl_options={'ca_certs': abs_path_ca_cert_path,
-                                                                          'ssl_version': ssl.PROTOCOL_TLSv1,
+                                                                          'ssl_version': ssl_version,
                                                                           'keyfile': abs_driver_keyfile,
                                                                           'certfile': abs_driver_certfile})
         with self.assertRaises(NoHostAvailable) as context:

tests/unit/io/test_twistedreactor.py

@@ -156,7 +156,7 @@ def test_client_connection_made(self):
         client_connection_made()
         """
         self.obj_ut._send_options_message = Mock()
-        self.obj_ut.client_connection_made()
+        self.obj_ut.client_connection_made(Mock())
         self.obj_ut._send_options_message.assert_called_with()
 
     @patch('twisted.internet.reactor.connectTCP')
@@ -168,7 +168,7 @@ def test_close(self, mock_connectTCP):
         self.obj_ut.add_connection()
         self.obj_ut.is_closed = False
         self.obj_ut.close()
-        self.obj_ut.connector.disconnect.assert_called_with()
+
         self.assertTrue(self.obj_ut.connected_event.is_set())
         self.assertTrue(self.obj_ut.error_all_requests.called)
 
@@ -217,6 +217,8 @@ def test_push(self, mock_connectTCP):
         Verifiy that push() calls transport.write(data).
         """
         self.obj_ut.add_connection()
+        transport_mock = Mock()
+        self.obj_ut.transport = transport_mock
         self.obj_ut.push('123 pickup')
         self.mock_reactor_cft.assert_called_with(
-            self.obj_ut.connector.transport.write, '123 pickup')
+            transport_mock.write, '123 pickup')