diff --git a/cassandra/cluster.py b/cassandra/cluster.py index d858f5835e..f608312ad5 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -53,6 +53,7 @@ EndPoint, DefaultEndPoint, DefaultEndPointFactory, SniEndPointFactory, UnixSocketEndPoint, ConnectionBusy, locally_supported_compressions) +from cassandra.ssl_session_cache import SSLSessionCache from cassandra.cqltypes import UserType import cassandra.cqltypes as types from cassandra.encoder import Encoder @@ -867,6 +868,8 @@ def default_retry_policy(self, policy): .. versionadded:: 3.17.0 """ + # ssl_session_cache is a property, defined with the rest of the TLS + # session resumption code below. sockopts = None """ An optional list of tuples which will be used as arguments to @@ -1167,6 +1170,12 @@ def token_metadata_enabled(self, enabled): _prepared_statements = None _prepared_statement_lock = None _idle_heartbeat = None + _ssl_session_cache = _NOT_SET + _ssl_session_cache_created = None + # Held only while the cache above is constructed, and nothing else is held + # underneath it: making one is not something that can take part in an + # ordering with the pool locks, which self._lock would. + _ssl_session_cache_lock = None _protocol_version_explicit = False _discount_down_events = True @@ -1222,7 +1231,8 @@ def __init__(self, application_info:Optional[ApplicationInfoBase]=None, client_routes_config:Optional[ClientRoutesConfig]=None, allow_control_connection_query_fallback:Optional[ControlConnectionQueryFallback]=ControlConnectionQueryFallback.Disabled, - driver_config_reporting_enabled=True + driver_config_reporting_enabled=True, + ssl_session_cache=_NOT_SET ): """ ``executor_threads`` defines the number of threads in a pool for handling asynchronous tasks such as @@ -1469,6 +1479,10 @@ def __init__(self, self.ssl_options = ssl_options self.ssl_context = ssl_context + + self._ssl_session_cache_lock = Lock() + self._ssl_session_cache = ssl_session_cache + # Materialized once: these are applied to every socket the cluster opens # and are read again to build the configuration report, so a one-shot # iterable would leave whichever consumer ran second with nothing at all. @@ -1681,6 +1695,125 @@ def add_execution_profile(self, name, profile, pool_wait_timeout=5): raise OperationTimedOut("Failed to create all new connection pools in the %ss timeout." % pool_wait_timeout, timeout=pool_wait_timeout) + @property + def ssl_session_cache(self): + """ + A :class:`~cassandra.ssl_session_cache.SSLSessionCache` shared by every + connection this cluster opens, letting them resume TLS sessions instead of + performing a full handshake each time. This matters most for the group of + per-shard connections opened to a node at once, and for reconnections. + + One is created on first use when :attr:`~Cluster.ssl_context` is set. + Nothing is settled before then: this answers against whatever + :attr:`~Cluster.ssl_context` and :attr:`~Cluster.connection_class` are in + force when it is read, so configuring TLS at any point still gets a cache, + and swapping in a connection class that cannot resume turns resumption off + rather than handing that class a keyword it does not take. + + A cache created here is reachable only through this attribute, so it and + the sessions in it go when the cluster does. A cache passed in stays the + caller's: :meth:`~.Cluster.shutdown` leaves its entries alone, so several + clusters -- at the same time or one after another -- can share the + sessions in it. Its entries hold the ``SSLContext`` they were established + with, bounded by the cache's + :attr:`~cassandra.ssl_session_cache.SSLSessionCache.max_size`; call + :meth:`~cassandra.ssl_session_cache.SSLSessionCache.clear` to release + them. + + Assigning it is honoured whenever, and a cache that cannot be used reads + back as :const:`None` rather than being left to fill with nothing. Why + it cannot be used is said once, by :meth:`~.Cluster.connect`, which is + the only place that says it: a cache assigned after that reads back as + :const:`None` just the same, without a second word about it. + + Pass ``ssl_session_cache=None`` to :class:`.Cluster` to turn resumption + off, or pass your own instance to size it or to share it between + clusters:: + + from cassandra.ssl_session_cache import SSLSessionCache + + cluster = Cluster(ssl_context=ssl_context, + ssl_session_cache=SSLSessionCache(max_size=64)) + + What resumption asks of the reactor, and of the server -- Scylla issues + session tickets only when told to, before 2026.3 -- is described under + :ref:`security`. Where any of that is missing no cache is created and + this reads as :const:`None`, and connections handshake in full as they + did before. Over TLS 1.2 a server that issues no tickets still assigns + a session id, so the cache may hold an entry it will not honour; + offering that costs nothing and the handshake simply completes in + full.""" + if not self._tls_session_resumption_available(): + return None + if self._ssl_session_cache is not _NOT_SET: + return self._ssl_session_cache + if self._ssl_session_cache_created is None: + # Reached from whichever threads are opening connections, which for + # a cluster configured with TLS after connect() is several pools at + # once. Two of them both finding nothing here would each make a + # cache and the later one would win, leaving the connections of the + # other to fill an object nothing can reach and never resume from. + with self._ssl_session_cache_lock: + if self._ssl_session_cache_created is None: + self._ssl_session_cache_created = SSLSessionCache() + return self._ssl_session_cache_created + + @ssl_session_cache.setter + def ssl_session_cache(self, cache): + self._ssl_session_cache = cache + + def _tls_session_resumption_available(self): + """ + Whether a session could be resumed at all: it has to be replayable onto + the same ``SSLContext``, and the reactor has to give the driver a chance + to offer it before the handshake. connection_class is not required to + derive from Connection, so one that does not report the capability is + treated as lacking it rather than raising. + """ + return (self.ssl_context is not None and + getattr(self.connection_class, + 'supports_tls_session_resumption', False)) + + def _report_tls_session_resumption(self): + """ + Say, once, why a cluster configured for TLS will not resume sessions. + + A cache the caller asked for and cannot have is a warning: asking for + resumption and silently getting none is worse than not having it, since + the attribute reads as None and nothing is ever cached, which is also + what a server that issues no tickets looks like. + + Where nothing was asked for it is only a note. Resumption is on by + default wherever it works, so a cluster that configured TLS and will + not get it has something worth finding -- a Python 3.12 or newer + without the libev extension resolves to the asyncio reactor, which + cannot restore a session -- but nobody asked, so this is not news to + interrupt anyone with. A cluster with no TLS at all is told nothing: + there is nothing there to resume. + """ + if self._tls_session_resumption_available(): + return + if self.ssl_context is None and self.ssl_options is None: + return + + if self.ssl_context is None: + reason = ('no ssl_context is configured, and a session cannot be ' + 'replayed onto the fresh context each connection builds ' + 'from ssl_options') + else: + reason = ('%s cannot restore a session before the handshake' % + getattr(self.connection_class, '__name__', + self.connection_class)) + + if (self._ssl_session_cache is _NOT_SET + or self._ssl_session_cache is None): + log.debug('TLS session resumption is unavailable here, so every ' + 'connection will perform a full handshake: %s.', reason) + else: + log.warning('ssl_session_cache is set but TLS session resumption ' + 'is unavailable here, so no sessions will be cached: ' + '%s.', reason) + def connection_factory(self, endpoint, host_conn = None, *args, **kwargs): """ Called to create a new connection with proper configuration. @@ -1702,6 +1835,13 @@ 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) + ssl_session_cache = self.ssl_session_cache + if ssl_session_cache is not None: + # Set only where resumption is possible, so this is also the test + # for that: a connection class that does not accept the keyword + # should not have to grow one for a cluster that will never cache a + # session. + kwargs_dict.setdefault('ssl_session_cache', 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) @@ -1761,6 +1901,7 @@ def connect(self, keyspace=None, wait_for_all_pools=False): self.contact_points, self.protocol_version) self.connection_class.initialize_reactor() _register_cluster_shutdown(self) + self._report_tls_session_resumption() try: self.control_connection.connect() @@ -1850,6 +1991,11 @@ def shutdown(self): if self.metrics_enabled and self.metrics: self.metrics.shutdown() + # Nothing to do here for ssl_session_cache: a cache created for this + # cluster is reachable only through it and goes when it does, and a + # cache the caller supplied is the caller's to empty -- deleting rows + # in it here would defeat sharing one so that sessions outlive a + # cluster. See the attribute's documentation. _discard_cluster_shutdown(self) def __enter__(self): diff --git a/cassandra/connection.py b/cassandra/connection.py index d0a75818b2..2d7a5e9738 100644 --- a/cassandra/connection.py +++ b/cassandra/connection.py @@ -173,6 +173,37 @@ def socket_family(self): """ return socket.AF_UNSPEC + _tls_session_cache_key_override = None + + @property + def tls_session_cache_key(self): + """ + A hashable value identifying the TLS peer this endpoint connects to, + used to look up cached TLS sessions (see + :class:`~cassandra.ssl_session_cache.SSLSessionCache`). Two endpoints + may share a key only if a + TLS session established with one is valid for the other. + + An endpoint built to reach a node that another one already describes -- + an alternate listener of the same server -- carries that node's key + here, so both share one cached session. Subclasses give their own + identity in :meth:`_default_tls_session_cache_key`. + """ + if self._tls_session_cache_key_override is not None: + return self._tls_session_cache_key_override + return self._default_tls_session_cache_key() + + @property + def _tls_session_cache_key_is_borrowed(self): + """ + Whether :attr:`tls_session_cache_key` names a node this endpoint is an + alternate route to, rather than this endpoint's own identity. + """ + return self._tls_session_cache_key_override is not None + + def _default_tls_session_cache_key(self): + return (self.address, self.port) + def resolve(self): """ Resolve the endpoint to an address/port. This is called @@ -287,6 +318,11 @@ def port(self): def ssl_options(self): return self._ssl_options + def _default_tls_session_cache_key(self): + # Several SNI endpoints share a proxy address and port, but each one + # presents a different server_name and therefore a different TLS peer. + return (self.address, self.port, self._server_name) + def resolve(self): try: resolved_addresses = socket.getaddrinfo(self._proxy_address, self._port, @@ -465,6 +501,11 @@ def port(self) -> Optional[int]: def host_id(self) -> uuid.UUID: return self._host_id + def _default_tls_session_cache_key(self): + # The proxy address this endpoint resolves to may change between + # connections; the TLS peer is identified by the node behind it. + return (self._host_id, self._original_address, self._original_port) + def resolve(self) -> Tuple[str, int]: """ Resolve endpoint by delegating to the handler. @@ -815,6 +856,23 @@ class Connection(object): ssl_context = None last_error = None + # Whether this connection implementation can restore a cached TLS session + # before the handshake. True here because the accessors below speak the + # stdlib ssl API, which is what the asyncore and libev reactors use. A + # reactor that establishes TLS some other way sets this to False until it + # overrides those accessors -- asyncio hands the handshake to + # loop.create_connection(), which offers no point to restore a session at + # all. + supports_tls_session_resumption = True + + _ssl_session_cache = None + _tls_session_offered = None + _tls_handshake_began_at = None + + # RFC 8446 section 4.6.1: "Clients MUST NOT cache tickets for longer than + # 7 days, regardless of the ticket_lifetime". + _MAX_TLS_SESSION_LIFETIME = 7 * 24 * 60 * 60 + # The current number of operations that are in flight. More precisely, # the number of request IDs that are currently in use. # This includes orphaned requests. @@ -943,13 +1001,22 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None, 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, - session_id=None, driver_config_reporter: Optional[DriverConfigReporter] = None): + session_id=None, driver_config_reporter: Optional[DriverConfigReporter] = 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 + # A TLS session can only be replayed onto the SSLContext it was + # established with -- the stdlib ssl module rejects anything else with + # "Session refers to a different SSLContext". Connections that derive + # their own context from ssl_options below therefore have nothing to + # gain from the cache, and would only fill it with sessions no one can + # use, so resumption is limited to a caller-supplied context. + if ssl_context is not None and self.supports_tls_session_resumption: + self._ssl_session_cache = ssl_session_cache self.sockopts = sockopts self.compression = compression self.cql_version = cql_version @@ -1089,17 +1156,16 @@ def _wrap_socket_from_context(self): # Extract a subset of names from self.ssl_options which apply to SSLContext.wrap_socket (or at least the parts # of it that don't involve building an SSLContext under the covers) - wrap_socket_opt_names = ['server_side', 'do_handshake_on_connect', 'suppress_ragged_eofs', 'server_hostname'] + wrap_socket_opt_names = ['server_side', 'do_handshake_on_connect', 'suppress_ragged_eofs'] opts = {k:self.ssl_options.get(k, None) for k in wrap_socket_opt_names if k in self.ssl_options} - # PYTHON-1186: set the server_hostname only if the SSLContext has - # check_hostname enabled and it is not already provided by the EndPoint ssl options - #opts['server_hostname'] = self.endpoint.address - if (self.ssl_context.check_hostname and 'server_hostname' not in opts): - server_hostname = self.endpoint.address + server_hostname = self._tls_server_hostname() + if server_hostname is not None: opts['server_hostname'] = server_hostname - return self.ssl_context.wrap_socket(self._socket, **opts) + ssl_sock = self.ssl_context.wrap_socket(self._socket, **opts) + self._restore_tls_session(ssl_sock) + return ssl_sock def _initiate_connection(self, sockaddr): if self.features.shard_id is not None: @@ -1113,6 +1179,238 @@ def _initiate_connection(self, sockaddr): self._socket.connect(sockaddr) + # TLS session resumption. Everything a reactor establishing TLS by other + # means than the stdlib ssl module has to reimplement is in the three + # accessors below -- _set_tls_session, _get_resumable_tls_session and + # _tls_negotiated_version -- and nothing else here touches the socket, so + # the policy around them is shared by every reactor that has them. + + def _tls_server_hostname(self): + """ + The name ``wrap_socket`` is given, which is the name the peer + certificate is verified against when the context checks hostnames. + + PYTHON-1186: the endpoint's ssl_options may provide it (an SNI proxy + needs it for routing); otherwise it is the endpoint address, and only + when the context actually checks hostnames. + """ + if 'server_hostname' in self.ssl_options: + return self.ssl_options['server_hostname'] + if getattr(self.ssl_context, 'check_hostname', False): + return self.endpoint.address + return None + + def _tls_session_cache_key(self): + # The SSLContext is part of the key because a session cannot be + # replayed onto a different one, and a cache may be shared by several + # clusters. It is held strongly: a cached session already keeps its + # context alive on its own -- CPython's SSLSession holds a reference to + # the context it was established with -- so holding it weakly here + # would buy nothing. + # The verified name is part of it because a resumed handshake carries no + # Certificate, so that name is never checked again: offering a session + # to a connection expecting a different name would silently skip + # hostname verification for it. Deriving the name from the same place + # _wrap_socket_from_context does is what keeps the two from drifting. + return (self.ssl_context, self.endpoint.tls_session_cache_key, + self._tls_server_hostname()) + + def _restore_tls_session(self, sock): + """ + Offer the session cached for this endpoint, if any, on *sock*, which + must not have begun its handshake yet. Not offering one only costs a + full handshake, so failures here are logged and ignored. + """ + # Set below only if a session is actually offered, so that this always + # describes the attempt in flight: _connect_socket may come back here + # for another address, and an earlier attempt's session is not this + # one's to retract. + self._tls_session_offered = None + # Taken for every attempt, offered or not: what is stored afterwards + # was issued during the handshake this is about to begin, and its + # lifetime runs from then rather than from when the CQL handshake + # finishes reading it. + self._tls_handshake_began_at = time.monotonic() + if self._ssl_session_cache is None: + return + + try: + session = self._ssl_session_cache.get(self._tls_session_cache_key()) + if session is not None: + self._set_tls_session(sock, session) + self._tls_session_offered = session + log.debug("Offering a cached TLS session to %s", self.endpoint) + except Exception as exc: + log.debug("Could not offer a cached TLS session to %s: %s", self.endpoint, exc) + + def _discard_tls_session(self): + """ + Drop the session offered on this connection, after a handshake it took + part in failed -- during _connect_socket, or, where ssl_options defer + the handshake past it, wherever the reactor meets the failure. + + A cached session should never be able to fail a handshake -- RFC 5077 + section 3.2 and RFC 8446 section 4.6.1 both have the server fall back to + a full one when it will not resume -- but nothing stores a fresh session + for a connection that never came up, so an entry that does provoke a + failure would otherwise be offered again by every later connection until + its lifetime ran out. + + Only the session this connection offered is dropped: another connection + may have stored a session the peer issued in its place, and removing + that would cost every later connection a full handshake for a session + that never failed anything. A store of the same session is not that, + and leaves the entry retractable. + + Nothing is dropped for an endpoint that borrowed its key from another. + An alternate listener resumes the node's session and stores back to it, + which is the point of sharing the key, but the entry is not its to + remove: it would cost the node's other pools and its control connection + a full handshake apiece, and a pool filling against that listener would + do it again on every retry. A session that genuinely cannot resume + fails on the endpoint that owns the key as well, and goes from there. + """ + offered, self._tls_session_offered = self._tls_session_offered, None + if offered is None or self._ssl_session_cache is None: + # Nothing was offered on this connection -- there was nothing + # cached, or setting it on the socket was refused -- so there is + # nothing of ours to retract. Going on would hand discard() no + # session to compare against, which tells it to drop whatever is + # there, including one a sibling connection stored in the meantime. + return + + if self.endpoint._tls_session_cache_key_is_borrowed: + log.debug("Leaving the TLS session %s offered where it is: the key " + "is another endpoint's", self.endpoint) + return + + try: + self._ssl_session_cache.discard(self._tls_session_cache_key(), offered) + log.debug("Dropped the cached TLS session offered to %s", self.endpoint) + except Exception as exc: + log.debug("Could not drop the cached TLS session of %s: %s", self.endpoint, exc) + + def _store_tls_session(self): + """ + Cache this connection's TLS session so that later connections to the + same peer can resume it. Called once the CQL handshake has completed, + which is late enough to have read a TLS 1.3 session ticket from a + server that sends one with the handshake, Scylla among them. + """ + # Reaching here is the TLS handshake having stood: the CQL one is + # complete, so there is nothing left to retract and the attribute goes + # back to meaning "offered, and not yet known to be good". + offered, self._tls_session_offered = self._tls_session_offered, None + if self._ssl_session_cache is None: + return + + try: + session = self._get_resumable_tls_session() + if session is None: + # Every connection samples at this same point in the CQL + # handshake, so reaching here is not something the next one + # retries: a peer that has not produced a ticket by now will + # not have for the next connection either, and nothing is ever + # cached for it. Scylla produces one well before this -- the + # TLS handshake plus the OPTIONS exchange -- so a peer that + # deferred its ticket past this point is what would call for a + # later hook than this one. + return + lifetime = self._tls_session_lifetime(session) + if lifetime is None: + return + self._ssl_session_cache.set(self._tls_session_cache_key(), + session, lifetime, offered=offered) + log.debug("Cached the TLS session of %s for resumption, for %ss", + self.endpoint, int(lifetime)) + except Exception as exc: + log.debug("Could not cache the TLS session of %s: %s", self.endpoint, exc) + + def _tls_session_lifetime(self, session): + """ + How much longer, in seconds, *session* may be offered, or ``None`` if + it must not be cached at all. + + A ticket's lifetime is the one the server announced; + ``SSLSession.timeout`` is the local context's default and says nothing + about what the peer will still accept, so it is only used where the + server announced nothing. RFC 8446 section 4.6.1 also caps a client at + seven days however long a lifetime the server asked for. + + A zero lifetime means opposite things in the two RFCs that define + tickets, so the negotiated version has to decide: RFC 8446 section 4.6.1 + (TLS 1.3) says discard the ticket immediately, while RFC 5077 section 3.3 + (TLS 1.2) reserves zero for "lifetime unspecified" and leaves retention + to local policy -- for which the local timeout is the only figure + available. + + What is left of that lifetime is what the entry gets. The peer issued + the ticket during the handshake, while this runs once the CQL handshake + has completed -- a startup exchange later, and an authentication one + after that -- so stamping the announced lifetime here would hand the + entry every second of that on top of what the peer allowed. The age is + measured from a monotonic mark taken as the TLS handshake began, not + from ``SSLSession.time``: that is a wall-clock stamp, and subtracting it + from ``time.time()`` would let a clock step landing in between decide + the answer -- far enough forward and nothing is cached at all, backward + and the age disappears. The deadline the cache keeps is monotonic too, + so nothing after this point can skew it either. + """ + if session.has_ticket: + lifetime = session.ticket_lifetime_hint + if not lifetime: + if self._tls_negotiated_version() == 'TLSv1.3': + return None + lifetime = session.timeout + elif self._tls_negotiated_version() == 'TLSv1.3': + # TLS 1.3 resumes only from a ticket, whose pre-shared key is the + # whole mechanism; the session id a TLS 1.3 handshake carries is + # legacy_session_id_echo (RFC 8446 section 4.1.3), which a server + # echoes for the middlebox compatibility mode of appendix D.4 and + # which resumes nothing. OpenSSL does not report such an id as the + # session's -- one appears only once a NewSessionTicket has been + # read, which is what defers the store until then -- so this is + # unreachable there; it is here so the rule follows from the + # protocol rather than from what one library chooses to expose. + return None + else: + lifetime = session.timeout + + lifetime = min(lifetime, self._MAX_TLS_SESSION_LIFETIME) + if self._tls_handshake_began_at is not None: + lifetime -= time.monotonic() - self._tls_handshake_began_at + return lifetime if lifetime > 0 else None + + def _set_tls_session(self, sock, session): + sock.session = session + + def _tls_negotiated_version(self): + """ + The name of the TLS version in force on this connection, as + ``SSLSocket.version`` reports it -- ``'TLSv1.3'`` and so on -- or + :const:`None` if there is no handshake to ask about. + + Only the retention rules need this: which RFC defines the tickets the + peer issues, and so what a lifetime of zero in one means, is decided by + the version, and no property of the session itself distinguishes them. + """ + return self._socket.version() + + def _get_resumable_tls_session(self): + session = getattr(self._socket, 'session', None) + if session is None: + return None + # There has to be something to offer on the next connection: a ticket + # (RFC 5077 for TLS 1.2, RFC 8446 for TLS 1.3) or a session id. A TLS + # 1.3 server sends its NewSessionTicket after the handshake as a + # separate message, and until that has been read the session carries + # neither, which is what keeps an empty one from being stored. Whether + # an id alone is worth anything is not decided here: that depends on + # the negotiated version, which _tls_session_lifetime reads. + if not (session.has_ticket or session.id): + return None + return session + # PYTHON-1331 # # Allow implementations specific to an event loop to add additional behaviours @@ -1154,12 +1452,22 @@ def _connect_socket(self): # run that here. if self._check_hostname: self._validate_hostname() + # The handshake stood, so there is nothing left to retract -- + # but what was offered is kept, because the store still reads + # it to tell a session the peer reissued from the one this + # connection offered and got back unchanged. sockerr = None break except socket.error as err: if self._socket: self._socket.close() self._socket = None + # Only for a TLS failure: a connection refused or reset says + # nothing about the session, and dropping it would cost a later + # connection a full handshake for no reason. Whether anything + # was offered to retract is _discard_tls_session's own business. + if isinstance(err, ssl.SSLError): + self._discard_tls_session() sockerr = err if sockerr: @@ -1198,6 +1506,15 @@ def defunct(self, exc): log.debug("Defuncting connection (%s) to %s: %s", id(self), self.endpoint, exc) + if isinstance(exc, ssl.SSLError): + # ssl_options may carry do_handshake_on_connect=False, which leaves + # the handshake to the first read or write and so to the reactor, + # where a failure arrives here rather than at _connect_socket. + # Nothing is retracted for a connection that got as far as the CQL + # handshake, which cleared what it had offered; only a TLS error + # counts, for the same reason it does there. + self._discard_tls_session() + self.last_error = exc self.close() self.error_all_cp_sessions(exc) @@ -1700,6 +2017,7 @@ def _handle_startup_response(self, startup_response, did_authenticate=False): if ProtocolVersion.has_checksumming_support(self.protocol_version): self._enable_checksumming() + self._store_tls_session() self.connected_event.set() elif isinstance(startup_response, AuthenticateMessage): log.debug("Got AuthenticateMessage on new connection (%s) from %s: %s", @@ -1756,6 +2074,7 @@ def _handle_auth_response(self, auth_response): self.authenticator.on_authentication_success(auth_response.token) if self._compressor: self.compressor = self._compressor + self._store_tls_session() 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..20fe79b851 100644 --- a/cassandra/io/asyncioreactor.py +++ b/cassandra/io/asyncioreactor.py @@ -118,8 +118,16 @@ 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``). + + TLS session resumption (:attr:`.Cluster.ssl_session_cache`) is not + available on this reactor: the handshake happens inside + ``loop.create_connection(..., ssl=...)``, which offers no point at which + a cached session could be restored. """ + # See the note on TLS session resumption above. + supports_tls_session_resumption = False + _loop = None _pid = os.getpid() diff --git a/cassandra/pool.py b/cassandra/pool.py index 2cd376d293..96a49c5962 100644 --- a/cassandra/pool.py +++ b/cassandra/pool.py @@ -721,6 +721,13 @@ def _get_shard_aware_endpoint(self): endpoint = copy.copy(self.host.endpoint) endpoint._port = self.host.sharding_info.shard_aware_port + if endpoint is not None: + # Another listener of this same node, with the same TLS + # credentials, so it offers and refreshes the session cached for + # the node rather than one of its own. + endpoint._tls_session_cache_key_override = \ + self.host.endpoint.tls_session_cache_key + return endpoint def _open_connection_to_missing_shard(self, shard_id): diff --git a/cassandra/ssl_session_cache.py b/cassandra/ssl_session_cache.py new file mode 100644 index 0000000000..96af5fa219 --- /dev/null +++ b/cassandra/ssl_session_cache.py @@ -0,0 +1,240 @@ +# Copyright 2026 ScyllaDB, 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. +""" +Storage for TLS sessions, so that connections can resume one instead of +performing a full handshake. + +The policy around this -- what may be offered to whom, for how long, and what +a handshake hands back -- lives with the connections in +:mod:`cassandra.connection`; what is here is only the keeping of them. +""" + +import time +from collections import OrderedDict +from threading import Lock + + +class SSLSessionCache(object): + """ + A thread-safe, bounded cache of TLS sessions, keyed by TLS peer identity. + + TLS clients can skip the expensive part of a handshake by replaying a + session established earlier with the same peer (RFC 5077 session tickets + for TLS 1.2, RFC 8446 pre-shared keys for TLS 1.3). OpenSSL never does + this on its own -- the client has to hold on to the session and offer it + on the next connection -- so the driver keeps one of these caches per + :class:`~.Cluster` and reuses sessions across every connection it opens, + most importantly the burst of per-shard connections opened to a node at + once. + + A cached session is not consumed by being used: the same session can be + replayed by any number of concurrent connections, and each successful + handshake stores back whatever the peer handed over -- a fresh session + where one was issued, otherwise the same one again, which keeps the + deadline it already had rather than starting a new one. An entry whose + lifetime has run out is never handed out again, and is dropped when it + is looked up or when room is needed; entries otherwise go only by being + replaced or, once the cache is full, by having been used least + recently. A session the server declines for any other reason simply + results in a full handshake, which is what would have happened anyway. + + Instances are safe to use from multiple threads, and may be shared + between clusters -- which is what makes sessions outlive the cluster that + established them, so that a cluster replacing an earlier one resumes + instead of handshaking in full. A cache the driver created for a cluster + lives and dies with it; one supplied to :class:`~.Cluster` belongs to + whoever supplied it, and the driver removes an entry from it only to + replace it, because its lifetime ran out, or to make room. Note that an + entry keeps the ``SSLContext`` its session was established with alive -- + CPython's ``SSLSession`` holds a reference to it -- so a long-lived cache + holds the contexts of at most :attr:`max_size` peers. :meth:`clear` drops + everything, for a caller that wants them gone sooner. + """ + + def __init__(self, max_size=1024): + """ + :param max_size: maximum number of peers to keep sessions for. When + exceeded, the least recently used entry is evicted. + """ + # Anything but a positive integer is rejected outright rather than + # compared against: a float such as nan or inf would pass a `< 1` check + # and then leave the cache growing without bound, while True is an int + # that passes it and would quietly cap the cache at one entry. + if (not isinstance(max_size, int) or isinstance(max_size, bool) + or max_size < 1): + raise ValueError( + "max_size must be a positive integer, got %r" % (max_size,)) + self._max_size = max_size + self._sessions = OrderedDict() + self._lock = Lock() + + @property + def max_size(self): + """The maximum number of peers this cache keeps sessions for.""" + return self._max_size + + def get(self, key): + """ + Return the cached session for *key*, or :const:`None` if there is none + or its lifetime has run out. A session that is still live stays in the + cache; an expired one is dropped. + """ + with self._lock: + entry = self._sessions.get(key) + if entry is None: + return None + session, expires_at = entry + if expires_at is not None and time.monotonic() >= expires_at: + del self._sessions[key] + return None + self._sessions.move_to_end(key) + return session + + def set(self, key, session, lifetime=None, offered=None): + """ + Store *session* as the session to offer for *key*, replacing any + previous one. A :const:`None` session is ignored. + + A session the peer handed back unchanged keeps the deadline the entry + already had, rather than starting a new one: its lifetime runs from + when the peer issued it and not from when it was last replayed, so + re-stamping a full lifetime on every reuse would let one ticket be + offered for as long as connections keep being opened. Resuming below + TLS 1.3 is exactly that case -- an abbreviated handshake hands back the + session that was offered, same id and same ticket -- while TLS 1.3 + normally issues a fresh one, which starts its own lifetime. Comparing + the two here is what makes the rule hold: what is cached, what the + caller offered and what is replacing them are all read under the one + lock that also stores the result, so a connection storing concurrently + cannot land in between. + + :param lifetime: how much longer, in seconds, the session may be + offered. Once it has passed, the entry is dropped rather than + returned. :const:`None` means no limit, which callers should + reserve for sessions that carry no lifetime of their own. + :param offered: the session the caller offered on the handshake it is + storing the result of, if any. Storing that same session back when + the entry no longer holds it is not a new session arriving, and is + skipped: see below. + """ + if session is None: + return + expires_at = None if lifetime is None else time.monotonic() + lifetime + with self._lock: + previous = self._sessions.get(key) + if previous is not None and self._is_same_session(previous[0], session): + # The entry keeps both its deadline and the object holding it. + # ``SSLSocket.session`` builds a new wrapper on every access, so + # what arrives here for a session already cached is another + # handle on the same credential; swapping one for the other + # changes nothing except the identity that :meth:`discard` + # compares against, and a connection that offered the entry + # would then be unable to retract what it offered. + session, expires_at = previous + elif offered is not None and self._is_same_session(offered, session): + # The peer handed this caller back the very session it offered, + # but the entry no longer holds it: another connection opened + # alongside stored a session the peer reissued to it, or the + # deadline passed and a lookup dropped the entry. Either way + # this caller has nothing to add -- and storing it would put a + # deadline running from now on a session the peer issued at + # some earlier point, which is the one thing the rule above + # exists to prevent. + return + self._sessions[key] = (session, expires_at) + self._sessions.move_to_end(key) + if len(self._sessions) > self._max_size: + # Whose lifetime has run out and which was used least recently + # are independent once peers announce different lifetimes, so + # evicting purely by recency can drop a live entry and keep a + # dead one. Take the dead ones first. + self._drop_expired_unlocked() + while len(self._sessions) > self._max_size: + self._sessions.popitem(last=False) + + @staticmethod + def _is_same_session(cached, session): + """ + Whether *session* is the one already cached, so that the entry's + deadline is not a new store's to move. + + ``SSLSocket.session`` builds a new object on each access, so identity + cannot answer this on its own; a session id can, and is what tells a + ticket the peer reissued from the one it handed back -- including the + TLS 1.3 server that resumes without issuing one. + + An id of no length is a case of its own. RFC 5077 section 3.4 lets a + server issue a ticket and send an empty session id with it, and + ``SSLSession`` exposes no ticket to compare instead, so two such + sessions cannot be told apart at all. They are reported as the same + one, which is the conservative reading: the deadline then stays where + it is, where calling them different would re-stamp a full lifetime on + what may well be the ticket already held -- the one thing this + comparison exists to prevent. What that costs is resumption, never + correctness: a reissued ticket inherits its predecessor's deadline, and + one arriving where the entry has since gone is not stored at all, which + the next connection puts right by offering nothing and storing afresh. + An object carrying no id at all is not a session this can recognise, + and is taken to be new. + """ + if cached is session: + return True + cached_id = getattr(cached, 'id', None) + return cached_id is not None and cached_id == getattr(session, 'id', None) + + def _drop_expired_unlocked(self): + now = time.monotonic() + for key in [key for key, (_, expires_at) in self._sessions.items() + if expires_at is not None and now >= expires_at]: + del self._sessions[key] + + def discard(self, key, session=None): + """ + Drop the session cached for *key*, if any. + + Give *session* to drop it only while that is still the cached one. A + caller acting on a session it read earlier needs this: by the time it + decides to drop it, another connection may have stored a session the + peer issued in its place, and that one is not the caller's to remove. + + The comparison is by identity, which :meth:`set` is what makes + dependable: a store of the session already cached keeps the object + that is there, so an entry changes identity only when it changes + credential. + """ + with self._lock: + entry = self._sessions.get(key) + if entry is None: + return + if session is not None and entry[0] is not session: + return + del self._sessions[key] + + def clear(self): + """Drop all cached sessions.""" + with self._lock: + self._sessions.clear() + + def __len__(self): + with self._lock: + return len(self._sessions) + + def __repr__(self): + # The size is read without the lock, unlike __len__: a repr has to be + # safe to take from inside the cache's own methods -- a log line + # formatting %r under the lock would otherwise wait for itself -- and a + # count that another thread has moved on from is no worse than one it + # moves on from a moment later. + return "<%s max_size=%d size=%d>" % ( + self.__class__.__name__, self._max_size, len(self._sessions)) diff --git a/docs/api/cassandra/cluster.rst b/docs/api/cassandra/cluster.rst index cf9cc59fc4..f0149244a6 100644 --- a/docs/api/cassandra/cluster.rst +++ b/docs/api/cassandra/cluster.rst @@ -43,6 +43,8 @@ Clusters and Sessions .. autoattribute:: ssl_options + .. autoattribute:: ssl_session_cache + .. autoattribute:: sockopts .. autoattribute:: max_schema_agreement_wait diff --git a/docs/api/cassandra/ssl-session-cache.rst b/docs/api/cassandra/ssl-session-cache.rst new file mode 100644 index 0000000000..32aafec78e --- /dev/null +++ b/docs/api/cassandra/ssl-session-cache.rst @@ -0,0 +1,7 @@ +cassandra.ssl_session_cache +=========================== + +.. module:: cassandra.ssl_session_cache + +.. autoclass:: SSLSessionCache + :members: diff --git a/docs/api/index.rst b/docs/api/index.rst index f63534e532..6b921ffbf2 100644 --- a/docs/api/index.rst +++ b/docs/api/index.rst @@ -19,6 +19,7 @@ Core Driver cassandra/decoder cassandra/concurrent cassandra/connection + cassandra/ssl-session-cache cassandra/util cassandra/timestamps cassandra/io/asyncioreactor diff --git a/docs/security.rst b/docs/security.rst index 3cbbcbbb40..caa10bf288 100644 --- a/docs/security.rst +++ b/docs/security.rst @@ -65,6 +65,71 @@ keystore files with these instructions: * `Scylla TLS/SSL Guide `_ +TLS Session Resumption +^^^^^^^^^^^^^^^^^^^^^^ +A shard-aware driver opens one connection per shard to every node, and a full +TLS handshake on each is the expensive part of that. Whenever +:attr:`.Cluster.ssl_context` is set, the driver caches the TLS session each +connection establishes and offers it on the next one, so the connections that +follow resume instead of handshaking in full. This is on by default and needs no +configuration. + +There is one thing to weigh before leaving it on. A resumed handshake carries +no Certificate message, so nothing about the server's certificate is checked +again while a cached session is being offered -- not its expiry, and not a +revocation list the ``SSLContext`` carries. A certificate that expires or is +revoked goes on being accepted by connections that resume, until the cached +entry goes: that is the lifetime the server announced with the ticket, which +the driver caps at seven days. Connections that handshake in full verify as +they always have. The hostname is not re-checked either, which is why the +driver keys each cached session by the name verified when it was established, +so a session is never offered to a connection expecting a different one. Where +that window is not acceptable, turn resumption off with +``ssl_session_cache=None``. + +It does need the server to issue something to resume from. Scylla sends session +tickets only when ``enable_session_tickets`` is set in its +``client_encryption_options``, which defaults to true from Scylla 2026.3 and to +false in the releases before it: + +.. code-block:: yaml + + client_encryption_options: + enabled: true + certificate: /path/to/scylla.crt + keyfile: /path/to/scylla.key + enable_session_tickets: true + +Without that, nothing resumes and every connection performs a full handshake, as +it did before. + +Resumption also needs a reactor that can offer a session before the handshake +begins: the ``libev`` reactor, and ``asyncore`` on the Python versions that still +ship it, which is up to 3.11. The ``asyncio`` reactor performs its handshake +inside ``loop.create_connection()``, leaving no point at which to restore a +session, so resumption is unavailable there -- worth knowing, because that is the +default reactor on Python 3.12 and newer when the libev extension is not +installed. It is unavailable too with the deprecated +:attr:`.Cluster.ssl_options`-only configuration below, since each of those +connections builds its own ``SSLContext`` and a session cannot be replayed onto a +different one. + +Where resumption is unavailable, no cache is created and +:attr:`.Cluster.ssl_session_cache` reads as ``None``; asking for one anyway is +reported when :meth:`.Cluster.connect` is called. To turn resumption off, or to +size the cache or share it between clusters, see +:attr:`.Cluster.ssl_session_cache`: + +.. code-block:: python + + from cassandra.cluster import Cluster + from cassandra.ssl_session_cache import SSLSessionCache + + cluster = Cluster(ssl_context=ssl_context, ssl_session_cache=None) + + cluster = Cluster(ssl_context=ssl_context, + ssl_session_cache=SSLSessionCache(max_size=64)) + SSL Configuration Examples ^^^^^^^^^^^^^^^^^^^^^^^^^^ Here, we'll describe the server and driver configuration necessary to set up SSL to meet various goals, such as the client verifying the server and the server verifying the client. We'll also include Python code demonstrating how to use servers and drivers configured in these ways. diff --git a/tests/integration/standard/test_tls_resumption.py b/tests/integration/standard/test_tls_resumption.py new file mode 100644 index 0000000000..f0f2e3af3a --- /dev/null +++ b/tests/integration/standard/test_tls_resumption.py @@ -0,0 +1,231 @@ +# Copyright 2026 ScyllaDB, 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. +""" +TLS session resumption against a real, TLS-enabled Scylla cluster. + +The mechanics of resumption are covered by +``tests/unit/test_tls_resumption.py`` against a local TLS server. What needs +a real cluster is whether the *server* accepts one session offered by several +connections at once, which is the case DRIVER-165 is about: a pool opens one +connection per shard and they all offer the same cached session. +""" + +import logging +import shutil +import ssl +import tempfile +import unittest + +from cassandra.cluster import Cluster +from cassandra.ssl_session_cache import SSLSessionCache +from tests.integration import (use_singledc, get_cluster, remove_cluster, + start_cluster_wait_for_up, KEEP_TEST_CLUSTER, + SCYLLA_VERSION, TestCluster) +from tests.tls_certificates import HAVE_CRYPTOGRAPHY, write_self_signed_cert +from tests.util import wait_until + +log = logging.getLogger(__name__) + +_cert_dir = None +_cert_path = None +_key_path = None + + + +def setup_module(): + """ + Restart the shared cluster with client encryption enabled, the way other + modules in this directory reconfigure it (see test_custom_cluster). + teardown_module drops it so the next module gets a clean one. + """ + if SCYLLA_VERSION is None: + raise unittest.SkipTest( + 'client_encryption_options are configured the Scylla way here; ' + 'set SCYLLA_VERSION to run this') + # Asked of the class TestCluster will actually use, not of the selector: + # tests.integration sets Cluster.connection_class from EVENT_LOOP_MANAGER + # only when it resolved one, and leaves the driver's own default in place + # otherwise -- which, with no selector and no libev, is the asyncio reactor. + reactor = Cluster.connection_class + if not getattr(reactor, 'supports_tls_session_resumption', False): + raise unittest.SkipTest( + '%s cannot restore a cached TLS session before the handshake, so ' + 'there is no resumption here to test' + % getattr(reactor, '__name__', reactor)) + if not HAVE_CRYPTOGRAPHY: + raise unittest.SkipTest( + 'cryptography is required to generate a server certificate') + + global _cert_dir, _cert_path, _key_path + _cert_dir = tempfile.mkdtemp(prefix='tls_resumption_') + try: + use_singledc(start=False) + ccm_cluster = get_cluster() + ccm_cluster.stop() + # The certificate has to name every node, so it can only be issued once + # the cluster exists. + _cert_path, _key_path = write_self_signed_cert( + _cert_dir, [node.address() for node in ccm_cluster.nodelist()]) + ccm_cluster.set_configuration_options({ + # Per-shard connections go to this port, which is where resumption + # has to pay off; Scylla leaves it unset by default. + 'native_shard_aware_transport_port_ssl': 19142, + 'client_encryption_options': { + 'enabled': True, + 'certificate': _cert_path, + 'keyfile': _key_path, + # Set rather than assumed: this defaults to false before + # Scylla 2026.3, and without it the server issues no + # NewSessionTicket and nothing can be resumed. + 'enable_session_tickets': True, + } + }) + start_cluster_wait_for_up(ccm_cluster) + except Exception: + # pytest skips teardown_module when setup_module raises, so undo both + # halves here: the cluster would otherwise be left stopped and still + # configured for TLS for every module that runs after this one, and the + # key and certificate would be left behind on disk. + try: + remove_cluster() + finally: + _discard_certificate() + raise + + +def _discard_certificate(): + """ + Remove the key and certificate, unless the cluster that was configured to + use them is being kept. + + KEEP_TEST_CLUSTER makes remove_cluster() a no-op, and what it leaves behind + still names these files in its client_encryption_options: deleting them + would leave a cluster that cannot be started again, which is the one thing + that mode exists to avoid. + """ + global _cert_dir + if _cert_dir is None: + return + if KEEP_TEST_CLUSTER: + log.info('Keeping the TLS key and certificate in %s: the cluster kept ' + 'by KEEP_TEST_CLUSTER is still configured to use them.', + _cert_dir) + else: + shutil.rmtree(_cert_dir, ignore_errors=True) + _cert_dir = None + + +def teardown_module(): + try: + remove_cluster() + finally: + _discard_certificate() + + +def make_ssl_context(): + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + context.load_verify_locations(_cert_path) + context.verify_mode = ssl.CERT_REQUIRED + context.check_hostname = True + return context + + +def resumption_of_every_connection(cluster): + """ + What OpenSSL reports for each of the cluster's live connections: a list of + ``session_reused`` flags, one per connection that still has a socket. + + A connection that went away between being collected here and being read is + left out rather than reported: SSLSocket.session_reused answers None once + the socket is closed, which reads as "did not resume" and would fail an + assertion about a connection that is no longer there. The count is + asserted separately, so leaving one out cannot quietly shrink the sample. + """ + return [bool(connection._socket.session_reused) + for holder in cluster.get_connection_holders() + for connection in holder.get_connections() + if connection._socket is not None + and not (connection.is_closed or connection.is_defunct)] + + +def expected_connection_count(cluster): + """ + One control connection, plus one pool connection per shard of every host + the driver considers up. + """ + return 1 + sum(host.sharding_info.shards_count if host.sharding_info else 1 + for host in cluster.metadata.all_hosts() if host.is_up) + + +def collect_resumption(cluster): + """ + Wait for the pools to fill, then report whether each connection resumed a + TLS session. The wait and the count assertion matter: per-shard + connections are opened in the background, so an assertion made too early + would run against a fraction of them -- or against the control connection + alone -- and pass without testing anything. + """ + expected = expected_connection_count(cluster) + wait_until(lambda: len(resumption_of_every_connection(cluster)) >= expected, 0.5, 40) + + resumed = resumption_of_every_connection(cluster) + log.info('%d of %d connections resumed a TLS session (expected at least %d)', + sum(resumed), len(resumed), expected) + assert len(resumed) >= expected, \ + 'inspected %d connections, expected at least %d' % (len(resumed), expected) + return resumed + + +class TLSSessionResumptionTests(unittest.TestCase): + + def setUp(self): + # Cluster.sessions is a WeakSet and HostConnection keeps only a + # weakref.proxy to its session, so a Session nobody holds is collected + # and takes the pools -- everything worth inspecting -- with it. + self._sessions = [] + + def connect(self, **kwargs): + cluster = TestCluster(**kwargs) + self.addCleanup(cluster.shutdown) + self._sessions.append(cluster.connect(wait_for_all_pools=True)) + return cluster + + def test_resumption_is_on_by_default_with_an_ssl_context(self): + cluster = self.connect(ssl_context=make_ssl_context()) + + assert isinstance(cluster.ssl_session_cache, SSLSessionCache) + assert len(cluster.ssl_session_cache) > 0 + + def test_every_connection_resumes_from_a_warmed_cache(self): + # Warm a cache, then hand it to a second cluster using the same + # SSLContext. Every connection that cluster opens -- including the + # whole batch of per-shard connections opened at once, which reach the + # node on its shard-aware port -- then has a session to offer, so the + # server has to accept the same one from all of them concurrently. + context = make_ssl_context() + cache = SSLSessionCache() + self.connect(ssl_context=context, ssl_session_cache=cache) + + cluster = self.connect(ssl_context=context, ssl_session_cache=cache) + + assert all(collect_resumption(cluster)) + + def test_nothing_resumes_when_the_cache_is_disabled(self): + context = make_ssl_context() + self.connect(ssl_context=context, ssl_session_cache=SSLSessionCache()) + + cluster = self.connect(ssl_context=context, ssl_session_cache=None) + + assert cluster.ssl_session_cache is None + assert not any(collect_resumption(cluster)) diff --git a/tests/tls_certificates.py b/tests/tls_certificates.py new file mode 100644 index 0000000000..b6362f8dff --- /dev/null +++ b/tests/tls_certificates.py @@ -0,0 +1,78 @@ +# Copyright 2026 ScyllaDB, 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. +""" +Certificates for tests that need a TLS server, generated rather than checked in +so that nothing expires in the repository. + +``cryptography`` is optional, so this imports it defensively and reports what it +found in :data:`HAVE_CRYPTOGRAPHY`; a suite that needs a certificate skips on +that rather than failing to import. +""" + +import datetime +import ipaddress +import os + +try: + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.x509.oid import NameOID + HAVE_CRYPTOGRAPHY = True +except ImportError: # pragma: no cover - depends on the environment + HAVE_CRYPTOGRAPHY = False + + +def write_self_signed_cert(directory, addresses=('127.0.0.1',)): + """ + Write a self-signed certificate naming every address in *addresses*, and + its key, into *directory*. Returns ``(cert_path, key_path)``. + + Every address a client will connect to has to be named: the client verifies + hostnames, so a certificate covering only the first of them would leave it + unable to reach the rest. Callers against a cluster pass every node's + address for that reason; one against a loopback server takes the default. + """ + if not HAVE_CRYPTOGRAPHY: + raise RuntimeError('cryptography is required to generate a certificate') + + addresses = list(addresses) + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, addresses[0])]) + now = datetime.datetime.now(datetime.timezone.utc) + certificate = ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(minutes=5)) + .not_valid_after(now + datetime.timedelta(days=1)) + .add_extension( + x509.SubjectAlternativeName( + [x509.IPAddress(ipaddress.ip_address(address)) + for address in addresses]), + critical=False) + .sign(key, hashes.SHA256()) + ) + + cert_path = os.path.join(directory, 'server.crt') + key_path = os.path.join(directory, 'server.key') + with open(cert_path, 'wb') as f: + f.write(certificate.public_bytes(serialization.Encoding.PEM)) + with open(key_path, 'wb') as f: + f.write(key.private_bytes(serialization.Encoding.PEM, + serialization.PrivateFormat.TraditionalOpenSSL, + serialization.NoEncryption())) + return cert_path, key_path diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 74ed346c68..755a3f4888 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -14,8 +14,13 @@ import unittest from concurrent.futures import Future +import gc import logging import socket +import ssl +import threading +import time +import weakref from types import SimpleNamespace from unittest.mock import patch, Mock @@ -25,7 +30,9 @@ 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 (Connection, ConnectionBusy, ConnectionException, + DefaultEndPoint) +from cassandra.ssl_session_cache import SSLSessionCache from cassandra.driver_config import DriverConfigReporter from cassandra.pool import Host from cassandra.policies import HostDistance, RetryPolicy, RoundRobinPolicy, DowngradingConsistencyRetryPolicy, SimpleConvictionPolicy @@ -1167,3 +1174,353 @@ def test_no_warning_adding_lbp_ep_to_cluster_with_contact_points(self): ) patched_logger.warning.assert_not_called() + + +class _ResumableConnection(Connection): + supports_tls_session_resumption = True + + +class _NonResumableConnection(Connection): + supports_tls_session_resumption = False + + +class ClusterSSLSessionCacheTest(unittest.TestCase): + + def make_cluster(self, connection_class=_ResumableConnection, **kwargs): + cluster = Cluster(connection_class=connection_class, **kwargs) + # Every Cluster starts a _Scheduler thread in __init__, so one that is + # constructed and dropped leaks it for the rest of the session. + self.addCleanup(cluster.shutdown) + return cluster + + def test_cache_is_created_for_an_ssl_context(self): + cluster = self.make_cluster(ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)) + + assert isinstance(cluster.ssl_session_cache, SSLSessionCache) + # Made once and kept: a second one would take the sessions of whatever + # was opened against the first. + assert cluster.ssl_session_cache is cluster.ssl_session_cache + + def test_no_cache_without_tls(self): + assert self.make_cluster().ssl_session_cache is None + + def test_no_cache_for_ssl_options_only(self): + # Each connection builds its own SSLContext from ssl_options, and a + # session cannot be replayed onto a different context. + with patch('cassandra.cluster.warn'): + cluster = self.make_cluster(ssl_options={'ca_certs': '/dev/null'}) + + assert cluster.ssl_session_cache is None + + def test_no_cache_for_a_reactor_that_cannot_resume(self): + cluster = self.make_cluster(connection_class=_NonResumableConnection, + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)) + + assert cluster.ssl_session_cache is None + + def test_no_cache_for_a_connection_class_that_reports_nothing(self): + # connection_class is not required to derive from Connection (see + # test_set_connection_class), so a class without the capability + # attribute must be treated as unable to resume, not blow up. + cluster = self.make_cluster(connection_class='not a connection class', + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)) + + assert cluster.ssl_session_cache is None + + def test_a_supplied_cache_is_used(self): + cache = SSLSessionCache(max_size=7) + cluster = self.make_cluster(ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), + ssl_session_cache=cache) + + assert cluster.ssl_session_cache is cache + + def test_warns_when_a_supplied_cache_cannot_be_used(self): + # Asking for resumption and silently getting none is worse than not + # having it: the attribute reads as None and nothing is ever cached. + with patch('cassandra.cluster.warn'): + cluster = self.make_cluster(ssl_options={'ca_certs': '/dev/null'}, + ssl_session_cache=SSLSessionCache()) + + with patch('cassandra.cluster.log') as logger: + cluster._report_tls_session_resumption() + + logger.warning.assert_called_once() + assert 'ssl_session_cache' in logger.warning.call_args[0][0] + assert 'ssl_context' in logger.warning.call_args[0][1] + + def test_warns_when_the_reactor_cannot_resume(self): + cluster = self.make_cluster( + connection_class=_NonResumableConnection, + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), + ssl_session_cache=SSLSessionCache()) + + with patch('cassandra.cluster.log') as logger: + cluster._report_tls_session_resumption() + + logger.warning.assert_called_once() + assert '_NonResumableConnection' in logger.warning.call_args[0][1] + + def test_warns_about_a_cache_assigned_to_a_cluster_that_cannot_use_it(self): + # Assigning is as much asking for one as passing it in, and nothing + # about the answer depends on which was done when. + cluster = self.make_cluster(connection_class=_NonResumableConnection, + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)) + cluster.ssl_session_cache = SSLSessionCache() + + with patch('cassandra.cluster.log') as logger: + cluster._report_tls_session_resumption() + + logger.warning.assert_called_once() + assert '_NonResumableConnection' in logger.warning.call_args[0][1] + assert cluster.ssl_session_cache is None + + def test_does_not_warn_about_a_cache_the_caller_never_asked_for(self): + # The cache here was made for this cluster, so turning TLS off is not + # something to complain about: nobody asked for resumption. + cluster = self.make_cluster(ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)) + assert cluster.ssl_session_cache is not None + + cluster.ssl_context = None + with patch('cassandra.cluster.log') as logger: + cluster._report_tls_session_resumption() + + logger.warning.assert_not_called() + assert cluster.ssl_session_cache is None + + def test_an_unusable_cache_is_not_kept_or_passed_on(self): + # Warning and then handing the cache to every connection anyway is the + # worst of both: a connection class that does not take the keyword + # cannot even be constructed. + with patch('cassandra.cluster.log'): + cluster = self.make_cluster( + connection_class=_NonResumableConnection, + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), + ssl_session_cache=SSLSessionCache()) + + assert cluster.ssl_session_cache is None + kwargs = cluster._make_connection_kwargs(DefaultEndPoint('127.0.0.1'), {}) + assert 'ssl_session_cache' not in kwargs + + def test_does_not_warn_where_resumption_works_or_was_declined(self): + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + clusters = [ + self.make_cluster(ssl_context=context, + ssl_session_cache=SSLSessionCache()), + self.make_cluster(ssl_context=context, ssl_session_cache=None), + self.make_cluster(), + ] + + with patch('cassandra.cluster.log') as logger: + for cluster in clusters: + cluster._report_tls_session_resumption() + + logger.warning.assert_not_called() + + def test_notes_at_debug_where_nobody_asked_for_a_cache(self): + # Resumption is on by default wherever it works, so a cluster that + # configured TLS and will not get it has something worth finding. + # Nobody asked for it, though, so it is a note rather than a warning. + cluster = self.make_cluster(connection_class=_NonResumableConnection, + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)) + + with patch('cassandra.cluster.log') as logger: + cluster._report_tls_session_resumption() + + logger.warning.assert_not_called() + said = [call[0][0] for call in logger.debug.call_args_list] + assert any('_NonResumableConnection' in call[0][1] + for call in logger.debug.call_args_list), said + + def test_says_nothing_to_a_cluster_with_no_tls(self): + # Nothing there to resume, so nothing to report at any level. + cluster = self.make_cluster(connection_class=_NonResumableConnection) + + with patch('cassandra.cluster.log') as logger: + cluster._report_tls_session_resumption() + + logger.warning.assert_not_called() + logger.debug.assert_not_called() + + def test_connect_says_why_a_cache_it_cannot_use_will_stay_empty(self): + # connect() is the only caller, so without this nothing holds the + # wiring: the method could be left in place and never reached. + cluster = self.make_cluster( + connection_class=_NonResumableConnection, + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), + ssl_session_cache=SSLSessionCache()) + cluster.idle_heartbeat_interval = 0 + + with patch('cassandra.cluster.log') as logger: + with patch.object(cluster.control_connection, 'connect'), \ + patch.object(cluster, '_populate_hosts'), \ + patch.object(cluster.profile_manager, 'check_supported'), \ + patch.object(cluster, '_new_session'), \ + patch.object(cluster, '_set_default_dbaas_consistency'): + cluster.connect() + + said = [call[0][0] for call in logger.warning.call_args_list] + assert any('ssl_session_cache' in message for message in said), said + + def test_resumption_can_be_turned_off(self): + cluster = self.make_cluster(ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), + ssl_session_cache=None) + + assert cluster.ssl_session_cache is None + + def test_assigning_none_turns_resumption_off(self): + # The documented opt-out, taken after construction rather than at it: + # nothing may put a cache back afterwards, or the attribute would not + # mean what it says. + cluster = self.make_cluster(ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)) + assert cluster.ssl_session_cache is not None + + cluster.ssl_session_cache = None + + assert cluster.ssl_session_cache is None + kwargs = cluster._make_connection_kwargs(DefaultEndPoint('127.0.0.1'), {}) + assert 'ssl_session_cache' not in kwargs + + def test_assigning_none_turns_off_a_cache_given_to_the_constructor(self): + # The same, over the top of a cache the caller passed in: the last + # thing they said about it is the one that counts. + cluster = self.make_cluster(ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), + ssl_session_cache=SSLSessionCache()) + assert cluster.ssl_session_cache is not None + + cluster.ssl_session_cache = None + + assert cluster.ssl_session_cache is None + + def test_shutdown_leaves_a_supplied_cache_alone(self): + # The cache belongs to whoever passed it in, and the point of passing + # one in is that its sessions outlive a cluster: a cluster replacing + # this one resumes rather than handshaking in full. A cache created + # for a cluster needs no shutdown hook either -- it is reachable only + # through the cluster, so it goes when the cluster does. + cache = SSLSessionCache() + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + cluster = self.make_cluster(ssl_context=context, ssl_session_cache=cache) + session = object() + cache.set((context, ('10.0.0.1', 9042), None), session) + + cluster.shutdown() + + assert cache.get((context, ('10.0.0.1', 9042), None)) is session + + def test_threads_reading_it_at_once_all_get_the_same_cache(self): + # A cluster given TLS after connect() first reaches the creating branch + # from whichever pools are opening connections, several at a time. Two + # of them each making one would leave the connections of the loser + # filling a cache nothing else can reach. + cluster = self.make_cluster(ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)) + started = threading.Barrier(8) + seen = [] + + # Widen the window the check-then-act leaves: without the lock the + # sleep guarantees every thread finds nothing there. + real = SSLSessionCache + + def slow_cache(*args, **kwargs): + time.sleep(0.05) + return real(*args, **kwargs) + + def read(): + started.wait(timeout=10) + seen.append(cluster.ssl_session_cache) + + with patch('cassandra.cluster.SSLSessionCache', slow_cache): + threads = [threading.Thread(target=read) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=10) + + assert len(seen) == 8 + assert all(cache is seen[0] for cache in seen), seen + assert cluster.ssl_session_cache is seen[0] + + def test_a_cache_created_here_goes_when_the_cluster_does(self): + # Built without make_cluster, whose addCleanup would hold the cluster. + cluster = Cluster(connection_class=_ResumableConnection, + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)) + cache = weakref.ref(cluster.ssl_session_cache) + cluster.shutdown() + + del cluster + gc.collect() + + assert cache() is None + + def test_the_answer_follows_a_connection_class_that_can_resume(self): + # Both inputs are public attributes, so a decision kept from the + # constructor would leave resumption off on a reactor that supports it. + cluster = self.make_cluster(connection_class=_NonResumableConnection, + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)) + assert cluster.ssl_session_cache is None + + cluster.connection_class = _ResumableConnection + + assert isinstance(cluster.ssl_session_cache, SSLSessionCache) + + def test_the_answer_follows_a_connection_class_that_cannot(self): + # Otherwise the keyword goes to a class that may not take it. + cluster = self.make_cluster(ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)) + assert cluster.ssl_session_cache is not None + + cluster.connection_class = _NonResumableConnection + + assert cluster.ssl_session_cache is None + kwargs = cluster._make_connection_kwargs(DefaultEndPoint('127.0.0.1'), {}) + assert 'ssl_session_cache' not in kwargs + + def test_the_answer_follows_a_context_set_after_construction(self): + cluster = self.make_cluster() + assert cluster.ssl_session_cache is None + + cluster.ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + + assert isinstance(cluster.ssl_session_cache, SSLSessionCache) + + def test_a_declined_cache_stays_declined(self): + # Every read settles the question again, so the answer has to keep + # being no: nothing may put a cache here behind the caller's back on + # some later look. + cluster = self.make_cluster(ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), + ssl_session_cache=None) + + assert [cluster.ssl_session_cache for _ in range(3)] == [None] * 3 + + def test_a_supplied_cache_is_returned_every_time(self): + # Connections are given whatever this answers, one read per connection, + # so an answer that varied would have them filling different caches and + # resuming from none of them. + cache = SSLSessionCache() + cluster = self.make_cluster(ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), + ssl_session_cache=cache) + + assert all(cluster.ssl_session_cache is cache for _ in range(3)) + + def test_cache_is_passed_to_connections(self): + cluster = self.make_cluster(ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)) + + kwargs = cluster._make_connection_kwargs(DefaultEndPoint('127.0.0.1'), {}) + + assert kwargs['ssl_session_cache'] is cluster.ssl_session_cache + + def test_no_cache_keyword_when_resumption_is_inactive(self): + # A connection class that does not accept the keyword should not be + # handed one for a cluster that will never cache a session. + cluster = self.make_cluster() + + kwargs = cluster._make_connection_kwargs(DefaultEndPoint('127.0.0.1'), {}) + + assert 'ssl_session_cache' not in kwargs + + def test_an_explicitly_passed_cache_still_reaches_the_connection(self): + cache = SSLSessionCache() + cluster = self.make_cluster() + + kwargs = cluster._make_connection_kwargs(DefaultEndPoint('127.0.0.1'), + {'ssl_session_cache': cache}) + + assert kwargs['ssl_session_cache'] is cache diff --git a/tests/unit/test_connection.py b/tests/unit/test_connection.py index fcea10dfaf..b5074f7546 100644 --- a/tests/unit/test_connection.py +++ b/tests/unit/test_connection.py @@ -12,6 +12,8 @@ # See the License for the specific language governing permissions and # limitations under the License. import itertools +import copy +import ssl import unittest import uuid from io import BytesIO @@ -27,9 +29,11 @@ ConnectionException, ConnectionShutdown, DefaultEndPoint, ShardAwarePortGenerator, DRIVER_NAME, DRIVER_VERSION) from cassandra.driver_config import DRIVER_CONFIG_OPTION, SESSION_ID_OPTION +from cassandra.ssl_session_cache import SSLSessionCache from cassandra.marshal import uint8_pack, uint32_pack, int32_pack from cassandra.protocol import (write_stringmultimap, write_int, write_string, - read_stringmap, SupportedMessage, ProtocolHandler, + read_stringmap, AuthSuccessMessage, ReadyMessage, + SupportedMessage, ProtocolHandler, ResultMessage, RESULT_KIND_SET_KEYSPACE) from tests.unit.utils import StubReporter, ThrowingReporter @@ -906,6 +910,566 @@ def test_timer_collision(self): tm.service_timeouts() +class TlsSessionResumptionTest(unittest.TestCase): + """ + Connection-level wiring of :class:`~.SSLSessionCache`. The end-to-end + behaviour against a real TLS server lives in + ``tests/unit/test_tls_resumption.py``. + """ + + def make_connection(self, endpoint=None, **kwargs): + c = Connection(endpoint or DefaultEndPoint('1.2.3.4'), **kwargs) + c._socket = Mock() + return c + + def make_ssl_connection(self, cache=None, **kwargs): + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + return context, self.make_connection( + ssl_context=context, + ssl_session_cache=SSLSessionCache() if cache is None else cache, + **kwargs) + + def test_cache_is_used_with_a_supplied_ssl_context(self): + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + + assert connection._ssl_session_cache is cache + + def test_cache_is_ignored_without_tls(self): + connection = self.make_connection(ssl_session_cache=SSLSessionCache()) + + assert connection._ssl_session_cache is None + + def test_cache_is_ignored_for_a_context_derived_from_ssl_options(self): + # Each such connection builds its own SSLContext, and a session cannot + # be replayed onto a different context, so there is nothing to cache. + connection = self.make_connection( + ssl_options={'ca_certs': None, 'check_hostname': False}, + ssl_session_cache=SSLSessionCache()) + + assert connection.ssl_context is not None + assert connection._ssl_session_cache is None + + def test_cache_is_ignored_when_the_reactor_cannot_resume(self): + class NoResumptionConnection(Connection): + supports_tls_session_resumption = False + + connection = NoResumptionConnection( + DefaultEndPoint('1.2.3.4'), + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), + ssl_session_cache=SSLSessionCache()) + + assert connection._ssl_session_cache is None + + def test_cache_key_separates_endpoints_and_contexts(self): + context, connection = self.make_ssl_connection() + other_endpoint_connection = self.make_connection( + ssl_context=context, ssl_session_cache=connection._ssl_session_cache) + other_endpoint_connection.endpoint = DefaultEndPoint('5.6.7.8') + _, other_context_connection = self.make_ssl_connection() + + assert connection._tls_session_cache_key() == \ + (context, ('1.2.3.4', 9042), '1.2.3.4') + assert connection._tls_session_cache_key() != \ + other_endpoint_connection._tls_session_cache_key() + assert connection._tls_session_cache_key() != \ + other_context_connection._tls_session_cache_key() + + def test_cache_key_separates_verified_hostnames(self): + # A resumed handshake sends no Certificate, so the name the peer was + # verified against is never re-checked. Two connections to one address + # that verify different names must not share a session. + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + one = self.make_connection(ssl_context=context, + ssl_options={'server_hostname': 'one.example'}, + ssl_session_cache=SSLSessionCache()) + other = self.make_connection(ssl_context=context, + ssl_options={'server_hostname': 'other.example'}, + ssl_session_cache=one._ssl_session_cache) + + assert one._tls_session_cache_key() != other._tls_session_cache_key() + + def test_cache_key_uses_the_name_wrap_socket_is_given(self): + # The key has to be derived from the same value _wrap_socket_from_context + # passes to wrap_socket, or the two can drift apart. + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + context.check_hostname = False + connection = self.make_connection(ssl_context=context, + ssl_options={'server_hostname': 'sni.example'}, + ssl_session_cache=SSLSessionCache()) + connection.ssl_context = Mock(check_hostname=False) + + connection._wrap_socket_from_context() + + _, kwargs = connection.ssl_context.wrap_socket.call_args + assert kwargs['server_hostname'] == 'sni.example' + assert connection._tls_session_cache_key()[2] == 'sni.example' + + def test_cache_key_falls_back_to_the_endpoint_address(self): + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + checking = self.make_connection(ssl_context=context, + ssl_session_cache=SSLSessionCache()) + context_without_checks = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + context_without_checks.check_hostname = False + not_checking = self.make_connection(ssl_context=context_without_checks, + ssl_session_cache=SSLSessionCache()) + + assert context.check_hostname is True + assert checking._tls_session_cache_key()[2] == '1.2.3.4' + # Nothing is verified, so there is no name to pin the session to. + assert not_checking._tls_session_cache_key()[2] is None + + def test_cache_key_follows_an_endpoint_that_names_another_node(self): + # A shard-aware connection reaches the same node on a different port, + # and its endpoint carries that node's key + # (HostConnection._get_shard_aware_endpoint), so both share a session. + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + node = Connection(DefaultEndPoint('1.2.3.4', 9042), ssl_context=context, + ssl_session_cache=SSLSessionCache()) + alias = DefaultEndPoint('1.2.3.4', 19142) + alias._tls_session_cache_key_override = node.endpoint.tls_session_cache_key + shard_aware = Connection(alias, ssl_context=context, + ssl_session_cache=node._ssl_session_cache) + + assert shard_aware._tls_session_cache_key() == node._tls_session_cache_key() + + def test_cache_key_without_an_override_follows_the_endpoint(self): + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + node = Connection(DefaultEndPoint('1.2.3.4', 9042), ssl_context=context, + ssl_session_cache=SSLSessionCache()) + other_port = Connection(DefaultEndPoint('1.2.3.4', 19142), ssl_context=context, + ssl_session_cache=node._ssl_session_cache) + + assert other_port._tls_session_cache_key() != node._tls_session_cache_key() + + def test_restore_offers_the_cached_session(self): + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + session = object() + cache.set(connection._tls_session_cache_key(), session) + sock = Mock() + + connection._restore_tls_session(sock) + + assert sock.session is session + # The session stays available for the next connection. + assert cache.get(connection._tls_session_cache_key()) is session + + def test_restore_is_a_no_op_without_a_cached_session(self): + _, connection = self.make_ssl_connection() + sock = Mock(spec=[]) + + connection._restore_tls_session(sock) + + assert not hasattr(sock, 'session') + + def test_restore_tolerates_a_rejected_session(self): + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + cache.set(connection._tls_session_cache_key(), object()) + sock = Mock() + type(sock).session = property( + lambda self: None, + Mock(side_effect=ValueError("Session refers to a different SSLContext"))) + + # A rejected session must cost a full handshake, not the connection. + connection._restore_tls_session(sock) + + def test_discard_without_having_offered_anything_keeps_the_cache(self): + # Reached when there was nothing cached to offer, or when setting the + # session on the socket was refused. Passing no session to discard() + # would tell it to drop whatever is there, which may be one a sibling + # connection stored while this one was failing. + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + cache.set(connection._tls_session_cache_key(), 'stored-by-a-sibling') + + connection._discard_tls_session() + + assert cache.get(connection._tls_session_cache_key()) == 'stored-by-a-sibling' + + def test_discard_after_a_refused_session_keeps_the_cache(self): + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + cache.set(connection._tls_session_cache_key(), 'the-only-session') + connection._set_tls_session = Mock(side_effect=ValueError('refused')) + + connection._restore_tls_session(Mock()) + connection._discard_tls_session() + + assert cache.get(connection._tls_session_cache_key()) == 'the-only-session' + + def test_an_alternate_listener_does_not_retract_the_nodes_session(self): + # The shard-aware endpoint borrows the node's key so it can resume what + # the node established. A failure there is not grounds for dropping + # that entry: the node's other pools and its control connection resume + # from it, and a pool filling against this listener would drop it again + # on every retry. + cache = SSLSessionCache() + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + node = DefaultEndPoint('1.2.3.4', 9042) + alias = copy.copy(node) + alias._port = 19142 + alias._tls_session_cache_key_override = node.tls_session_cache_key + + owner = self.make_connection(endpoint=node, ssl_context=context, + ssl_session_cache=cache) + listener = self.make_connection(endpoint=alias, ssl_context=context, + ssl_session_cache=cache) + assert listener._tls_session_cache_key() == owner._tls_session_cache_key() + session = Mock(has_ticket=True, id=b'\x01' * 32) + cache.set(owner._tls_session_cache_key(), session, lifetime=3600) + + listener._tls_session_offered = session + listener._discard_tls_session() + + assert cache.get(owner._tls_session_cache_key()) is session + + # The endpoint that owns the key still retracts, so a session that + # really cannot resume is not kept for ever. + owner._tls_session_offered = session + owner._discard_tls_session() + + assert cache.get(owner._tls_session_cache_key()) is None + + def test_store_caches_a_session_carrying_a_ticket(self): + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + session = Mock(has_ticket=True, id=b'', ticket_lifetime_hint=7200, + time=time.time(), timeout=7200) + connection._socket.session = session + + connection._store_tls_session() + + assert cache.get(connection._tls_session_cache_key()) is session + + def test_store_caches_a_session_carrying_only_an_id(self): + # Below TLS 1.3 a session id is offerable on its own, whether or not + # the server turns out to honour it. + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + session = Mock(has_ticket=False, id=b'\x01' * 32, ticket_lifetime_hint=0, + time=time.time(), timeout=7200) + connection._socket.session = session + + connection._store_tls_session() + + assert cache.get(connection._tls_session_cache_key()) is session + + def test_store_skips_a_tls13_ticket_with_a_zero_lifetime(self): + # RFC 8446 4.6.1: a ticket announced with a lifetime of zero is to be + # discarded immediately. + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + connection._socket.session = Mock(has_ticket=True, id=b'x' * 32, + ticket_lifetime_hint=0, + time=time.time(), timeout=7200) + connection._socket.version.return_value = 'TLSv1.3' + + connection._store_tls_session() + + assert cache.get(connection._tls_session_cache_key()) is None + + def test_store_keeps_a_tls12_ticket_with_an_unspecified_lifetime(self): + # RFC 5077 3.3 reserves a zero hint for "lifetime unspecified" and + # leaves retention to local policy, so the ticket is still usable and + # the local timeout is what there is to go on. + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + session = Mock(has_ticket=True, id=b'x' * 32, ticket_lifetime_hint=0, + time=time.time(), timeout=300) + connection._socket.session = session + connection._socket.version.return_value = 'TLSv1.2' + + connection._store_tls_session() + + assert cache.get(connection._tls_session_cache_key()) is session + + def test_store_caps_the_lifetime_at_seven_days(self): + # RFC 8446 4.6.1: no ticket may be kept longer than 7 days, whatever + # lifetime the server asked for. + _, connection = self.make_ssl_connection() + connection._ssl_session_cache.set = Mock() + connection._socket.session = Mock(has_ticket=True, id=b'', + ticket_lifetime_hint=30 * 24 * 3600, + time=time.time(), timeout=7200) + + connection._store_tls_session() + + _, kwargs = connection._ssl_session_cache.set.call_args + lifetime = kwargs.get('lifetime', connection._ssl_session_cache.set.call_args[0][-1]) + assert 7 * 24 * 3600 - 5 < lifetime <= 7 * 24 * 3600 + + def test_store_uses_the_announced_ticket_lifetime_not_the_local_timeout(self): + # SSLSession.timeout is the local context default and says nothing about + # what the peer will still accept. + _, connection = self.make_ssl_connection() + connection._ssl_session_cache.set = Mock() + connection._socket.session = Mock(has_ticket=True, id=b'', + ticket_lifetime_hint=60, + time=time.time(), timeout=7200) + + connection._store_tls_session() + + lifetime = connection._ssl_session_cache.set.call_args[0][-1] + assert 55 < lifetime <= 60 + + def test_store_ignores_the_sessions_wall_clock_stamp(self): + # SSLSession.time is wall clock, so reducing the lifetime by + # time.time() - session.time would let a clock step landing between the + # handshake and the store decide the answer. The age is taken from a + # monotonic mark instead, which these connections never set, so the + # announced lifetime comes through whole whatever the stamp says. + for stamp in (time.time() - 10_000, time.time() + 10_000, 0): + _, connection = self.make_ssl_connection() + connection._ssl_session_cache.set = Mock() + connection._socket.session = Mock(has_ticket=True, id=b'', + ticket_lifetime_hint=100, + time=stamp, timeout=7200) + + connection._store_tls_session() + + lifetime = connection._ssl_session_cache.set.call_args[0][-1] + assert lifetime == 100, stamp + + def test_store_charges_the_ticket_for_the_time_since_the_handshake(self): + # The peer issued it during the handshake; the store happens a startup + # exchange and perhaps an authentication later, and that time is the + # peer's to spend, not the entry's. + _, connection = self.make_ssl_connection() + connection._ssl_session_cache.set = Mock() + connection._socket.session = Mock(has_ticket=True, id=b'\x01' * 32, + ticket_lifetime_hint=7200, + time=time.time(), timeout=7200) + connection._tls_handshake_began_at = time.monotonic() - 100 + + connection._store_tls_session() + + lifetime = connection._ssl_session_cache.set.call_args[0][-1] + assert 7099 < lifetime <= 7100 + + def test_store_skips_a_ticket_that_expired_before_it_could_be_cached(self): + # A short-lived ticket and a slow startup leave nothing worth keeping. + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + connection._socket.session = Mock(has_ticket=True, id=b'\x01' * 32, + ticket_lifetime_hint=30, + time=time.time(), timeout=7200) + connection._tls_handshake_began_at = time.monotonic() - 60 + + connection._store_tls_session() + + assert cache.get(connection._tls_session_cache_key()) is None + + def test_store_skips_an_id_only_session_with_a_zero_timeout(self): + # The only way a lifetime can be nothing once the announced one is + # taken whole. + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + connection._socket.session = Mock(has_ticket=False, id=b'x' * 32, + ticket_lifetime_hint=0, + time=time.time(), timeout=0) + + connection._store_tls_session() + + assert cache.get(connection._tls_session_cache_key()) is None + + def test_store_falls_back_to_the_timeout_for_an_id_only_session(self): + # A session that resumes by id carries no announced lifetime, so the + # local timeout is all there is to go on. + _, connection = self.make_ssl_connection() + connection._ssl_session_cache.set = Mock() + connection._socket.session = Mock(has_ticket=False, id=b'x' * 32, + ticket_lifetime_hint=0, + time=time.time(), timeout=300) + + connection._store_tls_session() + + lifetime = connection._ssl_session_cache.set.call_args[0][-1] + assert 295 < lifetime <= 300 + + def test_store_hands_the_cache_the_session_that_was_offered(self): + # What this connection offered is half of the decision the cache makes, + # so it has to reach it: without it, a connection whose entry was + # dropped meanwhile re-creates it with a fresh full lifetime, on a + # session the peer issued long enough ago to have expired. + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + session = Mock(has_ticket=True, id=b'\x01' * 32, ticket_lifetime_hint=7200, + time=time.time(), timeout=7200) + connection._socket.session = session + # The state a resumed connection is in by the time it stores: it + # offered this session, and the entry has since gone. + connection._tls_session_offered = session + + connection._store_tls_session() + + assert cache.get(connection._tls_session_cache_key()) is None + assert len(cache) == 0 + + def test_store_skips_an_id_only_session_on_tls13(self): + # TLS 1.3 resumes only from a ticket, so an id on its own is nothing to + # offer however it got there. OpenSSL reports no id until a ticket has + # been read, so this state does not arise with it -- see + # test_the_session_is_only_stored_once_the_ticket_has_arrived in + # tests/unit/test_tls_resumption.py, which holds that against a real + # server -- and the rule is asserted here so it does not rest on that. + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + connection._socket.session = Mock(has_ticket=False, id=b'\x01' * 32, + ticket_lifetime_hint=0, + time=time.time(), timeout=7200) + connection._socket.version.return_value = 'TLSv1.3' + + connection._store_tls_session() + + assert cache.get(connection._tls_session_cache_key()) is None + + def test_store_skips_a_session_with_nothing_to_offer(self): + # This is a TLS 1.3 session read before the server's NewSessionTicket + # has arrived: no ticket and no id, so it could never resume and must + # not displace a usable entry. + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + cache.set(connection._tls_session_cache_key(), 'earlier-session') + connection._socket.session = Mock(has_ticket=False, id=b'') + + connection._store_tls_session() + + assert cache.get(connection._tls_session_cache_key()) == 'earlier-session' + + def test_store_tolerates_a_failure(self): + _, connection = self.make_ssl_connection() + connection._ssl_session_cache.set = Mock(side_effect=RuntimeError('boom')) + connection._socket.session = Mock(has_ticket=True, id=b'', + ticket_lifetime_hint=7200, + time=time.time(), timeout=7200) + + # _store_tls_session runs inside @defunct_on_error-wrapped handlers; + # a caching failure must never take the connection down. + connection._store_tls_session() + + # Asserted so the failure has to come from the cache: a session the + # accessors choke on would raise before ever reaching it, and the test + # would pass without covering what it names. + connection._ssl_session_cache.set.assert_called_once() + + def test_store_is_a_no_op_without_a_cache(self): + connection = self.make_connection() + + connection._store_tls_session() + + def test_the_accessors_are_the_whole_reactor_specific_surface(self): + # A reactor that establishes TLS by other means than an ssl.SSLSocket + # has no socket to read a session, a ticket or a version off, and + # reimplements the three accessors instead. Nothing else in the policy + # may reach for a socket, so this connection deliberately has none: + # anything that did would come back with no session cached and none + # offered. + session = Mock(has_ticket=True, id=b'\x01' * 32, + ticket_lifetime_hint=0, time=time.time(), timeout=7200) + + class OwnTransport(Connection): + offered = None + + def _set_tls_session(self, sock, restored): + self.offered = restored + + def _get_resumable_tls_session(self): + return session + + def _tls_negotiated_version(self): + # A zero ticket lifetime means "unspecified" here, not + # "discard", so the session is cached with the local timeout. + return 'TLSv1.2' + + cache = SSLSessionCache() + connection = OwnTransport( + DefaultEndPoint('1.2.3.4'), + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), + ssl_session_cache=cache) + + connection._store_tls_session() + assert cache.get(connection._tls_session_cache_key()) is session + + connection._restore_tls_session(sock=None) + assert connection.offered is session + + def test_a_tls_failure_in_the_reactor_drops_the_session_it_offered(self): + # ssl_options may carry do_handshake_on_connect=False, which leaves the + # handshake to the first read or write: the failure then arrives at + # defunct() rather than at _connect_socket, and would otherwise leave + # the session to be offered again by every later connection. + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + session = Mock(has_ticket=True, id=b'\x01' * 32, ticket_lifetime_hint=7200, + time=time.time(), timeout=7200) + cache.set(connection._tls_session_cache_key(), session, lifetime=7200) + connection._tls_session_offered = session + connection.close = Mock() + + connection.defunct(ssl.SSLError('handshake failure')) + + assert cache.get(connection._tls_session_cache_key()) is None + + def test_a_connection_that_started_up_retracts_nothing_later(self): + # Storing is the point the TLS handshake is known to have stood, so + # what was offered stops being in flight there. A TLS error long + # afterwards says nothing about a session that already worked. + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + session = Mock(has_ticket=True, id=b'\x01' * 32, ticket_lifetime_hint=7200, + time=time.time(), timeout=7200) + cache.set(connection._tls_session_cache_key(), session, lifetime=7200) + connection._socket.session = session + connection._tls_session_offered = session + connection._store_tls_session() + assert connection._tls_session_offered is None + connection.close = Mock() + + connection.defunct(ssl.SSLError('much later')) + + assert cache.get(connection._tls_session_cache_key()) is session + + def test_a_non_tls_failure_in_the_reactor_keeps_the_session(self): + # A reset or refused connection says nothing about the session, and + # dropping it would cost a later connection a full handshake. + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + session = Mock(has_ticket=True, id=b'\x01' * 32, ticket_lifetime_hint=7200, + time=time.time(), timeout=7200) + cache.set(connection._tls_session_cache_key(), session, lifetime=7200) + connection._tls_session_offered = session + connection.close = Mock() + + connection.defunct(OSError('connection reset')) + + assert cache.get(connection._tls_session_cache_key()) is session + + def test_session_is_stored_once_the_connection_is_ready(self): + _, connection = self.make_ssl_connection() + connection._compressor = None + connection._store_tls_session = Mock() + connection.defunct = Mock() + + connection._handle_startup_response(ReadyMessage()) + + connection.defunct.assert_not_called() + connection._store_tls_session.assert_called_once_with() + + def test_session_is_stored_once_authentication_succeeds(self): + _, connection = self.make_ssl_connection() + connection._compressor = None + connection._store_tls_session = Mock() + connection.authenticator = Mock() + connection.defunct = Mock() + + connection._handle_auth_response(AuthSuccessMessage(token=None)) + + connection.defunct.assert_not_called() + connection._store_tls_session.assert_called_once_with() + + class DefaultEndPointTest(unittest.TestCase): def test_default_endpoint_properties(self): diff --git a/tests/unit/test_endpoints.py b/tests/unit/test_endpoints.py index 1b6367dc2d..87d487945f 100644 --- a/tests/unit/test_endpoints.py +++ b/tests/unit/test_endpoints.py @@ -9,8 +9,10 @@ import unittest import itertools +import uuid -from cassandra.connection import DefaultEndPoint, SniEndPointFactory +from cassandra.connection import (ClientRoutesEndPoint, DefaultEndPoint, + SniEndPointFactory, UnixSocketEndPoint) from unittest.mock import patch @@ -53,3 +55,69 @@ def test_endpoint_resolve(self): for i in range(10): (address, _) = endpoint.resolve() assert address == next(it) + + def test_tls_session_cache_key_distinguishes_server_names(self): + # All SNI endpoints behind a proxy share an address and port, so the + # server name has to be part of the key or they would share sessions. + one = self.endpoint_factory.create_from_sni('node1') + other = self.endpoint_factory.create_from_sni('node2') + + assert one.tls_session_cache_key != other.tls_session_cache_key + assert one.tls_session_cache_key == \ + self.endpoint_factory.create_from_sni('node1').tls_session_cache_key + assert one.tls_session_cache_key != DefaultEndPoint( + 'proxy.datastax.com', 30002).tls_session_cache_key + + +class TlsSessionCacheKeyTest(unittest.TestCase): + + def test_default_endpoint_key(self): + assert DefaultEndPoint('10.0.0.1', 9042).tls_session_cache_key == ('10.0.0.1', 9042) + assert DefaultEndPoint('10.0.0.1', 9042).tls_session_cache_key != \ + DefaultEndPoint('10.0.0.1', 9142).tls_session_cache_key + + def test_unix_socket_endpoint_key(self): + assert UnixSocketEndPoint('/tmp/a').tls_session_cache_key != \ + UnixSocketEndPoint('/tmp/b').tls_session_cache_key + + def test_client_routes_endpoint_key_follows_the_node_not_the_route(self): + host_id = uuid.uuid4() + endpoint = ClientRoutesEndPoint(host_id, handler=None, + original_address='10.0.0.1', + original_port=9042) + other = ClientRoutesEndPoint(uuid.uuid4(), handler=None, + original_address='10.0.0.1', + original_port=9042) + + assert endpoint.tls_session_cache_key == (host_id, '10.0.0.1', 9042) + assert endpoint.tls_session_cache_key != other.tls_session_cache_key + + def test_an_override_replaces_the_endpoints_own_identity(self): + # An endpoint built to reach a node another one already describes -- the + # shard-aware port alias -- carries that node's key so both share one + # cached session. + node = DefaultEndPoint('10.0.0.1', 9042) + alias = DefaultEndPoint('10.0.0.1', 19142) + assert alias.tls_session_cache_key != node.tls_session_cache_key + + alias._tls_session_cache_key_override = node.tls_session_cache_key + + assert alias.tls_session_cache_key == node.tls_session_cache_key + + def test_an_override_applies_to_every_endpoint_type(self): + # The override lives on the base property, so a subclass that gives its + # own identity still honours it. + endpoints = [DefaultEndPoint('10.0.0.1'), + UnixSocketEndPoint('/tmp/a'), + ClientRoutesEndPoint(uuid.uuid4(), None, '10.0.0.1', 9042), + SniEndPointFactory("proxy", 30002).create_from_sni('node1')] + for endpoint in endpoints: + endpoint._tls_session_cache_key_override = ('the', 'node') + assert endpoint.tls_session_cache_key == ('the', 'node'), endpoint + + def test_keys_are_hashable(self): + # Keys are used as dict keys in SSLSessionCache. + for endpoint in (DefaultEndPoint('10.0.0.1'), + UnixSocketEndPoint('/tmp/a'), + ClientRoutesEndPoint(uuid.uuid4(), None, '10.0.0.1', 9042)): + hash(endpoint.tls_session_cache_key) diff --git a/tests/unit/test_shard_aware.py b/tests/unit/test_shard_aware.py index 5c0b06c25d..fb906fe1ec 100644 --- a/tests/unit/test_shard_aware.py +++ b/tests/unit/test_shard_aware.py @@ -140,6 +140,27 @@ class OptionsHolder(object): assert shard_info.shard_id_from_token(Murmur3Token.from_key(b"e").value) == 4 assert shard_info.shard_id_from_token(Murmur3Token.from_key(b"100000").value) == 2 + def test_shard_aware_endpoint_carries_the_nodes_tls_identity(self): + """ + The alternate listener must resume from the session cached for the node, + not key on its own port. + """ + host = MagicMock() + host.endpoint = DefaultEndPoint("1.2.3.4") + session = MockSession(ssl_context=object()) + pool = HostConnection(host=host, host_distance=HostDistance.REMOTE, + session=session) + try: + for f in session.futures: + f.result() + shard_aware_endpoint = pool._get_shard_aware_endpoint() + assert shard_aware_endpoint.port == 19045 + assert (shard_aware_endpoint.tls_session_cache_key == + host.endpoint.tls_session_cache_key) + finally: + pool.shutdown() + session.cluster.executor.shutdown(wait=True) + def test_advanced_shard_aware_port(self): """ Test that on given a `shard_aware_port` on the OPTIONS message (ShardInfo class) diff --git a/tests/unit/test_ssl_session_cache.py b/tests/unit/test_ssl_session_cache.py new file mode 100644 index 0000000000..d8ae7e9934 --- /dev/null +++ b/tests/unit/test_ssl_session_cache.py @@ -0,0 +1,386 @@ +# Copyright 2026 ScyllaDB, 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. + +import threading +import time +import unittest +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from cassandra.ssl_session_cache import SSLSessionCache + + +class _Session(object): + """ + Stands in for ssl.SSLSession, which only a real handshake can produce. + Only the session id matters here: it is how the cache tells a session the + peer reissued from the one it handed back unchanged. + """ + + def __init__(self, id=b'\x01' * 32): + self.id = id + + +class SSLSessionCacheTest(unittest.TestCase): + + def test_get_missing_key_returns_none(self): + assert SSLSessionCache().get(('10.0.0.1', 9042)) is None + + def test_set_then_get(self): + cache = SSLSessionCache() + session = object() + cache.set(('10.0.0.1', 9042), session) + + assert cache.get(('10.0.0.1', 9042)) is session + assert cache.get(('10.0.0.2', 9042)) is None + + def test_get_does_not_consume_the_session(self): + # Sessions are replayable: a burst of per-shard connections to one + # node must all be able to offer the same cached session. + cache = SSLSessionCache() + session = object() + cache.set(('10.0.0.1', 9042), session) + + assert [cache.get(('10.0.0.1', 9042)) for _ in range(10)] == [session] * 10 + assert len(cache) == 1 + + def test_set_replaces_the_previous_session(self): + cache = SSLSessionCache() + older, newer = object(), object() + cache.set(('10.0.0.1', 9042), older) + cache.set(('10.0.0.1', 9042), newer) + + assert cache.get(('10.0.0.1', 9042)) is newer + assert len(cache) == 1 + + def test_none_session_is_ignored(self): + cache = SSLSessionCache() + session = object() + cache.set(('10.0.0.1', 9042), session) + cache.set(('10.0.0.1', 9042), None) + + assert cache.get(('10.0.0.1', 9042)) is session + assert len(cache) == 1 + + def test_evicts_least_recently_used_key(self): + cache = SSLSessionCache(max_size=2) + first, second, third = object(), object(), object() + cache.set('first', first) + cache.set('second', second) + + # Touching 'first' makes 'second' the least recently used. + assert cache.get('first') is first + cache.set('third', third) + + assert len(cache) == 2 + assert cache.get('second') is None + assert cache.get('first') is first + assert cache.get('third') is third + + def test_set_refreshes_recency(self): + cache = SSLSessionCache(max_size=2) + cache.set('first', object()) + cache.set('second', object()) + cache.set('first', object()) + cache.set('third', object()) + + assert cache.get('second') is None + assert cache.get('first') is not None + + def test_expired_entry_is_not_returned_and_is_dropped(self): + cache = SSLSessionCache() + cache.set('key', object(), lifetime=-1) + + assert cache.get('key') is None + assert len(cache) == 0 + + def test_live_entry_is_returned(self): + cache = SSLSessionCache() + session = object() + cache.set('key', session, lifetime=3600) + + assert cache.get('key') is session + + def test_the_same_session_again_keeps_the_deadline_it_had(self): + # What a resumed handshake below TLS 1.3 stores back: the session that + # was offered, whose lifetime runs from when the peer issued it and not + # from when it was last replayed. + cache = SSLSessionCache() + cache.set('key', _Session(), lifetime=0.05) + cache.set('key', _Session(), lifetime=3600) + + time.sleep(0.06) + + assert cache.get('key') is None + + def test_a_reissued_session_starts_its_own_deadline(self): + # What TLS 1.3 normally stores back: a fresh ticket, which is entitled + # to the lifetime the peer announced with it. + cache = SSLSessionCache() + cache.set('key', _Session(b'first'), lifetime=-1) + reissued = _Session(b'second') + cache.set('key', reissued, lifetime=3600) + + assert cache.get('key') is reissued + + def test_the_offered_session_does_not_revive_an_entry_that_was_dropped(self): + # The deadline passed between the offer and the store, and a lookup + # dropped the entry. Storing the offered session back would put a full + # fresh lifetime on a session the peer issued long enough ago to have + # expired. + cache = SSLSessionCache() + session = _Session() + cache.set('key', session, lifetime=-1) + assert cache.get('key') is None + + cache.set('key', session, lifetime=3600, offered=session) + + assert cache.get('key') is None + assert len(cache) == 0 + + def test_the_offered_session_does_not_displace_a_siblings(self): + # Connections to one node are opened together: another may have stored + # a session the peer reissued to it between this one's offer and its + # store. That entry is fresher than what this caller has to say. + cache = SSLSessionCache() + offered, reissued = _Session(b'offered'), _Session(b'reissued') + cache.set('key', offered, lifetime=3600) + cache.set('key', reissued, lifetime=3600) + deadline = cache._sessions['key'][1] + + cache.set('key', offered, lifetime=3600, offered=offered) + + assert cache.get('key') is reissued + assert cache._sessions['key'][1] == deadline + + def test_a_reissued_session_still_replaces_a_siblings(self): + # The other side of it: what the peer issued to this connection is new, + # and is entitled to the lifetime announced with it. + cache = SSLSessionCache() + offered, theirs = _Session(b'offered'), _Session(b'theirs') + cache.set('key', theirs, lifetime=3600) + mine = _Session(b'mine') + + cache.set('key', mine, lifetime=3600, offered=offered) + + assert cache.get('key') is mine + + def test_tickets_that_carry_no_session_id_are_not_told_apart(self): + # RFC 5077 3.4 lets a server issue a ticket and send an empty session + # id with it, and SSLSession exposes no ticket to compare instead, so + # two of them are indistinguishable. Reading them as the same session + # keeps the deadline where it is; reading them as different would + # re-stamp a full lifetime on what may be the ticket already held. + cache = SSLSessionCache() + first, second = _Session(b''), _Session(b'') + cache.set('key', first, lifetime=3600) + deadline = cache._sessions['key'][1] + + cache.set('key', second, lifetime=7200) + + # One session as far as this can tell, so the entry keeps both the + # deadline it had and the object holding it. + assert cache.get('key') is first + assert cache._sessions['key'][1] == deadline + + def test_an_empty_id_costs_a_store_where_the_entry_has_gone(self): + # The other half of that reading, and the reason it is the safe one: + # the offered ticket and the one that came back cannot be told apart, + # so this store is skipped rather than reviving a deadline. The next + # connection offers nothing and caches whatever it is given. + cache = SSLSessionCache() + offered = _Session(b'') + cache.set('key', offered, lifetime=-1) + assert cache.get('key') is None + + cache.set('key', _Session(b''), lifetime=3600, offered=offered) + assert len(cache) == 0 + + fresh = _Session(b'') + cache.set('key', fresh, lifetime=3600) + assert cache.get('key') is fresh + + def test_an_object_carrying_no_id_is_taken_to_be_new(self): + # Distinct from an empty id: nothing the driver caches is in this + # position, since every SSLSession has the attribute. + cache = SSLSessionCache() + cache.set('key', object(), lifetime=-1) + session = object() + cache.set('key', session, lifetime=3600) + + assert cache.get('key') is session + + def test_a_lifetime_replaces_the_previous_one(self): + cache = SSLSessionCache() + cache.set('key', object(), lifetime=-1) + session = object() + cache.set('key', session, lifetime=3600) + + assert cache.get('key') is session + + def test_a_dead_entry_is_evicted_before_a_live_one(self): + # Whose lifetime has run out and which was used least recently are + # independent once peers announce different lifetimes. + cache = SSLSessionCache(max_size=3) + cache.set('live-1', 'A', lifetime=3600) + cache.set('live-2', 'B', lifetime=3600) + cache.set('expired', 'C', lifetime=-1) + + cache.set('fourth', 'D', lifetime=3600) + + assert cache.get('live-1') == 'A' + assert cache.get('live-2') == 'B' + assert cache.get('fourth') == 'D' + assert len(cache) == 3 + + def test_the_lru_still_goes_when_nothing_has_expired(self): + cache = SSLSessionCache(max_size=2) + cache.set('first', 'A', lifetime=3600) + cache.set('second', 'B', lifetime=3600) + + cache.set('third', 'C', lifetime=3600) + + assert cache.get('first') is None + assert cache.get('second') == 'B' + assert cache.get('third') == 'C' + + def test_a_dead_entry_lingers_until_it_is_looked_up_or_room_is_needed(self): + # Documented rather than swept eagerly: nothing walks the cache on a + # timer, so an entry nobody asks for and nobody needs room for stays. + cache = SSLSessionCache(max_size=8) + cache.set('expired', 'C', lifetime=-1) + + assert len(cache) == 1 + assert cache.get('expired') is None + assert len(cache) == 0 + + def test_discard(self): + cache = SSLSessionCache() + cache.set('key', object()) + cache.discard('key') + + assert cache.get('key') is None + assert len(cache) == 0 + cache.discard('key') # discarding what is not there is fine + + def test_storing_the_same_session_keeps_the_object_that_holds_it(self): + # SSLSocket.session builds a new wrapper on every access, so what a + # resumed connection stores back is another handle on the credential + # already cached. Keeping the one that is there is what lets the + # connection that offered it retract it: identity is how discard() + # tells its own session from one the peer issued in its place. + cache = SSLSessionCache() + offered = _Session() + cache.set('key', offered, lifetime=3600) + cache.set('key', _Session(), lifetime=3600) # a sibling resumes + + assert cache.get('key') is offered + + cache.discard('key', offered) + assert cache.get('key') is None + + def test_a_session_the_peer_issued_in_its_place_is_not_retractable(self): + # The other side of it: that one never failed anything, and every later + # connection would pay a full handshake for dropping it. + cache = SSLSessionCache() + offered = _Session(b'offered') + cache.set('key', offered, lifetime=3600) + reissued = _Session(b'reissued') + cache.set('key', reissued, lifetime=3600) + + cache.discard('key', offered) + + assert cache.get('key') is reissued + + def test_discard_of_a_named_session_spares_a_newer_one(self): + # A connection acting on a session it read earlier must not remove the + # fresh one another connection stored under the same key meanwhile. + cache = SSLSessionCache() + older, newer = object(), object() + cache.set('key', older) + cache.set('key', newer) + + cache.discard('key', older) + + assert cache.get('key') is newer + + def test_discard_of_a_named_session_removes_it_when_still_current(self): + cache = SSLSessionCache() + session = object() + cache.set('key', session) + + cache.discard('key', session) + + assert cache.get('key') is None + + def test_clear(self): + cache = SSLSessionCache() + cache.set('key', object()) + cache.clear() + + assert len(cache) == 0 + assert cache.get('key') is None + + def test_rejects_invalid_max_size(self): + # A float would pass a plain `< 1` check and then never bound the cache + # (nan and inf compare False against every limit), and True is an int + # that passes it and would cap the cache at a single entry. + for max_size in (0, -1, float('nan'), float('inf'), 2.5, '8', None, + True, False): + with pytest.raises(ValueError): + SSLSessionCache(max_size=max_size) + + def test_repr(self): + cache = SSLSessionCache(max_size=7) + cache.set('key', object()) + + assert repr(cache) == '' + + def test_repr_can_be_taken_while_the_cache_is_locked(self): + # So that a log line formatting %r from inside one of the cache's own + # methods does not wait for the lock that method is holding. + cache = SSLSessionCache(max_size=7) + cache.set('key', object()) + taken = [] + + def under_the_lock(): + with cache._lock: + taken.append(repr(cache)) + + # Daemon: if this ever does deadlock the assertion below reports it + # rather than the suite hanging at exit waiting for the thread. + thread = threading.Thread(target=under_the_lock, daemon=True) + thread.start() + thread.join(timeout=5) + + assert not thread.is_alive(), 'repr() deadlocked against the cache lock' + assert taken == [''] + + def test_concurrent_access_keeps_the_cache_bounded(self): + cache = SSLSessionCache(max_size=8) + + def hammer(worker): + for i in range(500): + key = (worker + i) % 32 + cache.set(key, object()) + cache.get(key) + assert len(cache) <= 8 + + # result() re-raises whatever a worker hit, with its own traceback. + with ThreadPoolExecutor(max_workers=8) as pool: + for future in [pool.submit(hammer, worker) for worker in range(8)]: + future.result() + + assert len(cache) <= 8 diff --git a/tests/unit/test_tls_resumption.py b/tests/unit/test_tls_resumption.py new file mode 100644 index 0000000000..fa706517e5 --- /dev/null +++ b/tests/unit/test_tls_resumption.py @@ -0,0 +1,434 @@ +# Copyright 2026 ScyllaDB, 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. +""" +TLS session resumption exercised against a real TLS server on loopback. + +These tests drive the actual code paths a connection uses -- restoring a +cached session onto the socket before the handshake, and storing the +negotiated session afterwards -- and check the outcome the way OpenSSL +reports it, through ``SSLSocket.session_reused``. No Cassandra or Scylla +server is involved: the peer speaks TLS and echoes bytes, which is all the +socket-level code under test needs. +""" + +import gc +import socket +import ssl +import tempfile +import threading +import time +import unittest +import weakref + +import pytest +from unittest.mock import Mock + +from cassandra.connection import Connection, DefaultEndPoint +from cassandra.ssl_session_cache import SSLSessionCache +from tests.tls_certificates import HAVE_CRYPTOGRAPHY, write_self_signed_cert + + + +def _wait_for_the_monotonic_clock_to_move(): + """ + Block until ``time.monotonic()`` reports a new value. + + The deadline a store computes works out to the mark taken when the + handshake began plus the lifetime the peer announced -- the "now" in + ``lifetime - (now - mark)`` and the one in ``now + lifetime`` cancel out -- + so two handshakes the clock cannot tell apart are given the same deadline. + That is the right answer, there being no difference to record, but it + leaves nothing for a test comparing two deadlines to see. Windows resolves + this clock to about ten milliseconds, which two loopback handshakes fit + inside comfortably. + """ + started = time.monotonic() + while time.monotonic() == started: + time.sleep(0.001) + + +class _TLSEchoServer(object): + """ + A TLS server on loopback that echoes back whatever a client sends. Each + accepted connection is served on its own thread, so a batch of clients can + handshake concurrently. + """ + + def __init__(self, cert_path, key_path, tls_version): + self.context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + self.context.load_cert_chain(cert_path, key_path) + self.context.minimum_version = tls_version + self.context.maximum_version = tls_version + + self._listener = socket.socket() + self._listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self._listener.bind(('127.0.0.1', 0)) + self._listener.listen(16) + self._listener.settimeout(0.1) + self.port = self._listener.getsockname()[1] + + self._stop = threading.Event() + self._accept_thread = threading.Thread(target=self._accept_loop, daemon=True) + self._accept_thread.start() + + def _accept_loop(self): + while not self._stop.is_set(): + try: + client, _ = self._listener.accept() + except socket.timeout: + continue + except OSError: + return + threading.Thread(target=self._serve, args=(client,), daemon=True).start() + + def _serve(self, client): + try: + tls_client = self.context.wrap_socket(client, server_side=True) + while True: + data = tls_client.recv(64) + if not data: + return + tls_client.sendall(data) + except OSError: + pass + finally: + try: + client.close() + except OSError: + pass + + def close(self): + self._stop.set() + self._accept_thread.join(timeout=5) + self._listener.close() + + +class _SocketOnlyConnection(Connection): + """ + A connection that performs only the socket and TLS part of setup. The CQL + handshake is stood in for by an echo exchange, which is enough to have a + TLS 1.3 server's NewSessionTicket read off the socket, exactly as the + OPTIONS/STARTUP exchange does in a real connection. + """ + + def __init__(self, *args, **kwargs): + Connection.__init__(self, *args, **kwargs) + self._connect_socket() + + def exchange(self): + self._socket.sendall(b'ping') + assert self._socket.recv(4) == b'ping' + + def close(self): + if self._socket is not None: + try: + self._socket.close() + except OSError: + pass + + @property + def session_reused(self): + return self._socket.session_reused + + +@unittest.skipIf(not HAVE_CRYPTOGRAPHY, + 'cryptography is required to generate a test certificate') +class TlsResumptionTest(unittest.TestCase): + + tls_version = ssl.TLSVersion.TLSv1_2 + + @classmethod + def setUpClass(cls): + cls._cert_dir = tempfile.TemporaryDirectory(prefix='tls_resumption_') + cls.addClassCleanup(cls._cert_dir.cleanup) + cls._cert_path, cls._key_path = write_self_signed_cert(cls._cert_dir.name) + # A second pair, for a server the client context will not trust. + cls._untrusted_dir = tempfile.TemporaryDirectory(prefix='tls_untrusted_') + cls.addClassCleanup(cls._untrusted_dir.cleanup) + cls._untrusted_cert, cls._untrusted_key = write_self_signed_cert( + cls._untrusted_dir.name) + + def setUp(self): + self.server = _TLSEchoServer(self._cert_path, self._key_path, self.tls_version) + self.addCleanup(self.server.close) + self.cache = SSLSessionCache() + self.connections = [] + + def make_ssl_context(self): + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + context.load_verify_locations(self._cert_path) + context.verify_mode = ssl.CERT_REQUIRED + context.check_hostname = True + return context + + def untrusted_server(self): + """ + A TLS server whose certificate the client context does not trust, so + the handshake fails during verification. + + Failing that way rather than by feeding a listener non-TLS bytes keeps + the failure a TLS one on every platform: bytes sent and the connection + then closed is a race between OpenSSL reading the bad record and the + socket reporting the close, and Windows reports the close first + (WSAECONNABORTED), which is not a TLS error at all. + """ + server = _TLSEchoServer(self._untrusted_cert, self._untrusted_key, + self.tls_version) + self.addCleanup(server.close) + return server + + def connect(self, ssl_context, cache=None, exchange=True, ssl_options=None): + connection = _SocketOnlyConnection( + DefaultEndPoint('127.0.0.1', self.server.port), + ssl_context=ssl_context, + ssl_options=ssl_options, + ssl_session_cache=self.cache if cache is None else cache, + connect_timeout=10) + self.connections.append(connection) + self.addCleanup(connection.close) + if exchange: + connection.exchange() + return connection + + def test_a_second_connection_resumes_the_first_session(self): + context = self.make_ssl_context() + + first = self.connect(context) + assert not first.session_reused + first._store_tls_session() + assert len(self.cache) == 1 + + second = self.connect(context) + + assert second.session_reused + + def test_concurrent_connections_all_resume_one_cached_session(self): + # This is the case DRIVER-165 is about: a pool opens one connection per + # shard at once, and they all have to be able to offer the session + # cached by an earlier connection to the same node. + context = self.make_ssl_context() + self.connect(context)._store_tls_session() + + resumed = [] + barrier = threading.Barrier(4) + + def connect_and_record(): + barrier.wait() + resumed.append(self.connect(context, exchange=False).session_reused) + + threads = [threading.Thread(target=connect_and_record) for _ in range(4)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=30) + + assert resumed == [True] * 4 + + def test_no_resumption_without_a_cache(self): + context = self.make_ssl_context() + self.connect(context)._store_tls_session() + + without_cache = _SocketOnlyConnection( + DefaultEndPoint('127.0.0.1', self.server.port), + ssl_context=context, ssl_session_cache=None, connect_timeout=10) + self.addCleanup(without_cache.close) + + assert not without_cache.session_reused + + def test_a_cached_session_pins_its_context_until_the_cache_drops_it(self): + # A real SSLSession holds a strong reference to the SSLContext it was + # established with, so an entry keeps that context -- and everything + # reachable from it -- alive. A cache the driver created for a cluster + # is dropped with it; one the caller supplied outlives it, holding the + # contexts of at most max_size peers until clear() drops them. + # Built directly rather than through self.connect(), whose bookkeeping + # would hold the connection, and so the context, itself. + context = self.make_ssl_context() + connection = _SocketOnlyConnection( + DefaultEndPoint('127.0.0.1', self.server.port), + ssl_context=context, ssl_session_cache=self.cache, + connect_timeout=10) + connection.exchange() + connection._store_tls_session() + key = connection._tls_session_cache_key() + assert self.cache.get(key) is not None + weak = weakref.ref(context) + + connection.close() + del context, connection, key + gc.collect() + # The entry still holds it, which is the retention the cache documents. + assert weak() is not None + + self.cache.clear() + gc.collect() + + assert weak() is None + + def test_a_connection_that_resumed_remembers_what_it_offered(self): + # Nothing is left to retract once the handshake stands, but the store + # reads what was offered: handing it to the cache is what tells a + # session the peer reissued from the one that came back unchanged. + context = self.make_ssl_context() + first = self.connect(context) + first._store_tls_session() + offered = self.cache.get(first._tls_session_cache_key()) + + resumed = self.connect(context) + + assert resumed.session_reused + assert resumed._tls_session_offered.id == offered.id + + def test_an_attempt_that_offers_nothing_clears_what_came_before(self): + context = self.make_ssl_context() + connection = self.connect(context) + connection._tls_session_offered = 'from an earlier address' + + # Nothing cached for this key, so nothing is offered. + connection._restore_tls_session(Mock()) + + assert connection._tls_session_offered is None + + def test_a_failed_handshake_drops_the_session_it_offered(self): + # Nothing stores a session for a connection that never came up, so an + # entry that provokes a handshake failure would be offered again by + # every later connection until its lifetime ran out. + context = self.make_ssl_context() + donor = self.connect(context) + donor._store_tls_session() + + rejecting = self.untrusted_server() + endpoint = DefaultEndPoint('127.0.0.1', rejecting.port) + # A Connection built without connecting, just to ask for the key the + # failing connection below will use. + key = Connection(endpoint, ssl_context=context, + ssl_session_cache=self.cache)._tls_session_cache_key() + self.cache.set(key, self.cache.get(donor._tls_session_cache_key())) + assert self.cache.get(key) is not None + + # _connect_socket re-raises as socket.error(errno, ...), so the + # SSLError type does not survive -- only its message. + with pytest.raises(OSError, match='SSL'): + _SocketOnlyConnection(endpoint, ssl_context=context, + ssl_session_cache=self.cache, connect_timeout=10) + + assert self.cache.get(key) is None + + def test_a_failed_handshake_spares_a_session_stored_meanwhile(self): + # Connections to one node are opened together, so another may store a + # fresh session under this key between the offer and the failure. That + # one did not fail anything and has to stay. + context = self.make_ssl_context() + donor = self.connect(context) + donor._store_tls_session() + + rejecting = self.untrusted_server() + endpoint = DefaultEndPoint('127.0.0.1', rejecting.port) + key = Connection(endpoint, ssl_context=context, + ssl_session_cache=self.cache)._tls_session_cache_key() + self.cache.set(key, self.cache.get(donor._tls_session_cache_key())) + + # Stand in for the connection that succeeds while this one is failing. + class Refresher(_SocketOnlyConnection): + def _set_tls_session(self, sock, session): + super()._set_tls_session(sock, session) + self._ssl_session_cache.set(key, 'stored-by-another-connection') + + with pytest.raises(OSError, match='SSL'): + Refresher(endpoint, ssl_context=context, + ssl_session_cache=self.cache, connect_timeout=10) + + assert self.cache.get(key) == 'stored-by-another-connection' + + def test_a_session_is_not_offered_to_a_different_server_name(self): + # A resumed handshake carries no Certificate, so the name the peer was + # verified against is never checked again. A session established for + # one name must therefore never be offered to a connection expecting + # another, even though both reach the same address and port. + context = self.make_ssl_context() + context.check_hostname = False + self.connect(context, ssl_options={'server_hostname': 'one.example'})._store_tls_session() + + same_name = self.connect(context, ssl_options={'server_hostname': 'one.example'}) + other_name = self.connect(context, ssl_options={'server_hostname': 'other.example'}) + + assert same_name.session_reused + assert not other_name.session_reused + + def test_a_session_is_not_offered_to_a_different_context(self): + # A session can only be replayed onto the context it was established + # with -- the stdlib ssl module rejects anything else -- so the context + # is part of the cache key. + self.connect(self.make_ssl_context())._store_tls_session() + + second = self.connect(self.make_ssl_context()) + + assert not second.session_reused + + def test_what_a_resumed_handshake_stores_back(self): + # SSLSocket.session builds a new object on every access, so comparing + # object identity here would pass whatever happened. What matters is + # whether the peer issued a new session: below TLS 1.3 an abbreviated + # handshake hands back the one that was offered, and its deadline has + # to stay where it was rather than start again on every reuse. + context = self.make_ssl_context() + first = self.connect(context) + first._store_tls_session() + # Ask the connection for its key rather than rebuilding it here, so this + # test does not depend on the key's shape. + key = first._tls_session_cache_key() + first_id = self.cache.get(key).id + first_deadline = self.cache._sessions[key][1] + + # Otherwise both handshakes may begin within one tick of the clock, and + # a deadline computed afresh is indistinguishable from an inherited one. + _wait_for_the_monotonic_clock_to_move() + resumed = self.connect(context) + assert resumed.session_reused + resumed._store_tls_session() + + renewed = self.tls_version >= ssl.TLSVersion.TLSv1_3 + assert (self.cache.get(key).id != first_id) is renewed + assert (self.cache._sessions[key][1] > first_deadline) is renewed + + +@unittest.skipUnless(ssl.HAS_TLSv1_3, 'this build of OpenSSL has no TLS 1.3') +class Tls13ResumptionTest(TlsResumptionTest): + """ + The same coverage over TLS 1.3, plus what is specific to it. + + Every inherited test drives a TLS 1.3 server, so a build without it has to + skip the class rather than fail each handshake. The guard sits here and not + on the base class: skipping a base skips its subclasses too, which would + take these tests out on a build that has TLS 1.3 but not 1.2. + """ + + tls_version = ssl.TLSVersion.TLSv1_3 + + def test_the_session_is_only_stored_once_the_ticket_has_arrived(self): + # A TLS 1.3 server sends its NewSessionTicket after the handshake, so a + # session read before the first application-data exchange carries no + # ticket and must not be cached. + connection = self.connect(self.make_ssl_context(), exchange=False) + + assert connection._socket.version() == 'TLSv1.3' + assert connection._get_resumable_tls_session() is None + connection._store_tls_session() + assert len(self.cache) == 0 + + connection.exchange() + + assert connection._get_resumable_tls_session() is not None + connection._store_tls_session() + assert len(self.cache) == 1