From 05b7044391d41b23afb2cc359015319d3121c8da Mon Sep 17 00:00:00 2001 From: sylwiaszunejko Date: Wed, 12 Aug 2026 13:39:44 +0200 Subject: [PATCH 1/6] Add SSLSessionCache and per-endpoint TLS session cache keys TLS clients can skip the expensive part of a handshake by replaying a session established earlier with the same peer (RFC 5077 tickets for TLS 1.2, RFC 8446 PSKs for TLS 1.3), but OpenSSL never does this on its own: the client has to hold on to the session and offer it explicitly on the next connection. Add the storage half of that: a bounded, thread-safe LRU of TLS sessions keyed by TLS peer identity, plus an EndPoint.tls_session_cache_key property that produces the key. A cached session is not consumed by being used -- one session can be replayed by any number of concurrent connections -- so get() leaves the entry in place and each successful handshake stores a fresh session back over it. Entries carry the lifetime the caller gives them and are dropped once it runs out, so a session is never offered past the point the peer said it would honour it; what that lifetime should be is for the caller to work out, since it depends on how the session resumes. SNI endpoints add the server name to their key, since they all share a proxy address and port but are distinct TLS peers. Client-routes endpoints key on the node's host_id rather than the proxy address they happen to resolve to at the moment. Nothing uses the cache yet. Refs DRIVER-165 --- cassandra/connection.py | 207 +++++++++++++++++++- docs/api/cassandra/connection.rst | 3 + tests/unit/test_endpoints.py | 70 ++++++- tests/unit/test_ssl_session_cache.py | 273 +++++++++++++++++++++++++++ 4 files changed, 551 insertions(+), 2 deletions(-) create mode 100644 tests/unit/test_ssl_session_cache.py diff --git a/cassandra/connection.py b/cassandra/connection.py index af95891a3b..8869a6dafb 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 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/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_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 From c8172b571a38329f96d726d9c1a9acd83e70db52 Mon Sep 17 00:00:00 2001 From: sylwiaszunejko Date: Wed, 12 Aug 2026 15:54:44 +0200 Subject: [PATCH 2/6] Resume TLS sessions on new connections Offer the cached session for the endpoint before the handshake, and store the negotiated session once the connection is up, so that the next connection to the same node -- in particular the burst of per-shard connections a pool opens at once -- can skip the certificate exchange and signature of a full handshake. The session is stored from the ReadyMessage / AuthSuccessMessage handlers rather than right after the handshake. A TLS 1.3 server sends its NewSessionTicket as a post-handshake message, so a session read straight after connect() carries no ticket and would not resume; by the time the CQL handshake has completed the ticket has been read off the socket. Storing is idempotent, so nothing needs to track whether it already happened, and every failure in this path is logged and dropped: resumption is an optimisation, and both call sites are wrapped in @defunct_on_error, where a raised exception would kill a healthy connection. How long a session may be offered is worked out here, because it depends on how the session resumes: a ticket's lifetime is the one the server announced, while SSLSession.timeout is only the local context's default and says nothing about what the peer will still accept, so it is used solely for a session that resumes by id. RFC 8446 section 4.6.1 also caps the client at seven days however long the server asked for. A zero lifetime is read against the negotiated version, since the two RFCs disagree on it: TLS 1.3 says discard the ticket immediately, while RFC 5077 section 3.3 reserves zero for "lifetime unspecified" and leaves retention to local policy, so a TLS 1.2 ticket is kept and timed by the local timeout. OpenSSL does not apply either limit on the client's behalf -- it will offer an expired ticket and let the server refuse it. The announced lifetime is taken whole rather than reduced by the session's age: this connection established the session moments ago, so that age is one CQL handshake, and SSLSession.time is a wall-clock stamp, so subtracting it would let a clock step landing in between decide the answer -- far enough forward and nothing is cached at all. The deadline the cache keeps is monotonic, so nothing after the store can skew it either. A pool reaches a shard-aware node on a second port, which would otherwise key those connections separately from the one the control connection established, leaving the whole per-shard burst to handshake in full. The endpoint alias that _get_shard_aware_endpoint already builds for that port therefore carries the node's cache key, so both listeners share one session and nothing has to be threaded through the connection factory. The port stays part of the key by default, so two unrelated TLS servers on one address still cannot share a session; only an endpoint that names another node is exempt. The context is named first in the key so that the sessions belonging to one can be found and released: a cached session keeps a strong reference to the SSLContext it was established with, so entries left in a cache shared between clusters would hold a departed cluster's certificate chain and trust store. The key also carries the name wrap_socket() is given, which is the name the peer certificate is verified against. A resumed handshake sends 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. Both the key and wrap_socket() take the name from one accessor so the two cannot drift apart. A session offered on a connection whose handshake then failed is dropped from the cache. Both RFCs have a server fall back to a full handshake rather than fail when it will not resume, so this should not happen; but nothing stores a fresh session for a connection that never came up, so an entry that did provoke a failure would otherwise be offered again by every later connection until its lifetime ran out. Only a TLS error counts: a refused or reset connection says nothing about the session. And only the session this connection offered goes: connections to one node are opened together, so another may have stored a fresh one under the same key in the meantime, and that one failed nothing. The session accessors are kept separate from the policy around them, so that a reactor whose TLS does not go through the stdlib ssl module can take part by overriding just those. Connections whose SSLContext is derived from ssl_options do not participate, because a session cannot be replayed onto a different context and each of those connections builds its own. The asyncio reactor opts out entirely: its handshake happens inside loop.create_connection(), with no point at which a session could be restored. Refs DRIVER-165 --- cassandra/connection.py | 232 +++++++++++++++++++- cassandra/io/asyncioreactor.py | 8 + cassandra/pool.py | 7 + tests/unit/test_connection.py | 387 ++++++++++++++++++++++++++++++++- tests/unit/test_shard_aware.py | 21 ++ 5 files changed, 645 insertions(+), 10 deletions(-) diff --git a/cassandra/connection.py b/cassandra/connection.py index 8869a6dafb..f62fe06302 100644 --- a/cassandra/connection.py +++ b/cassandra/connection.py @@ -1020,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. @@ -1106,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 @@ -1250,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: @@ -1274,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 @@ -1315,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: @@ -1855,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", @@ -1911,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/tests/unit/test_connection.py b/tests/unit/test_connection.py index 5962db1189..106b462c48 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/io/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_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) From bfdd105158bb761270afb408f4d9a6235ea46d66 Mon Sep 17 00:00:00 2001 From: sylwiaszunejko Date: Wed, 12 Aug 2026 16:02:50 +0200 Subject: [PATCH 3/6] Enable TLS session resumption from Cluster Create an SSLSessionCache per Cluster whenever TLS is configured through ssl_context, and hand it to every connection the cluster opens, so that resumption is on by default with no configuration. Pass ssl_session_cache=None to turn it off, or an instance of your own to size it or share it between clusters. No cache is created where resumption cannot work: the deprecated ssl_options-only path, whose per-connection SSLContexts a session cannot be replayed onto, and reactors that report they cannot restore a session before the handshake, which today means asyncio. connection_class is not required to derive from Connection, so one that does not report the capability at all is treated as lacking it rather than raising. A cache supplied for one of those configurations is warned about and then dropped. Asking for resumption and silently getting none is worse than not having it: an unusable cache left in place 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, which is also what a server that issues no tickets looks like. So the attribute holds a cache only where one will actually be used, and that is what decides whether connections are given it. Releasing the registration is the last thing shutdown does, and the cache is the caller's object, so a failure there is logged rather than allowed to leave the cluster half torn down. A cluster registers its SSLContext with the cache as it connects and gives that registration back as it shuts down, at which point the sessions established with that context are dropped -- otherwise they would hold its certificate chain and trust store for as long as the cache lived. The registrations are counted rather than assumed to be one apiece: sessions can only be replayed onto the context they came from, so clusters sharing a cache in order to share sessions are also sharing one context, and the first of them to shut down must not take sessions the others are still using. Registering as connections are about to use the cache, rather than in __init__, is what makes the two directions match: both attributes it reads can still be set after construction -- which is how a cluster configured with TLS afterwards gets a cache at all -- and a cluster that is constructed and never connected would otherwise hold a share of the context that nothing gives back. The pair taken is remembered, so shutdown releases that rather than whatever the attributes say by then. Refs DRIVER-165 --- cassandra/cluster.py | 130 +++++++++++++++++++- docs/api/cassandra/cluster.rst | 2 + tests/unit/test_cluster.py | 218 ++++++++++++++++++++++++++++++++- 3 files changed, 347 insertions(+), 3 deletions(-) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 7260bd08b6..7cae39873c 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,43 @@ 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. + """ + sockopts = None """ An optional list of tuples which will be used as arguments to @@ -1166,6 +1204,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 +1262,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 +1510,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 +1751,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 +1792,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 +1857,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 +1947,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/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/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 From 30ebffde4e64c8b7034f76be4e611263181b3bc2 Mon Sep 17 00:00:00 2001 From: sylwiaszunejko Date: Thu, 13 Aug 2026 09:30:42 +0200 Subject: [PATCH 4/6] Test TLS session resumption against a real TLS server Stand up a TLS server on loopback and connect to it with the driver's own socket setup, so the restore-before-handshake and store-after-startup paths run for real and the result is read back the way OpenSSL reports it, through SSLSocket.session_reused. Covers TLS 1.2 and TLS 1.3. Two of these pin down behaviour that is easy to regress: that four connections opened at once all resume from the single cached session -- the per-shard burst DRIVER-165 is about -- and that on TLS 1.3 nothing is cached until the server's NewSessionTicket has actually been read off the socket. Refs DRIVER-165 --- tests/unit/test_connection.py | 2 +- tests/unit/test_tls_resumption.py | 439 ++++++++++++++++++++++++++++++ 2 files changed, 440 insertions(+), 1 deletion(-) create mode 100644 tests/unit/test_tls_resumption.py diff --git a/tests/unit/test_connection.py b/tests/unit/test_connection.py index 106b462c48..2c38f8c3f9 100644 --- a/tests/unit/test_connection.py +++ b/tests/unit/test_connection.py @@ -836,7 +836,7 @@ class TlsSessionResumptionTest(unittest.TestCase): """ Connection-level wiring of :class:`~.SSLSessionCache`. The end-to-end behaviour against a real TLS server lives in - ``tests/unit/io/test_tls_resumption.py``. + ``tests/unit/test_tls_resumption.py``. """ def make_connection(self, **kwargs): 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 From f10d444fb7016f46dc24c6f3e71dbfb2a6a9b75f Mon Sep 17 00:00:00 2001 From: sylwiaszunejko Date: Thu, 13 Aug 2026 10:22:49 +0200 Subject: [PATCH 5/6] Add an integration test for TLS session resumption Restart the cluster with client encryption on, warm a session cache with one cluster, then hand it to a second one and require every connection it opens to have resumed -- which is the question only a real server can answer: whether it accepts one session offered concurrently by the whole batch of per-shard connections. The cluster is given a shard-aware TLS port, since that is the port those per-shard connections use and therefore where resumption has to pay off; Scylla leaves it unset by default. The certificate names every node rather than only the contact point, or the driver could not build pools to the rest of the cluster and the test would quietly examine a single host. Each Session is held for the duration of a test: Cluster.sessions is a WeakSet, so a dropped Session takes its pools -- everything worth inspecting -- with it and leaves only the control connection behind. The number of connections collected is asserted before their resumption flags, so the test cannot pass by examining almost nothing. Follows the reconfigure-and-remove pattern the other modules here use for cluster-level options, and generates the server certificate with cryptography so the test does not depend on an openssl binary. Refs DRIVER-165 --- .../standard/test_tls_resumption.py | 245 ++++++++++++++++++ 1 file changed, 245 insertions(+) create mode 100644 tests/integration/standard/test_tls_resumption.py 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)) From f636dd4466aae9fe86b4623a84f453bad1113e7f Mon Sep 17 00:00:00 2001 From: sylwiaszunejko Date: Thu, 13 Aug 2026 11:01:46 +0200 Subject: [PATCH 6/6] Document the server-side requirement for TLS session resumption Scylla only issues session tickets when enable_session_tickets is set in client_encryption_options, and that is off by default -- without it the cache stays empty and every connection performs a full handshake, with no indication of why. Refs DRIVER-165 --- cassandra/cluster.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 7cae39873c..c228146b60 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -902,6 +902,15 @@ def default_retry_policy(self, policy): ``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