diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 7260bd08b6..c228146b60 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -51,7 +51,8 @@ from cassandra.connection import (ClientRoutesEndPointFactory, ConnectionException, ConnectionShutdown, ConnectionHeartbeat, ProtocolVersionUnsupported, EndPoint, DefaultEndPoint, DefaultEndPointFactory, - SniEndPointFactory, ConnectionBusy, locally_supported_compressions) + SniEndPointFactory, ConnectionBusy, locally_supported_compressions, + SSLSessionCache) from cassandra.cqltypes import UserType import cassandra.cqltypes as types from cassandra.encoder import Encoder @@ -866,6 +867,52 @@ def default_retry_policy(self, policy): .. versionadded:: 3.17.0 """ + ssl_session_cache = None + """ + A :class:`~cassandra.connection.SSLSessionCache` shared by every + connection this cluster opens, letting them resume TLS sessions instead of + performing a full handshake each time. This matters most for the group of + per-shard connections opened to a node at once, and for reconnections. + + One is created automatically when :attr:`~Cluster.ssl_context` is set. + That decision is made while the :class:`.Cluster` is being constructed, as + it is for the other state derived from the TLS configuration, so setting + ``ssl_context`` afterwards leaves resumption off; assign a cache here + yourself if you configure TLS that way, any time before + :meth:`~.Cluster.connect`. + + Pass ``ssl_session_cache=None`` to :class:`.Cluster` to turn resumption + off, or pass your own instance to size it or to share it between + clusters:: + + from cassandra.connection import SSLSessionCache + + cluster = Cluster(ssl_context=ssl_context, + ssl_session_cache=SSLSessionCache(max_size=64)) + + Resumption is available when TLS is configured through + :attr:`~Cluster.ssl_context` and the reactor establishes TLS with the + standard library's ``ssl`` module: the ``libev`` and ``asyncore`` reactors, + which is to say the default one. + + It is not available with the deprecated :attr:`~Cluster.ssl_options`-only + configuration, because each connection builds its own ``SSLContext`` and a + session cannot be replayed onto a different one; nor on the ``asyncio`` + reactor, which performs the handshake inside + ``loop.create_connection()``, leaving no point at which to restore a + session. In those cases no cache is created and connections handshake in + full. + + It equally requires the server to hand out something it will honour later. + Scylla issues session tickets only when ``enable_session_tickets`` is set + in its ``client_encryption_options``, which is off by default; without it + nothing resumes and every connection performs a full handshake, as it would + have anyway. Over TLS 1.3 the cache then stays empty, while over TLS 1.2 + such a server still assigns a session id, so the cache may hold an entry it + will not honour -- offering that costs nothing and the handshake simply + completes in full. + """ + sockopts = None """ An optional list of tuples which will be used as arguments to @@ -1166,6 +1213,9 @@ def token_metadata_enabled(self, enabled): _prepared_statements = None _prepared_statement_lock = None _idle_heartbeat = None + # (cache, ssl_context) registered with the cache, see + # _acquire_tls_session_context. + _acquired_tls_session_context = None _protocol_version_explicit = False _discount_down_events = True @@ -1221,7 +1271,8 @@ def __init__(self, application_info:Optional[ApplicationInfoBase]=None, client_routes_config:Optional[ClientRoutesConfig]=None, allow_control_connection_query_fallback:Optional[ControlConnectionQueryFallback]=ControlConnectionQueryFallback.Disabled, - driver_config_reporting_enabled=True + driver_config_reporting_enabled=True, + ssl_session_cache=_NOT_SET ): """ ``executor_threads`` defines the number of threads in a pool for handling asynchronous tasks such as @@ -1468,6 +1519,45 @@ def __init__(self, self.ssl_options = ssl_options self.ssl_context = ssl_context + + # TLS sessions can only be resumed where the session can be replayed + # onto the same SSLContext and the reactor gives the driver a chance to + # offer it before the handshake. connection_class is not required to + # derive from Connection, so treat one that does not report the + # capability as lacking it. + resumable = (ssl_context is not None and + getattr(self.connection_class, + 'supports_tls_session_resumption', False)) + + if ssl_session_cache is _NOT_SET: + self.ssl_session_cache = SSLSessionCache() if resumable else None + else: + self.ssl_session_cache = ssl_session_cache + if ssl_session_cache is not None and not resumable: + # Asking for resumption and silently getting none is worse than + # not having it: the cache stays reachable and empty, with + # nothing to explain why. + if ssl_context is None: + reason = ('no ssl_context is configured, and a session ' + 'cannot be replayed onto the fresh context each ' + 'connection builds from ssl_options') + else: + reason = ('%s cannot restore a session before the ' + 'handshake' % + getattr(self.connection_class, '__name__', + self.connection_class)) + log.warning('ssl_session_cache was supplied but TLS session ' + 'resumption is unavailable here, so no sessions ' + 'will be cached: %s.', reason) + # Dropped rather than kept unused, so that this attribute means + # "the cache these connections use" throughout: a cache left + # here would be handed to every connection -- which a + # connection class that does not take the keyword cannot even + # accept -- and would sit reachable and empty for anyone + # reading it back. + self.ssl_session_cache = None + + self.sockopts = sockopts self.cql_version = cql_version self.max_schema_agreement_wait = max_schema_agreement_wait @@ -1670,6 +1760,26 @@ def add_execution_profile(self, name, profile, pool_wait_timeout=5): raise OperationTimedOut("Failed to create all new connection pools in the %ss timeout." % pool_wait_timeout, timeout=pool_wait_timeout) + def _acquire_tls_session_context(self): + """ + Tell the session cache that this cluster's sessions are in use, so that + they outlive any other cluster sharing the context giving up its own. + + Done here rather than in __init__ because both attributes it reads can + be set afterwards -- which is how a cluster configured with TLS after + construction gets a cache at all -- and because a cluster that is + constructed and never connected would otherwise hold a registration + that nothing ever gives back, leaving a shared cache unable to drop + that context. The pair is remembered so shutdown() releases what was + actually taken, whatever the attributes say by then. + """ + if self.ssl_session_cache is None or self.ssl_context is None: + return + + self._acquired_tls_session_context = (self.ssl_session_cache, + self.ssl_context) + self.ssl_session_cache.acquire_context(self.ssl_context) + def connection_factory(self, endpoint, host_conn = None, *args, **kwargs): """ Called to create a new connection with proper configuration. @@ -1691,6 +1801,12 @@ def _make_connection_kwargs(self, endpoint, kwargs_dict): kwargs_dict.setdefault('sockopts', self.sockopts) kwargs_dict.setdefault('ssl_options', self.ssl_options) kwargs_dict.setdefault('ssl_context', self.ssl_context) + if self.ssl_session_cache is not None: + # Set only where resumption is possible, so this is also the test + # for that: a connection class that does not accept the keyword + # should not have to grow one for a cluster that will never cache a + # session. + kwargs_dict.setdefault('ssl_session_cache', self.ssl_session_cache) kwargs_dict.setdefault('cql_version', self.cql_version) kwargs_dict.setdefault('protocol_version', self.protocol_version) kwargs_dict.setdefault('user_type_map', self._user_types) @@ -1750,6 +1866,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._acquire_tls_session_context() try: self.control_connection.connect() @@ -1839,6 +1956,24 @@ def shutdown(self): if self.metrics_enabled and self.metrics: self.metrics.shutdown() + if self._acquired_tls_session_context is not None: + # Entries left behind would hold this cluster's certificate chain + # and trust store for as long as the cache lives, since a cached + # session keeps its SSLContext alive. They only go once every + # cluster sharing this context has shut down too: clusters sharing + # a cache to share sessions share the context those sessions + # belong to. + cache, ssl_context = self._acquired_tls_session_context + self._acquired_tls_session_context = None + try: + cache.release_context(ssl_context) + except Exception as exc: + # The cache is the caller's object and giving a registration + # back is the last thing shutdown does; a problem in it must + # not leave the cluster half torn down. + log.warning('Could not release the TLS session cache entries ' + 'of this cluster: %s', exc) + _discard_cluster_shutdown(self) def __enter__(self): diff --git a/cassandra/connection.py b/cassandra/connection.py index af95891a3b..f62fe06302 100644 --- a/cassandra/connection.py +++ b/cassandra/connection.py @@ -21,7 +21,7 @@ import socket import struct import sys -from threading import Thread, Event, RLock, Condition +from threading import Thread, Event, Lock, RLock, Condition import time import ssl import uuid @@ -173,6 +173,28 @@ 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:`~.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() + + 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 +309,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 +492,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. @@ -793,6 +825,179 @@ def generate(self, shard_id: int, total_shards: int): DefaultShardAwarePortGenerator = ShardAwarePortGenerator(DEFAULT_LOCAL_PORT_LOW, DEFAULT_LOCAL_PORT_HIGH) +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 a fresh session back, so the entry keeps rolling + forward. 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 may be shared between clusters, and are safe to use from + multiple threads. + """ + + 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() + # SSLContext -> how many clusters are still using it. See + # acquire_context. + self._context_owners = {} + 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): + """ + Store *session* as the session to offer for *key*, replacing any + previous one. A :const:`None` session is ignored. + + :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. + """ + if session is None: + return + expires_at = None if lifetime is None else time.monotonic() + lifetime + with self._lock: + 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) + + 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 fresh session + under the same key, and that one is not the caller's to remove. + """ + 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 acquire_context(self, ssl_context): + """ + Register a user of *ssl_context*, whose sessions are to be kept until + every user has released it again. + + A cached session holds a strong reference to the ``SSLContext`` it was + established with, so an entry keeps that context -- and the certificate + chain, trust store and OpenSSL state reachable from it -- alive, which + is why a cache outliving its clusters cannot simply keep everything. + But sessions can only be replayed onto the context they came from, so + clusters sharing this cache to share sessions are also sharing one + context: dropping a context's sessions as soon as any one of them shuts + down would take sessions the others are still using. Counting the + users is what tells those two cases apart. + """ + with self._lock: + self._context_owners[ssl_context] = \ + self._context_owners.get(ssl_context, 0) + 1 + + def release_context(self, ssl_context): + """ + Give up one registration made by :meth:`acquire_context`, dropping + every session established with *ssl_context* once the last one goes. + + Does nothing for a context that was never registered. Entries keyed + the way :meth:`.Connection._tls_session_cache_key` builds them -- a + tuple naming its context first -- are the only ones that can belong to + a context, so anything else a caller has put in the cache is left + alone. + """ + with self._lock: + owners = self._context_owners.get(ssl_context) + if owners is None: + return + if owners > 1: + self._context_owners[ssl_context] = owners - 1 + return + + del self._context_owners[ssl_context] + for key in [k for k in self._sessions + if isinstance(k, tuple) and k and k[0] is ssl_context]: + 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): + return "<%s max_size=%d size=%d>" % ( + self.__class__.__name__, self._max_size, len(self)) + + class Connection(object): CALLBACK_ERR_THREAD_THRESHOLD = 100 @@ -815,6 +1020,22 @@ 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 + + # 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. @@ -901,13 +1122,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 @@ -1045,17 +1275,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: @@ -1069,6 +1298,186 @@ def _initiate_connection(self, sockaddr): self._socket.connect(sockaddr) + # TLS session resumption. Reactors that do not use the stdlib ssl module + # override the two accessors (_set_tls_session and + # _get_resumable_tls_session); the policy around them is shared. + + 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 and named first in the key: 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, and naming it first lets + # SSLSessionCache.release_context release a departing cluster's entries + # without touching anyone else's. + # 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 + 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. + + 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 fresh one under the same key in the meantime, and + removing that would cost every later connection a full handshake for a + session that never failed anything. + """ + 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 + + 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. + """ + 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) + 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. + + The announced lifetime is taken whole rather than reduced by the + session's age. This connection established the session itself moments + ago, so that age is the length of a CQL handshake against a lifetime of + hours; and ``SSLSession.time`` is a wall-clock stamp, so subtracting it + from ``time.time()`` would let a clock step landing between the + handshake and here decide the answer -- a step forward large enough + makes the remainder zero and caches nothing at all, a step backward + hides whatever age there was. The deadline the cache keeps is + monotonic, so nothing after this point can skew it either. + """ + if session.has_ticket: + lifetime = session.ticket_lifetime_hint + if not lifetime: + if self._socket.version() == 'TLSv1.3': + return None + lifetime = session.timeout + else: + lifetime = session.timeout + + lifetime = min(lifetime, self._MAX_TLS_SESSION_LIFETIME) + return lifetime if lifetime > 0 else None + + def _set_tls_session(self, sock, session): + sock.session = session + + 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, below TLS 1.3, 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, so this is also what defers the store on + # TLS 1.3 without having to recognise a protocol version by name. + 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 @@ -1110,12 +1519,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 and + # no reason to keep hold of what was offered for the life of + # the connection. + self._tls_session_offered = None 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: @@ -1650,6 +2069,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", @@ -1706,6 +2126,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 1d90e3233f..d71448ac56 100644 --- a/cassandra/pool.py +++ b/cassandra/pool.py @@ -710,6 +710,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/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/connection.rst b/docs/api/cassandra/connection.rst index f9ec4eef61..76fc0247c6 100644 --- a/docs/api/cassandra/connection.rst +++ b/docs/api/cassandra/connection.rst @@ -21,3 +21,6 @@ Low Level Connection Info .. autoclass:: SniEndPointFactory .. autoclass:: UnixSocketEndPoint + +.. autoclass:: SSLSessionCache + :members: diff --git a/tests/integration/standard/test_tls_resumption.py b/tests/integration/standard/test_tls_resumption.py new file mode 100644 index 0000000000..2d0fb663ee --- /dev/null +++ b/tests/integration/standard/test_tls_resumption.py @@ -0,0 +1,245 @@ +# 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 datetime +import ipaddress +import logging +import os +import ssl +import tempfile +import unittest + +from cassandra.connection import SSLSessionCache +from tests import EVENT_LOOP_MANAGER +from tests.integration import (use_singledc, get_cluster, remove_cluster, + start_cluster_wait_for_up, SCYLLA_VERSION, + TestCluster) +from tests.util import wait_until + +log = logging.getLogger(__name__) + +_cert_dir = None +_cert_path = None +_key_path = None + + +def _write_self_signed_cert(directory, addresses): + """ + Write a certificate valid for every address in *addresses*, and its key, + into *directory*. Returns ``(cert_path, key_path)``. + + Every node of the cluster has to be covered: the client verifies hostnames, + so a certificate naming only the contact point would leave the driver + unable to build pools to the rest of the cluster. + """ + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.x509.oid import NameOID + + 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 + + +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') + # This reads the environment rather than asking the connection class + # whether it can resume, so it only holds while the reactor is selected + # explicitly, as CI does. Issue #984 tracks keying the skip off + # connection_class.supports_tls_session_resumption instead. + if 'asyncio' in EVENT_LOOP_MANAGER: + raise unittest.SkipTest( + 'the asyncio reactor performs the TLS handshake inside ' + 'loop.create_connection() and cannot resume sessions') + try: + import cryptography # noqa: F401 + except ImportError: + raise unittest.SkipTest( + 'cryptography is required to generate a server certificate') from None + + global _cert_dir, _cert_path, _key_path + _cert_dir = tempfile.TemporaryDirectory(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.name, [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, + # Off by default in Scylla; 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: + _cert_dir.cleanup() + _cert_dir = None + raise + + +def teardown_module(): + try: + remove_cluster() + finally: + if _cert_dir is not None: + _cert_dir.cleanup() + + +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. + """ + return [bool(connection._socket.session_reused) + for holder in cluster.get_connection_holders() + for connection in holder.get_connections()] + + +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/unit/test_cluster.py b/tests/unit/test_cluster.py index 35dc354465..11eed7775f 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -16,6 +16,7 @@ from concurrent.futures import Future import logging import socket +import ssl from types import SimpleNamespace from unittest.mock import patch, Mock @@ -25,7 +26,8 @@ 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, SSLSessionCache) from cassandra.driver_config import DriverConfigReporter from cassandra.pool import Host from cassandra.policies import HostDistance, RetryPolicy, RoundRobinPolicy, DowngradingConsistencyRetryPolicy, SimpleConvictionPolicy @@ -1127,3 +1129,217 @@ 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) + + 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 cache stays reachable and empty either way. + with patch('cassandra.cluster.log') as logger: + with patch('cassandra.cluster.warn'): + self.make_cluster(ssl_options={'ca_certs': '/dev/null'}, + ssl_session_cache=SSLSessionCache()) + + 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): + with patch('cassandra.cluster.log') as logger: + self.make_cluster(connection_class=_NonResumableConnection, + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), + ssl_session_cache=SSLSessionCache()) + + logger.warning.assert_called_once() + assert '_NonResumableConnection' in logger.warning.call_args[0][1] + + 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) + with patch('cassandra.cluster.log') as logger: + self.make_cluster(ssl_context=context, + ssl_session_cache=SSLSessionCache()) + self.make_cluster(ssl_context=context, ssl_session_cache=None) + self.make_cluster() + + logger.warning.assert_not_called() + + 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_shutdown_releases_only_this_clusters_sessions(self): + # A cached session holds its SSLContext alive, so a cluster has to take + # its own entries with it -- and leave a shared cache's others behind. + cache = SSLSessionCache() + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + other_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + cluster = self.make_cluster(ssl_context=context, ssl_session_cache=cache) + # Taken when connections are about to use the cache; connect() does + # this for a cluster that is really used. + cluster._acquire_tls_session_context() + cache.set((context, ('10.0.0.1', 9042), None), object()) + theirs = object() + cache.set((other_context, ('10.0.0.1', 9042), None), theirs) + + cluster.shutdown() + + assert len(cache) == 1 + assert cache.get((other_context, ('10.0.0.1', 9042), None)) is theirs + + def test_a_cache_assigned_after_construction_is_released(self): + # The documented way to get a cache when TLS is configured late. The + # registration is taken when connections are about to use it, not in + # __init__, or a cache assigned afterwards would never be released. + cache = SSLSessionCache() + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + cluster = self.make_cluster(ssl_context=context) + cluster.ssl_session_cache = cache + cluster._acquire_tls_session_context() + cache.set((context, ('10.0.0.1', 9042), None), object()) + + cluster.shutdown() + + assert len(cache) == 0 + + def test_a_cluster_that_never_connects_pins_nothing(self): + # Registering in __init__ would leave a cluster that is constructed and + # dropped holding a share of the context for good, and a shared cache + # could then never drop it. + cache = SSLSessionCache() + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + self.make_cluster(ssl_context=context, ssl_session_cache=cache) + user = self.make_cluster(ssl_context=context, ssl_session_cache=cache) + user._acquire_tls_session_context() + cache.set((context, ('10.0.0.1', 9042), None), object()) + + user.shutdown() + + assert len(cache) == 0 + + def test_release_uses_the_pair_that_was_acquired(self): + # Whatever the attributes say by shutdown, what was taken is what is + # given back. + first, second = SSLSessionCache(), SSLSessionCache() + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + cluster = self.make_cluster(ssl_context=context, ssl_session_cache=first) + cluster._acquire_tls_session_context() + first.set((context, ('10.0.0.1', 9042), None), object()) + cluster.ssl_session_cache = second + + cluster.shutdown() + + assert len(first) == 0 + + def test_shutdown_survives_a_cache_that_raises(self): + # Giving the registration back is the last thing shutdown does, and the + # cache is the caller's object: a problem in it must not leave the + # cluster half torn down. + class Broken(SSLSessionCache): + def release_context(self, ssl_context): + raise RuntimeError('boom') + + cluster = self.make_cluster(ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), + ssl_session_cache=Broken()) + cluster._acquire_tls_session_context() + + with patch('cassandra.cluster._discard_cluster_shutdown') as discard: + with patch('cassandra.cluster.log') as logger: + cluster.shutdown() + + discard.assert_called_once_with(cluster) + logger.warning.assert_called_once() + + 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 5962db1189..2c38f8c3f9 100644 --- a/tests/unit/test_connection.py +++ b/tests/unit/test_connection.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. import itertools +import ssl import unittest import uuid from io import BytesIO @@ -25,12 +26,13 @@ from cassandra.connection import (Connection, HEADER_DIRECTION_TO_CLIENT, ProtocolError, locally_supported_compressions, ConnectionHeartbeat, HeartbeatFuture, _Frame, Timer, TimerManager, ConnectionException, ConnectionShutdown, DefaultEndPoint, ShardAwarePortGenerator, - DRIVER_NAME, DRIVER_VERSION) + DRIVER_NAME, DRIVER_VERSION, SSLSessionCache) from cassandra.driver_config import (DriverConfigReporter, DRIVER_CONFIG_OPTION, DRIVER_CONFIG_SCHEMA_VERSION, SESSION_ID_OPTION) 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 ThrowingReporter @@ -830,6 +832,387 @@ 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, **kwargs): + c = Connection(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_names_the_context_first(self): + # SSLSessionCache.release_context relies on this to release one + # cluster's entries without touching anyone else's. + context, connection = self.make_ssl_connection() + + assert connection._tls_session_cache_key()[0] is context + + 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_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 session was + # established by this connection moments ago, so the announced lifetime + # is what remains. + 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_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_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_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 902b48a276..cf7bc55ec2 100644 --- a/tests/unit/test_shard_aware.py +++ b/tests/unit/test_shard_aware.py @@ -95,6 +95,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..fb43154a43 --- /dev/null +++ b/tests/unit/test_ssl_session_cache.py @@ -0,0 +1,273 @@ +# 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 unittest +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from cassandra.connection import SSLSessionCache + + +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_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_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_releasing_the_last_owner_drops_that_contexts_sessions(self): + # A cache may be shared between clusters, so a departing one must take + # only its own context's entries with it. + cache = SSLSessionCache() + one, other = object(), object() + cache.acquire_context(one) + cache.acquire_context(other) + cache.set((one, ('10.0.0.1', 9042), None), object()) + cache.set((one, ('10.0.0.2', 9042), None), object()) + theirs = object() + cache.set((other, ('10.0.0.1', 9042), None), theirs) + + cache.release_context(one) + + assert len(cache) == 1 + assert cache.get((other, ('10.0.0.1', 9042), None)) is theirs + + def test_sessions_survive_while_another_owner_holds_the_context(self): + # Clusters sharing a cache to share sessions share the context those + # sessions belong to, so one shutting down must not take them. + cache = SSLSessionCache() + context = object() + cache.acquire_context(context) + cache.acquire_context(context) + session = object() + cache.set((context, ('10.0.0.1', 9042), None), session) + + cache.release_context(context) + assert cache.get((context, ('10.0.0.1', 9042), None)) is session + + cache.release_context(context) + assert len(cache) == 0 + + def test_release_leaves_keys_it_could_not_have_created(self): + # A cache is not restricted to the driver's own keys -- the + # concurrency test below uses plain ints -- so anything that is not a + # tuple naming a context cannot belong to one. + cache = SSLSessionCache() + context = object() + cache.acquire_context(context) + for key in (7, 'a string', (), (object(), 'other context')): + cache.set(key, 'not this context') + cache.set((context, ('10.0.0.1', 9042), None), object()) + + cache.release_context(context) + + assert len(cache) == 4 + assert cache.get(7) == 'not this context' + + def test_releasing_an_unregistered_context_does_nothing(self): + cache = SSLSessionCache() + session = object() + cache.set((object(), ('10.0.0.1', 9042), None), session) + + cache.release_context(object()) + + assert len(cache) == 1 + + 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_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..65a7df8ded --- /dev/null +++ b/tests/unit/test_tls_resumption.py @@ -0,0 +1,439 @@ +# 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 datetime +import gc +import ipaddress +import os +import socket +import ssl +import tempfile +import threading +import unittest +import weakref + +import pytest +from unittest.mock import Mock + +from cassandra.connection import Connection, DefaultEndPoint, SSLSessionCache + +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 +except ImportError: # pragma: no cover - depends on the environment + x509 = None + + +def _write_self_signed_cert(directory): + """ + Write a self-signed certificate valid for 127.0.0.1, and its key, into + *directory*. Returns ``(cert_path, key_path)``. + """ + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, '127.0.0.1')]) + 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('127.0.0.1'))]), + critical=False) + .sign(key, hashes.SHA256()) + ) + + cert_path = os.path.join(directory, 'cert.pem') + key_path = os.path.join(directory, 'key.pem') + 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 + + +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(x509 is None, '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_stops_pinning_its_context_once_discarded(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. This is what Cluster.shutdown() + # releases its registration for. + # 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: this is the retention being guarded against. + assert weak() is not None + + self.cache.acquire_context(weak()) + self.cache.release_context(weak()) + gc.collect() + + assert weak() is None + + def test_a_connection_that_came_up_holds_nothing_to_retract(self): + # Kept only for as long as it could be needed: after the handshake + # stands there is nothing to retract, and nothing reads it again. + context = self.make_ssl_context() + self.connect(context)._store_tls_session() + + resumed = self.connect(context) + + assert resumed.session_reused + assert resumed._tls_session_offered is None + + 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_resumed_connections_keep_refreshing_the_cache(self): + 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_session = self.cache.get(key) + + resumed = self.connect(context) + assert resumed.session_reused + resumed._store_tls_session() + + assert self.cache.get(key) is not first_session + + +class Tls13ResumptionTest(TlsResumptionTest): + """ + The same coverage over TLS 1.3, plus what is specific to it. + + This assumes the local OpenSSL offers TLS 1.3; it is not guarded on + ``ssl.HAS_TLSv1_3``, so a build without it fails here rather than skipping. + Issue #984 tracks adding that guard. + """ + + 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