Add PyOpenSSL TLS session caching for Eventlet and Twisted

- EventletConnection: restore cached session before handshake via set_session(),
  store session after do_handshake() via _cache_pyopenssl_session()
- TwistedConnection: pass ssl_session_cache to _SSLCreator, restore cached
  session in clientConnectionForTLS(), store after handshake in info_callback()
- All operations wrapped in try/except for error tolerance
- Debug logging for session reuse and restore/store failures

Author: sylwiaszunejko

cassandra/io/eventletreactor.py

@@ -108,6 +108,16 @@ 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'))
+        # Apply cached TLS session for resumption (PyOpenSSL)
+        if self._ssl_session_cache:
+            cached_session = self._ssl_session_cache.get(
+                self._ssl_session_cache_key())
+            if cached_session:
+                try:
+                    self._socket.set_session(cached_session)
+                    log.debug("Using cached TLS session for %s", self.endpoint)
+                except Exception:
+                    log.debug("Could not restore TLS session for %s", self.endpoint)
 
     def _initiate_connection(self, sockaddr):
         if self.uses_legacy_ssl_options:
@@ -116,6 +126,8 @@ def _initiate_connection(self, sockaddr):
             self._socket.connect(sockaddr)
             if self.ssl_context or self.ssl_options:
                 self._socket.do_handshake()
+                # Store TLS session after successful handshake (PyOpenSSL)
+                self._cache_pyopenssl_session()
 
     def _match_hostname(self):
         if self.uses_legacy_ssl_options:
@@ -126,6 +138,19 @@ def _match_hostname(self):
                 raise Exception("Hostname verification failed! Certificate name '{}' "
                                 "doesn't endpoint '{}'".format(cert_name, self.endpoint.address))
 
+    def _cache_pyopenssl_session(self):
+        """Store the PyOpenSSL TLS session in the cache after a successful handshake."""
+        if self._ssl_session_cache is not None:
+            try:
+                session = self._socket.get_session()
+                if session:
+                    self._ssl_session_cache.set(
+                        self._ssl_session_cache_key(), session)
+                    if self._socket.session_reused():
+                        log.debug("TLS session was reused for %s", self.endpoint)
+            except Exception:
+                log.debug("Could not cache TLS session for %s", self.endpoint)
+
     def close(self):
         with self.lock:
             if self.is_closed:

cassandra/io/twistedreactor.py

@@ -139,11 +139,12 @@ def _on_loop_timer(self):
 
 @implementer(IOpenSSLClientConnectionCreator)
 class _SSLCreator(object):
-    def __init__(self, endpoint, ssl_context, ssl_options, check_hostname, timeout):
+    def __init__(self, endpoint, ssl_context, ssl_options, check_hostname, timeout, ssl_session_cache=None):
         self.endpoint = endpoint
         self.ssl_options = ssl_options
         self.check_hostname = check_hostname
         self.timeout = timeout
+        self.ssl_session_cache = ssl_session_cache
 
         if ssl_context:
             self.context = ssl_context
@@ -170,12 +171,36 @@ def info_callback(self, connection, where, ret):
             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)))
+                return
+            # Store TLS session after successful handshake (PyOpenSSL)
+            if self.ssl_session_cache is not None:
+                try:
+                    session = connection.get_session()
+                    if session:
+                        self.ssl_session_cache.set(
+                            self.endpoint.tls_session_cache_key, session)
+                        if connection.session_reused():
+                            log.debug("TLS session was reused for %s", self.endpoint)
+                except Exception:
+                    log.debug("Could not cache TLS session for %s", self.endpoint)
 
     def clientConnectionForTLS(self, tlsProtocol):
         connection = SSL.Connection(self.context, None)
         connection.set_app_data(tlsProtocol)
         if self.ssl_options and "server_hostname" in self.ssl_options:
             connection.set_tlsext_host_name(self.ssl_options['server_hostname'].encode('ascii'))
+
+        # Apply cached TLS session for resumption (PyOpenSSL)
+        if self.ssl_session_cache is not None:
+            cached_session = self.ssl_session_cache.get(
+                self.endpoint.tls_session_cache_key)
+            if cached_session:
+                try:
+                    connection.set_session(cached_session)
+                    log.debug("Using cached TLS session for %s", self.endpoint)
+                except Exception:
+                    log.debug("Could not restore TLS session for %s", self.endpoint)
+
         return connection
 
 
@@ -241,6 +266,7 @@ def add_connection(self):
                 self.ssl_options,
                 self._check_hostname,
                 self.connect_timeout,
+                ssl_session_cache=self._ssl_session_cache,
             )
 
             endpoint = SSL4ClientEndpoint(

tests/unit/io/test_twistedreactor.py

@@ -13,9 +13,9 @@
 # limitations under the License.
 
 import unittest
-from unittest.mock import Mock, patch
+from unittest.mock import Mock, patch, MagicMock
 
-from cassandra.connection import DefaultEndPoint
+from cassandra.connection import DefaultEndPoint, SSLSessionCache
 
 try:
     from twisted.test import proto_helpers
@@ -197,3 +197,91 @@ def test_push(self, mock_connectTCP):
         self.obj_ut.push('123 pickup')
         self.mock_reactor_cft.assert_called_with(
             transport_mock.write, '123 pickup')
+
+
+class TestSSLCreatorInfoCallback(unittest.TestCase):
+    """Verify that _SSLCreator.info_callback does not cache TLS sessions
+    when hostname verification fails."""
+
+    def setUp(self):
+        if twistedreactor is None:
+            raise unittest.SkipTest("Twisted libraries not available")
+        from OpenSSL import SSL
+        self.SSL = SSL
+
+    def _make_creator(self, check_hostname, endpoint_address, cert_cn,
+                      ssl_session_cache=None):
+        from cassandra.io.twistedreactor import _SSLCreator
+        endpoint = Mock()
+        endpoint.address = endpoint_address
+        endpoint.tls_session_cache_key = (endpoint_address, 9042)
+        ssl_ctx = Mock()
+        creator = _SSLCreator(
+            endpoint=endpoint,
+            ssl_context=ssl_ctx,
+            ssl_options=None,
+            check_hostname=check_hostname,
+            timeout=5,
+            ssl_session_cache=ssl_session_cache,
+        )
+
+        # Build a mock OpenSSL connection
+        connection = Mock()
+        subject = Mock()
+        subject.commonName = cert_cn
+        cert = Mock()
+        cert.get_subject.return_value = subject
+        connection.get_peer_certificate.return_value = cert
+        session = Mock()
+        connection.get_session.return_value = session
+        connection.session_reused.return_value = False
+        transport = Mock()
+        connection.get_app_data.return_value = transport
+
+        return creator, connection, transport, session
+
+    def test_hostname_mismatch_does_not_cache(self):
+        """When hostname verification fails, the session must NOT be cached."""
+        cache = SSLSessionCache()
+        creator, connection, transport, session = self._make_creator(
+            check_hostname=True,
+            endpoint_address='good.example.com',
+            cert_cn='evil.example.com',
+            ssl_session_cache=cache,
+        )
+
+        creator.info_callback(connection, self.SSL.SSL_CB_HANDSHAKE_DONE, 0)
+
+        transport.failVerification.assert_called_once()
+        assert cache.size() == 0, "Session was cached despite hostname mismatch"
+
+    def test_hostname_match_caches_session(self):
+        """When hostname matches, the session should be cached."""
+        cache = SSLSessionCache()
+        creator, connection, transport, session = self._make_creator(
+            check_hostname=True,
+            endpoint_address='good.example.com',
+            cert_cn='good.example.com',
+            ssl_session_cache=cache,
+        )
+
+        creator.info_callback(connection, self.SSL.SSL_CB_HANDSHAKE_DONE, 0)
+
+        transport.failVerification.assert_not_called()
+        assert cache.size() == 1
+        assert cache.get(('good.example.com', 9042)) is session
+
+    def test_no_check_hostname_caches_session(self):
+        """When check_hostname is False, always cache regardless of CN."""
+        cache = SSLSessionCache()
+        creator, connection, transport, session = self._make_creator(
+            check_hostname=False,
+            endpoint_address='good.example.com',
+            cert_cn='evil.example.com',
+            ssl_session_cache=cache,
+        )
+
+        creator.info_callback(connection, self.SSL.SSL_CB_HANDSHAKE_DONE, 0)
+
+        transport.failVerification.assert_not_called()
+        assert cache.size() == 1