Refactor TLS session caching to use abstractions and endpoint-based API

Changes requested by @dkropachev:
1. Changed set_session() to accept Endpoint objects instead of host/port
2. Added configuration option to cache by host only or by host+port
3. Created abstract base classes (TLSSessionCache, TLSSessionCacheOptions)
4. Moved implementations to cassandra/tls.py module
5. Updated Cluster to use TLSSessionCacheOptions for configuration

Benefits:
- More flexible caching strategies (by host or by host+port)
- Cleaner separation of concerns with abstractions
- Easier to extend with custom implementations
- Single configuration object instead of multiple parameters

Co-authored-by: dkropachev <40304587+dkropachev@users.noreply.github.com>

Author: Copilot

cassandra/cluster.py

@@ -200,6 +200,74 @@ def _connection_reduce_fn(val,import_fn):
 _DEFAULT_TLS_SESSION_CACHE_TTL = 3600  # 1 hour in seconds
 
 
+# Abstract base classes for TLS session caching - implementations in cassandra.tls
+from abc import ABC, abstractmethod
+
+
+class TLSSessionCache(ABC):
+    """
+    Abstract base class for TLS session caching.
+    
+    Implementations should provide thread-safe caching of TLS sessions
+    to enable session resumption for faster reconnections.
+    """
+    
+    @abstractmethod
+    def get_session(self, endpoint):
+        """
+        Get a cached TLS session for the given endpoint.
+        
+        Args:
+            endpoint: The EndPoint object representing the connection target
+            
+        Returns:
+            ssl.SSLSession object if a valid cached session exists, None otherwise
+        """
+        pass
+    
+    @abstractmethod
+    def set_session(self, endpoint, session):
+        """
+        Store a TLS session for the given endpoint.
+        
+        Args:
+            endpoint: The EndPoint object representing the connection target
+            session: The ssl.SSLSession object to cache
+        """
+        pass
+    
+    @abstractmethod
+    def clear_expired(self):
+        """Remove all expired sessions from the cache."""
+        pass
+    
+    @abstractmethod
+    def clear(self):
+        """Clear all sessions from the cache."""
+        pass
+    
+    @abstractmethod
+    def size(self):
+        """Return the current number of cached sessions."""
+        pass
+
+
+class TLSSessionCacheOptions(ABC):
+    """
+    Abstract base class for TLS session cache configuration options.
+    """
+    
+    @abstractmethod
+    def create_cache(self):
+        """
+        Build and return a TLSSessionCache implementation.
+        
+        Returns:
+            TLSSessionCache: A configured session cache instance
+        """
+        pass
+
+
 class NoHostAvailable(Exception):
     """
     Raised when an operation is attempted but all connections are
@@ -879,33 +947,28 @@ def default_retry_policy(self, policy):
     .. versionadded:: 3.17.0
     """
 
-    tls_session_cache_enabled = True
+    tls_session_cache_options = None
     """
-    Enable TLS session caching for faster reconnections. When enabled, TLS sessions
-    are cached per endpoint and reused for subsequent connections to the same server.
-    This reduces handshake latency and CPU usage during reconnections.
+    TLS session cache configuration. Set to an instance of :class:`~.TLSSessionCacheOptions`
+    to configure session caching behavior. If None (default), a default configuration will
+    be used when SSL/TLS is enabled.
     
-    Defaults to True when SSL/TLS is enabled. Set to False to disable session caching.
+    The default configuration caches sessions by host and port, with a maximum of 100 sessions
+    and a TTL of 3600 seconds (1 hour).
     
-    .. versionadded:: 3.30.0
-    """
-
-    tls_session_cache_size = _DEFAULT_TLS_SESSION_CACHE_SIZE
-    """
-    Maximum number of TLS sessions to cache. When the cache is full, the least
-    recently used session is evicted.
+    To disable session caching entirely, set this to False.
     
-    Defaults to 100.
-    
-    .. versionadded:: 3.30.0
-    """
-
-    tls_session_cache_ttl = _DEFAULT_TLS_SESSION_CACHE_TTL
-    """
-    Time-to-live for cached TLS sessions in seconds. Sessions older than this
-    are not reused and are removed from the cache.
+    Example::
     
-    Defaults to 3600 seconds (1 hour).
+        from cassandra.tls import DefaultTLSSessionCacheOptions
+        
+        # Cache by host only (ignoring port)
+        options = DefaultTLSSessionCacheOptions(
+            max_size=200,
+            ttl=7200,
+            cache_by_host_only=True
+        )
+        cluster = Cluster(ssl_context=ssl_context, tls_session_cache_options=options)
     
     .. versionadded:: 3.30.0
     """
@@ -1239,9 +1302,7 @@ def __init__(self,
                  idle_heartbeat_timeout=30,
                  no_compact=False,
                  ssl_context=None,
-                 tls_session_cache_enabled=True,
-                 tls_session_cache_size=_DEFAULT_TLS_SESSION_CACHE_SIZE,
-                 tls_session_cache_ttl=_DEFAULT_TLS_SESSION_CACHE_TTL,
+                 tls_session_cache_options=None,
                  endpoint_factory=None,
                  application_name=None,
                  application_version=None,
@@ -1458,18 +1519,24 @@ def __init__(self,
 
         self.ssl_options = ssl_options
         self.ssl_context = ssl_context
-        self.tls_session_cache_enabled = tls_session_cache_enabled
-        self.tls_session_cache_size = tls_session_cache_size
-        self.tls_session_cache_ttl = tls_session_cache_ttl
+        self.tls_session_cache_options = tls_session_cache_options
 
         # Initialize TLS session cache if SSL is enabled
         self._tls_session_cache = None
-        if (ssl_context or ssl_options) and tls_session_cache_enabled:
-            from cassandra.connection import TLSSessionCache
-            self._tls_session_cache = TLSSessionCache(
-                max_size=tls_session_cache_size,
-                ttl=tls_session_cache_ttl
-            )
+        if (ssl_context or ssl_options) and tls_session_cache_options is not False:
+            from cassandra.tls import DefaultTLSSessionCacheOptions
+            
+            # Use provided options or create default
+            if tls_session_cache_options is None:
+                cache_options = DefaultTLSSessionCacheOptions(
+                    max_size=_DEFAULT_TLS_SESSION_CACHE_SIZE,
+                    ttl=_DEFAULT_TLS_SESSION_CACHE_TTL,
+                    cache_by_host_only=False
+                )
+            else:
+                cache_options = tls_session_cache_options
+            
+            self._tls_session_cache = cache_options.create_cache()
 
         self.sockopts = sockopts
         self.cql_version = cql_version

cassandra/connection.py

@@ -128,121 +128,6 @@ def decompress(byts):
 frame_header_v3 = struct.Struct('>BhBi')
 
 
-# Named tuple for TLS session cache entries
-_SessionCacheEntry = namedtuple('_SessionCacheEntry', ['session', 'timestamp'])
-
-
-class TLSSessionCache:
-    """
-    Thread-safe cache for TLS sessions to enable session resumption.
-    
-    This cache stores TLS sessions per endpoint (host:port) to allow
-    quick TLS renegotiation when reconnecting to the same server.
-    Sessions are automatically expired after a TTL and the cache has
-    a maximum size with LRU eviction using OrderedDict.
-    
-    TLS session resumption works with both TLS 1.2 and TLS 1.3:
-    - TLS 1.2: Session IDs (RFC 5246) and optionally Session Tickets (RFC 5077)
-    - TLS 1.3: Session Tickets (RFC 8446)
-    
-    Python's ssl.SSLSession API handles both versions transparently, so no
-    version-specific checks are needed.
-    """
-    
-    def __init__(self, max_size=100, ttl=3600):
-        """
-        Initialize the TLS session cache.
-        
-        Args:
-            max_size: Maximum number of sessions to cache (default: 100)
-            ttl: Time-to-live for cached sessions in seconds (default: 3600)
-        """
-        self._sessions = OrderedDict()  # OrderedDict for O(1) LRU eviction
-        self._lock = RLock()
-        self._max_size = max_size
-        self._ttl = ttl
-    
-    def _make_key(self, host, port):
-        """Create a cache key from host and port."""
-        return (host, port)
-    
-    def get_session(self, host, port):
-        """
-        Get a cached TLS session for the given endpoint.
-        
-        Args:
-            host: The hostname or IP address
-            port: The port number
-            
-        Returns:
-            ssl.SSLSession object if a valid cached session exists, None otherwise
-        """
-        key = self._make_key(host, port)
-        with self._lock:
-            if key not in self._sessions:
-                return None
-            
-            entry = self._sessions[key]
-            
-            # Check if session has expired
-            if time.time() - entry.timestamp > self._ttl:
-                del self._sessions[key]
-                return None
-            
-            # Move to end to mark as recently used (LRU)
-            self._sessions.move_to_end(key)
-            return entry.session
-    
-    def set_session(self, host, port, session):
-        """
-        Store a TLS session for the given endpoint.
-        
-        Args:
-            host: The hostname or IP address
-            port: The port number
-            session: The ssl.SSLSession object to cache
-        """
-        if session is None:
-            return
-        
-        key = self._make_key(host, port)
-        current_time = time.time()
-        
-        with self._lock:
-            # If key already exists, just update it
-            if key in self._sessions:
-                self._sessions[key] = _SessionCacheEntry(session, current_time)
-                self._sessions.move_to_end(key)
-                return
-            
-            # If cache is at max size, remove least recently used entry (first item)
-            if len(self._sessions) >= self._max_size:
-                self._sessions.popitem(last=False)
-            
-            # Store session with creation time
-            self._sessions[key] = _SessionCacheEntry(session, current_time)
-    
-    def clear_expired(self):
-        """Remove all expired sessions from the cache."""
-        current_time = time.time()
-        with self._lock:
-            expired_keys = [
-                key for key, entry in self._sessions.items()
-                if current_time - entry.timestamp > self._ttl
-            ]
-            for key in expired_keys:
-                del self._sessions[key]
-    
-    def clear(self):
-        """Clear all sessions from the cache."""
-        with self._lock:
-            self._sessions.clear()
-    
-    def size(self):
-        """Return the current number of cached sessions."""
-        with self._lock:
-            return len(self._sessions)
-
 
 class EndPoint(object):
     """
@@ -1037,8 +922,7 @@ def _wrap_socket_from_context(self):
         # Note: Session resumption works with both TLS 1.2 and TLS 1.3
         # Python's ssl module handles both transparently via SSLSession objects
         if self.tls_session_cache:
-            cached_session = self.tls_session_cache.get_session(
-                self.endpoint.address, self.endpoint.port)
+            cached_session = self.tls_session_cache.get_session(self.endpoint)
             if cached_session:
                 opts['session'] = cached_session
                 log.debug("Using cached TLS session for %s:%s", 
@@ -1109,8 +993,7 @@ def _connect_socket(self):
                 # This ensures we only cache sessions for connections that actually succeeded
                 if self.tls_session_cache and self.ssl_context and hasattr(self._socket, 'session'):
                     if self._socket.session:
-                        self.tls_session_cache.set_session(
-                            self.endpoint.address, self.endpoint.port, self._socket.session)
+                        self.tls_session_cache.set_session(self.endpoint, self._socket.session)
                         # Track if the session was reused
                         self.session_reused = self._socket.session_reused
                         if self.session_reused: