Add SSLSessionCache to Cluster for TLS ticket caching

Add ssl_session_cache to Cluster so all managed connections share a
single TLS session store.

- Auto-create SSLSessionCache when ssl_context or ssl_options are set;
  pass ssl_session_cache=None to opt out
- Accept explicit ssl_session_cache to allow a custom instance
- Warn when Twisted/Eventlet connection classes are used (pyOpenSSL
  does not support stdlib ssl session resumption; cache has no effect)
- Pass the cache through connection_factory kwargs to each Connection

Includes unit tests for auto-creation, opt-out, custom cache injection,
factory propagation, and pyOpenSSL warnings.

Author: sylwiaszunejko

cassandra/cluster.py

@@ -52,7 +52,8 @@
 from cassandra.connection import (ClientRoutesEndPointFactory, ConnectionException, ConnectionShutdown,
                                   ConnectionHeartbeat, ProtocolVersionUnsupported,
                                   EndPoint, DefaultEndPoint, DefaultEndPointFactory,
-                                  SniEndPointFactory, ConnectionBusy, locally_supported_compressions)
+                                  SniEndPointFactory, ConnectionBusy, locally_supported_compressions,
+                                  SSLSessionCache)
 from cassandra.cqltypes import UserType
 import cassandra.cqltypes as types
 from cassandra.encoder import Encoder
@@ -876,6 +877,26 @@ def default_retry_policy(self, policy):
     .. versionadded:: 3.17.0
     """
 
+    ssl_session_cache = None
+    """
+    An optional :class:`.connection.SSLSessionCache` instance used to enable TLS
+    session resumption (via session tickets or PSK) for all connections managed
+    by this cluster.
+
+    When :attr:`~Cluster.ssl_context` or :attr:`~Cluster.ssl_options` are set,
+    a cache is created automatically so that reconnections to the same host can
+    skip the full TLS handshake.  Set this to :const:`None` explicitly to
+    disable session caching.
+
+    Note: TLS 1.2 sessions are cached immediately after connect.  TLS 1.3
+    sessions are cached after the CQL handshake completes (Ready / AuthSuccess),
+    because session tickets are sent asynchronously by the server.
+
+    Note: only the stdlib ``ssl`` reactor paths are supported (asyncore, libev,
+    gevent, asyncio).  Twisted and Eventlet connections use pyOpenSSL and are
+    not covered by this cache.
+    """
+
     sockopts = None
     """
     An optional list of tuples which will be used as arguments to
@@ -1217,7 +1238,8 @@ def __init__(self,
                  metadata_request_timeout: Optional[float] = None,
                  column_encryption_policy=None,
                  application_info:Optional[ApplicationInfoBase]=None,
-                 client_routes_config:Optional[ClientRoutesConfig]=None
+                 client_routes_config:Optional[ClientRoutesConfig]=None,
+                 ssl_session_cache=_NOT_SET
                  ):
         """
         ``executor_threads`` defines the number of threads in a pool for handling asynchronous tasks such as
@@ -1461,6 +1483,30 @@ def __init__(self,
 
         self.ssl_options = ssl_options
         self.ssl_context = ssl_context
+
+        # Auto-create a session cache when TLS is enabled, unless the caller
+        # explicitly passed ssl_session_cache (including None to opt out).
+        if ssl_session_cache is _NOT_SET:
+            if ssl_context is not None or ssl_options is not None:
+                self.ssl_session_cache = SSLSessionCache()
+            else:
+                self.ssl_session_cache = None
+        else:
+            self.ssl_session_cache = ssl_session_cache
+
+        # Warn when the session cache won't be used because the connection
+        # class uses pyOpenSSL instead of the stdlib ssl module.
+        if self.ssl_session_cache is not None:
+            uses_twisted = TwistedConnection and issubclass(self.connection_class, TwistedConnection)
+            uses_eventlet = EventletConnection and issubclass(self.connection_class, EventletConnection)
+            if uses_twisted or uses_eventlet:
+                log.warning(
+                    "ssl_session_cache is set but the connection class %s uses "
+                    "pyOpenSSL, which does not support stdlib ssl session "
+                    "resumption.  The cache will have no effect.",
+                    self.connection_class.__name__,
+                )
+
         self.sockopts = sockopts
         self.cql_version = cql_version
         self.max_schema_agreement_wait = max_schema_agreement_wait
@@ -1706,6 +1752,7 @@ def _make_connection_kwargs(self, endpoint, kwargs_dict):
         kwargs_dict.setdefault('sockopts', self.sockopts)
         kwargs_dict.setdefault('ssl_options', self.ssl_options)
         kwargs_dict.setdefault('ssl_context', self.ssl_context)
+        kwargs_dict.setdefault('ssl_session_cache', self.ssl_session_cache)
         kwargs_dict.setdefault('cql_version', self.cql_version)
         kwargs_dict.setdefault('protocol_version', self.protocol_version)
         kwargs_dict.setdefault('user_type_map', self._user_types)

tests/unit/test_cluster.py

@@ -23,6 +23,7 @@
     InvalidRequest, Unauthorized, AuthenticationFailed, OperationTimedOut, UnsupportedOperation, RequestValidationException, ConfigurationException, ProtocolVersion
 from cassandra.cluster import _Scheduler, Session, Cluster, default_lbp_factory, \
     ExecutionProfile, _ConfigMode, EXEC_PROFILE_DEFAULT
+from cassandra.connection import SSLSessionCache
 from cassandra.pool import Host
 from cassandra.policies import HostDistance, RetryPolicy, RoundRobinPolicy, DowngradingConsistencyRetryPolicy, SimpleConvictionPolicy
 from cassandra.query import SimpleStatement, named_tuple_factory, tuple_factory
@@ -634,3 +635,95 @@ def test_no_warning_adding_lbp_ep_to_cluster_with_contact_points(self):
             )
 
         patched_logger.warning.assert_not_called()
+
+
+class TestSSLSessionCacheAutoCreation(unittest.TestCase):
+
+    def test_cache_created_when_ssl_context_set(self):
+        import ssl
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.check_hostname = False
+        ctx.verify_mode = ssl.CERT_NONE
+        cluster = Cluster(contact_points=['127.0.0.1'], ssl_context=ctx)
+        assert isinstance(cluster.ssl_session_cache, SSLSessionCache)
+
+    def test_cache_created_when_ssl_options_set(self):
+        cluster = Cluster(contact_points=['127.0.0.1'], ssl_options={'ca_certs': '/dev/null'})
+        assert isinstance(cluster.ssl_session_cache, SSLSessionCache)
+
+    def test_no_cache_when_tls_not_enabled(self):
+        cluster = Cluster(contact_points=['127.0.0.1'])
+        assert cluster.ssl_session_cache is None
+
+    def test_explicit_none_disables_cache(self):
+        import ssl
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.check_hostname = False
+        ctx.verify_mode = ssl.CERT_NONE
+        cluster = Cluster(contact_points=['127.0.0.1'], ssl_context=ctx,
+                          ssl_session_cache=None)
+        assert cluster.ssl_session_cache is None
+
+    def test_explicit_custom_cache_used(self):
+        import ssl
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.check_hostname = False
+        ctx.verify_mode = ssl.CERT_NONE
+        custom = SSLSessionCache()
+        cluster = Cluster(contact_points=['127.0.0.1'], ssl_context=ctx,
+                          ssl_session_cache=custom)
+        assert cluster.ssl_session_cache is custom
+
+    def test_cache_passed_to_connection_factory(self):
+        import ssl
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.check_hostname = False
+        ctx.verify_mode = ssl.CERT_NONE
+        endpoint = Mock(address='127.0.0.1')
+        with patch.object(Cluster.connection_class, 'factory', autospec=True, return_value='connection') as factory:
+            cluster = Cluster(contact_points=['127.0.0.1'], ssl_context=ctx)
+            cluster.connection_factory(endpoint)
+
+        assert factory.call_args.kwargs['ssl_session_cache'] is cluster.ssl_session_cache
+
+    def test_warning_for_eventlet_connection_class(self):
+        """A warning is logged when ssl_session_cache is set with EventletConnection."""
+        import ssl
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.check_hostname = False
+        ctx.verify_mode = ssl.CERT_NONE
+
+        # Create a real class so issubclass() works throughout Cluster.__init__
+        from cassandra.connection import Connection as BaseConn
+        class FakeEventletConnection(BaseConn):
+            pass
+
+        with patch('cassandra.cluster.EventletConnection', FakeEventletConnection, create=True), \
+             patch('cassandra.cluster.log') as patched_logger:
+            Cluster(contact_points=['127.0.0.1'], ssl_context=ctx,
+                    connection_class=FakeEventletConnection)
+
+        # At least one warning about pyOpenSSL
+        warning_calls = [c for c in patched_logger.warning.call_args_list
+                         if 'pyOpenSSL' in str(c)]
+        assert len(warning_calls) == 1
+
+    def test_warning_for_twisted_connection_class(self):
+        """A warning is logged when ssl_session_cache is set with TwistedConnection."""
+        import ssl
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.check_hostname = False
+        ctx.verify_mode = ssl.CERT_NONE
+
+        from cassandra.connection import Connection as BaseConn
+        class FakeTwistedConnection(BaseConn):
+            pass
+
+        with patch('cassandra.cluster.TwistedConnection', FakeTwistedConnection, create=True), \
+             patch('cassandra.cluster.log') as patched_logger:
+            Cluster(contact_points=['127.0.0.1'], ssl_context=ctx,
+                    connection_class=FakeTwistedConnection)
+
+        warning_calls = [c for c in patched_logger.warning.call_args_list
+                         if 'pyOpenSSL' in str(c)]
+        assert len(warning_calls) == 1