Add PyOpenSSL TLS session caching for Eventlet and Twisted

- EventletConnection: restore cached session before handshake via set_session()
- TwistedConnection: pass ssl_session_cache to _SSLCreator, restore cached
  session in clientConnectionForTLS()
- Both reactors: defer session storage to _cache_tls_session_if_needed() override
  called at ReadyMessage / AuthSuccessMessage time, ensuring TLS 1.3 session
  tickets (which arrive after the first application-data exchange) are captured
- Skip caching when session_reused() is True (abbreviated handshake)
- 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,22 @@ 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 is not None:
+            cached_session = self._ssl_session_cache.get(
+                self._ssl_session_cache_key())
+            if cached_session is not None:
+                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)
+
+        # Return the SSL.Connection so that the base-class _connect_socket() can
+        # store it in self._socket via:  self._socket = self._wrap_socket_from_context()
+        # Without this return the assignment would overwrite self._socket with None,
+        # breaking every subsequent call on the socket.
+        return self._socket
 
     def _initiate_connection(self, sockaddr):
         if self.uses_legacy_ssl_options:
@@ -116,6 +132,11 @@ def _initiate_connection(self, sockaddr):
             self._socket.connect(sockaddr)
             if self.ssl_context or self.ssl_options:
                 self._socket.do_handshake()
+                # No early session caching here.  _cache_tls_session_if_needed() is
+                # called by the base class at ReadyMessage / AuthSuccessMessage time,
+                # which is the correct point for both TLS 1.2 and TLS 1.3 (for TLS 1.3
+                # the resumable session ticket is only available after the first
+                # application-data exchange).
 
     def _match_hostname(self):
         if self.uses_legacy_ssl_options:
@@ -126,6 +147,35 @@ def _match_hostname(self):
                 raise Exception("Hostname verification failed! Certificate name '{}' "
                                 "doesn't endpoint '{}'".format(cert_name, self.endpoint.address))
 
+    def _cache_tls_session_if_needed(self):
+        """
+        PyOpenSSL override of :meth:`.Connection._cache_tls_session_if_needed`.
+
+        Uses ``SSL.Connection.get_session()`` / ``session_reused()`` instead of
+        the stdlib ``ssl.SSLSocket.session`` / ``session_reused`` attributes.
+        Called by the base class at ReadyMessage / AuthSuccessMessage time so
+        that TLS 1.3 session tickets — which arrive after the first
+        application-data exchange — are captured correctly.
+
+        Falls back to the base-class implementation for the legacy
+        ``ssl_options``-only path, which uses a stdlib ``ssl.SSLSocket``.
+        """
+        if self.uses_legacy_ssl_options:
+            super(EventletConnection, self)._cache_tls_session_if_needed()
+            return
+        if self._ssl_session_cache is None or not (self.ssl_context or self.ssl_options):
+            return
+        try:
+            if self._socket.session_reused():
+                log.debug("TLS session was reused for %s", self.endpoint)
+            else:
+                session = self._socket.get_session()
+                if session is not None:
+                    self._ssl_session_cache.set(
+                        self._ssl_session_cache_key(), session)
+        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,15 @@ 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
+        # Populated by info_callback after every successful handshake; read by
+        # TwistedConnection._cache_tls_session_if_needed() after the first CQL exchange.
+        self._ssl_connection = None
 
         if ssl_context:
             self.context = ssl_context
@@ -170,12 +174,41 @@ 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 the live SSL.Connection so that TwistedConnection._cache_tls_session_if_needed()
+            # can read the final session after the first CQL exchange.  Session caching is
+            # deliberately deferred to that point: for TLS 1.3 the resumable session ticket
+            # is only available after the first application-data record, which arrives with
+            # the CQL ReadyMessage or AuthSuccessMessage.
+            self._ssl_connection = connection
+            # For TLS 1.2 the session is already available at handshake completion; cache
+            # it immediately.  For TLS 1.3 get_session() returns None here and the deferred
+            # path in TwistedConnection._cache_tls_session_if_needed() handles it instead.
+            if self.ssl_session_cache is not None and not connection.session_reused():
+                session = connection.get_session()
+                if session is not None:
+                    self.ssl_session_cache.set(self.endpoint.tls_session_cache_key, session)
 
     def clientConnectionForTLS(self, tlsProtocol):
+        # Reset any reference from a previous connection attempt on this creator.
+        self._ssl_connection = None
+
         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
 
 
@@ -212,6 +245,7 @@ def __init__(self, *args, **kwargs):
         self.is_closed = True
         self.connector = None
         self.transport = None
+        self._ssl_creator = None  # set in add_connection() when SSL is used
 
         reactor.callFromThread(self.add_connection)
         self._loop.maybe_start()
@@ -241,7 +275,9 @@ def add_connection(self):
                 self.ssl_options,
                 self._check_hostname,
                 self.connect_timeout,
+                ssl_session_cache=self._ssl_session_cache,
             )
+            self._ssl_creator = ssl_connection_creator
 
             endpoint = SSL4ClientEndpoint(
                 reactor,
@@ -259,6 +295,31 @@ def add_connection(self):
             )
         connectProtocol(endpoint, TwistedConnectionProtocol(self))
 
+    def _cache_tls_session_if_needed(self):
+        """
+        PyOpenSSL override of :meth:`.Connection._cache_tls_session_if_needed`.
+
+        Called by the base class at ReadyMessage / AuthSuccessMessage time —
+        after the first application-data exchange.  For TLS 1.3 this is the
+        earliest point at which the resumable session ticket is available;
+        for TLS 1.2 the session is also valid here.
+        """
+        if self._ssl_session_cache is None or self._ssl_creator is None:
+            return
+        ssl_conn = self._ssl_creator._ssl_connection
+        if ssl_conn is None:
+            return
+        try:
+            if ssl_conn.session_reused():
+                log.debug("TLS session was reused for %s", self.endpoint)
+            else:
+                session = ssl_conn.get_session()
+                if session is not None:
+                    self._ssl_session_cache.set(
+                        self.endpoint.tls_session_cache_key, session)
+        except Exception:
+            log.debug("Could not cache TLS session for %s", self.endpoint)
+
     def client_connection_made(self, transport):
         """
         Called by twisted protocol when a connection attempt has

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