Add TLS session resumption via SSLSessionCache

cassandra/cluster.py

@@ -51,7 +51,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
@@ -866,6 +867,69 @@ def default_retry_policy(self, policy):
     .. versionadded:: 3.17.0
     """
 
+    ssl_session_cache = None
+    """
+    A :class:`~cassandra.connection.SSLSessionCache` shared by every
+    connection this cluster opens, letting them resume TLS sessions instead of
+    performing a full handshake each time.  This matters most for the group of
+    per-shard connections opened to a node at once, and for reconnections.
+
+    One is created automatically when :attr:`~Cluster.ssl_context` is set.
+    That is settled again when :meth:`~.Cluster.connect` is called, against
+    whatever :attr:`~Cluster.ssl_context` and :attr:`~Cluster.connection_class`
+    are in force by then, so configuring TLS after construction still gets a
+    cache -- and swapping in a connection class that cannot resume still turns
+    resumption off rather than handing the class a keyword it does not take.
+
+    A cache created here is reachable only through this attribute, so it and
+    the sessions in it go when the cluster does.  A cache passed in stays the
+    caller's: :meth:`~.Cluster.shutdown` leaves its entries alone, so several
+    clusters -- at the same time or one after another -- can share the
+    sessions in it.  Its entries hold the ``SSLContext`` they were established
+    with, bounded by the cache's
+    :attr:`~cassandra.connection.SSLSessionCache.max_size`; call
+    :meth:`~cassandra.connection.SSLSessionCache.clear` to release them.
+
+    Assigning this attribute is honoured up to :meth:`~.Cluster.connect`,
+    which is where the decision is settled: a cache put here that cannot be
+    used is dropped rather than left to fill with nothing, and the reason is
+    logged once for the cluster.
+
+    Pass ``ssl_session_cache=None`` to :class:`.Cluster` to turn resumption
+    off, or pass your own instance to size it or to share it between
+    clusters::
+
+        from cassandra.connection import SSLSessionCache
+
+        cluster = Cluster(ssl_context=ssl_context,
+                          ssl_session_cache=SSLSessionCache(max_size=64))
+
+    Resumption is available when TLS is configured through
+    :attr:`~Cluster.ssl_context` and the reactor establishes TLS with the
+    standard library's ``ssl`` module: the ``libev`` reactor, and ``asyncore``
+    on the Python versions that still ship it, which is up to 3.11.  Which of
+    them is the default depends on what can be imported -- libev first, then
+    asyncore, then asyncio -- so on Python 3.12 and newer without the libev
+    extension the default is the ``asyncio`` reactor, and resumption is off.
+
+    It is not available with the deprecated :attr:`~Cluster.ssl_options`-only
+    configuration, because each connection builds its own ``SSLContext`` and a
+    session cannot be replayed onto a different one; nor on the ``asyncio``
+    reactor, which performs the handshake inside
+    ``loop.create_connection()``, leaving no point at which to restore a
+    session.  In those cases no cache is created and connections handshake in
+    full.
+
+    It equally requires the server to hand out something it will honour later.
+    Scylla issues session tickets only when ``enable_session_tickets`` is set
+    in its ``client_encryption_options``, which is off by default; without it
+    nothing resumes and every connection performs a full handshake, as it would
+    have anyway.  Over TLS 1.3 the cache then stays empty, while over TLS 1.2
+    such a server still assigns a session id, so the cache may hold an entry it
+    will not honour -- offering that costs nothing and the handshake simply
+    completes in full.
+    """
+
     sockopts = None
     """
     An optional list of tuples which will be used as arguments to
@@ -1221,7 +1285,8 @@ def __init__(self,
                  application_info:Optional[ApplicationInfoBase]=None,
                  client_routes_config:Optional[ClientRoutesConfig]=None,
                  allow_control_connection_query_fallback:Optional[ControlConnectionQueryFallback]=ControlConnectionQueryFallback.Disabled,
-                 driver_config_reporting_enabled=True
+                 driver_config_reporting_enabled=True,
+                 ssl_session_cache=_NOT_SET
                  ):
         """
         ``executor_threads`` defines the number of threads in a pool for handling asynchronous tasks such as
@@ -1468,6 +1533,14 @@ def __init__(self,
 
         self.ssl_options = ssl_options
         self.ssl_context = ssl_context
+
+        self._ssl_session_cache_warned = False
+        self._ssl_session_cache_explicit = ssl_session_cache is not _NOT_SET
+        self._ssl_session_cache_requested = (
+            ssl_session_cache if self._ssl_session_cache_explicit else None)
+        self.ssl_session_cache = None
+        self._decide_tls_session_cache()
+
         # Materialized once: these are applied to every socket the cluster opens
         # and are read again to build the configuration report, so a one-shot
         # iterable would leave whichever consumer ran second with nothing at all.
@@ -1680,6 +1753,62 @@ def add_execution_profile(self, name, profile, pool_wait_timeout=5):
             raise OperationTimedOut("Failed to create all new connection pools in the %ss timeout." % pool_wait_timeout,
                                     timeout=pool_wait_timeout)
 
+    def _decide_tls_session_cache(self):
+        """
+        Settle whether this cluster caches TLS sessions, and in what.
+
+        Run at construction, so that :attr:`ssl_session_cache` reads as
+        documented straight away, and again from :meth:`connect`, because both
+        inputs are public attributes that can be reassigned in between: a
+        decision made from the pair given to the constructor would leave
+        resumption off on a reactor that does support it, or hand the keyword
+        to a connection class that does not take it.
+
+        Resumption needs the session to be replayable onto the same
+        ``SSLContext``, and a reactor that gives the driver a chance to offer it
+        before the handshake.  connection_class is not required to derive from
+        Connection, so one that does not report the capability is treated as
+        lacking it.
+        """
+        resumable = (self.ssl_context is not None and
+                     getattr(self.connection_class,
+                             'supports_tls_session_resumption', False))
+
+        if resumable:
+            if self.ssl_session_cache is None:
+                self.ssl_session_cache = (self._ssl_session_cache_requested
+                                          if self._ssl_session_cache_explicit
+                                          else SSLSessionCache())
+            return
+
+        wanted = (self.ssl_session_cache is not None
+                  or self._ssl_session_cache_requested is not None)
+        if not wanted or self._ssl_session_cache_warned:
+            self.ssl_session_cache = None
+            return
+
+        # Asking for resumption and silently getting none is worse than not
+        # having it: the cache stays reachable and empty, with nothing to
+        # explain why.
+        if self.ssl_context is None:
+            reason = ('no ssl_context is configured, and a session cannot be '
+                      'replayed onto the fresh context each connection builds '
+                      'from ssl_options')
+        else:
+            reason = ('%s cannot restore a session before the handshake' %
+                      getattr(self.connection_class, '__name__',
+                              self.connection_class))
+        log.warning('ssl_session_cache is set but TLS session resumption is '
+                    'unavailable here, so no sessions will be cached: %s.',
+                    reason)
+        # Dropped rather than kept unused, so that this attribute means "the
+        # cache these connections use" throughout: a cache left here would be
+        # handed to every connection -- which a connection class that does not
+        # take the keyword cannot even accept -- and would sit reachable and
+        # empty for anyone reading it back.
+        self.ssl_session_cache = None
+        self._ssl_session_cache_warned = True
+
     def connection_factory(self, endpoint, host_conn = None, *args, **kwargs):
         """
         Called to create a new connection with proper configuration.
@@ -1701,6 +1830,12 @@ 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)
+        if self.ssl_session_cache is not None:
+            # Set only where resumption is possible, so this is also the test
+            # for that: a connection class that does not accept the keyword
+            # should not have to grow one for a cluster that will never cache a
+            # session.
+            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)
@@ -1760,6 +1895,9 @@ def connect(self, keyspace=None, wait_for_all_pools=False):
                           self.contact_points, self.protocol_version)
                 self.connection_class.initialize_reactor()
                 _register_cluster_shutdown(self)
+                # Both inputs are public attributes, so the decision is settled
+                # against the pair actually in force before anything is opened.
+                self._decide_tls_session_cache()
 
                 try:
                     self.control_connection.connect()
@@ -1849,6 +1987,11 @@ def shutdown(self):
         if self.metrics_enabled and self.metrics:
             self.metrics.shutdown()
 
+        # Nothing to do here for ssl_session_cache: a cache created for this
+        # cluster is reachable only through it and goes when it does, and a
+        # cache the caller supplied is the caller's to empty -- deleting rows
+        # in it here would defeat sharing one so that sessions outlive a
+        # cluster.  See the attribute's documentation.
         _discard_cluster_shutdown(self)
 
     def __enter__(self):