From 72fcb6b1a91fe572a908233b923b0baec3e70ef8 Mon Sep 17 00:00:00 2001 From: sylwiaszunejko Date: Tue, 8 Sep 2026 12:49:05 +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 back over it whatever the peer handed over. 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. Whether a store moves an existing deadline is decided here rather than asked of the caller: a session the peer handed back unchanged keeps the deadline it had, since its lifetime runs from when the peer issued it and not from when it was last replayed, and re-stamping a full lifetime on every reuse would let one ticket be offered for as long as connections keep being opened. Below TLS 1.3 an abbreviated handshake hands back the offered session itself, so that is the ordinary case; TLS 1.3 normally issues a fresh one, which is entitled to a lifetime of its own. The two are told apart by session id, compared under the same lock that stores the result, so no concurrent store can land in between. A caller can also say which session it offered on the handshake it is storing the result of, and a store of that same session is skipped where the entry no longer holds it: another connection has stored a session the peer reissued to it, or the deadline passed and a lookup dropped the entry. In neither case has this caller anything to add, while storing it would put a deadline running from now on a session the peer issued at some earlier point -- and would replace a fresher session with an older one. 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 | 218 ++++++++++++++++++- docs/api/cassandra/connection.rst | 3 + tests/unit/test_endpoints.py | 70 ++++++- tests/unit/test_ssl_session_cache.py | 302 +++++++++++++++++++++++++++ 4 files changed, 591 insertions(+), 2 deletions(-) create mode 100644 tests/unit/test_ssl_session_cache.py diff --git a/cassandra/connection.py b/cassandra/connection.py index b4ea59b23c..1c075c0f7b 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,190 @@ 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 back whatever the peer handed over -- a fresh session + where one was issued, otherwise the same one again, which keeps the + deadline it already had rather than starting a new one. An entry whose lifetime has run out is never handed out again, and + is dropped when it is looked up or when room is needed; entries otherwise + go only by being replaced or, once the cache is full, by having been used + least recently. A session the server declines for any other reason simply + results in a full handshake, which is what would have happened anyway. + + Instances are safe to use from multiple threads, and may be shared + between clusters -- which is what makes sessions outlive the cluster that + established them, so that a cluster replacing an earlier one resumes + instead of handshaking in full. A cache the driver created for a cluster + lives and dies with it; one supplied to :class:`~.Cluster` belongs to + whoever supplied it, and the driver removes an entry from it only to + replace it, because its lifetime ran out, or to make room. Note that an + entry keeps the ``SSLContext`` its session was established with alive -- + CPython's ``SSLSession`` holds a reference to it -- so a long-lived cache + holds the contexts of at most :attr:`max_size` peers. :meth:`clear` drops + everything, for a caller that wants them gone sooner. + """ + + def __init__(self, max_size=1024): + """ + :param max_size: maximum number of peers to keep sessions for. When + exceeded, the least recently used entry is evicted. + """ + # Anything but a positive integer is rejected outright rather than + # compared against: a float such as nan or inf would pass a `< 1` check + # and then leave the cache growing without bound, while True is an int + # that passes it and would quietly cap the cache at one entry. + if (not isinstance(max_size, int) or isinstance(max_size, bool) + or max_size < 1): + raise ValueError( + "max_size must be a positive integer, got %r" % (max_size,)) + self._max_size = max_size + self._sessions = OrderedDict() + self._lock = Lock() + + @property + def max_size(self): + """The maximum number of peers this cache keeps sessions for.""" + return self._max_size + + def get(self, key): + """ + Return the cached session for *key*, or :const:`None` if there is none + or its lifetime has run out. A session that is still live stays in the + cache; an expired one is dropped. + """ + with self._lock: + entry = self._sessions.get(key) + if entry is None: + return None + session, expires_at = entry + if expires_at is not None and time.monotonic() >= expires_at: + del self._sessions[key] + return None + self._sessions.move_to_end(key) + return session + + def set(self, key, session, lifetime=None, offered=None): + """ + Store *session* as the session to offer for *key*, replacing any + previous one. A :const:`None` session is ignored. + + A session the peer handed back unchanged keeps the deadline the entry + already had, rather than starting a new one: its lifetime runs from + when the peer issued it and not from when it was last replayed, so + re-stamping a full lifetime on every reuse would let one ticket be + offered for as long as connections keep being opened. Resuming below + TLS 1.3 is exactly that case -- an abbreviated handshake hands back the + session that was offered, same id and same ticket -- while TLS 1.3 + normally issues a fresh one, which starts its own lifetime. Comparing + the two here is what makes the rule hold: what is cached, what the + caller offered and what is replacing them are all read under the one + lock that also stores the result, so a connection storing concurrently + cannot land in between. + + :param lifetime: how much longer, in seconds, the session may be + offered. Once it has passed, the entry is dropped rather than + returned. :const:`None` means no limit, which callers should + reserve for sessions that carry no lifetime of their own. + :param offered: the session the caller offered on the handshake it is + storing the result of, if any. Storing that same session back when + the entry no longer holds it is not a new session arriving, and is + skipped: see below. + """ + if session is None: + return + expires_at = None if lifetime is None else time.monotonic() + lifetime + with self._lock: + previous = self._sessions.get(key) + if previous is not None and self._is_same_session(previous[0], session): + expires_at = previous[1] + elif offered is not None and self._is_same_session(offered, session): + # The peer handed this caller back the very session it offered, + # but the entry no longer holds it: another connection opened + # alongside stored a session the peer reissued to it, or the + # deadline passed and a lookup dropped the entry. Either way + # this caller has nothing to add -- and storing it would put a + # deadline running from now on a session the peer issued at + # some earlier point, which is the one thing the rule above + # exists to prevent. + return + self._sessions[key] = (session, expires_at) + self._sessions.move_to_end(key) + if len(self._sessions) > self._max_size: + # Whose lifetime has run out and which was used least recently + # are independent once peers announce different lifetimes, so + # evicting purely by recency can drop a live entry and keep a + # dead one. Take the dead ones first. + self._drop_expired_unlocked() + while len(self._sessions) > self._max_size: + self._sessions.popitem(last=False) + + @staticmethod + def _is_same_session(cached, session): + """ + Whether *session* is the one already cached, so that the entry's + deadline is not a new store's to move. + + ``SSLSocket.session`` builds a new object on each access, so identity + cannot answer this on its own; a session id can, and is what tells a + ticket the peer reissued from the one it handed back -- including the + TLS 1.3 server that resumes without issuing one. Something with no id + at all is not a session this can recognise, and is taken to be new. + """ + if cached is session: + return True + cached_id = getattr(cached, 'id', None) + return cached_id is not None and cached_id == getattr(session, 'id', None) + + def _drop_expired_unlocked(self): + now = time.monotonic() + for key in [key for key, (_, expires_at) in self._sessions.items() + if expires_at is not None and now >= expires_at]: + del self._sessions[key] + + def discard(self, key, session=None): + """ + Drop the session cached for *key*, if any. + + Give *session* to drop it only while that is still the cached one. A + caller acting on a session it read earlier needs this: by the time it + decides to drop it, another connection may have stored a 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 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..2ff252f162 --- /dev/null +++ b/tests/unit/test_ssl_session_cache.py @@ -0,0 +1,302 @@ +# 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 time +import unittest +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from cassandra.connection import SSLSessionCache + + +class _Session(object): + """ + Stands in for ssl.SSLSession, which only a real handshake can produce. + Only the session id matters here: it is how the cache tells a session the + peer reissued from the one it handed back unchanged. + """ + + def __init__(self, id=b'\x01' * 32): + self.id = id + + +class SSLSessionCacheTest(unittest.TestCase): + + def test_get_missing_key_returns_none(self): + assert SSLSessionCache().get(('10.0.0.1', 9042)) is None + + def test_set_then_get(self): + cache = SSLSessionCache() + session = object() + cache.set(('10.0.0.1', 9042), session) + + assert cache.get(('10.0.0.1', 9042)) is session + assert cache.get(('10.0.0.2', 9042)) is None + + def test_get_does_not_consume_the_session(self): + # Sessions are replayable: a burst of per-shard connections to one + # node must all be able to offer the same cached session. + cache = SSLSessionCache() + session = object() + cache.set(('10.0.0.1', 9042), session) + + assert [cache.get(('10.0.0.1', 9042)) for _ in range(10)] == [session] * 10 + assert len(cache) == 1 + + def test_set_replaces_the_previous_session(self): + cache = SSLSessionCache() + older, newer = object(), object() + cache.set(('10.0.0.1', 9042), older) + cache.set(('10.0.0.1', 9042), newer) + + assert cache.get(('10.0.0.1', 9042)) is newer + assert len(cache) == 1 + + def test_none_session_is_ignored(self): + cache = SSLSessionCache() + session = object() + cache.set(('10.0.0.1', 9042), session) + cache.set(('10.0.0.1', 9042), None) + + assert cache.get(('10.0.0.1', 9042)) is session + assert len(cache) == 1 + + def test_evicts_least_recently_used_key(self): + cache = SSLSessionCache(max_size=2) + first, second, third = object(), object(), object() + cache.set('first', first) + cache.set('second', second) + + # Touching 'first' makes 'second' the least recently used. + assert cache.get('first') is first + cache.set('third', third) + + assert len(cache) == 2 + assert cache.get('second') is None + assert cache.get('first') is first + assert cache.get('third') is third + + def test_set_refreshes_recency(self): + cache = SSLSessionCache(max_size=2) + cache.set('first', object()) + cache.set('second', object()) + cache.set('first', object()) + cache.set('third', object()) + + assert cache.get('second') is None + assert cache.get('first') is not None + + def test_expired_entry_is_not_returned_and_is_dropped(self): + cache = SSLSessionCache() + cache.set('key', object(), lifetime=-1) + + assert cache.get('key') is None + assert len(cache) == 0 + + def test_live_entry_is_returned(self): + cache = SSLSessionCache() + session = object() + cache.set('key', session, lifetime=3600) + + assert cache.get('key') is session + + def test_the_same_session_again_keeps_the_deadline_it_had(self): + # What a resumed handshake below TLS 1.3 stores back: the session that + # was offered, whose lifetime runs from when the peer issued it and not + # from when it was last replayed. + cache = SSLSessionCache() + cache.set('key', _Session(), lifetime=0.05) + cache.set('key', _Session(), lifetime=3600) + + time.sleep(0.06) + + assert cache.get('key') is None + + def test_a_reissued_session_starts_its_own_deadline(self): + # What TLS 1.3 normally stores back: a fresh ticket, which is entitled + # to the lifetime the peer announced with it. + cache = SSLSessionCache() + cache.set('key', _Session(b'first'), lifetime=-1) + reissued = _Session(b'second') + cache.set('key', reissued, lifetime=3600) + + assert cache.get('key') is reissued + + def test_the_offered_session_does_not_revive_an_entry_that_was_dropped(self): + # The deadline passed between the offer and the store, and a lookup + # dropped the entry. Storing the offered session back would put a full + # fresh lifetime on a session the peer issued long enough ago to have + # expired. + cache = SSLSessionCache() + session = _Session() + cache.set('key', session, lifetime=-1) + assert cache.get('key') is None + + cache.set('key', session, lifetime=3600, offered=session) + + assert cache.get('key') is None + assert len(cache) == 0 + + def test_the_offered_session_does_not_displace_a_siblings(self): + # Connections to one node are opened together: another may have stored + # a session the peer reissued to it between this one's offer and its + # store. That entry is fresher than what this caller has to say. + cache = SSLSessionCache() + offered, reissued = _Session(b'offered'), _Session(b'reissued') + cache.set('key', offered, lifetime=3600) + cache.set('key', reissued, lifetime=3600) + deadline = cache._sessions['key'][1] + + cache.set('key', offered, lifetime=3600, offered=offered) + + assert cache.get('key') is reissued + assert cache._sessions['key'][1] == deadline + + def test_a_reissued_session_still_replaces_a_siblings(self): + # The other side of it: what the peer issued to this connection is new, + # and is entitled to the lifetime announced with it. + cache = SSLSessionCache() + offered, theirs = _Session(b'offered'), _Session(b'theirs') + cache.set('key', theirs, lifetime=3600) + mine = _Session(b'mine') + + cache.set('key', mine, lifetime=3600, offered=offered) + + assert cache.get('key') is mine + + def test_a_session_with_no_id_is_taken_to_be_new(self): + # Nothing the driver caches is in this position -- every SSLSession + # OpenSSL produces carries an id -- so the safe reading of a session + # this cannot recognise is that it replaces what was there. + cache = SSLSessionCache() + cache.set('key', object(), lifetime=-1) + session = object() + cache.set('key', session, lifetime=3600) + + assert cache.get('key') is session + + def test_a_lifetime_replaces_the_previous_one(self): + cache = SSLSessionCache() + cache.set('key', object(), lifetime=-1) + session = object() + cache.set('key', session, lifetime=3600) + + assert cache.get('key') is session + + def test_a_dead_entry_is_evicted_before_a_live_one(self): + # Whose lifetime has run out and which was used least recently are + # independent once peers announce different lifetimes. + cache = SSLSessionCache(max_size=3) + cache.set('live-1', 'A', lifetime=3600) + cache.set('live-2', 'B', lifetime=3600) + cache.set('expired', 'C', lifetime=-1) + + cache.set('fourth', 'D', lifetime=3600) + + assert cache.get('live-1') == 'A' + assert cache.get('live-2') == 'B' + assert cache.get('fourth') == 'D' + assert len(cache) == 3 + + def test_the_lru_still_goes_when_nothing_has_expired(self): + cache = SSLSessionCache(max_size=2) + cache.set('first', 'A', lifetime=3600) + cache.set('second', 'B', lifetime=3600) + + cache.set('third', 'C', lifetime=3600) + + assert cache.get('first') is None + assert cache.get('second') == 'B' + assert cache.get('third') == 'C' + + def test_a_dead_entry_lingers_until_it_is_looked_up_or_room_is_needed(self): + # Documented rather than swept eagerly: nothing walks the cache on a + # timer, so an entry nobody asks for and nobody needs room for stays. + cache = SSLSessionCache(max_size=8) + cache.set('expired', 'C', lifetime=-1) + + assert len(cache) == 1 + assert cache.get('expired') is None + assert len(cache) == 0 + + def test_discard(self): + cache = SSLSessionCache() + cache.set('key', object()) + cache.discard('key') + + assert cache.get('key') is None + assert len(cache) == 0 + cache.discard('key') # discarding what is not there is fine + + def test_discard_of_a_named_session_spares_a_newer_one(self): + # A connection acting on a session it read earlier must not remove the + # fresh one another connection stored under the same key meanwhile. + cache = SSLSessionCache() + older, newer = object(), object() + cache.set('key', older) + cache.set('key', newer) + + cache.discard('key', older) + + assert cache.get('key') is newer + + def test_discard_of_a_named_session_removes_it_when_still_current(self): + cache = SSLSessionCache() + session = object() + cache.set('key', session) + + cache.discard('key', session) + + assert cache.get('key') is None + + def test_clear(self): + cache = SSLSessionCache() + cache.set('key', object()) + cache.clear() + + assert len(cache) == 0 + assert cache.get('key') is None + + def test_rejects_invalid_max_size(self): + # A float would pass a plain `< 1` check and then never bound the cache + # (nan and inf compare False against every limit), and True is an int + # that passes it and would cap the cache at a single entry. + for max_size in (0, -1, float('nan'), float('inf'), 2.5, '8', None, + True, False): + with pytest.raises(ValueError): + SSLSessionCache(max_size=max_size) + + def test_repr(self): + cache = SSLSessionCache(max_size=7) + cache.set('key', object()) + + assert repr(cache) == '' + + def test_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 f199f72b44ec03f247244b2b986a1dbf4513b68b Mon Sep 17 00:00:00 2001 From: sylwiaszunejko Date: Tue, 8 Sep 2026 12:49:05 +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. The version also rules out caching a TLS 1.3 session that carries only an id: resumption there is the ticket's pre-shared key, while the id a TLS 1.3 handshake carries is the legacy_session_id_echo a server sends back for the middlebox compatibility mode of RFC 8446 appendix D.4, which resumes nothing. OpenSSL reports no id at all until a NewSessionTicket has been read, so that session is not one it produces; stating the rule against the negotiated version rather than against what one library exposes is what makes it hold regardless. 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 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 there: a cached session already keeps its context alive on its own -- CPython's SSLSession holds a reference to the context it was established with -- so holding it weakly here would buy nothing. The 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. What was offered is kept past a handshake that succeeded as well, because the store hands it to the cache. Telling a session the peer reissued from the one that came back unchanged needs both sides of the exchange, and where connections to a node are opened together only the connection that offered one knows the second. Three accessors are the whole of what a reactor whose TLS does not go through the stdlib ssl module has to reimplement to take part: the policy around them asks one for the session to store, one for the negotiated version and one to restore a session onto a socket, and reads nothing off a socket itself. 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 | 257 ++++++++++++++++++- cassandra/io/asyncioreactor.py | 8 + cassandra/pool.py | 7 + tests/unit/test_connection.py | 453 ++++++++++++++++++++++++++++++++- tests/unit/test_shard_aware.py | 21 ++ 5 files changed, 736 insertions(+), 10 deletions(-) diff --git a/cassandra/connection.py b/cassandra/connection.py index 1c075c0f7b..20678347e6 100644 --- a/cassandra/connection.py +++ b/cassandra/connection.py @@ -1031,6 +1031,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. @@ -1157,13 +1173,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 @@ -1303,17 +1328,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: @@ -1327,6 +1351,211 @@ def _initiate_connection(self, sockaddr): self._socket.connect(sockaddr) + # TLS session resumption. Everything a reactor establishing TLS by other + # means than the stdlib ssl module has to reimplement is in the three + # accessors below -- _set_tls_session, _get_resumable_tls_session and + # _tls_negotiated_version -- and nothing else here touches the socket, so + # the policy around them is shared by every reactor that has them. + + def _tls_server_hostname(self): + """ + The name ``wrap_socket`` is given, which is the name the peer + certificate is verified against when the context checks hostnames. + + PYTHON-1186: the endpoint's ssl_options may provide it (an SNI proxy + needs it for routing); otherwise it is the endpoint address, and only + when the context actually checks hostnames. + """ + if 'server_hostname' in self.ssl_options: + return self.ssl_options['server_hostname'] + if getattr(self.ssl_context, 'check_hostname', False): + return self.endpoint.address + return None + + def _tls_session_cache_key(self): + # The SSLContext is part of the key because a session cannot be + # replayed onto a different one, and a cache may be shared by several + # clusters. It is held strongly: a cached session already keeps its + # context alive on its own -- CPython's SSLSession holds a reference to + # the context it was established with -- so holding it weakly here + # would buy nothing. + # The verified name is part of it because a resumed handshake carries no + # Certificate, so that name is never checked again: offering a session + # to a connection expecting a different name would silently skip + # hostname verification for it. Deriving the name from the same place + # _wrap_socket_from_context does is what keeps the two from drifting. + return (self.ssl_context, self.endpoint.tls_session_cache_key, + self._tls_server_hostname()) + + def _restore_tls_session(self, sock): + """ + Offer the session cached for this endpoint, if any, on *sock*, which + must not have begun its handshake yet. Not offering one only costs a + full handshake, so failures here are logged and ignored. + """ + # Set below only if a session is actually offered, so that this always + # describes the attempt in flight: _connect_socket may come back here + # for another address, and an earlier attempt's session is not this + # one's to retract. + self._tls_session_offered = None + 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, + offered=self._tls_session_offered) + log.debug("Cached the TLS session of %s for resumption, for %ss", + self.endpoint, int(lifetime)) + except Exception as exc: + log.debug("Could not cache the TLS session of %s: %s", self.endpoint, exc) + + def _tls_session_lifetime(self, session): + """ + How much longer, in seconds, *session* may be offered, or ``None`` if + it must not be cached at all. + + A ticket's lifetime is the one the server announced; + ``SSLSession.timeout`` is the local context's default and says nothing + about what the peer will still accept, so it is only used where the + server announced nothing. RFC 8446 section 4.6.1 also caps a client at + seven days however long a lifetime the server asked for. + + A zero lifetime means opposite things in the two RFCs that define + tickets, so the negotiated version has to decide: RFC 8446 section 4.6.1 + (TLS 1.3) says discard the ticket immediately, while RFC 5077 section 3.3 + (TLS 1.2) reserves zero for "lifetime unspecified" and leaves retention + to local policy -- for which the local timeout is the only figure + available. + + 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._tls_negotiated_version() == 'TLSv1.3': + return None + lifetime = session.timeout + elif self._tls_negotiated_version() == 'TLSv1.3': + # TLS 1.3 resumes only from a ticket, whose pre-shared key is the + # whole mechanism; the session id a TLS 1.3 handshake carries is + # legacy_session_id_echo (RFC 8446 section 4.1.3), which a server + # echoes for the middlebox compatibility mode of appendix D.4 and + # which resumes nothing. OpenSSL does not report such an id as the + # session's -- one appears only once a NewSessionTicket has been + # read, which is what defers the store until then -- so this is + # unreachable there; it is here so the rule follows from the + # protocol rather than from what one library chooses to expose. + return None + else: + lifetime = session.timeout + + lifetime = min(lifetime, self._MAX_TLS_SESSION_LIFETIME) + return lifetime if lifetime > 0 else None + + def _set_tls_session(self, sock, session): + sock.session = session + + def _tls_negotiated_version(self): + """ + The name of the TLS version in force on this connection, as + ``SSLSocket.version`` reports it -- ``'TLSv1.3'`` and so on -- or + :const:`None` if there is no handshake to ask about. + + Only the retention rules need this: which RFC defines the tickets the + peer issues, and so what a lifetime of zero in one means, is decided by + the version, and no property of the session itself distinguishes them. + """ + return self._socket.version() + + def _get_resumable_tls_session(self): + session = getattr(self._socket, 'session', None) + if session is None: + return None + # There has to be something to offer on the next connection: a ticket + # (RFC 5077 for TLS 1.2, RFC 8446 for TLS 1.3) or a session id. A TLS + # 1.3 server sends its NewSessionTicket after the handshake as a + # separate message, and until that has been read the session carries + # neither, which is what keeps an empty one from being stored. Whether + # an id alone is worth anything is not decided here: that depends on + # the negotiated version, which _tls_session_lifetime reads. + if not (session.has_ticket or session.id): + return None + return session + # PYTHON-1331 # # Allow implementations specific to an event loop to add additional behaviours @@ -1368,12 +1597,22 @@ def _connect_socket(self): # run that here. if self._check_hostname: self._validate_hostname() + # The handshake stood, so there is nothing left to retract -- + # but what was offered is kept, because the store still reads + # it to tell a session the peer reissued from the one this + # connection offered and got back unchanged. sockerr = None break except socket.error as err: if self._socket: self._socket.close() self._socket = None + # Only for a TLS failure: a connection refused or reset says + # nothing about the session, and dropping it would cost a later + # connection a full handshake for no reason. Whether anything + # was offered to retract is _discard_tls_session's own business. + if isinstance(err, ssl.SSLError): + self._discard_tls_session() sockerr = err if sockerr: @@ -1914,6 +2153,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", @@ -1970,6 +2210,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 fcea10dfaf..d8c7f1fad8 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,11 +26,12 @@ 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 DRIVER_CONFIG_OPTION, 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 StubReporter, ThrowingReporter @@ -906,6 +908,453 @@ 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_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_hands_the_cache_the_session_that_was_offered(self): + # What this connection offered is half of the decision the cache makes, + # so it has to reach it: without it, a connection whose entry was + # dropped meanwhile re-creates it with a fresh full lifetime, on a + # session the peer issued long enough ago to have expired. + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + session = Mock(has_ticket=True, id=b'\x01' * 32, ticket_lifetime_hint=7200, + time=time.time(), timeout=7200) + connection._socket.session = session + # The state a resumed connection is in by the time it stores: it + # offered this session, and the entry has since gone. + connection._tls_session_offered = session + + connection._store_tls_session() + + assert cache.get(connection._tls_session_cache_key()) is None + assert len(cache) == 0 + + def test_store_skips_an_id_only_session_on_tls13(self): + # TLS 1.3 resumes only from a ticket, so an id on its own is nothing to + # offer however it got there. OpenSSL reports no id until a ticket has + # been read, so this state does not arise with it -- see + # test_the_session_is_only_stored_once_the_ticket_has_arrived in + # tests/unit/test_tls_resumption.py, which holds that against a real + # server -- and the rule is asserted here so it does not rest on that. + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + connection._socket.session = Mock(has_ticket=False, id=b'\x01' * 32, + ticket_lifetime_hint=0, + time=time.time(), timeout=7200) + connection._socket.version.return_value = 'TLSv1.3' + + connection._store_tls_session() + + assert cache.get(connection._tls_session_cache_key()) is None + + def test_store_skips_a_session_with_nothing_to_offer(self): + # This is a TLS 1.3 session read before the server's NewSessionTicket + # has arrived: no ticket and no id, so it could never resume and must + # not displace a usable entry. + cache = SSLSessionCache() + _, connection = self.make_ssl_connection(cache) + cache.set(connection._tls_session_cache_key(), 'earlier-session') + connection._socket.session = Mock(has_ticket=False, id=b'') + + connection._store_tls_session() + + assert cache.get(connection._tls_session_cache_key()) == 'earlier-session' + + def test_store_tolerates_a_failure(self): + _, connection = self.make_ssl_connection() + connection._ssl_session_cache.set = Mock(side_effect=RuntimeError('boom')) + connection._socket.session = Mock(has_ticket=True, id=b'', + ticket_lifetime_hint=7200, + time=time.time(), timeout=7200) + + # _store_tls_session runs inside @defunct_on_error-wrapped handlers; + # a caching failure must never take the connection down. + connection._store_tls_session() + + # Asserted so the failure has to come from the cache: a session the + # accessors choke on would raise before ever reaching it, and the test + # would pass without covering what it names. + connection._ssl_session_cache.set.assert_called_once() + + def test_store_is_a_no_op_without_a_cache(self): + connection = self.make_connection() + + connection._store_tls_session() + + def test_the_accessors_are_the_whole_reactor_specific_surface(self): + # A reactor that establishes TLS by other means than an ssl.SSLSocket + # has no socket to read a session, a ticket or a version off, and + # reimplements the three accessors instead. Nothing else in the policy + # may reach for a socket, so this connection deliberately has none: + # anything that did would come back with no session cached and none + # offered. + session = Mock(has_ticket=True, id=b'\x01' * 32, + ticket_lifetime_hint=0, time=time.time(), timeout=7200) + + class OwnTransport(Connection): + offered = None + + def _set_tls_session(self, sock, restored): + self.offered = restored + + def _get_resumable_tls_session(self): + return session + + def _tls_negotiated_version(self): + # A zero ticket lifetime means "unspecified" here, not + # "discard", so the session is cached with the local timeout. + return 'TLSv1.2' + + cache = SSLSessionCache() + connection = OwnTransport( + DefaultEndPoint('1.2.3.4'), + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), + ssl_session_cache=cache) + + connection._store_tls_session() + assert cache.get(connection._tls_session_cache_key()) is session + + connection._restore_tls_session(sock=None) + assert connection.offered is session + + def test_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 af27a84011..4499d2eae5 100644 --- a/tests/unit/test_shard_aware.py +++ b/tests/unit/test_shard_aware.py @@ -139,6 +139,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 548b06a30480b1810df89cba474a2f558534e458 Mon Sep 17 00:00:00 2001 From: sylwiaszunejko Date: Tue, 8 Sep 2026 15:52:18 +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 -- which is also what the default connection class resolves to on Python 3.12 and newer with no libev extension installed, asyncore having left the standard library there. 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. That is settled at construction, so the attribute reads as documented straight away, and again from connect(), against whatever ssl_context and connection_class are in force by then. Both are public attributes: a decision kept from the constructor would leave resumption off on a reactor that does support it, or hand the keyword to a connection class that does not take it. What the caller asked for is remembered, so a cache supplied and then declined is restored rather than lost if a later decision turns resumption back on, and an explicit None still means off. A cache put there for one of those configurations is warned about and then dropped, whether it was passed to the constructor or assigned to the attribute afterwards: what is decided is the attribute as it stands, so both arrive at the same place. 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. The reason is given once for the cluster: both decisions usually see the same pair, and saying it again at connect() would be noise. Whose the cache is settles what becomes of it, and that is recorded at construction. One created here is reachable only through the attribute, so it and the sessions in it go when the cluster does and shutdown has nothing to do. One the caller supplied stays the caller's: shutdown leaves its entries alone, which is what lets clusters share sessions -- at the same time, or one after another, so that a cluster replacing an earlier one resumes rather than handshaking in full -- and keeps the driver from deleting rows in an object it does not own. Its entries hold the SSLContext their session was established with, bounded by the cache's max_size, and clear() is there for a caller who wants them gone sooner. Refs DRIVER-165 --- cassandra/cluster.py | 138 ++++++++++++++++++- docs/api/cassandra/cluster.rst | 2 + tests/unit/test_cluster.py | 244 ++++++++++++++++++++++++++++++++- 3 files changed, 381 insertions(+), 3 deletions(-) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 57fcf46331..ff74d07e67 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,60 @@ 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 is settled again when :meth:`~.Cluster.connect` is called, against + whatever :attr:`~Cluster.ssl_context` and :attr:`~Cluster.connection_class` + are in force by then, so configuring TLS after construction still gets a + cache -- and swapping in a connection class that cannot resume still turns + resumption off rather than handing the class a keyword it does not take. + + A cache created here is reachable only through this attribute, so it and + the sessions in it go when the cluster does. A cache passed in stays the + caller's: :meth:`~.Cluster.shutdown` leaves its entries alone, so several + clusters -- at the same time or one after another -- can share the + sessions in it. Its entries hold the ``SSLContext`` they were established + with, bounded by the cache's + :attr:`~cassandra.connection.SSLSessionCache.max_size`; call + :meth:`~cassandra.connection.SSLSessionCache.clear` to release them. + + Assigning this attribute is honoured up to :meth:`~.Cluster.connect`, + which is where the decision is settled: a cache put here that cannot be + used is dropped rather than left to fill with nothing, and the reason is + logged once for the cluster. + + 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`` reactor, and ``asyncore`` + on the Python versions that still ship it, which is up to 3.11. Which of + them is the default depends on what can be imported -- libev first, then + asyncore, then asyncio -- so on Python 3.12 and newer without the libev + extension the default is the ``asyncio`` reactor, and resumption is off. + + 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 @@ -1221,7 +1276,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 +1524,14 @@ def __init__(self, self.ssl_options = ssl_options self.ssl_context = ssl_context + + self._ssl_session_cache_warned = False + self._ssl_session_cache_explicit = ssl_session_cache is not _NOT_SET + self._ssl_session_cache_requested = ( + ssl_session_cache if self._ssl_session_cache_explicit else None) + self.ssl_session_cache = None + self._decide_tls_session_cache() + # Materialized once: these are applied to every socket the cluster opens # and are read again to build the configuration report, so a one-shot # iterable would leave whichever consumer ran second with nothing at all. @@ -1680,6 +1744,62 @@ 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 _decide_tls_session_cache(self): + """ + Settle whether this cluster caches TLS sessions, and in what. + + Run at construction, so that :attr:`ssl_session_cache` reads as + documented straight away, and again from :meth:`connect`, because both + inputs are public attributes that can be reassigned in between: a + decision made from the pair given to the constructor would leave + resumption off on a reactor that does support it, or hand the keyword + to a connection class that does not take it. + + Resumption needs the session to be replayable onto the same + ``SSLContext``, and a reactor that gives the driver a chance to offer it + before the handshake. connection_class is not required to derive from + Connection, so one that does not report the capability is treated as + lacking it. + """ + resumable = (self.ssl_context is not None and + getattr(self.connection_class, + 'supports_tls_session_resumption', False)) + + if resumable: + if self.ssl_session_cache is None: + self.ssl_session_cache = (self._ssl_session_cache_requested + if self._ssl_session_cache_explicit + else SSLSessionCache()) + return + + wanted = (self.ssl_session_cache is not None + or self._ssl_session_cache_requested is not None) + if not wanted or self._ssl_session_cache_warned: + self.ssl_session_cache = None + return + + # 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 self.ssl_context is None: + reason = ('no ssl_context is configured, and a session cannot be ' + 'replayed onto the fresh context each connection builds ' + 'from ssl_options') + else: + reason = ('%s cannot restore a session before the handshake' % + getattr(self.connection_class, '__name__', + self.connection_class)) + log.warning('ssl_session_cache is set 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._ssl_session_cache_warned = True + def connection_factory(self, endpoint, host_conn = None, *args, **kwargs): """ Called to create a new connection with proper configuration. @@ -1701,6 +1821,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) @@ -1760,6 +1886,9 @@ 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) + # Both inputs are public attributes, so the decision is settled + # against the pair actually in force before anything is opened. + self._decide_tls_session_cache() try: self.control_connection.connect() @@ -1849,6 +1978,11 @@ def shutdown(self): if self.metrics_enabled and self.metrics: self.metrics.shutdown() + # Nothing to do here for ssl_session_cache: a cache created for this + # cluster is reachable only through it and goes when it does, and a + # cache the caller supplied is the caller's to empty -- deleting rows + # in it here would defeat sharing one so that sessions outlive a + # cluster. See the attribute's documentation. _discard_cluster_shutdown(self) def __enter__(self): diff --git a/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 74ed346c68..7d66069ffb 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -14,8 +14,11 @@ import unittest from concurrent.futures import Future +import gc import logging import socket +import ssl +import weakref from types import SimpleNamespace from unittest.mock import patch, Mock @@ -25,7 +28,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 @@ -1167,3 +1171,241 @@ 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_warned_about_once(self): + # __init__ decides and says so; connect() decides again against the + # same pair, and repeating itself would only be noise. + with patch('cassandra.cluster.log') as logger: + cluster = self.make_cluster( + connection_class=_NonResumableConnection, + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), + ssl_session_cache=SSLSessionCache()) + cluster._decide_tls_session_cache() + + logger.warning.assert_called_once() + + def test_warns_about_a_cache_assigned_to_a_cluster_that_cannot_use_it(self): + # Nothing was supplied at construction, so nothing was decided about a + # cache then; connect() settles it against the attribute as it stands, + # and a cache put there by hand is denied as loudly as a supplied one. + cluster = self.make_cluster(connection_class=_NonResumableConnection, + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)) + cluster.ssl_session_cache = SSLSessionCache() + + with patch('cassandra.cluster.log') as logger: + cluster._decide_tls_session_cache() + + logger.warning.assert_called_once() + assert '_NonResumableConnection' in logger.warning.call_args[0][1] + assert cluster.ssl_session_cache is None + + def test_an_unusable_cache_is_not_kept_or_passed_on(self): + # Warning and then handing the cache to every connection anyway is the + # worst of both: a connection class that does not take the keyword + # cannot even be constructed. + with patch('cassandra.cluster.log'): + cluster = self.make_cluster( + connection_class=_NonResumableConnection, + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), + ssl_session_cache=SSLSessionCache()) + + assert cluster.ssl_session_cache is None + kwargs = cluster._make_connection_kwargs(DefaultEndPoint('127.0.0.1'), {}) + assert 'ssl_session_cache' not in kwargs + + def test_does_not_warn_where_resumption_works_or_was_declined(self): + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + 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_leaves_a_supplied_cache_alone(self): + # The cache belongs to whoever passed it in, and the point of passing + # one in is that its sessions outlive a cluster: a cluster replacing + # this one resumes rather than handshaking in full. A cache created + # for a cluster needs no shutdown hook either -- it is reachable only + # through the cluster, so it goes when the cluster does. + cache = SSLSessionCache() + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + cluster = self.make_cluster(ssl_context=context, ssl_session_cache=cache) + session = object() + cache.set((context, ('10.0.0.1', 9042), None), session) + + cluster.shutdown() + + assert cache.get((context, ('10.0.0.1', 9042), None)) is session + + def test_a_cache_created_here_goes_when_the_cluster_does(self): + # Built without make_cluster, whose addCleanup would hold the cluster. + cluster = Cluster(connection_class=_ResumableConnection, + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)) + cache = weakref.ref(cluster.ssl_session_cache) + cluster.shutdown() + + del cluster + gc.collect() + + assert cache() is None + + def test_the_decision_follows_a_connection_class_that_can_resume(self): + # Both inputs are public attributes, so a decision kept from the + # constructor would leave resumption off on a reactor that supports it. + cluster = self.make_cluster(connection_class=_NonResumableConnection, + ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)) + assert cluster.ssl_session_cache is None + + cluster.connection_class = _ResumableConnection + cluster._decide_tls_session_cache() + + assert isinstance(cluster.ssl_session_cache, SSLSessionCache) + + def test_the_decision_follows_a_connection_class_that_cannot(self): + # Otherwise the keyword goes to a class that may not take it. + cluster = self.make_cluster(ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)) + assert cluster.ssl_session_cache is not None + + cluster.connection_class = _NonResumableConnection + with patch('cassandra.cluster.log'): + cluster._decide_tls_session_cache() + + assert cluster.ssl_session_cache is None + kwargs = cluster._make_connection_kwargs(DefaultEndPoint('127.0.0.1'), {}) + assert 'ssl_session_cache' not in kwargs + + def test_the_decision_follows_a_context_set_after_construction(self): + cluster = self.make_cluster() + assert cluster.ssl_session_cache is None + + cluster.ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + cluster._decide_tls_session_cache() + + assert isinstance(cluster.ssl_session_cache, SSLSessionCache) + + def test_a_declined_cache_stays_declined_when_re_decided(self): + cluster = self.make_cluster(ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), + ssl_session_cache=None) + + cluster._decide_tls_session_cache() + + assert cluster.ssl_session_cache is None + + def test_a_supplied_cache_survives_being_re_decided(self): + cache = SSLSessionCache() + cluster = self.make_cluster(ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), + ssl_session_cache=cache) + + cluster._decide_tls_session_cache() + + assert cluster.ssl_session_cache is cache + + 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 72eb00b31792439207fd1f3290c9f7dd0c300f9c 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, the latter skipped where the local OpenSSL does not offer it -- skipping the subclass rather than the base, since a skipped base would take its subclasses with it. 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 | 452 ++++++++++++++++++++++++++++++ 2 files changed, 453 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 d8c7f1fad8..789ecd75c1 100644 --- a/tests/unit/test_connection.py +++ b/tests/unit/test_connection.py @@ -912,7 +912,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..315ecc3b48 --- /dev/null +++ b/tests/unit/test_tls_resumption.py @@ -0,0 +1,452 @@ +# 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_pins_its_context_until_the_cache_drops_it(self): + # A real SSLSession holds a strong reference to the SSLContext it was + # established with, so an entry keeps that context -- and everything + # reachable from it -- alive. A cache the driver created for a cluster + # is dropped with it; one the caller supplied outlives it, holding the + # contexts of at most max_size peers until clear() drops them. + # Built directly rather than through self.connect(), whose bookkeeping + # would hold the connection, and so the context, itself. + context = self.make_ssl_context() + connection = _SocketOnlyConnection( + DefaultEndPoint('127.0.0.1', self.server.port), + ssl_context=context, ssl_session_cache=self.cache, + connect_timeout=10) + connection.exchange() + connection._store_tls_session() + key = connection._tls_session_cache_key() + assert self.cache.get(key) is not None + weak = weakref.ref(context) + + connection.close() + del context, connection, key + gc.collect() + # The entry still holds it, which is the retention the cache documents. + assert weak() is not None + + self.cache.clear() + gc.collect() + + assert weak() is None + + def test_a_connection_that_resumed_remembers_what_it_offered(self): + # Nothing is left to retract once the handshake stands, but the store + # reads what was offered: handing it to the cache is what tells a + # session the peer reissued from the one that came back unchanged. + context = self.make_ssl_context() + first = self.connect(context) + first._store_tls_session() + offered = self.cache.get(first._tls_session_cache_key()) + + resumed = self.connect(context) + + assert resumed.session_reused + assert resumed._tls_session_offered.id == offered.id + + def test_an_attempt_that_offers_nothing_clears_what_came_before(self): + context = self.make_ssl_context() + connection = self.connect(context) + connection._tls_session_offered = 'from an earlier address' + + # Nothing cached for this key, so nothing is offered. + connection._restore_tls_session(Mock()) + + assert connection._tls_session_offered is None + + def test_a_failed_handshake_drops_the_session_it_offered(self): + # Nothing stores a session for a connection that never came up, so an + # entry that provokes a handshake failure would be offered again by + # every later connection until its lifetime ran out. + context = self.make_ssl_context() + donor = self.connect(context) + donor._store_tls_session() + + rejecting = self.untrusted_server() + endpoint = DefaultEndPoint('127.0.0.1', rejecting.port) + # A Connection built without connecting, just to ask for the key the + # failing connection below will use. + key = Connection(endpoint, ssl_context=context, + ssl_session_cache=self.cache)._tls_session_cache_key() + self.cache.set(key, self.cache.get(donor._tls_session_cache_key())) + assert self.cache.get(key) is not None + + # _connect_socket re-raises as socket.error(errno, ...), so the + # SSLError type does not survive -- only its message. + with pytest.raises(OSError, match='SSL'): + _SocketOnlyConnection(endpoint, ssl_context=context, + ssl_session_cache=self.cache, connect_timeout=10) + + assert self.cache.get(key) is None + + def test_a_failed_handshake_spares_a_session_stored_meanwhile(self): + # Connections to one node are opened together, so another may store a + # fresh session under this key between the offer and the failure. That + # one did not fail anything and has to stay. + context = self.make_ssl_context() + donor = self.connect(context) + donor._store_tls_session() + + rejecting = self.untrusted_server() + endpoint = DefaultEndPoint('127.0.0.1', rejecting.port) + key = Connection(endpoint, ssl_context=context, + ssl_session_cache=self.cache)._tls_session_cache_key() + self.cache.set(key, self.cache.get(donor._tls_session_cache_key())) + + # Stand in for the connection that succeeds while this one is failing. + class Refresher(_SocketOnlyConnection): + def _set_tls_session(self, sock, session): + super()._set_tls_session(sock, session) + self._ssl_session_cache.set(key, 'stored-by-another-connection') + + with pytest.raises(OSError, match='SSL'): + Refresher(endpoint, ssl_context=context, + ssl_session_cache=self.cache, connect_timeout=10) + + assert self.cache.get(key) == 'stored-by-another-connection' + + def test_a_session_is_not_offered_to_a_different_server_name(self): + # A resumed handshake carries no Certificate, so the name the peer was + # verified against is never checked again. A session established for + # one name must therefore never be offered to a connection expecting + # another, even though both reach the same address and port. + context = self.make_ssl_context() + context.check_hostname = False + self.connect(context, ssl_options={'server_hostname': 'one.example'})._store_tls_session() + + same_name = self.connect(context, ssl_options={'server_hostname': 'one.example'}) + other_name = self.connect(context, ssl_options={'server_hostname': 'other.example'}) + + assert same_name.session_reused + assert not other_name.session_reused + + def test_a_session_is_not_offered_to_a_different_context(self): + # A session can only be replayed onto the context it was established + # with -- the stdlib ssl module rejects anything else -- so the context + # is part of the cache key. + self.connect(self.make_ssl_context())._store_tls_session() + + second = self.connect(self.make_ssl_context()) + + assert not second.session_reused + + def test_what_a_resumed_handshake_stores_back(self): + # SSLSocket.session builds a new object on every access, so comparing + # object identity here would pass whatever happened. What matters is + # whether the peer issued a new session: below TLS 1.3 an abbreviated + # handshake hands back the one that was offered, and its deadline has + # to stay where it was rather than start again on every reuse. + context = self.make_ssl_context() + first = self.connect(context) + first._store_tls_session() + # Ask the connection for its key rather than rebuilding it here, so this + # test does not depend on the key's shape. + key = first._tls_session_cache_key() + first_id = self.cache.get(key).id + first_deadline = self.cache._sessions[key][1] + + resumed = self.connect(context) + assert resumed.session_reused + resumed._store_tls_session() + + renewed = self.tls_version >= ssl.TLSVersion.TLSv1_3 + assert (self.cache.get(key).id != first_id) is renewed + assert (self.cache._sessions[key][1] > first_deadline) is renewed + + +@unittest.skipUnless(ssl.HAS_TLSv1_3, 'this build of OpenSSL has no TLS 1.3') +class Tls13ResumptionTest(TlsResumptionTest): + """ + The same coverage over TLS 1.3, plus what is specific to it. + + Every inherited test drives a TLS 1.3 server, so a build without it has to + skip the class rather than fail each handshake. The guard sits here and not + on the base class: skipping a base skips its subclasses too, which would + take these tests out on a build that has TLS 1.3 but not 1.2. + """ + + tls_version = ssl.TLSVersion.TLSv1_3 + + def test_the_session_is_only_stored_once_the_ticket_has_arrived(self): + # A TLS 1.3 server sends its NewSessionTicket after the handshake, so a + # session read before the first application-data exchange carries no + # ticket and must not be cached. + connection = self.connect(self.make_ssl_context(), exchange=False) + + assert connection._socket.version() == 'TLSv1.3' + assert connection._get_resumable_tls_session() is None + connection._store_tls_session() + assert len(self.cache) == 0 + + connection.exchange() + + assert connection._get_resumable_tls_session() is not None + connection._store_tls_session() + assert len(self.cache) == 1 From b308baf8141b7eac2d12ce6e2b97431b76c80b07 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. Whether there is anything here to test at all depends on the reactor, so the skip asks the connection class Cluster will instantiate rather than reading EVENT_LOOP_MANAGER: with no selector set and no libev to import, that class is the asyncio reactor, which cannot restore a session before the handshake. 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 | 247 ++++++++++++++++++ 1 file changed, 247 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..d55b695134 --- /dev/null +++ b/tests/integration/standard/test_tls_resumption.py @@ -0,0 +1,247 @@ +# 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.cluster import Cluster +from cassandra.connection import SSLSessionCache +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') + # Asked of the class TestCluster will actually use, not of the selector: + # tests.integration sets Cluster.connection_class from EVENT_LOOP_MANAGER + # only when it resolved one, and leaves the driver's own default in place + # otherwise -- which, with no selector and no libev, is the asyncio reactor. + reactor = Cluster.connection_class + if not getattr(reactor, 'supports_tls_session_resumption', False): + raise unittest.SkipTest( + '%s cannot restore a cached TLS session before the handshake, so ' + 'there is no resumption here to test' + % getattr(reactor, '__name__', reactor)) + 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 637f384c5890f1afda30a53604f2d9ccc1c3dce5 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 ff74d07e67..3fd34656e9 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -919,6 +919,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