diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 12ade2018f..e1974e884c 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -53,7 +53,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 @@ -916,6 +917,45 @@ def default_retry_policy(self, policy): .. versionadded:: 3.17.0 """ + ssl_session_cache = None + """ + An optional :class:`~cassandra.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` is 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: automatic caching is **not** enabled for the legacy + :attr:`~Cluster.ssl_options` path because each connection builds a fresh + ``SSLContext``, making session reuse impossible. If you migrate to + ``ssl_context``, the cache will be created automatically. + + You may also pass a custom :class:`~cassandra.connection.SSLSessionCache` + instance with specific ``max_size`` and ``ttl`` parameters:: + + from cassandra.connection import SSLSessionCache + + cluster = Cluster( + ssl_context=ssl_context, + ssl_session_cache=SSLSessionCache(max_size=200, ttl=7200), + ) + + 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. + + Works with stdlib ``ssl`` (asyncore, libev, gevent) and PyOpenSSL + (Twisted, Eventlet). + + Note: the ``asyncio`` event loop (``EVENT_LOOP_MANAGER=asyncio``) is **not** + supported. ``AsyncioConnection`` performs the TLS handshake via + ``loop.create_connection(..., ssl=...)``, which offers no hook to restore a + cached session before the handshake, so the cache is neither populated nor + used on that reactor. + """ + sockopts = None """ An optional list of tuples which will be used as arguments to @@ -1268,7 +1308,8 @@ def __init__(self, column_encryption_policy=None, application_info:Optional[ApplicationInfoBase]=None, client_routes_config:Optional[ClientRoutesConfig]=None, - allow_control_connection_query_fallback:Optional[ControlConnectionQueryFallback]=ControlConnectionQueryFallback.Disabled + allow_control_connection_query_fallback:Optional[ControlConnectionQueryFallback]=ControlConnectionQueryFallback.Disabled, + ssl_session_cache=_NOT_SET ): """ ``executor_threads`` defines the number of threads in a pool for handling asynchronous tasks such as @@ -1517,6 +1558,23 @@ def __init__(self, self.ssl_options = ssl_options self.ssl_context = ssl_context + + # Auto-create a session cache when TLS is enabled via ssl_context, + # unless the caller explicitly passed ssl_session_cache (including None + # to opt out). The legacy ssl_options-only path is excluded because it + # builds a fresh SSLContext per connection, making session reuse + # impossible. The asyncio reactor is also skipped because + # AsyncioConnection establishes TLS via loop.create_connection(..., + # ssl=...) and offers no hook to restore/store sessions, so an + # auto-created cache would stay empty. + if ssl_session_cache is _NOT_SET: + if ssl_context is not None and self._connection_class_supports_tls_resumption(): + self.ssl_session_cache = SSLSessionCache() + else: + self.ssl_session_cache = None + else: + self.ssl_session_cache = ssl_session_cache + self.sockopts = sockopts self.cql_version = cql_version self.max_schema_agreement_wait = max_schema_agreement_wait @@ -1751,6 +1809,23 @@ def connection_factory(self, endpoint, host_conn = None, *args, **kwargs): kwargs = self._make_connection_kwargs(endpoint, kwargs) return self.connection_class.factory(endpoint, self.connect_timeout, host_conn, *args, **kwargs) + def _connection_class_supports_tls_resumption(self): + """ + Return ``False`` for reactors that cannot restore/store TLS sessions, + so a session cache is not auto-created where it would stay empty. + + Currently only the asyncio reactor is unsupported: it establishes TLS + via ``loop.create_connection(..., ssl=...)`` with no hook to restore a + cached session before the handshake. + """ + try: + from cassandra.io.asyncioreactor import AsyncioConnection + except (ImportError, DependencyException): + # asyncio not available at all; treat as "supported" since the + # AsyncioConnection class cannot be selected in that environment. + return True + return not issubclass(self.connection_class, AsyncioConnection) + def _make_connection_factory(self, host, *args, **kwargs): kwargs = self._make_connection_kwargs(host.endpoint, kwargs) return partial(self.connection_class.factory, host.endpoint, self.connect_timeout, *args, **kwargs) @@ -1764,6 +1839,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) @@ -5329,7 +5405,7 @@ def _execute_after_prepare(self, host, connection, pool, response): new_metadata_id = response.result_metadata_id if new_metadata_id is not None: self.prepared_statement.result_metadata_id = new_metadata_id - + # use self._query to re-use the same host and # at the same time properly borrow the connection if pool is None and connection is not None and connection.is_control_connection: diff --git a/cassandra/connection.py b/cassandra/connection.py index 25508e32ac..80a1bbbb73 100644 --- a/cassandra/connection.py +++ b/cassandra/connection.py @@ -13,7 +13,7 @@ # limitations under the License. from __future__ import absolute_import # to enable import io from stdlib -from collections import defaultdict, deque +from collections import defaultdict, deque, namedtuple, OrderedDict import errno from functools import wraps, partial, total_ordering from heapq import heappush, heappop @@ -22,7 +22,7 @@ import socket import struct import sys -from threading import Thread, Event, RLock, Condition +from threading import Thread, Event, Lock, RLock, Condition import time import ssl import uuid @@ -163,6 +163,15 @@ def socket_family(self): """ return socket.AF_UNSPEC + @property + def tls_session_cache_key(self): + """ + Returns the cache key components for TLS session caching. + This is a tuple that uniquely identifies this endpoint for TLS session purposes. + Subclasses may override this to include additional components (e.g., SNI server name). + """ + return (self.address, self.port) + def resolve(self): """ Resolve the endpoint to an address/port. This is called @@ -277,6 +286,14 @@ def port(self): def ssl_options(self): return self._ssl_options + @property + def tls_session_cache_key(self): + """ + Returns the cache key including server_name for SNI endpoints. + This prevents cache collisions when multiple SNI endpoints use the same proxy. + """ + return (self.address, self.port, self._server_name) + def resolve(self): try: resolved_addresses = socket.getaddrinfo(self._proxy_address, self._port, @@ -395,6 +412,14 @@ def port(self): def socket_family(self): return socket.AF_UNIX + @property + def tls_session_cache_key(self): + """ + Returns the cache key for Unix socket endpoints. + Since Unix sockets don't have a port, only the path is used. + """ + return (self._unix_socket_path,) + def resolve(self): return self.address, None @@ -455,6 +480,14 @@ def port(self) -> Optional[int]: def host_id(self) -> uuid.UUID: return self._host_id + @property + def tls_session_cache_key(self): + """ + Returns the cache key for Client Routes endpoints. + Uses host_id and original address for uniqueness. + """ + return (str(self._host_id), self._original_address, self._original_port) + def resolve(self) -> Tuple[str, int]: """ Resolve endpoint by delegating to the handler. @@ -783,6 +816,163 @@ def generate(self, shard_id: int, total_shards: int): DefaultShardAwarePortGenerator = ShardAwarePortGenerator(DEFAULT_LOCAL_PORT_LOW, DEFAULT_LOCAL_PORT_HIGH) +_SessionCacheEntry = namedtuple('_SessionCacheEntry', ['session', 'timestamp']) + + +class SSLSessionCache(object): + """ + A thread-safe cache of TLS session objects, keyed by connection TLS + identity, with LRU eviction and TTL expiration. + + When TLS is enabled, the driver stores the negotiated session after each + successful handshake and reuses it for subsequent connections to the same + host, enabling TLS session resumption (tickets / PSK) without any extra + configuration. + + This cache is created automatically by :class:`.Cluster` when + ``ssl_context`` is set (not for the legacy ``ssl_options`` path, where each + connection builds a fresh ``SSLContext``, making session reuse impossible). + Pass ``ssl_session_cache=None`` to :class:`.Cluster` to opt out. + + Works with both the stdlib ``ssl`` module (asyncore, libev, gevent + reactors) and PyOpenSSL (Twisted and Eventlet reactors). The asyncio + reactor is not supported: it establishes TLS via + ``loop.create_connection(..., ssl=...)`` with no hook to restore a cached + session before the handshake, so no cache is auto-created for it. + + 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) + """ + + # Cleanup expired sessions every N set() calls + _EXPIRY_CLEANUP_INTERVAL = 100 + + def __init__(self, max_size=100, ttl=3600): + """ + Initialize the TLS session cache. + + :param max_size: Maximum number of sessions to cache. Must be at + least ``1``. When full, the least recently used entry is evicted. + Default: ``100``. + :param ttl: Time-to-live for cached sessions in seconds. Must be + greater than ``0``. Expired entries are lazily removed on access + and periodically during :meth:`set`. Default: ``3600`` (one hour). + """ + if max_size < 1: + raise ValueError("max_size must be >= 1, got %r" % (max_size,)) + if ttl <= 0: + raise ValueError("ttl must be > 0, got %r" % (ttl,)) + self._sessions = OrderedDict() + self._lock = Lock() + self._max_size = max_size + self._ttl = ttl + self._operation_count = 0 + + @property + def max_size(self): + return self._max_size + + @property + def ttl(self): + return self._ttl + + def get(self, key): + """ + Return the cached TLS session for *key*, or ``None`` if none + is stored or if the entry has expired. Accessing an entry + marks it as recently used. + """ + with self._lock: + entry = self._sessions.get(key) + if entry is None: + return None + if time.monotonic() - entry.timestamp > self._ttl: + del self._sessions[key] + return None + # Mark as recently used + self._sessions.move_to_end(key) + return entry.session + + def set(self, key, session): + """ + Store *session* for *key*. ``None`` sessions are silently ignored. + """ + if session is None: + return + + current_time = time.monotonic() + with self._lock: + self._operation_count += 1 + if self._operation_count >= self._EXPIRY_CLEANUP_INTERVAL: + self._operation_count = 0 + self._clear_expired_unlocked(current_time) + + if key in self._sessions: + self._sessions[key] = _SessionCacheEntry(session, current_time) + self._sessions.move_to_end(key) + return + + if len(self._sessions) >= self._max_size: + # Prefer evicting an expired entry over the LRU live one so + # that a still-valid session is not displaced by a stale one. + # Locate the first expired key (two-pass: collect then delete to + # avoid mutating the OrderedDict while iterating it). + expired_key = next( + (k for k, entry in self._sessions.items() + if current_time - entry.timestamp > self._ttl), + None, + ) + if expired_key is not None: + del self._sessions[expired_key] + else: + self._sessions.popitem(last=False) + + self._sessions[key] = _SessionCacheEntry(session, current_time) + + def clear(self): + """Clear all sessions from the cache.""" + with self._lock: + self._sessions.clear() + + def clear_expired(self): + """Remove all expired sessions from the cache.""" + current_time = time.monotonic() + with self._lock: + self._clear_expired_unlocked(current_time) + + def size(self): + """Return the current number of cached sessions.""" + with self._lock: + return len(self._sessions) + + def _clear_expired_unlocked(self, current_time=None): + """Remove all expired sessions (must be called with lock held).""" + if current_time is None: + current_time = time.monotonic() + 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 _snapshot_sessions(self): + """Return a list of cached session objects (for testing only).""" + with self._lock: + return [entry.session for entry in self._sessions.values()] + + def __repr__(self): + with self._lock: + return "<%s max_size=%d ttl=%d size=%d>" % ( + self.__class__.__name__, + self._max_size, + self._ttl, + len(self._sessions), + ) + + class Connection(object): CALLBACK_ERR_THREAD_THRESHOLD = 100 @@ -803,6 +993,11 @@ class Connection(object): endpoint = None ssl_options = None ssl_context = None + _ssl_session_cache = None + _tls_session_cached = False + # Set to True once the TLS session for this connection has been stored in + # the cache. Safe to never reset because Connection objects are created + # per physical TCP connection and are never recycled across reconnections. last_error = None # The current number of operations that are in flight. More precisely, @@ -880,13 +1075,15 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None, cql_version=None, protocol_version=ProtocolVersion.MAX_SUPPORTED, is_control_connection=False, user_type_map=None, connect_timeout=None, allow_beta_protocol_version=False, no_compact=False, ssl_context=None, owning_pool=None, shard_id=None, total_shards=None, - on_orphaned_stream_released=None, application_info: Optional[ApplicationInfoBase] = None): + on_orphaned_stream_released=None, application_info: Optional[ApplicationInfoBase] = None, + ssl_session_cache=None): # TODO next major rename host to endpoint and remove port kwarg. self.endpoint = host if isinstance(host, EndPoint) else DefaultEndPoint(host, port) self.authenticator = authenticator self.ssl_options = ssl_options.copy() if ssl_options else {} self.ssl_context = ssl_context + self._ssl_session_cache = ssl_session_cache self.sockopts = sockopts self.compression = compression self.cql_version = cql_version @@ -1032,7 +1229,28 @@ def _wrap_socket_from_context(self): server_hostname = self.endpoint.address opts['server_hostname'] = server_hostname - return self.ssl_context.wrap_socket(self._socket, **opts) + ssl_sock = self.ssl_context.wrap_socket(self._socket, **opts) + + # Restore a previously cached session to enable TLS session resumption + # (session tickets / PSK). The session must be set *after* + # wrap_socket() (which only creates the SSLSocket) but *before* + # connect(), because connect() triggers the actual TLS handshake + # (via do_handshake_on_connect, which defaults to True). + # _initiate_connection, called after this method returns, performs + # the connect(). + if self._ssl_session_cache is not None: + cache_key = self._ssl_session_cache_key() + cached_session = self._ssl_session_cache.get(cache_key) + if cached_session is not None: + try: + ssl_sock.session = cached_session + log.debug("TLS session restore attempted for %s key=%s", self.endpoint, cache_key) + except (AttributeError, ssl.SSLError, ValueError) as e: + log.debug("Could not restore TLS session for %s: %s", self.endpoint, e) + else: + log.debug("No cached TLS session found for %s key=%s", self.endpoint, cache_key) + + return ssl_sock def _initiate_connection(self, sockaddr): if self.features.shard_id is not None: @@ -1046,6 +1264,80 @@ def _initiate_connection(self, sockaddr): self._socket.connect(sockaddr) + def _cache_tls_session_if_needed(self, initial=False): + """ + Store the current TLS session in the cache (if any) so that future + connections to the same endpoint can resume it. + + The current session is cached on every handshake, including resumed + ones: a resumed TLS 1.2 handshake may issue a *replacement* session + ticket, so re-storing keeps the freshest resumable session rather than + a consumed/older one. + + TLS 1.3 delivers the session ticket asynchronously, after the handshake + completes. Right after ``connect()`` the ``SSLSession`` may already be + present -- either still ticketless (``has_ticket`` is ``False``) or, on + a *resumed* handshake, carrying the old (potentially single-use) ticket + that was used for resumption. Caching in either case is wrong: a + ticketless session is non-resumable, and the old resumption ticket may + be consumed and about to be rotated by the server's replacement + ``NewSessionTicket``. Storing it now (and setting the + ``_tls_session_cached`` guard) would suppress the later store and let a + future reconnect retry a consumed ticket. Every *initial* TLS 1.3 store + is therefore skipped; the write is deferred to the second attempt from + the Ready/AuthSuccess handlers once the CQL handshake has completed and + the replacement ticket has arrived (see _handle_startup_response and + _handle_auth_response). + + ``initial`` is ``True`` when called from :meth:`_connect_socket` + immediately after the handshake, and ``False`` when called from the + Ready/AuthSuccess handlers after the first application-data exchange. + For TLS 1.3 the stdlib ``ssl`` path uses it to skip the premature + initial store (even when the resumed session already reports + ``has_ticket``); PyOpenSSL subclasses, which cannot inspect + ``has_ticket``, likewise use it to skip the premature initial attempt. + """ + if self._tls_session_cached: + # Already stored for this connection (called from both + # _connect_socket and the Ready/AuthSuccess handlers). + return + if self._ssl_session_cache is not None and self.ssl_context is not None: + session = getattr(self._socket, 'session', None) + if session is None: + # No session available yet (e.g. TLS 1.3 before the ticket + # arrives); a later call will retry. + return + tls_version = getattr(self._socket, 'version', lambda: None)() + if tls_version == 'TLSv1.3' and ( + initial or not getattr(session, 'has_ticket', False)): + # TLS 1.3 exposes the SSLSession before the server's (possibly + # replacement) NewSessionTicket is delivered. Skip every + # *initial* store: on a resumed handshake the session still + # carries the old, potentially single-use ticket that is about + # to be rotated, so caching it now (and setting the guard) could + # let a later reconnect retry a consumed ticket. On non-initial + # calls, still defer while the session is ticketless. Either way + # the write is deferred to the Ready/AuthSuccess retry once the + # replacement ticket has arrived. + return + cache_key = self._ssl_session_cache_key() + log.debug("TLS caching session for %s: tls_version=%s has_ticket=%s key=%s", + self.endpoint, tls_version, + getattr(session, 'has_ticket', None), + cache_key) + self._ssl_session_cache.set(cache_key, session) + self._tls_session_cached = True + + def _ssl_session_cache_key(self): + """ + Return a cache key that matches the TLS peer identity. + + Delegates to the endpoint's ``tls_session_cache_key`` property, which + returns appropriate components for each endpoint type (e.g., includes + ``server_name`` for SNI endpoints to prevent cache collisions). + """ + return self.endpoint.tls_session_cache_key + # PYTHON-1331 # # Allow implementations specific to an event loop to add additional behaviours @@ -1087,6 +1379,19 @@ def _connect_socket(self): # run that here. if self._check_hostname: self._validate_hostname() + + # Cache the negotiated TLS session for future resumption, only + # after hostname validation has succeeded so that a rejected + # peer never leaves a usable session in the cache. + # For TLS 1.2 the session is available right after connect(). + # For TLS 1.3 the server sends the session ticket + # asynchronously after the first application-data exchange, + # so socket.session may still be None here; a second + # attempt is made in _cache_tls_session_if_needed() after + # the CQL handshake completes (see _handle_startup_response + # and _handle_auth_response). + self._cache_tls_session_if_needed(initial=True) + sockerr = None break except socket.error as err: @@ -1583,6 +1888,9 @@ def _handle_startup_response(self, startup_response, did_authenticate=False): if ProtocolVersion.has_checksumming_support(self.protocol_version): self._enable_checksumming() + # TLS 1.3: the session ticket is sent after the first + # application-data exchange, so try caching it now. + self._cache_tls_session_if_needed() self.connected_event.set() elif isinstance(startup_response, AuthenticateMessage): log.debug("Got AuthenticateMessage on new connection (%s) from %s: %s", @@ -1639,6 +1947,9 @@ def _handle_auth_response(self, auth_response): self.authenticator.on_authentication_success(auth_response.token) if self._compressor: self.compressor = self._compressor + # TLS 1.3: the session ticket is sent after the first + # application-data exchange, so try caching it now. + self._cache_tls_session_if_needed() self.connected_event.set() elif isinstance(auth_response, AuthChallengeMessage): response = self.authenticator.evaluate_challenge(auth_response.challenge) diff --git a/cassandra/io/asyncioreactor.py b/cassandra/io/asyncioreactor.py index 92ab972e7d..44170216cd 100644 --- a/cassandra/io/asyncioreactor.py +++ b/cassandra/io/asyncioreactor.py @@ -118,6 +118,12 @@ class AsyncioConnection(Connection): Supports SSL connections via asyncio's native TLS transport, which avoids the incompatibility between ``ssl.SSLSocket`` and asyncio's low-level socket methods (``sock_sendall``, ``sock_recv``). + + Note: TLS session resumption (:attr:`.Cluster.ssl_session_cache`) is not + supported on this reactor. TLS is established through + ``loop.create_connection(..., ssl=...)``, which exposes no hook to restore a + cached session before the handshake, so sessions are neither restored nor + stored here. """ _loop = None diff --git a/cassandra/io/eventletreactor.py b/cassandra/io/eventletreactor.py index 234a4a574c..6d8bfc7386 100644 --- a/cassandra/io/eventletreactor.py +++ b/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 as e: + log.debug("Could not restore TLS session for %s: %s", self.endpoint, e) + + # 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,15 +132,64 @@ 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: super(EventletConnection, self)._match_hostname() else: - cert_name = self._socket.get_peer_certificate().get_subject().commonName - if cert_name != self.endpoint.address: + cert = self._socket.get_peer_certificate() + cert_name = cert.get_subject().commonName if cert is not None else None + if cert is None or cert_name != self.endpoint.address: raise Exception("Hostname verification failed! Certificate name '{}' " - "doesn't endpoint '{}'".format(cert_name, self.endpoint.address)) + "doesn't match endpoint '{}'".format(cert_name, self.endpoint.address)) + + def _cache_tls_session_if_needed(self, initial=False): + """ + PyOpenSSL override of :meth:`.Connection._cache_tls_session_if_needed`. + + Uses ``SSL.Connection.get_session()`` instead of the stdlib + ``ssl.SSLSocket.session`` attribute. + + The base ``_connect_socket()`` also invokes this immediately after + ``do_handshake()`` (with ``initial=True``). pyOpenSSL exposes no way to + tell whether a TLS 1.3 ``NewSessionTicket`` has arrived yet + (``get_session()`` can return a non-``None`` but ticketless, + non-resumable session at that point), and caching then would set the + ``_tls_session_cached`` guard and block the later store of the real + resumable session. The initial invocation is therefore skipped; the + session is stored only from the ReadyMessage / AuthSuccessMessage + handlers, after the first application-data exchange, which is the + earliest reliable moment for both TLS 1.2 and TLS 1.3. + + 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(initial=initial) + return + if initial: + # Defer to the post-application-data call (see docstring). + return + if self._tls_session_cached: + return + if self._ssl_session_cache is None or not (self.ssl_context or self.ssl_options): + return + # Note: pyOpenSSL's SSL.Connection exposes no session_reused(), so we cannot + # skip resumed handshakes; the per-connection ``_tls_session_cached`` guard + # above still prevents storing twice for the same connection. + try: + session = self._socket.get_session() + if session is not None: + self._ssl_session_cache.set( + self._ssl_session_cache_key(), session) + self._tls_session_cached = True + except Exception as e: + log.debug("Could not cache TLS session for %s: %s", self.endpoint, e) def close(self): with self.lock: diff --git a/cassandra/io/twistedreactor.py b/cassandra/io/twistedreactor.py index 446200bf63..7708722d69 100644 --- a/cassandra/io/twistedreactor.py +++ b/cassandra/io/twistedreactor.py @@ -139,11 +139,16 @@ 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 + # Set in clientConnectionForTLS() to the single SSL.Connection this creator + # drives; read by TwistedConnection._cache_tls_session_if_needed() after the + # first CQL exchange. + self._ssl_connection = None if ssl_context: self.context = ssl_context @@ -167,15 +172,80 @@ def verify_callback(self, connection, x509, errnum, errdepth, ok): def info_callback(self, connection, where, ret): if where & SSL.SSL_CB_HANDSHAKE_DONE: - 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))) + transport = connection.get_app_data() + endpoint = connection._cassandra_endpoint + cert = connection.get_peer_certificate() + # Read check_hostname from the connection (not this creator) because a + # shared ssl_context means set_info_callback() only keeps the last + # creator's bound callback, so self.check_hostname could belong to a + # different creator and bypass or misapply hostname verification. + if connection._cassandra_check_hostname and ( + cert is None or + endpoint.address != cert.get_subject().commonName): + transport.failVerification(Failure(ConnectionException("Hostname verification failed", endpoint))) + return + # For TLS 1.2 (and earlier) the session is fully established at + # handshake completion, so cache it immediately. For TLS 1.3 the + # server sends the NewSessionTicket *after* the handshake, so + # get_session() here may return None or a ticketless, non-resumable + # session; caching that and setting the guard flag would suppress + # the deferred path and leave a ticketless session cached + # indefinitely. TLS 1.3 is therefore left entirely to the deferred + # path in TwistedConnection._cache_tls_session_if_needed(), which + # runs after the first CQL exchange once the ticket has arrived. + # Note: pyOpenSSL's SSL.Connection exposes no session_reused(), so we cannot + # detect resumed handshakes here; the per-connection guard flag stored on the + # SSL.Connection prevents the deferred path from storing the same session again. + # Both the guard flag and the cache reference live on the connection + # (not this creator) because a shared ssl_context means + # set_info_callback() only keeps the last creator's bound callback, + # so creator-scoped state would belong to the wrong instance -- e.g. + # storing one cluster's session into another cluster's cache. + ssl_session_cache = connection._cassandra_ssl_session_cache + if ssl_session_cache is not None and \ + connection.get_protocol_version_name() != 'TLSv1.3': + session = connection.get_session() + if session is not None: + ssl_session_cache.set(endpoint.tls_session_cache_key, session) + connection._cassandra_session_cached = True def clientConnectionForTLS(self, tlsProtocol): connection = SSL.Connection(self.context, None) connection.set_app_data(tlsProtocol) + # Twisted may overwrite app_data with its own state, so stash the + # per-connection endpoint as a dedicated attribute on this specific + # SSL.Connection. info_callback recovers it from here because it runs off + # the shared SSL.Context and cannot rely on this creator's attributes. + connection._cassandra_endpoint = self.endpoint + # Likewise stash this creator's cache on the connection so info_callback + # writes to the correct cache even when a shared ssl_context binds it to + # a different creator instance. + connection._cassandra_ssl_session_cache = self.ssl_session_cache + # Same reasoning for the hostname check policy: keep it on the connection + # so info_callback applies this creator's check_hostname rather than that + # of whichever creator last bound the shared context's info_callback. + connection._cassandra_check_hostname = self.check_hostname + # Per-connection "session already stored" guard. Kept on the connection + # (not this creator) so it stays correct even when a shared ssl_context + # means info_callback fires bound to a different creator instance. + connection._cassandra_session_cached = False + # This creator drives exactly one connection, so this reference unambiguously + # identifies it for TwistedConnection._cache_tls_session_if_needed(). + self._ssl_connection = connection 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 as e: + log.debug("Could not restore TLS session for %s: %s", self.endpoint, e) + return connection @@ -212,6 +282,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 +312,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 +332,46 @@ def add_connection(self): ) connectProtocol(endpoint, TwistedConnectionProtocol(self)) + def _cache_tls_session_if_needed(self, initial=False): + """ + PyOpenSSL override of :meth:`.Connection._cache_tls_session_if_needed`. + + Twisted drives connection setup through its own machinery rather than + the base ``_connect_socket()``, so this is only ever called from the + ReadyMessage / AuthSuccessMessage handlers (``initial`` is always + ``False``) — 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. The premature + initial attempt is skipped defensively should it ever be invoked. + """ + if initial: + return + if self._ssl_session_cache is None or self._ssl_creator is None: + return + if self._tls_session_cached: + return + ssl_conn = self._ssl_creator._ssl_connection + if ssl_conn is None: + return + if getattr(ssl_conn, '_cassandra_session_cached', False): + # info_callback already stored this connection's session eagerly + # (TLS 1.2 path). Mirror the flag so the driver-level invariant + # ("True once the session has been stored") holds for this path too. + self._tls_session_cached = True + return + # Note: pyOpenSSL's SSL.Connection exposes no session_reused(), so we cannot + # skip resumed handshakes; the per-connection guards above still prevent + # storing the same session more than once for this connection. + try: + session = ssl_conn.get_session() + if session is not None: + self._ssl_session_cache.set( + self.endpoint.tls_session_cache_key, session) + self._tls_session_cached = True + ssl_conn._cassandra_session_cached = True + except Exception as e: + log.debug("Could not cache TLS session for %s: %s", self.endpoint, e) + def client_connection_made(self, transport): """ Called by twisted protocol when a connection attempt has diff --git a/tests/integration/standard/test_tls_resumption.py b/tests/integration/standard/test_tls_resumption.py new file mode 100644 index 0000000000..1ae8750ff8 --- /dev/null +++ b/tests/integration/standard/test_tls_resumption.py @@ -0,0 +1,595 @@ +# Copyright DataStax, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +""" +Integration test for TLS session ticket resumption. + +Verifies that after the initial TLS handshake learns session tickets, the +driver reuses them on reconnection instead of performing full handshakes. +The test wraps the ``SSLSessionCache`` with a tracking layer that counts +``set()`` calls as a sanity check on ticket activity (``set()`` runs for +both new and resumed handshakes, so the count is not asserted to stay +fixed). Actual proof of resumption comes from the handshake itself, via +``ssl.SSLSocket.session_reused`` (stdlib) or ``SSL_session_reused()`` +(PyOpenSSL), which report whether the last handshake reused a prior session. + +Requires a live Scylla / Cassandra cluster with TLS enabled. +""" + +import logging +import os +import shutil +import socket +import ssl +import subprocess +import tempfile +import threading +import time +import unittest + +from cassandra.connection import SSLSessionCache + +try: + from OpenSSL import SSL +except ImportError: + SSL = None + +from tests import EVENT_LOOP_MANAGER +from tests.integration import ( + get_cluster, remove_cluster, use_single_node, start_cluster_wait_for_up, + TestCluster, CASSANDRA_IP, SCYLLA_VERSION +) + +log = logging.getLogger(__name__) + + +def _generate_ssl_certs(cert_dir): + """ + Generate a minimal self-signed CA and a server cert/key signed by that CA. + + Writes into *cert_dir*: + - ca.key / ca.crt : self-signed CA + - cassandra.key / cassandra.csr / cassandra.crt : server cert + + :param cert_dir: directory to write files into (must already exist) + :raises unittest.SkipTest: if ``openssl`` is not on PATH + :raises RuntimeError: if any openssl command fails + """ + if shutil.which("openssl") is None: + raise unittest.SkipTest("openssl not found on PATH; skipping TLS resumption test") + + san_cnf = os.path.join(cert_dir, "san.cnf") + with open(san_cnf, "w") as f: + f.write("subjectAltName=IP:127.0.0.1\n") + + def _run(cmd): + result = subprocess.run(cmd, cwd=cert_dir, capture_output=True, text=True) + if result.returncode != 0: + raise RuntimeError( + "openssl command failed: %s\n%s" % (" ".join(cmd), result.stderr) + ) + + _run(["openssl", "req", "-x509", "-newkey", "rsa:2048", + "-keyout", "ca.key", "-out", "ca.crt", + "-days", "1", "-nodes", "-subj", "/CN=Test CA"]) + + _run(["openssl", "req", "-newkey", "rsa:2048", + "-keyout", "cassandra.key", "-out", "cassandra.csr", + "-nodes", "-subj", "/CN=127.0.0.1"]) + + _run(["openssl", "x509", "-req", + "-in", "cassandra.csr", "-CA", "ca.crt", "-CAkey", "ca.key", + "-CAcreateserial", "-out", "cassandra.crt", + "-days", "1", "-extfile", "san.cnf"]) + + log.info("Generated SSL certs in %s", cert_dir) + +USES_PYOPENSSL = "twisted" in EVENT_LOOP_MANAGER or "eventlet" in EVENT_LOOP_MANAGER +# The asyncio reactor performs TLS via asyncio's native SSL transport rather +# than wrapping the socket, so ``conn._socket`` stays a plain TCP socket (no +# ``version()``/``session``) and ``ssl_session_cache`` is not supported. +USES_ASYNCIO = "asyncio" in EVENT_LOOP_MANAGER + +# --------------------------------------------------------------------------- +# Tracking wrapper around SSLSessionCache +# --------------------------------------------------------------------------- + +class _TrackingSSLSessionCache(SSLSessionCache): + """ + A thin wrapper around :class:`SSLSessionCache` that counts how many TLS + sessions are stored via :meth:`set`. + + :meth:`Connection._cache_tls_session_if_needed` stores the current session + on *every* successful handshake — including resumed ones, since a resumed + TLS 1.2 handshake may issue a replacement ticket — so this count grows on + each (re)connection and cannot by itself distinguish resumption from + renegotiation. The test therefore uses it only to confirm that tickets + were learned at all, and verifies actual reuse via + ``ssl.SSLSocket.session_reused`` (see ``_count_resumed_connections``). + """ + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._ticket_count_lock = threading.Lock() + self._ticket_count = 0 + + def set(self, key, session): + if session is not None: + with self._ticket_count_lock: + self._ticket_count += 1 + super().set(key, session) + + @property + def ticket_count(self): + """Total number of TLS sessions stored so far.""" + with self._ticket_count_lock: + return self._ticket_count + +def _make_ssl_context(ca_cert_path, force_tls13=False): + """Return a client ``ssl.SSLContext`` that trusts the test CA at *ca_cert_path*. + + By default TLS 1.2 is required: Python's ``ssl.SSLSocket.session_reused`` + is only reliable in the raw ``_server_supports_tls_resumption`` probe on + TLS 1.2, because a TLS 1.3 ``SSLSession`` captured immediately after the + handshake has no ticket yet and cannot be resumed. The driver itself only + caches *ticketed* sessions, so its own TLS 1.3 reconnections do resume and + report ``session_reused`` correctly. + + Pass ``force_tls13=True`` (stdlib path only) to pin TLS 1.3 and exercise + the driver's delayed-ticket caching path. + """ + if USES_PYOPENSSL: + ssl_context = SSL.Context(SSL.TLS_CLIENT_METHOD) + ssl_context.load_verify_locations(ca_cert_path) + # Actually verify the server certificate chain against the test CA so + # the Twisted/Eventlet paths exercise verification (matching the stdlib + # branch's CERT_REQUIRED intent). + ssl_context.set_verify(SSL.VERIFY_PEER, + lambda conn, cert, errno, depth, ok: bool(ok)) + else: + ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ssl_context.load_verify_locations(ca_cert_path) + ssl_context.verify_mode = ssl.CERT_REQUIRED + ssl_context.check_hostname = False + if force_tls13: + # Pin TLS 1.3 to exercise the asynchronous NewSessionTicket path. + ssl_context.minimum_version = ssl.TLSVersion.TLSv1_3 + ssl_context.maximum_version = ssl.TLSVersion.TLSv1_3 + else: + # Restrict to TLS 1.2 so that session_reused is reliable. + ssl_context.maximum_version = ssl.TLSVersion.TLSv1_2 + return ssl_context + + +def _tls_negotiates_v13(ca_cert_path, host, port=9042): + """ + Return ``True`` if a raw TLS 1.3 handshake with the server at + *host*:*port* succeeds (i.e. the server supports TLS 1.3). + """ + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.load_verify_locations(ca_cert_path) + ctx.check_hostname = False + ctx.minimum_version = ssl.TLSVersion.TLSv1_3 + try: + with socket.create_connection((host, port), timeout=5) as raw: + with ctx.wrap_socket(raw) as tls: + return tls.version() == 'TLSv1.3' + except (OSError, ssl.SSLError): + return False + + +def _server_supports_tls_resumption(ca_cert_path, host, port=9042): + """ + Return ``True`` if the server at *host*:*port* supports TLS 1.2 session + resumption (session-ID or session-ticket based). + + Makes two raw TLS 1.2 connections. The second reuses the session from + the first by setting ``ssl.SSLSocket.session`` before calling + ``do_handshake()``. Returns ``True`` only when ``session_reused`` is + ``True`` on the second connection. + """ + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.load_verify_locations(ca_cert_path) + ctx.check_hostname = False + ctx.maximum_version = ssl.TLSVersion.TLSv1_2 + try: + with socket.create_connection((host, port), timeout=5) as s1: + with ctx.wrap_socket(s1, do_handshake_on_connect=False) as ssl1: + ssl1.do_handshake() + session = ssl1.session + if session is None: + return False + time.sleep(0.1) + with socket.create_connection((host, port), timeout=5) as s2: + with ctx.wrap_socket(s2, do_handshake_on_connect=False) as ssl2: + ssl2.session = session + ssl2.do_handshake() + return bool(ssl2.session_reused) + except (OSError, ssl.SSLError): + return False + + +def _wait_for_all_connections(session, timeout=30): + """ + Wait until every host's connection pool is fully connected — i.e., has + at least one live connection per shard (or one connection if the host + has no sharding info), or raise on timeout. + + Checking ``len(pool._connections) > 0`` is insufficient for Scylla + clusters: the pool connects the first shard via the standard port, then + immediately fires off shard-aware connections to the remaining shards + asynchronously. Those connections complete in the background and their + TLS sessions are cached only after they finish. Waiting for all shards + ensures the cache is fully populated before we snapshot the ticket count. + """ + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + pools = session.get_pools() + if pools and all( + not p.is_shutdown and ( + len(p._connections) >= ( + p.host.sharding_info.shards_count + if p.host.sharding_info else 1 + ) + ) + for p in pools + ): + return + time.sleep(0.1) + raise RuntimeError( + "Timed out after %ds waiting for all connection pools to open" % timeout + ) + + +def _close_all_connections(session): + """ + Shut down every connection pool in the session and trigger + re-creation so the driver reconnects to all hosts. + """ + for pool in list(session._pools.values()): + pool.shutdown() + session.update_created_pools() + + +def _count_resumed_connections(session): + """ + Inspect every live pool connection and return ``(resumed, total)`` where + *resumed* counts connections whose last TLS handshake was an abbreviated + (resumed) handshake, per ``ssl.SSLSocket.session_reused``. + + Only meaningful on the stdlib ``ssl`` reactors; PyOpenSSL's ``SSL.Connection`` + exposes no ``session_reused``, so those sockets are skipped (not counted). + """ + resumed = 0 + total = 0 + for pool in list(session._pools.values()): + for conn in list(pool._connections.values()): + reused = getattr(getattr(conn, '_socket', None), 'session_reused', None) + if not isinstance(reused, bool): + continue + total += 1 + if reused: + resumed += 1 + return resumed, total + + +def _iter_live_ssl_sockets(session): + """Yield the live TLS socket of every connection in every pool.""" + for pool in list(session._pools.values()): + for conn in list(pool._connections.values()): + sock = getattr(conn, '_socket', None) + if sock is not None: + yield sock + + +def _pyopenssl_connection(conn): + """ + Return the pyOpenSSL ``SSL.Connection`` backing a driver connection, or + ``None`` if it cannot be located. + + Eventlet stores the ``SSL.Connection`` directly as ``conn._socket``; + Twisted keeps it on the SSL creator as ``conn._ssl_creator._ssl_connection``. + """ + sock = getattr(conn, '_socket', None) + if sock is not None and hasattr(sock, 'get_session'): + return sock + creator = getattr(conn, '_ssl_creator', None) + if creator is not None: + return getattr(creator, '_ssl_connection', None) + return None + + +def _count_pyopenssl_resumed(session): + """ + Return ``(resumed, total)`` for the PyOpenSSL reactors (Twisted, Eventlet). + + pyOpenSSL exposes no ``session_reused()`` Python method, but OpenSSL's + underlying ``SSL_session_reused()`` is reachable through the low-level cffi + binding (``OpenSSL._util.lib``) using the ``SSL *`` pointer that pyOpenSSL + keeps on ``SSL.Connection._ssl``. It returns ``1`` when the last handshake + resumed a prior session. + + This is a best-effort check. Both ``OpenSSL._util`` and the ``_ssl`` cffi + pointer are private pyOpenSSL internals that may change across versions. + If the binding is unavailable the function returns ``(0, 0)`` and the caller + is expected to call ``skipTest`` rather than fail the assertion. + + Returns ``(0, 0)`` when the binding or pointer is unavailable so callers + can skip rather than fail on unexpected pyOpenSSL internals. + """ + try: + from OpenSSL._util import lib as _ossl_lib + except Exception: + return 0, 0 + # SSL_session_reused may be absent from the imported cffi binding; accessing + # it directly would raise AttributeError and fail the test, so honor the + # best-effort contract by returning (0, 0) for the caller to skip. + _session_reused = getattr(_ossl_lib, 'SSL_session_reused', None) + if _session_reused is None: + return 0, 0 + resumed = 0 + total = 0 + for pool in list(session._pools.values()): + for conn in list(pool._connections.values()): + ssl_conn = _pyopenssl_connection(conn) + ssl_ptr = getattr(ssl_conn, '_ssl', None) + if ssl_ptr is None: + continue + total += 1 + if _session_reused(ssl_ptr): + resumed += 1 + return resumed, total + + +def _count_resumed(session): + """ + Return ``(resumed, total)`` counting connections whose last TLS handshake + was resumed, using the appropriate mechanism for the active reactor + (``ssl.SSLSocket.session_reused`` for stdlib, ``SSL_session_reused()`` for + PyOpenSSL). + """ + if USES_PYOPENSSL: + return _count_pyopenssl_resumed(session) + return _count_resumed_connections(session) + + +def _teardown_ccm_cluster(): + """ + Stop and remove the shared CCM cluster, restoring it to a clean state. + + Registered via ``addClassCleanup`` so it runs even when ``setUpClass`` + fails partway through reconfiguring the cluster. + """ + ccm_cluster = get_cluster() + if ccm_cluster is not None: + ccm_cluster.stop() + remove_cluster() + + +# --------------------------------------------------------------------------- +# Test +# --------------------------------------------------------------------------- + +class TestTLSTicketResumption(unittest.TestCase): + """ + Verify that TLS session tickets are reused across reconnections. + + Follows the pattern of the Go driver's ``TestTLSTicketResumption``: + + 1. Connect to the cluster with a tracking session cache. + 2. Record the number of tickets learned during the initial handshakes. + 3. Close all connections, wait for the driver to re-establish them. + 4. Verify resumption directly for each reconnected socket via + ``session_reused`` / ``SSL_session_reused()`` (the driver deliberately + re-stores sessions after resumed handshakes, so new ``set()`` calls are + expected and not asserted against). + 5. Repeat the close/reconnect cycle once more for confidence. + + @test_category connection:ssl:tls_resumption + """ + + _cert_dir = None + + @classmethod + def setUpClass(cls): + if SCYLLA_VERSION is None: + raise unittest.SkipTest( + "TLS resumption test requires a local Scylla CCM cluster " + "(SCYLLA_VERSION is not set)") + + if USES_PYOPENSSL and SSL is None: + raise unittest.SkipTest( + "pyOpenSSL is not installed; cannot run the TLS resumption " + "test under the Twisted/Eventlet event loop") + + cls._cert_dir = tempfile.mkdtemp(prefix="tls_resumption_certs_") + cls.addClassCleanup(shutil.rmtree, cls._cert_dir, ignore_errors=True) + _generate_ssl_certs(cls._cert_dir) + + cls._server_cert_path = os.path.join(cls._cert_dir, "cassandra.crt") + cls._server_key_path = os.path.join(cls._cert_dir, "cassandra.key") + cls._ca_cert_path = os.path.join(cls._cert_dir, "ca.crt") + + use_single_node() + ccm_cluster = get_cluster() + # Register cluster teardown before mutating the shared CCM cluster so + # that a failure during stop/configure/start still restores cluster + # state (setUpClass failures skip tearDownClass, but registered class + # cleanups always run). + cls.addClassCleanup(_teardown_ccm_cluster) + ccm_cluster.stop() + + config_options = { + 'client_encryption_options': { + 'enabled': True, + 'certificate': cls._server_cert_path, + 'keyfile': cls._server_key_path, + 'truststore': cls._ca_cert_path, + } + } + ccm_cluster.set_configuration_options(config_options) + start_cluster_wait_for_up(ccm_cluster) + + def test_tls_ticket_resumption(self): + if USES_ASYNCIO: + self.skipTest( + "The asyncio reactor uses asyncio's native SSL transport and " + "does not support ssl_session_cache; TLS resumption is not " + "applicable") + if not USES_PYOPENSSL and not _server_supports_tls_resumption( + self.__class__._ca_cert_path, CASSANDRA_IP): + self.skipTest( + "Server at %s:9042 does not support TLS session resumption " + "(every connection shows session_reused=False); " + "skipping test" % CASSANDRA_IP + ) + + cache = _TrackingSSLSessionCache(max_size=1024, ttl=3600) + ssl_context = _make_ssl_context(self.__class__._ca_cert_path) + + cluster = TestCluster( + ssl_context=ssl_context, + ssl_session_cache=cache, + ) + session = cluster.connect(wait_for_all_pools=True) + try: + _wait_for_all_connections(session) + + tickets_after_initial = cache.ticket_count + self.assertGreater( + tickets_after_initial, 0, + "No TLS tickets were learned during initial connection — " + "the server may not support TLS session tickets", + ) + log.info("Initial connection: %d ticket(s) learned", + tickets_after_initial) + + # ── first reconnection cycle ─────────────────────────────── + _close_all_connections(session) + _wait_for_all_connections(session) + + tickets_after_reconnect1 = cache.ticket_count + log.info("After 1st reconnect: %d ticket(s) total", + tickets_after_reconnect1) + + # The driver re-stores the current session on every handshake, + # including resumed ones (a resumed TLS 1.2 handshake may issue a + # replacement ticket), so a growing ticket_count no longer proves + # renegotiation. Verify resumption directly instead: every + # reconnected connection must report a resumed handshake. On the + # stdlib ssl reactors this reads ssl.SSLSocket.session_reused; on + # the PyOpenSSL reactors it uses OpenSSL's SSL_session_reused() via + # the low-level binding (see _count_pyopenssl_resumed). + self._assert_all_resumed(session, "1st reconnect") + + # ── second reconnection cycle ────────────────────────────── + _close_all_connections(session) + _wait_for_all_connections(session) + + tickets_after_reconnect2 = cache.ticket_count + log.info("After 2nd reconnect: %d ticket(s) total", + tickets_after_reconnect2) + + self._assert_all_resumed(session, "2nd reconnect") + finally: + cluster.shutdown() + + def _assert_all_resumed(self, session, label): + """Assert every inspectable live connection resumed its TLS session.""" + resumed, total = _count_resumed(session) + if total == 0 and USES_PYOPENSSL: + self.skipTest( + "Cannot observe TLS session reuse on this pyOpenSSL build " + "(SSL_session_reused binding unavailable); skipping the " + "%s resumption assertion" % label) + self.assertGreater( + total, 0, "No inspectable TLS connections after %s" % label) + self.assertEqual( + resumed, total, + "TLS sessions were NOT reused after %s " + "(%d of %d connections resumed)" % (label, resumed, total), + ) + + def test_tls13_ticket_resumption(self): + # The main test pins TLS 1.2; this variant exercises the TLS 1.3 + # delayed-ticket path, where the NewSessionTicket arrives *after* the + # handshake. The fix defers caching until the session actually carries + # a ticket, so here we assert (1) every cached session has a ticket and + # (2) a subsequent reconnection resumes. This is a stdlib-ssl concern + # (``ssl.SSLSession.has_ticket``); the PyOpenSSL path has no equivalent + # introspection and is covered by test_tls_ticket_resumption. + if USES_PYOPENSSL: + self.skipTest( + "TLS 1.3 delayed-ticket caching is verified on the stdlib ssl " + "reactors; PyOpenSSL is covered by test_tls_ticket_resumption") + + if USES_ASYNCIO: + self.skipTest( + "The asyncio reactor uses asyncio's native SSL transport and " + "does not support ssl_session_cache; TLS resumption is not " + "applicable") + + ca_cert_path = self.__class__._ca_cert_path + if not _tls_negotiates_v13(ca_cert_path, CASSANDRA_IP): + self.skipTest( + "Server at %s:9042 does not negotiate TLS 1.3; skipping " + "TLS 1.3 resumption test" % CASSANDRA_IP) + + cache = _TrackingSSLSessionCache(max_size=1024, ttl=3600) + ssl_context = _make_ssl_context(ca_cert_path, force_tls13=True) + + cluster = TestCluster( + ssl_context=ssl_context, + ssl_session_cache=cache, + ) + session = cluster.connect(wait_for_all_pools=True) + try: + _wait_for_all_connections(session) + + # Sanity-check that every connection really negotiated TLS 1.3. + versions = {s.version() for s in _iter_live_ssl_sockets(session)} + self.assertEqual( + versions, {'TLSv1.3'}, + "Expected all connections on TLS 1.3, got %s" % (versions,)) + + # Core of the fix: only *ticketed* (resumable) sessions may be + # cached. A session cached before its NewSessionTicket arrived + # would report has_ticket=False and could never resume. + cached_sessions = cache._snapshot_sessions() + self.assertTrue( + cached_sessions, + "No TLS 1.3 sessions were cached — the delayed-ticket path " + "never stored a session") + for sess in cached_sessions: + self.assertTrue( + getattr(sess, 'has_ticket', False), + "A cached TLS 1.3 session has no ticket — it was cached " + "before the NewSessionTicket arrived and cannot resume") + + # End-to-end: reconnect through the driver and confirm the ticketed + # sessions actually resume. The driver sets the cached session + # before the handshake, so a genuine TLS 1.3 resumption reports + # session_reused=True. + _close_all_connections(session) + _wait_for_all_connections(session) + + resumed, total = _count_resumed_connections(session) + self.assertGreater( + total, 0, "No inspectable TLS connections after reconnect") + self.assertEqual( + resumed, total, + "TLS 1.3 sessions were NOT reused after reconnect " + "(%d of %d connections resumed)" % (resumed, total), + ) + finally: + cluster.shutdown() diff --git a/tests/unit/io/test_twistedreactor.py b/tests/unit/io/test_twistedreactor.py index 02bac10d8e..b58bce741d 100644 --- a/tests/unit/io/test_twistedreactor.py +++ b/tests/unit/io/test_twistedreactor.py @@ -15,7 +15,7 @@ import unittest from unittest.mock import Mock, patch -from cassandra.connection import DefaultEndPoint +from cassandra.connection import DefaultEndPoint, SSLSessionCache try: from twisted.test import proto_helpers @@ -196,3 +196,205 @@ 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 + # Default to a pre-TLS-1.3 handshake so the eager path caches; TLS 1.3 + # tests override this explicitly. + connection.get_protocol_version_name.return_value = 'TLSv1.2' + # info_callback reads the cache from the connection (not the creator); + # mirror clientConnectionForTLS() here. + connection._cassandra_ssl_session_cache = ssl_session_cache + # Likewise, info_callback reads the hostname check policy from the + # connection; mirror clientConnectionForTLS() here. + connection._cassandra_check_hostname = check_hostname + transport = Mock() + connection.get_app_data.return_value = transport + connection._cassandra_endpoint = endpoint + + 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 + + def test_tls13_not_cached_eagerly(self): + """On TLS 1.3 the eager path must not cache (or set the guard flag); + the ticket arrives later, so the deferred path handles it.""" + cache = SSLSessionCache() + creator, connection, transport, session = self._make_creator( + check_hostname=False, + endpoint_address='good.example.com', + cert_cn='good.example.com', + ssl_session_cache=cache, + ) + connection.get_protocol_version_name.return_value = 'TLSv1.3' + connection._cassandra_session_cached = False + + creator.info_callback(connection, self.SSL.SSL_CB_HANDSHAKE_DONE, 0) + + transport.failVerification.assert_not_called() + assert cache.size() == 0, "TLS 1.3 session was cached eagerly" + assert connection._cassandra_session_cached is False + + def test_uses_connection_cache_not_creator_cache(self): + """With a shared ssl_context, info_callback may be bound to a different + creator; it must write to the cache stashed on the connection, not the + creator's own cache.""" + creator_cache = SSLSessionCache() + connection_cache = SSLSessionCache() + # Creator carries creator_cache, but the connection belongs to another + # cluster whose cache is connection_cache. + creator, connection, transport, session = self._make_creator( + check_hostname=False, + endpoint_address='good.example.com', + cert_cn='good.example.com', + ssl_session_cache=creator_cache, + ) + connection._cassandra_ssl_session_cache = connection_cache + + creator.info_callback(connection, self.SSL.SSL_CB_HANDSHAKE_DONE, 0) + + assert connection_cache.get(('good.example.com', 9042)) is session + assert creator_cache.size() == 0, \ + "Session leaked into the creator's cache instead of the connection's" + + +class TestTwistedConnectionCacheGuard(unittest.TestCase): + """Verify the per-connection guard prevents storing a session twice, even + when info_callback (bound to a shared ssl_context) already stored it.""" + + def setUp(self): + if twistedreactor is None: + raise unittest.SkipTest("Twisted libraries not available") + + def _make_conn(self, cache, ssl_conn): + conn = TwistedConnection.__new__(TwistedConnection) + conn.endpoint = DefaultEndPoint('10.0.0.1', 9042) + conn._ssl_session_cache = cache + conn._tls_session_cached = False + creator = Mock() + creator._ssl_connection = ssl_conn + conn._ssl_creator = creator + return conn + + def test_no_double_store_when_info_callback_already_cached(self): + cache = SSLSessionCache() + first_session = Mock(name='session_1') + ssl_conn = Mock() + ssl_conn._cassandra_session_cached = True # info_callback stored it + ssl_conn.get_session.return_value = Mock(name='session_2') + cache.set(('10.0.0.1', 9042), first_session) + + conn = self._make_conn(cache, ssl_conn) + conn._cache_tls_session_if_needed() + + # Guard must prevent overwriting with a different session object. + assert cache.get(('10.0.0.1', 9042)) is first_session + + def test_stores_then_is_idempotent(self): + cache = SSLSessionCache() + session = Mock(name='session') + ssl_conn = Mock() + ssl_conn._cassandra_session_cached = False + ssl_conn.get_session.return_value = session + + conn = self._make_conn(cache, ssl_conn) + conn._cache_tls_session_if_needed() + assert cache.get(('10.0.0.1', 9042)) is session + assert conn._tls_session_cached is True + assert ssl_conn._cassandra_session_cached is True + + # A second call with a different reported session is a no-op. + ssl_conn.get_session.return_value = Mock(name='other') + conn._cache_tls_session_if_needed() + assert cache.get(('10.0.0.1', 9042)) is session + + def test_initial_call_is_skipped(self): + # The immediate post-handshake call (initial=True) must not store the + # session, since a TLS 1.3 ticket may not have arrived yet. + cache = SSLSessionCache() + ssl_conn = Mock() + ssl_conn._cassandra_session_cached = False + ssl_conn.get_session.return_value = Mock(name='premature_session') + + conn = self._make_conn(cache, ssl_conn) + conn._cache_tls_session_if_needed(initial=True) + assert cache.get(('10.0.0.1', 9042)) is None + assert conn._tls_session_cached is False diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 3d55bc1860..56039c7fb7 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -25,7 +25,7 @@ InvalidRequest, Unauthorized, AuthenticationFailed, OperationTimedOut, UnsupportedOperation, RequestValidationException, ConfigurationException, ProtocolVersion from cassandra.cluster import _Scheduler, Session, Cluster, ResultSet, SchemaAgreementScope, ControlConnectionQueryFallback, default_lbp_factory, \ ExecutionProfile, _ConfigMode, EXEC_PROFILE_DEFAULT -from cassandra.connection import ConnectionBusy, ConnectionException +from cassandra.connection import ConnectionBusy, ConnectionException, 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 @@ -1004,3 +1004,75 @@ 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_no_cache_when_only_ssl_options_set(self): + cluster = Cluster(contact_points=['127.0.0.1'], ssl_options={'ca_certs': '/dev/null'}) + assert cluster.ssl_session_cache is None + + 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_no_auto_cache_on_asyncio_reactor(self): + # The asyncio reactor cannot restore/store TLS sessions, so a cache + # must not be auto-created for it (an explicit cache is still honored). + import ssl + try: + from cassandra.io.asyncioreactor import AsyncioConnection + except Exception: + raise unittest.SkipTest("asyncio reactor not available") + 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, + connection_class=AsyncioConnection) + assert cluster.ssl_session_cache is None + + # But an explicitly-provided cache is still used. + custom = SSLSessionCache() + cluster = Cluster(contact_points=['127.0.0.1'], ssl_context=ctx, + connection_class=AsyncioConnection, + 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 diff --git a/tests/unit/test_connection.py b/tests/unit/test_connection.py index cf4607fbed..274c2f84e4 100644 --- a/tests/unit/test_connection.py +++ b/tests/unit/test_connection.py @@ -22,7 +22,9 @@ from cassandra.cluster import Cluster from cassandra.connection import (Connection, HEADER_DIRECTION_TO_CLIENT, ProtocolError, locally_supported_compressions, ConnectionHeartbeat, HeartbeatFuture, _Frame, Timer, TimerManager, - ConnectionException, ConnectionShutdown, DefaultEndPoint, ShardAwarePortGenerator) + ConnectionException, ConnectionShutdown, DefaultEndPoint, ShardAwarePortGenerator, + SniEndPoint, + SSLSessionCache) from cassandra.marshal import uint8_pack, uint32_pack, int32_pack from cassandra.protocol import (write_stringmultimap, write_int, write_string, SupportedMessage, ProtocolHandler, ResultMessage, @@ -633,3 +635,502 @@ def test_generate_is_repeatable_with_same_mock(self, mock_randrange): second_run = list(itertools.islice(gen.generate(0, 2), 5)) assert first_run == second_run + + +class TestSSLSessionCache(unittest.TestCase): + + @staticmethod + def _key(address, port, server_hostname=None): + return (address, port, server_hostname) + + def test_get_returns_none_when_empty(self): + cache = SSLSessionCache() + assert cache.get(self._key('127.0.0.1', 9042)) is None + + def test_set_and_get(self): + cache = SSLSessionCache() + session = object() # stand-in for ssl.SSLSession + cache.set(self._key('127.0.0.1', 9042), session) + assert cache.get(self._key('127.0.0.1', 9042)) is session + + def test_different_keys_are_independent(self): + cache = SSLSessionCache() + s1 = object() + s2 = object() + cache.set(self._key('127.0.0.1', 9042), s1) + cache.set(self._key('127.0.0.2', 9042), s2) + assert cache.get(self._key('127.0.0.1', 9042)) is s1 + assert cache.get(self._key('127.0.0.2', 9042)) is s2 + assert cache.get(self._key('127.0.0.1', 9043)) is None + + def test_sni_keys_are_independent_for_same_proxy(self): + cache = SSLSessionCache() + s1 = object() + s2 = object() + + cache.set(self._key('proxy.example.com', 9042, 'node-a'), s1) + cache.set(self._key('proxy.example.com', 9042, 'node-b'), s2) + + assert cache.get(self._key('proxy.example.com', 9042, 'node-a')) is s1 + assert cache.get(self._key('proxy.example.com', 9042, 'node-b')) is s2 + + def test_overwrite_existing_entry(self): + cache = SSLSessionCache() + old = object() + new = object() + cache.set(self._key('127.0.0.1', 9042), old) + cache.set(self._key('127.0.0.1', 9042), new) + assert cache.get(self._key('127.0.0.1', 9042)) is new + + def test_thread_safety(self): + """Concurrent set/get operations must not raise.""" + import threading + cache = SSLSessionCache() + errors = [] + + def writer(addr_suffix): + try: + for i in range(200): + cache.set(self._key('127.0.0.%d' % addr_suffix, 9042), object()) + except Exception as e: + errors.append(e) + + def reader(addr_suffix): + try: + for i in range(200): + cache.get(self._key('127.0.0.%d' % addr_suffix, 9042)) + except Exception as e: + errors.append(e) + + threads = [] + for n in range(5): + threads.append(threading.Thread(target=writer, args=(n,))) + threads.append(threading.Thread(target=reader, args=(n,))) + for t in threads: + t.start() + for t in threads: + t.join() + + assert errors == [] + + def test_ttl_expiration(self): + """Sessions expire after TTL.""" + cache = SSLSessionCache(max_size=10, ttl=1) + session = object() + with patch('cassandra.connection.time.monotonic', return_value=1000.0): + cache.set(self._key('127.0.0.1', 9042), session) + assert cache.get(self._key('127.0.0.1', 9042)) is session + + with patch('cassandra.connection.time.monotonic', return_value=1001.1): + assert cache.get(self._key('127.0.0.1', 9042)) is None + + def test_lru_eviction(self): + """LRU eviction when cache reaches max_size.""" + cache = SSLSessionCache(max_size=3, ttl=60) + + s1, s2, s3, s4 = object(), object(), object(), object() + cache.set(self._key('host1', 9042), s1) + cache.set(self._key('host2', 9042), s2) + cache.set(self._key('host3', 9042), s3) + assert cache.size() == 3 + + # Access host2 to make it recently used + cache.get(self._key('host2', 9042)) + + # Adding host4 should evict host1 (LRU) + cache.set(self._key('host4', 9042), s4) + assert cache.size() == 3 + assert cache.get(self._key('host1', 9042)) is None + assert cache.get(self._key('host2', 9042)) is s2 + assert cache.get(self._key('host3', 9042)) is s3 + assert cache.get(self._key('host4', 9042)) is s4 + + def test_none_session_not_cached(self): + """None sessions should be silently ignored.""" + cache = SSLSessionCache() + cache.set(self._key('127.0.0.1', 9042), None) + assert cache.size() == 0 + + def test_clear(self): + """clear() removes all entries.""" + cache = SSLSessionCache() + cache.set(self._key('host1', 9042), object()) + cache.set(self._key('host2', 9042), object()) + assert cache.size() == 2 + cache.clear() + assert cache.size() == 0 + + def test_clear_expired(self): + """clear_expired() removes only expired entries.""" + cache = SSLSessionCache(max_size=10, ttl=1) + with patch('cassandra.connection.time.monotonic', return_value=1000.0): + cache.set(self._key('host1', 9042), object()) + with patch('cassandra.connection.time.monotonic', return_value=1001.1): + cache.set(self._key('host2', 9042), object()) + assert cache.size() == 2 + cache.clear_expired() + assert cache.size() == 1 + assert cache.get(self._key('host1', 9042)) is None + assert cache.get(self._key('host2', 9042)) is not None + + def test_automatic_expired_cleanup(self): + """Expired sessions are cleaned during set() periodically.""" + with patch.object(SSLSessionCache, '_EXPIRY_CLEANUP_INTERVAL', 5): + cache = SSLSessionCache(max_size=10, ttl=1) + + with patch('cassandra.connection.time.monotonic', return_value=1000.0): + for i in range(3): + cache.set(self._key('host%d' % i, 9042), object()) + assert cache.size() == 3 + + with patch('cassandra.connection.time.monotonic', return_value=1001.1): + # Add sessions until cleanup triggers (at 5 operations) + for i in range(5): + cache.set(self._key('new%d' % i, 9042), object()) + + # Expired sessions should have been cleaned + assert cache.size() == 5 + + def test_custom_max_size_and_ttl(self): + """Cache respects custom max_size and ttl parameters.""" + cache = SSLSessionCache(max_size=50, ttl=7200) + assert cache.max_size == 50 + assert cache.ttl == 7200 + + def test_max_size_zero_raises(self): + """max_size=0 must raise ValueError.""" + with self.assertRaises(ValueError): + SSLSessionCache(max_size=0) + + def test_max_size_negative_raises(self): + """Negative max_size must raise ValueError.""" + with self.assertRaises(ValueError): + SSLSessionCache(max_size=-1) + + def test_ttl_zero_raises(self): + """ttl=0 must raise ValueError.""" + with self.assertRaises(ValueError): + SSLSessionCache(ttl=0) + + def test_ttl_negative_raises(self): + """Negative ttl must raise ValueError.""" + with self.assertRaises(ValueError): + SSLSessionCache(ttl=-10) + + def test_max_size_one_works(self): + """max_size=1 is the smallest valid cache — ensure it works.""" + cache = SSLSessionCache(max_size=1, ttl=60) + s1, s2 = object(), object() + cache.set(self._key('host1', 9042), s1) + assert cache.get(self._key('host1', 9042)) is s1 + # Adding a second entry should evict the first + cache.set(self._key('host2', 9042), s2) + assert cache.size() == 1 + assert cache.get(self._key('host1', 9042)) is None + assert cache.get(self._key('host2', 9042)) is s2 + + +class TestEndPointTLSSessionCacheKey(unittest.TestCase): + """Tests for tls_session_cache_key on endpoint classes.""" + + def test_default_endpoint_key(self): + endpoint = DefaultEndPoint('10.0.0.1', 9042) + assert endpoint.tls_session_cache_key == ('10.0.0.1', 9042) + + def test_default_endpoint_different_ports(self): + ep1 = DefaultEndPoint('10.0.0.1', 9042) + ep2 = DefaultEndPoint('10.0.0.1', 9043) + assert ep1.tls_session_cache_key != ep2.tls_session_cache_key + + def test_sni_endpoint_includes_server_name(self): + ep1 = SniEndPoint('proxy.example.com', 'server1', 9042) + ep2 = SniEndPoint('proxy.example.com', 'server2', 9042) + assert ep1.tls_session_cache_key == ('proxy.example.com', 9042, 'server1') + assert ep2.tls_session_cache_key == ('proxy.example.com', 9042, 'server2') + assert ep1.tls_session_cache_key != ep2.tls_session_cache_key + + def test_unix_socket_endpoint_key(self): + from cassandra.connection import UnixSocketEndPoint + ep = UnixSocketEndPoint('/var/run/scylla.sock') + assert ep.tls_session_cache_key == ('/var/run/scylla.sock',) + + def test_unix_socket_different_paths(self): + from cassandra.connection import UnixSocketEndPoint + ep1 = UnixSocketEndPoint('/var/run/scylla.sock') + ep2 = UnixSocketEndPoint('/tmp/scylla.sock') + assert ep1.tls_session_cache_key != ep2.tls_session_cache_key + + +class TestConnectionSSLSessionRestore(unittest.TestCase): + + @patch.object(Connection, '_connect_socket') + @patch.object(Connection, '_send_options_message') + def test_wrap_socket_restores_cached_session(self, _send, _connect): + """_wrap_socket_from_context sets ssl_sock.session from cache.""" + import ssl as _ssl + + mock_ssl_sock = Mock() + mock_ctx = Mock(spec=_ssl.SSLContext) + mock_ctx.check_hostname = False + mock_ctx.wrap_socket.return_value = mock_ssl_sock + + cached = Mock(name='cached_session') + cache = SSLSessionCache() + cache.set(('10.0.0.1', 9042), cached) + + conn = Connection.__new__(Connection) + conn.endpoint = DefaultEndPoint('10.0.0.1', 9042) + conn.ssl_context = mock_ctx + conn.ssl_options = {} + conn._ssl_session_cache = cache + + result = conn._wrap_socket_from_context() + assert result is mock_ssl_sock + assert mock_ssl_sock.session == cached + + @patch.object(Connection, '_connect_socket') + @patch.object(Connection, '_send_options_message') + def test_wrap_socket_tolerates_missing_cache(self, _send, _connect): + """No error when _ssl_session_cache is None.""" + import ssl as _ssl + + mock_ssl_sock = Mock() + mock_ctx = Mock(spec=_ssl.SSLContext) + mock_ctx.check_hostname = False + mock_ctx.wrap_socket.return_value = mock_ssl_sock + + conn = Connection.__new__(Connection) + conn.endpoint = DefaultEndPoint('10.0.0.1', 9042) + conn.ssl_context = mock_ctx + conn.ssl_options = {} + conn._ssl_session_cache = None + + result = conn._wrap_socket_from_context() + assert result is mock_ssl_sock + + @patch.object(Connection, '_connect_socket') + @patch.object(Connection, '_send_options_message') + def test_wrap_socket_handles_set_session_failure(self, _send, _connect): + """If setting session raises ssl.SSLError, it is silently ignored.""" + import ssl as _ssl + + mock_ssl_sock = Mock() + type(mock_ssl_sock).session = property( + fget=lambda self: None, + fset=Mock(side_effect=_ssl.SSLError("bad session")), + ) + mock_ctx = Mock(spec=_ssl.SSLContext) + mock_ctx.check_hostname = False + mock_ctx.wrap_socket.return_value = mock_ssl_sock + + cache = SSLSessionCache() + cache.set(('10.0.0.1', 9042), Mock(name='bad_cached')) + + conn = Connection.__new__(Connection) + conn.endpoint = DefaultEndPoint('10.0.0.1', 9042) + conn.ssl_context = mock_ctx + conn.ssl_options = {} + conn._ssl_session_cache = cache + + # Should NOT raise + result = conn._wrap_socket_from_context() + assert result is mock_ssl_sock + + @patch.object(Connection, '_connect_socket') + @patch.object(Connection, '_send_options_message') + def test_wrap_socket_handles_value_error_on_different_context(self, _send, _connect): + """If setting session raises ValueError (different SSLContext), it is silently ignored.""" + import ssl as _ssl + + mock_ssl_sock = Mock() + type(mock_ssl_sock).session = property( + fget=lambda self: None, + fset=Mock(side_effect=ValueError("Session refers to a different SSLContext")), + ) + mock_ctx = Mock(spec=_ssl.SSLContext) + mock_ctx.check_hostname = False + mock_ctx.wrap_socket.return_value = mock_ssl_sock + + cache = SSLSessionCache() + cache.set(('10.0.0.1', 9042), Mock(name='stale_cached')) + + conn = Connection.__new__(Connection) + conn.endpoint = DefaultEndPoint('10.0.0.1', 9042) + conn.ssl_context = mock_ctx + conn.ssl_options = {} + conn._ssl_session_cache = cache + + # Should NOT raise — falls back to full handshake + result = conn._wrap_socket_from_context() + assert result is mock_ssl_sock + + @patch.object(Connection, '_connect_socket') + @patch.object(Connection, '_send_options_message') + def test_wrap_socket_uses_sni_specific_cached_session(self, _send, _connect): + import ssl as _ssl + + mock_ssl_sock = Mock() + mock_ctx = Mock(spec=_ssl.SSLContext) + mock_ctx.check_hostname = False + mock_ctx.wrap_socket.return_value = mock_ssl_sock + + expected = Mock(name='node_b_session') + cache = SSLSessionCache() + cache.set(('proxy.example.com', 9042, 'node-a'), Mock(name='node_a_session')) + cache.set(('proxy.example.com', 9042, 'node-b'), expected) + + conn = Connection.__new__(Connection) + conn.endpoint = SniEndPoint('proxy.example.com', 'node-b', 9042) + conn.ssl_context = mock_ctx + conn.ssl_options = {'server_hostname': 'node-b'} + conn._ssl_session_cache = cache + + result = conn._wrap_socket_from_context() + assert result is mock_ssl_sock + assert mock_ssl_sock.session == expected + + +class TestConnectionCacheTLSSession(unittest.TestCase): + + def _make_conn(self): + conn = Connection.__new__(Connection) + conn.endpoint = DefaultEndPoint('10.0.0.1', 9042) + conn.ssl_context = Mock() + conn._ssl_session_cache = SSLSessionCache() + conn._socket = Mock() + conn._socket.session_reused = False + conn._socket.version = Mock(return_value='TLSv1.2') + return conn + + def test_cache_tls_session_stores_session(self): + conn = self._make_conn() + fake_session = Mock(name='ssl_session') + conn._socket.session = fake_session + + conn._cache_tls_session_if_needed() + assert conn._ssl_session_cache.get(('10.0.0.1', 9042)) is fake_session + + def test_cache_tls_session_initial_call_caches_tls12(self): + # For the stdlib ssl path the initial (post-handshake) call caches a + # TLS 1.2 session immediately; the ``initial`` flag is only honoured by + # the PyOpenSSL subclasses. + conn = self._make_conn() + fake_session = Mock(name='ssl_session') + conn._socket.session = fake_session + + conn._cache_tls_session_if_needed(initial=True) + assert conn._ssl_session_cache.get(('10.0.0.1', 9042)) is fake_session + assert conn._tls_session_cached is True + + def test_cache_tls_session_is_idempotent(self): + # A second call for the same connection must not store the session + # again (guarded by _tls_session_cached). + conn = self._make_conn() + first_session = Mock(name='ssl_session_1') + conn._socket.session = first_session + conn._cache_tls_session_if_needed() + assert conn._tls_session_cached is True + + # Even if the socket now reports a different session, the guard should + # prevent a second store. + conn._socket.session = Mock(name='ssl_session_2') + conn._cache_tls_session_if_needed() + assert conn._ssl_session_cache.get(('10.0.0.1', 9042)) is first_session + + def test_cache_tls_session_no_op_when_session_none(self): + conn = self._make_conn() + conn._socket.session = None + + conn._cache_tls_session_if_needed() + assert conn._ssl_session_cache.get(('10.0.0.1', 9042)) is None + + def test_cache_tls_session_stores_resumed_session(self): + # A resumed handshake (session_reused True) may carry a replacement + # ticket, so the current session must still be (re-)stored rather than + # skipped. + conn = self._make_conn() + conn._socket.session_reused = True + fake_session = Mock(name='resumed_ssl_session') + conn._socket.session = fake_session + + conn._cache_tls_session_if_needed() + assert conn._ssl_session_cache.get(('10.0.0.1', 9042)) is fake_session + assert conn._tls_session_cached is True + + def test_cache_tls_session_defers_tls13_without_ticket(self): + # TLS 1.3 exposes the SSLSession before the ticket arrives; caching + # must be deferred (and the guard left unset) until has_ticket is True. + conn = self._make_conn() + conn._socket.version = Mock(return_value='TLSv1.3') + ticketless = Mock(name='ticketless_session') + ticketless.has_ticket = False + conn._socket.session = ticketless + + conn._cache_tls_session_if_needed() + assert conn._ssl_session_cache.get(('10.0.0.1', 9042)) is None + assert conn._tls_session_cached is False + + # Once the ticket arrives, a later call stores the resumable session. + with_ticket = Mock(name='ticketed_session') + with_ticket.has_ticket = True + conn._socket.session = with_ticket + conn._cache_tls_session_if_needed() + assert conn._ssl_session_cache.get(('10.0.0.1', 9042)) is with_ticket + assert conn._tls_session_cached is True + + def test_cache_tls_session_defers_tls13_initial_with_ticket(self): + # On a resumed TLS 1.3 handshake the SSLSession already reports + # has_ticket=True, but that ticket is the (potentially single-use) one + # used for resumption and may be rotated by the server's replacement + # NewSessionTicket. The immediate post-handshake call (initial=True) + # must NOT cache it or set the guard, so the deferred Ready/AuthSuccess + # store can capture the replacement ticket. + conn = self._make_conn() + conn._socket.version = Mock(return_value='TLSv1.3') + old_ticket = Mock(name='resumption_ticket_session') + old_ticket.has_ticket = True + conn._socket.session = old_ticket + + conn._cache_tls_session_if_needed(initial=True) + assert conn._ssl_session_cache.get(('10.0.0.1', 9042)) is None + assert conn._tls_session_cached is False + + # The deferred (non-initial) call then stores the replacement ticket. + replacement = Mock(name='replacement_ticket_session') + replacement.has_ticket = True + conn._socket.session = replacement + conn._cache_tls_session_if_needed() + assert conn._ssl_session_cache.get(('10.0.0.1', 9042)) is replacement + assert conn._tls_session_cached is True + + def test_cache_tls_session_no_op_when_cache_none(self): + conn = self._make_conn() + conn._ssl_session_cache = None + conn._socket.session = Mock() + + # Should not raise + conn._cache_tls_session_if_needed() + + def test_cache_tls_session_no_op_when_no_ssl_context(self): + conn = self._make_conn() + conn.ssl_context = None + conn._socket.session = Mock() + + conn._cache_tls_session_if_needed() + assert conn._ssl_session_cache.get(('10.0.0.1', 9042)) is None + + def test_cache_tls_session_uses_sni_specific_key(self): + conn = Connection.__new__(Connection) + conn.endpoint = SniEndPoint('proxy.example.com', 'node-b', 9042) + conn.ssl_context = Mock() + conn.ssl_options = {'server_hostname': 'node-b'} + conn._ssl_session_cache = SSLSessionCache() + conn._socket = Mock() + conn._socket.session_reused = False + conn._socket.version = Mock(return_value='TLSv1.2') + fake_session = Mock(name='ssl_session') + conn._socket.session = fake_session + + conn._cache_tls_session_if_needed() + assert conn._ssl_session_cache.get(('proxy.example.com', 9042, 'node-b')) is fake_session + assert conn._ssl_session_cache.get(('proxy.example.com', 9042, 'node-a')) is None