From 150eb37486726f52064927e1d21d5b73b8f80c27 Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Wed, 10 Jun 2026 06:54:12 -0500
Subject: [PATCH 01/16] PYTHON-5272 Implement TLS session resumption for sync
 pool

Add _SSLSessionCache to cache TLS sessions per pool, enabling session
resumption on subsequent connections to the same server. This avoids
full asymmetric-key handshakes on every new connection, addressing the
OpenSSL 3.0 performance overhead seen in BF-36991.
---
 pymongo/asynchronous/pool.py  |  8 +++++-
 pymongo/pool_shared.py        | 47 ++++++++++++++++++++++++++++++++---
 pymongo/synchronous/pool.py   |  8 +++++-
 test/asynchronous/test_ssl.py | 27 ++++++++++++++++++++
 test/test_ssl.py              | 27 ++++++++++++++++++++
 5 files changed, 111 insertions(+), 6 deletions(-)

diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py
index 475f4bfa99..f2b6df2f3e 100644
--- a/pymongo/asynchronous/pool.py
+++ b/pymongo/asynchronous/pool.py
@@ -85,6 +85,7 @@
     _CancellationContext,
     _configured_protocol_interface,
     _raise_connection_failure,
+    _SSLSessionCache,
 )
 from pymongo.read_preferences import ReadPreference
 from pymongo.server_api import _add_to_command
@@ -754,6 +755,9 @@ def __init__(
         self._pending = 0
         self._max_connecting = self.opts.max_connecting
         self._client_id = client_id
+        self._ssl_session_cache: Optional[_SSLSessionCache] = (
+            _SSLSessionCache() if self.opts._ssl_context is not None else None
+        )
         # Log before publishing event to prevent potential listener preemption in tests
         if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
             _debug_log(
@@ -1040,7 +1044,9 @@ async def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> A
             )
 
         try:
-            networking_interface = await _configured_protocol_interface(self.address, self.opts)
+            networking_interface = await _configured_protocol_interface(
+                self.address, self.opts, self._ssl_session_cache
+            )
         # Catch KeyboardInterrupt, CancelledError, etc. and cleanup.
         except BaseException as error:
             async with self.lock:
diff --git a/pymongo/pool_shared.py b/pymongo/pool_shared.py
index a6f434885b..a66903d473 100644
--- a/pymongo/pool_shared.py
+++ b/pymongo/pool_shared.py
@@ -20,6 +20,7 @@
 import socket
 import ssl
 import sys
+import threading
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -46,6 +47,32 @@
     from pymongo.pyopenssl_context import _sslConn
     from pymongo.typings import _Address
 
+
+class _SSLSessionCache:
+    """Thread-safe cache for a single TLS session per pool, enabling session resumption."""
+
+    __slots__ = ("_session", "_lock")
+
+    def __init__(self) -> None:
+        self._session: Optional[Any] = None
+        self._lock = threading.Lock()
+
+    def get(self) -> Optional[Any]:
+        with self._lock:
+            return self._session
+
+    def set(self, session: Any) -> None:
+        with self._lock:
+            self._session = session
+
+
+def _get_ssl_session(ssl_sock: Any) -> Optional[Any]:
+    """Return the TLS session from an SSL socket, handling both PyOpenSSL and stdlib ssl."""
+    if hasattr(ssl_sock, "get_session"):
+        return ssl_sock.get_session()
+    return getattr(ssl_sock, "session", None)
+
+
 try:
     from fcntl import F_GETFD, F_SETFD, FD_CLOEXEC, fcntl
 
@@ -298,7 +325,9 @@ async def _async_configured_socket(
 
 
 async def _configured_protocol_interface(
-    address: _Address, options: PoolOptions
+    address: _Address,
+    options: PoolOptions,
+    ssl_session_cache: Optional[_SSLSessionCache] = None,  # noqa: ARG001
 ) -> AsyncNetworkingInterface:
     """Given (host, port) and PoolOptions, return a configured AsyncNetworkingInterface.
 
@@ -470,7 +499,11 @@ def _configured_socket(address: _Address, options: PoolOptions) -> Union[socket.
     return ssl_sock
 
 
-def _configured_socket_interface(address: _Address, options: PoolOptions) -> NetworkingInterface:
+def _configured_socket_interface(
+    address: _Address,
+    options: PoolOptions,
+    ssl_session_cache: Optional[_SSLSessionCache] = None,
+) -> NetworkingInterface:
     """Given (host, port) and PoolOptions, return a NetworkingInterface wrapping a configured socket.
 
     Can raise socket.error, ConnectionFailure, or _CertificateError.
@@ -485,13 +518,14 @@ def _configured_socket_interface(address: _Address, options: PoolOptions) -> Net
         return NetworkingInterface(sock)
 
     host = address[0]
+    session = ssl_session_cache.get() if ssl_session_cache is not None else None
     try:
         # We have to pass hostname / ip address to wrap_socket
         # to use SSLContext.check_hostname.
         if _has_sni(True):
-            ssl_sock = ssl_context.wrap_socket(sock, server_hostname=host)
+            ssl_sock = ssl_context.wrap_socket(sock, server_hostname=host, session=session)
         else:
-            ssl_sock = ssl_context.wrap_socket(sock)
+            ssl_sock = ssl_context.wrap_socket(sock, session=session)
     except _CertificateError:
         sock.close()
         # Raise _CertificateError directly like we do after match_hostname
@@ -515,5 +549,10 @@ def _configured_socket_interface(address: _Address, options: PoolOptions) -> Net
             ssl_sock.close()
             raise
 
+    if ssl_session_cache is not None:
+        new_session = _get_ssl_session(ssl_sock)
+        if new_session is not None:
+            ssl_session_cache.set(new_session)
+
     ssl_sock.settimeout(options.socket_timeout)
     return NetworkingInterface(ssl_sock)
diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py
index 938eca42bd..8e7dacef19 100644
--- a/pymongo/synchronous/pool.py
+++ b/pymongo/synchronous/pool.py
@@ -82,6 +82,7 @@
     _CancellationContext,
     _configured_socket_interface,
     _raise_connection_failure,
+    _SSLSessionCache,
 )
 from pymongo.read_preferences import ReadPreference
 from pymongo.server_api import _add_to_command
@@ -752,6 +753,9 @@ def __init__(
         self._pending = 0
         self._max_connecting = self.opts.max_connecting
         self._client_id = client_id
+        self._ssl_session_cache: Optional[_SSLSessionCache] = (
+            _SSLSessionCache() if self.opts._ssl_context is not None else None
+        )
         # Log before publishing event to prevent potential listener preemption in tests
         if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
             _debug_log(
@@ -1036,7 +1040,9 @@ def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> Connect
             )
 
         try:
-            networking_interface = _configured_socket_interface(self.address, self.opts)
+            networking_interface = _configured_socket_interface(
+                self.address, self.opts, self._ssl_session_cache
+            )
         # Catch KeyboardInterrupt, CancelledError, etc. and cleanup.
         except BaseException as error:
             with self.lock:
diff --git a/test/asynchronous/test_ssl.py b/test/asynchronous/test_ssl.py
index 7fe57e8503..834a20550a 100644
--- a/test/asynchronous/test_ssl.py
+++ b/test/asynchronous/test_ssl.py
@@ -128,6 +128,16 @@ def test_config_ssl(self):
     def test_use_pyopenssl_when_available(self):
         self.assertTrue(HAVE_PYSSL)
 
+    def test_ssl_session_cache(self):
+        from pymongo.pool_shared import _SSLSessionCache
+
+        cache = _SSLSessionCache()
+        self.assertIsNone(cache.get())
+        cache.set("session")
+        self.assertEqual(cache.get(), "session")
+        cache.set("new_session")
+        self.assertEqual(cache.get(), "new_session")
+
 
 class TestSSL(AsyncIntegrationTest):
     saved_port: int
@@ -673,6 +683,23 @@ async def test_pyopenssl_ignored_in_async(self):
         await client.admin.command("ping")  # command doesn't matter, just needs it to connect
         await client.close()
 
+    @async_client_context.require_tls
+    async def test_pool_has_ssl_session_cache(self):
+        from pymongo.pool_shared import _SSLSessionCache
+
+        pool = list(self.client._topology._servers.values())[0].pool
+        self.assertIsInstance(pool._ssl_session_cache, _SSLSessionCache)
+
+    @async_client_context.require_tls
+    @unittest.skipUnless(
+        _IS_SYNC and _HAVE_PYOPENSSL, "Session caching only applies to PyOpenSSL sync path"
+    )
+    async def test_tls_session_cached_after_connect(self):
+        await self.client.admin.command("ping")
+        pool = list(self.client._topology._servers.values())[0].pool
+        self.assertIsNotNone(pool._ssl_session_cache)
+        self.assertIsNotNone(pool._ssl_session_cache.get())
+
 
 if __name__ == "__main__":
     unittest.main()
diff --git a/test/test_ssl.py b/test/test_ssl.py
index 77bb086ecb..31f435d87a 100644
--- a/test/test_ssl.py
+++ b/test/test_ssl.py
@@ -128,6 +128,16 @@ def test_config_ssl(self):
     def test_use_pyopenssl_when_available(self):
         self.assertTrue(HAVE_PYSSL)
 
+    def test_ssl_session_cache(self):
+        from pymongo.pool_shared import _SSLSessionCache
+
+        cache = _SSLSessionCache()
+        self.assertIsNone(cache.get())
+        cache.set("session")
+        self.assertEqual(cache.get(), "session")
+        cache.set("new_session")
+        self.assertEqual(cache.get(), "new_session")
+
 
 class TestSSL(IntegrationTest):
     saved_port: int
@@ -671,6 +681,23 @@ def test_pyopenssl_ignored_in_async(self):
         client.admin.command("ping")  # command doesn't matter, just needs it to connect
         client.close()
 
+    @client_context.require_tls
+    def test_pool_has_ssl_session_cache(self):
+        from pymongo.pool_shared import _SSLSessionCache
+
+        pool = list(self.client._topology._servers.values())[0].pool
+        self.assertIsInstance(pool._ssl_session_cache, _SSLSessionCache)
+
+    @client_context.require_tls
+    @unittest.skipUnless(
+        _IS_SYNC and _HAVE_PYOPENSSL, "Session caching only applies to PyOpenSSL sync path"
+    )
+    def test_tls_session_cached_after_connect(self):
+        self.client.admin.command("ping")
+        pool = list(self.client._topology._servers.values())[0].pool
+        self.assertIsNotNone(pool._ssl_session_cache)
+        self.assertIsNotNone(pool._ssl_session_cache.get())
+
 
 if __name__ == "__main__":
     unittest.main()

From 92a8230c80127bb22577cfdccc5be74522191e16 Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Wed, 10 Jun 2026 07:02:17 -0500
Subject: [PATCH 02/16] PYTHON-5272 Add session reuse unit test for
 _configured_socket_interface

Verify that a pre-populated _SSLSessionCache passes the cached session
as session= to wrap_socket on the next connection, using mocks so no
live server is required.
---
 test/asynchronous/test_ssl.py | 32 ++++++++++++++++++++++++++++++++
 test/test_ssl.py              | 32 ++++++++++++++++++++++++++++++++
 2 files changed, 64 insertions(+)

diff --git a/test/asynchronous/test_ssl.py b/test/asynchronous/test_ssl.py
index 834a20550a..f36379df35 100644
--- a/test/asynchronous/test_ssl.py
+++ b/test/asynchronous/test_ssl.py
@@ -138,6 +138,38 @@ def test_ssl_session_cache(self):
         cache.set("new_session")
         self.assertEqual(cache.get(), "new_session")
 
+    @unittest.skipUnless(_IS_SYNC, "Session reuse only applies to sync path")
+    def test_tls_session_reused_on_second_connection(self):
+        """Cached TLS session is passed to wrap_socket on subsequent connections."""
+        import unittest.mock as mock
+
+        from pymongo.pool_shared import _configured_socket_interface, _SSLSessionCache
+
+        fake_session = object()
+        cache = _SSLSessionCache()
+        cache.set(fake_session)
+
+        fake_ssl_sock = mock.MagicMock()
+        fake_ssl_sock.getpeercert.return_value = {}
+
+        mock_ssl_context = mock.MagicMock()
+        mock_ssl_context.wrap_socket.return_value = fake_ssl_sock
+        mock_ssl_context.verify_mode = False
+        mock_ssl_context.check_hostname = False
+
+        mock_opts = mock.MagicMock()
+        mock_opts._ssl_context = mock_ssl_context
+        mock_opts.socket_timeout = None
+        mock_opts.tls_allow_invalid_hostnames = True
+
+        with mock.patch("pymongo.pool_shared._create_connection") as mock_create:
+            mock_create.return_value = mock.MagicMock()
+            _configured_socket_interface(("localhost", 27017), mock_opts, cache)
+
+        mock_ssl_context.wrap_socket.assert_called_once()
+        _, kwargs = mock_ssl_context.wrap_socket.call_args
+        self.assertIs(kwargs.get("session"), fake_session)
+
 
 class TestSSL(AsyncIntegrationTest):
     saved_port: int
diff --git a/test/test_ssl.py b/test/test_ssl.py
index 31f435d87a..b175a0ba96 100644
--- a/test/test_ssl.py
+++ b/test/test_ssl.py
@@ -138,6 +138,38 @@ def test_ssl_session_cache(self):
         cache.set("new_session")
         self.assertEqual(cache.get(), "new_session")
 
+    @unittest.skipUnless(_IS_SYNC, "Session reuse only applies to sync path")
+    def test_tls_session_reused_on_second_connection(self):
+        """Cached TLS session is passed to wrap_socket on subsequent connections."""
+        import unittest.mock as mock
+
+        from pymongo.pool_shared import _configured_socket_interface, _SSLSessionCache
+
+        fake_session = object()
+        cache = _SSLSessionCache()
+        cache.set(fake_session)
+
+        fake_ssl_sock = mock.MagicMock()
+        fake_ssl_sock.getpeercert.return_value = {}
+
+        mock_ssl_context = mock.MagicMock()
+        mock_ssl_context.wrap_socket.return_value = fake_ssl_sock
+        mock_ssl_context.verify_mode = False
+        mock_ssl_context.check_hostname = False
+
+        mock_opts = mock.MagicMock()
+        mock_opts._ssl_context = mock_ssl_context
+        mock_opts.socket_timeout = None
+        mock_opts.tls_allow_invalid_hostnames = True
+
+        with mock.patch("pymongo.pool_shared._create_connection") as mock_create:
+            mock_create.return_value = mock.MagicMock()
+            _configured_socket_interface(("localhost", 27017), mock_opts, cache)
+
+        mock_ssl_context.wrap_socket.assert_called_once()
+        _, kwargs = mock_ssl_context.wrap_socket.call_args
+        self.assertIs(kwargs.get("session"), fake_session)
+
 
 class TestSSL(IntegrationTest):
     saved_port: int

From 2e6f985df9e8fb5eb009a217e0503c17b655e02a Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Wed, 10 Jun 2026 12:45:01 -0500
Subject: [PATCH 03/16] PYTHON-5272 Extend TLS session resumption to asyncio
 path (Python 3.11+)

On Python 3.11+, SSLProtocol.__init__ creates the ssl.SSLObject via
wrap_bio before the handshake starts in connection_made. We temporarily
replace asyncio.sslproto.SSLProtocol with a subclass that sets
sslobj.session to the cached session immediately after super().__init__,
then restore the original class. With a pre-connected sock= parameter,
_make_ssl_transport is called synchronously inside create_connection
before the first await, so the swap is race-free in a single-threaded
event loop.

After the handshake, the session is retrieved via
transport.get_extra_info('ssl_object').session and stored in the pool's
_SSLSessionCache for the next connection.
---
 pymongo/pool_shared.py        | 50 ++++++++++++++++++++++++++++++++++-
 test/asynchronous/test_ssl.py | 23 ++++++++++++++++
 test/test_ssl.py              | 23 ++++++++++++++++
 3 files changed, 95 insertions(+), 1 deletion(-)

diff --git a/pymongo/pool_shared.py b/pymongo/pool_shared.py
index a66903d473..a009d9d0ba 100644
--- a/pymongo/pool_shared.py
+++ b/pymongo/pool_shared.py
@@ -73,6 +73,28 @@ def _get_ssl_session(ssl_sock: Any) -> Optional[Any]:
     return getattr(ssl_sock, "session", None)
 
 
+# asyncio's SSLProtocol does not expose a session= parameter in create_connection.
+# On Python 3.11+, wrap_bio() is called in SSLProtocol.__init__ and the handshake
+# starts later in connection_made(), so we can set sslobj.session between the two.
+# On older Python, _SSLPipe.do_handshake calls wrap_bio and starts the handshake
+# atomically; session injection there requires copying private internals, so we skip it.
+_ASYNCIO_SSL_SESSION_SUPPORTED = sys.version_info >= (3, 11)
+
+
+def _make_session_ssl_protocol(session: Any) -> Any:
+    """Return an SSLProtocol subclass that injects *session* before the handshake."""
+    import asyncio.sslproto as _sslproto
+
+    class _SessionSSLProtocol(_sslproto.SSLProtocol):
+        def __init__(self, *args: Any, **kwargs: Any) -> None:
+            super().__init__(*args, **kwargs)
+            sslobj = getattr(self, "_sslobj", None)
+            if sslobj is not None:
+                sslobj.session = session
+
+    return _SessionSSLProtocol
+
+
 try:
     from fcntl import F_GETFD, F_SETFD, FD_CLOEXEC, fcntl
 
@@ -327,7 +349,7 @@ async def _async_configured_socket(
 async def _configured_protocol_interface(
     address: _Address,
     options: PoolOptions,
-    ssl_session_cache: Optional[_SSLSessionCache] = None,  # noqa: ARG001
+    ssl_session_cache: Optional[_SSLSessionCache] = None,
 ) -> AsyncNetworkingInterface:
     """Given (host, port) and PoolOptions, return a configured AsyncNetworkingInterface.
 
@@ -347,6 +369,21 @@ async def _configured_protocol_interface(
         )
 
     host = address[0]
+    # On Python 3.11+, temporarily patch asyncio's SSLProtocol to inject the
+    # cached session before the handshake.  _make_ssl_transport (which
+    # instantiates SSLProtocol) is called synchronously inside
+    # create_connection before the first await, so the swap is race-free in a
+    # single-threaded event loop when the socket is pre-connected.
+    import asyncio.sslproto as _sslproto
+
+    session = (
+        ssl_session_cache.get()
+        if ssl_session_cache is not None and _ASYNCIO_SSL_SESSION_SUPPORTED
+        else None
+    )
+    original_ssl_protocol = _sslproto.SSLProtocol
+    if session is not None:
+        _sslproto.SSLProtocol = _make_session_ssl_protocol(session)  # type: ignore[misc]
     try:
         # We have to pass hostname / ip address to wrap_socket
         # to use SSLContext.check_hostname.
@@ -366,6 +403,10 @@ async def _configured_protocol_interface(
         # mismatch, will be turned into ServerSelectionTimeoutErrors later.
         details = _get_timeout_details(options)
         _raise_connection_failure(address, exc, "SSL handshake failed: ", timeout_details=details)
+    finally:
+        if session is not None:
+            _sslproto.SSLProtocol = original_ssl_protocol  # type: ignore[misc]
+
     if (
         ssl_context.verify_mode
         and not ssl_context.check_hostname
@@ -377,6 +418,13 @@ async def _configured_protocol_interface(
             transport.abort()
             raise
 
+    if ssl_session_cache is not None and _ASYNCIO_SSL_SESSION_SUPPORTED:
+        ssl_obj = transport.get_extra_info("ssl_object")
+        if ssl_obj is not None:
+            new_session = ssl_obj.session
+            if new_session is not None:
+                ssl_session_cache.set(new_session)
+
     return AsyncNetworkingInterface((transport, protocol))
 
 
diff --git a/test/asynchronous/test_ssl.py b/test/asynchronous/test_ssl.py
index f36379df35..43792d6cd4 100644
--- a/test/asynchronous/test_ssl.py
+++ b/test/asynchronous/test_ssl.py
@@ -170,6 +170,29 @@ def test_tls_session_reused_on_second_connection(self):
         _, kwargs = mock_ssl_context.wrap_socket.call_args
         self.assertIs(kwargs.get("session"), fake_session)
 
+    @unittest.skipUnless(
+        not _IS_SYNC and sys.version_info >= (3, 11),
+        "Async session injection requires Python 3.11+",
+    )
+    def test_async_tls_session_injected_into_sslobj(self):
+        """Cached TLS session is set on SSLObject before the handshake on Python 3.11+."""
+        import asyncio.sslproto as _sslproto
+        import unittest.mock as mock
+
+        from pymongo.pool_shared import _make_session_ssl_protocol, _SSLSessionCache
+
+        fake_session = mock.MagicMock()
+        patched_cls = _make_session_ssl_protocol(fake_session)
+
+        mock_sslobj = mock.MagicMock()
+        instance = patched_cls.__new__(patched_cls)
+        instance._sslobj = mock_sslobj
+        # Call __init__ via the patched class, bypassing the real SSLProtocol init.
+        with mock.patch.object(_sslproto.SSLProtocol, "__init__", lambda *a, **kw: None):
+            patched_cls.__init__(instance)
+
+        self.assertEqual(mock_sslobj.session, fake_session)
+
 
 class TestSSL(AsyncIntegrationTest):
     saved_port: int
diff --git a/test/test_ssl.py b/test/test_ssl.py
index b175a0ba96..eb2801cd83 100644
--- a/test/test_ssl.py
+++ b/test/test_ssl.py
@@ -170,6 +170,29 @@ def test_tls_session_reused_on_second_connection(self):
         _, kwargs = mock_ssl_context.wrap_socket.call_args
         self.assertIs(kwargs.get("session"), fake_session)
 
+    @unittest.skipUnless(
+        not _IS_SYNC and sys.version_info >= (3, 11),
+        "Async session injection requires Python 3.11+",
+    )
+    def test_async_tls_session_injected_into_sslobj(self):
+        """Cached TLS session is set on SSLObject before the handshake on Python 3.11+."""
+        import asyncio.sslproto as _sslproto
+        import unittest.mock as mock
+
+        from pymongo.pool_shared import _make_session_ssl_protocol, _SSLSessionCache
+
+        fake_session = mock.MagicMock()
+        patched_cls = _make_session_ssl_protocol(fake_session)
+
+        mock_sslobj = mock.MagicMock()
+        instance = patched_cls.__new__(patched_cls)
+        instance._sslobj = mock_sslobj
+        # Call __init__ via the patched class, bypassing the real SSLProtocol init.
+        with mock.patch.object(_sslproto.SSLProtocol, "__init__", lambda *a, **kw: None):
+            patched_cls.__init__(instance)
+
+        self.assertEqual(mock_sslobj.session, fake_session)
+
 
 class TestSSL(IntegrationTest):
     saved_port: int

From 17fbd40a883a6d0482315600ee048f8592ecee75 Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Wed, 10 Jun 2026 13:21:44 -0500
Subject: [PATCH 04/16] PYTHON-5272 Update stale skip message for session reuse
 test

---
 test/asynchronous/test_ssl.py | 2 +-
 test/test_ssl.py              | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/test/asynchronous/test_ssl.py b/test/asynchronous/test_ssl.py
index 43792d6cd4..e29a6548ea 100644
--- a/test/asynchronous/test_ssl.py
+++ b/test/asynchronous/test_ssl.py
@@ -138,7 +138,7 @@ def test_ssl_session_cache(self):
         cache.set("new_session")
         self.assertEqual(cache.get(), "new_session")
 
-    @unittest.skipUnless(_IS_SYNC, "Session reuse only applies to sync path")
+    @unittest.skipUnless(_IS_SYNC, "Tests sync wrap_socket path only")
     def test_tls_session_reused_on_second_connection(self):
         """Cached TLS session is passed to wrap_socket on subsequent connections."""
         import unittest.mock as mock
diff --git a/test/test_ssl.py b/test/test_ssl.py
index eb2801cd83..20a21958d2 100644
--- a/test/test_ssl.py
+++ b/test/test_ssl.py
@@ -138,7 +138,7 @@ def test_ssl_session_cache(self):
         cache.set("new_session")
         self.assertEqual(cache.get(), "new_session")
 
-    @unittest.skipUnless(_IS_SYNC, "Session reuse only applies to sync path")
+    @unittest.skipUnless(_IS_SYNC, "Tests sync wrap_socket path only")
     def test_tls_session_reused_on_second_connection(self):
         """Cached TLS session is passed to wrap_socket on subsequent connections."""
         import unittest.mock as mock

From 04733ea90617ccf3ab803d6f28b167525bdf0ab4 Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Wed, 10 Jun 2026 13:26:35 -0500
Subject: [PATCH 05/16] PYTHON-5272 Add changelog entry for TLS session
 resumption

---
 doc/changelog.rst | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/doc/changelog.rst b/doc/changelog.rst
index ebbc72047b..ea5627fc88 100644
--- a/doc/changelog.rst
+++ b/doc/changelog.rst
@@ -1,6 +1,14 @@
 Changelog
 =========
 
+Changes in Version 4.18.0
+------------------------
+
+- Improved TLS connection performance by reusing TLS sessions across connections
+  to the same server, avoiding a full handshake on each new connection.
+  Session reuse is active on the sync path unconditionally, and on the async
+  path when running Python 3.11 or later.
+
 Changes in Version 4.17.0 (2026/04/20)
 --------------------------------------
 

From 7bf71f05c3d57c771f18451ef0a1e4ef5dd55a90 Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Wed, 10 Jun 2026 13:36:10 -0500
Subject: [PATCH 06/16] PYTHON-5272 Expand changelog entry with per-case
 caching details

---
 doc/changelog.rst | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/doc/changelog.rst b/doc/changelog.rst
index ea5627fc88..54ad6a0ee8 100644
--- a/doc/changelog.rst
+++ b/doc/changelog.rst
@@ -7,7 +7,7 @@ Changes in Version 4.18.0
 - Improved TLS connection performance by reusing TLS sessions across connections
   to the same server, avoiding a full handshake on each new connection.
   Session reuse is active on the sync path unconditionally, and on the async
-  path when running Python 3.11 or later.
+  path on Python 3.11 or later.
 
 Changes in Version 4.17.0 (2026/04/20)
 --------------------------------------

From 7c2264e4c94b58faa7f0a978871fba372858601a Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Wed, 10 Jun 2026 13:55:31 -0500
Subject: [PATCH 07/16] PYTHON-5272 Fix changelog RST title underline length

---
 doc/changelog.rst | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/doc/changelog.rst b/doc/changelog.rst
index 54ad6a0ee8..54d561e066 100644
--- a/doc/changelog.rst
+++ b/doc/changelog.rst
@@ -2,7 +2,7 @@ Changelog
 =========
 
 Changes in Version 4.18.0
-------------------------
+-------------------------
 
 - Improved TLS connection performance by reusing TLS sessions across connections
   to the same server, avoiding a full handshake on each new connection.

From 5958f598338a26f2776afaa2a427a263251951f6 Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Wed, 10 Jun 2026 14:30:28 -0500
Subject: [PATCH 08/16] PYTHON-5272 Reference closed CPython issue for async
 TLS session workaround

---
 pymongo/pool_shared.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/pymongo/pool_shared.py b/pymongo/pool_shared.py
index a009d9d0ba..73a61b3491 100644
--- a/pymongo/pool_shared.py
+++ b/pymongo/pool_shared.py
@@ -73,7 +73,9 @@ def _get_ssl_session(ssl_sock: Any) -> Optional[Any]:
     return getattr(ssl_sock, "session", None)
 
 
-# asyncio's SSLProtocol does not expose a session= parameter in create_connection.
+# asyncio's create_connection does not support TLS session resumption natively.
+# https://github.com/python/cpython/issues/79152 tracks this; a patch was submitted
+# in 2018 but never merged, and the issue is now closed.
 # On Python 3.11+, wrap_bio() is called in SSLProtocol.__init__ and the handshake
 # starts later in connection_made(), so we can set sslobj.session between the two.
 # On older Python, _SSLPipe.do_handshake calls wrap_bio and starts the handshake

From c9f62fe8fb88f5964eeffb7d48d9555666aff468 Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Wed, 10 Jun 2026 14:43:51 -0500
Subject: [PATCH 09/16] PYTHON-5272 Fix concurrency bug in async SSL protocol
 patch restore

Capture _ORIGINAL_SSL_PROTOCOL once at module load time and always restore
to it unconditionally, so that concurrent connections from the same pool
cannot leave a stale SSLProtocol subclass active in asyncio.sslproto.
---
 pymongo/pool_shared.py | 17 +++++++++++------
 1 file changed, 11 insertions(+), 6 deletions(-)

diff --git a/pymongo/pool_shared.py b/pymongo/pool_shared.py
index 73a61b3491..a4cd233ce6 100644
--- a/pymongo/pool_shared.py
+++ b/pymongo/pool_shared.py
@@ -81,6 +81,12 @@ def _get_ssl_session(ssl_sock: Any) -> Optional[Any]:
 # On older Python, _SSLPipe.do_handshake calls wrap_bio and starts the handshake
 # atomically; session injection there requires copying private internals, so we skip it.
 _ASYNCIO_SSL_SESSION_SUPPORTED = sys.version_info >= (3, 11)
+if _ASYNCIO_SSL_SESSION_SUPPORTED:
+    import asyncio.sslproto as _asyncio_sslproto
+
+    # Capture the true original once at import time so concurrent connections
+    # always restore to it, not to a locally-captured (possibly stale) reference.
+    _ORIGINAL_SSL_PROTOCOL = _asyncio_sslproto.SSLProtocol
 
 
 def _make_session_ssl_protocol(session: Any) -> Any:
@@ -376,16 +382,15 @@ async def _configured_protocol_interface(
     # instantiates SSLProtocol) is called synchronously inside
     # create_connection before the first await, so the swap is race-free in a
     # single-threaded event loop when the socket is pre-connected.
-    import asyncio.sslproto as _sslproto
-
+    # Always restore to _ORIGINAL_SSL_PROTOCOL (not a locally captured value)
+    # so that concurrent connections can't leave a stale subclass in place.
     session = (
         ssl_session_cache.get()
         if ssl_session_cache is not None and _ASYNCIO_SSL_SESSION_SUPPORTED
         else None
     )
-    original_ssl_protocol = _sslproto.SSLProtocol
     if session is not None:
-        _sslproto.SSLProtocol = _make_session_ssl_protocol(session)  # type: ignore[misc]
+        _asyncio_sslproto.SSLProtocol = _make_session_ssl_protocol(session)  # type: ignore[misc]
     try:
         # We have to pass hostname / ip address to wrap_socket
         # to use SSLContext.check_hostname.
@@ -406,8 +411,8 @@ async def _configured_protocol_interface(
         details = _get_timeout_details(options)
         _raise_connection_failure(address, exc, "SSL handshake failed: ", timeout_details=details)
     finally:
-        if session is not None:
-            _sslproto.SSLProtocol = original_ssl_protocol  # type: ignore[misc]
+        if _ASYNCIO_SSL_SESSION_SUPPORTED:
+            _asyncio_sslproto.SSLProtocol = _ORIGINAL_SSL_PROTOCOL  # type: ignore[misc]
 
     if (
         ssl_context.verify_mode

From dc856d2ffb0579cc56538e1e6640b70fdb88488b Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Wed, 10 Jun 2026 16:19:59 -0500
Subject: [PATCH 10/16] PYTHON-5272 Fix PyPy failure: keep asyncio.sslproto
 import lazy

Module-level import of asyncio.sslproto on PyPy 3.11 changed GC timing
and surfaced a pre-existing unclosed-socket ResourceWarning as a test
error.  Use a module-level None sentinel instead and initialise it on
first use inside the function, which keeps the import lazy while still
capturing the true original SSLProtocol before any patching occurs.
---
 pymongo/pool_shared.py | 20 ++++++++++++--------
 1 file changed, 12 insertions(+), 8 deletions(-)

diff --git a/pymongo/pool_shared.py b/pymongo/pool_shared.py
index a4cd233ce6..7da64792bf 100644
--- a/pymongo/pool_shared.py
+++ b/pymongo/pool_shared.py
@@ -81,12 +81,10 @@ def _get_ssl_session(ssl_sock: Any) -> Optional[Any]:
 # On older Python, _SSLPipe.do_handshake calls wrap_bio and starts the handshake
 # atomically; session injection there requires copying private internals, so we skip it.
 _ASYNCIO_SSL_SESSION_SUPPORTED = sys.version_info >= (3, 11)
-if _ASYNCIO_SSL_SESSION_SUPPORTED:
-    import asyncio.sslproto as _asyncio_sslproto
-
-    # Capture the true original once at import time so concurrent connections
-    # always restore to it, not to a locally-captured (possibly stale) reference.
-    _ORIGINAL_SSL_PROTOCOL = _asyncio_sslproto.SSLProtocol
+# Captured lazily on first SSL async connection; never reset thereafter so
+# concurrent connections always restore to the true original, not a locally-
+# captured (possibly stale) reference.
+_ORIGINAL_SSL_PROTOCOL: Any = None
 
 
 def _make_session_ssl_protocol(session: Any) -> Any:
@@ -389,8 +387,14 @@ async def _configured_protocol_interface(
         if ssl_session_cache is not None and _ASYNCIO_SSL_SESSION_SUPPORTED
         else None
     )
-    if session is not None:
-        _asyncio_sslproto.SSLProtocol = _make_session_ssl_protocol(session)  # type: ignore[misc]
+    if _ASYNCIO_SSL_SESSION_SUPPORTED:
+        import asyncio.sslproto as _asyncio_sslproto
+
+        global _ORIGINAL_SSL_PROTOCOL  # noqa: PLW0603
+        if _ORIGINAL_SSL_PROTOCOL is None:
+            _ORIGINAL_SSL_PROTOCOL = _asyncio_sslproto.SSLProtocol
+        if session is not None:
+            _asyncio_sslproto.SSLProtocol = _make_session_ssl_protocol(session)  # type: ignore[misc]
     try:
         # We have to pass hostname / ip address to wrap_socket
         # to use SSLContext.check_hostname.

From 2a12ea097eb0656ccf0e9fb1d4b052c90a465304 Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Thu, 11 Jun 2026 06:17:55 -0500
Subject: [PATCH 11/16] PYTHON-5272 Inject TLS session via sslobject_class
 instead of monkey-patching SSLProtocol

Replace the asyncio.sslproto.SSLProtocol monkey-patch with a
_SessionSSLContext wrapper that intercepts wrap_bio() and temporarily
sets sslobject_class to a per-connection ssl.SSLObject subclass that
assigns self.session after __init__. The set/use/restore cycle inside
wrap_bio() is atomic because wrap_bio() is synchronous, so asyncio
cannot yield to another coroutine mid-call.

This removes all global state (_ORIGINAL_SSL_PROTOCOL, the Python 3.11
version guard) and private attribute access (_sslobj), and works on all
supported Python versions without concurrency hazards.
---
 doc/changelog.rst             |  2 -
 pymongo/pool_shared.py        | 83 +++++++++++++++++------------------
 test/asynchronous/test_ssl.py | 42 +++++++++++-------
 test/test_ssl.py              | 42 +++++++++++-------
 4 files changed, 90 insertions(+), 79 deletions(-)

diff --git a/doc/changelog.rst b/doc/changelog.rst
index 54d561e066..60c87c43e1 100644
--- a/doc/changelog.rst
+++ b/doc/changelog.rst
@@ -6,8 +6,6 @@ Changes in Version 4.18.0
 
 - Improved TLS connection performance by reusing TLS sessions across connections
   to the same server, avoiding a full handshake on each new connection.
-  Session reuse is active on the sync path unconditionally, and on the async
-  path on Python 3.11 or later.
 
 Changes in Version 4.17.0 (2026/04/20)
 --------------------------------------
diff --git a/pymongo/pool_shared.py b/pymongo/pool_shared.py
index 7da64792bf..99cf2002cb 100644
--- a/pymongo/pool_shared.py
+++ b/pymongo/pool_shared.py
@@ -76,29 +76,43 @@ def _get_ssl_session(ssl_sock: Any) -> Optional[Any]:
 # asyncio's create_connection does not support TLS session resumption natively.
 # https://github.com/python/cpython/issues/79152 tracks this; a patch was submitted
 # in 2018 but never merged, and the issue is now closed.
-# On Python 3.11+, wrap_bio() is called in SSLProtocol.__init__ and the handshake
-# starts later in connection_made(), so we can set sslobj.session between the two.
-# On older Python, _SSLPipe.do_handshake calls wrap_bio and starts the handshake
-# atomically; session injection there requires copying private internals, so we skip it.
-_ASYNCIO_SSL_SESSION_SUPPORTED = sys.version_info >= (3, 11)
-# Captured lazily on first SSL async connection; never reset thereafter so
-# concurrent connections always restore to the true original, not a locally-
-# captured (possibly stale) reference.
-_ORIGINAL_SSL_PROTOCOL: Any = None
+# We work around this by wrapping the SSLContext to intercept wrap_bio(), which
+# asyncio always calls before the TLS handshake regardless of Python version.
 
 
-def _make_session_ssl_protocol(session: Any) -> Any:
-    """Return an SSLProtocol subclass that injects *session* before the handshake."""
-    import asyncio.sslproto as _sslproto
+class _SessionSSLContext:
+    """Wraps an SSLContext to inject a cached TLS session via sslobject_class.
 
-    class _SessionSSLProtocol(_sslproto.SSLProtocol):
-        def __init__(self, *args: Any, **kwargs: Any) -> None:
-            super().__init__(*args, **kwargs)
-            sslobj = getattr(self, "_sslobj", None)
-            if sslobj is not None:
-                sslobj.session = session
+    ssl.SSLContext.sslobject_class (Python 3.7+) controls the type returned by
+    wrap_bio().  We create a per-connection subclass of ssl.SSLObject that sets
+    self.session after __init__, then temporarily apply it inside wrap_bio().
+    The patch-and-restore is race-free because wrap_bio() is synchronous.
+    """
+
+    __slots__ = ("_ctx", "_session_class")
+
+    def __init__(self, ctx: Any, session: Any) -> None:
+        self._ctx = ctx
+        base = ctx.sslobject_class
+        _session = session  # captured by inner class; avoids any name clash with outer self
 
-    return _SessionSSLProtocol
+        class _SessionSSLObject(base):  # type: ignore[misc,valid-type]
+            def __init__(self, *args: Any, **kwargs: Any) -> None:
+                super().__init__(*args, **kwargs)
+                self.session = _session
+
+        self._session_class = _SessionSSLObject
+
+    def wrap_bio(self, *args: Any, **kwargs: Any) -> Any:
+        orig = self._ctx.sslobject_class
+        self._ctx.sslobject_class = self._session_class
+        try:
+            return self._ctx.wrap_bio(*args, **kwargs)
+        finally:
+            self._ctx.sslobject_class = orig
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._ctx, name)
 
 
 try:
@@ -375,26 +389,12 @@ async def _configured_protocol_interface(
         )
 
     host = address[0]
-    # On Python 3.11+, temporarily patch asyncio's SSLProtocol to inject the
-    # cached session before the handshake.  _make_ssl_transport (which
-    # instantiates SSLProtocol) is called synchronously inside
-    # create_connection before the first await, so the swap is race-free in a
-    # single-threaded event loop when the socket is pre-connected.
-    # Always restore to _ORIGINAL_SSL_PROTOCOL (not a locally captured value)
-    # so that concurrent connections can't leave a stale subclass in place.
-    session = (
-        ssl_session_cache.get()
-        if ssl_session_cache is not None and _ASYNCIO_SSL_SESSION_SUPPORTED
-        else None
+    session = ssl_session_cache.get() if ssl_session_cache is not None else None
+    # Wrap the SSL context to inject the cached session via wrap_bio(), which
+    # asyncio calls before the TLS handshake on all supported Python versions.
+    effective_ssl_context = (
+        _SessionSSLContext(ssl_context, session) if session is not None else ssl_context
     )
-    if _ASYNCIO_SSL_SESSION_SUPPORTED:
-        import asyncio.sslproto as _asyncio_sslproto
-
-        global _ORIGINAL_SSL_PROTOCOL  # noqa: PLW0603
-        if _ORIGINAL_SSL_PROTOCOL is None:
-            _ORIGINAL_SSL_PROTOCOL = _asyncio_sslproto.SSLProtocol
-        if session is not None:
-            _asyncio_sslproto.SSLProtocol = _make_session_ssl_protocol(session)  # type: ignore[misc]
     try:
         # We have to pass hostname / ip address to wrap_socket
         # to use SSLContext.check_hostname.
@@ -402,7 +402,7 @@ async def _configured_protocol_interface(
             lambda: PyMongoProtocol(timeout=timeout),
             sock=sock,
             server_hostname=host,
-            ssl=ssl_context,
+            ssl=effective_ssl_context,
         )
     except _CertificateError:
         # Raise _CertificateError directly like we do after match_hostname
@@ -414,9 +414,6 @@ async def _configured_protocol_interface(
         # mismatch, will be turned into ServerSelectionTimeoutErrors later.
         details = _get_timeout_details(options)
         _raise_connection_failure(address, exc, "SSL handshake failed: ", timeout_details=details)
-    finally:
-        if _ASYNCIO_SSL_SESSION_SUPPORTED:
-            _asyncio_sslproto.SSLProtocol = _ORIGINAL_SSL_PROTOCOL  # type: ignore[misc]
 
     if (
         ssl_context.verify_mode
@@ -429,7 +426,7 @@ async def _configured_protocol_interface(
             transport.abort()
             raise
 
-    if ssl_session_cache is not None and _ASYNCIO_SSL_SESSION_SUPPORTED:
+    if ssl_session_cache is not None:
         ssl_obj = transport.get_extra_info("ssl_object")
         if ssl_obj is not None:
             new_session = ssl_obj.session
diff --git a/test/asynchronous/test_ssl.py b/test/asynchronous/test_ssl.py
index e29a6548ea..ad0fa8e7cb 100644
--- a/test/asynchronous/test_ssl.py
+++ b/test/asynchronous/test_ssl.py
@@ -170,28 +170,35 @@ def test_tls_session_reused_on_second_connection(self):
         _, kwargs = mock_ssl_context.wrap_socket.call_args
         self.assertIs(kwargs.get("session"), fake_session)
 
-    @unittest.skipUnless(
-        not _IS_SYNC and sys.version_info >= (3, 11),
-        "Async session injection requires Python 3.11+",
-    )
-    def test_async_tls_session_injected_into_sslobj(self):
-        """Cached TLS session is set on SSLObject before the handshake on Python 3.11+."""
-        import asyncio.sslproto as _sslproto
+    @unittest.skipUnless(not _IS_SYNC, "Tests async sslobject_class injection path only")
+    def test_async_tls_session_injected_via_sslobject_class(self):
+        """_SessionSSLContext sets sslobject_class on the real context for the wrap_bio() call."""
+        import ssl
         import unittest.mock as mock
 
-        from pymongo.pool_shared import _make_session_ssl_protocol, _SSLSessionCache
+        from pymongo.pool_shared import _SessionSSLContext
 
         fake_session = mock.MagicMock()
-        patched_cls = _make_session_ssl_protocol(fake_session)
+        real_ctx = mock.MagicMock()
+        real_ctx.sslobject_class = ssl.SSLObject
+        wrapped = _SessionSSLContext(real_ctx, fake_session)
+
+        observed_class = None
+
+        def capture_class(*args, **kwargs):
+            nonlocal observed_class
+            observed_class = real_ctx.sslobject_class
+            return mock.MagicMock()
 
-        mock_sslobj = mock.MagicMock()
-        instance = patched_cls.__new__(patched_cls)
-        instance._sslobj = mock_sslobj
-        # Call __init__ via the patched class, bypassing the real SSLProtocol init.
-        with mock.patch.object(_sslproto.SSLProtocol, "__init__", lambda *a, **kw: None):
-            patched_cls.__init__(instance)
+        real_ctx.wrap_bio.side_effect = capture_class
+        wrapped.wrap_bio("incoming", "outgoing", server_side=False)
 
-        self.assertEqual(mock_sslobj.session, fake_session)
+        # sslobject_class was our session-injecting subclass during the call
+        assert observed_class is not None
+        self.assertTrue(issubclass(observed_class, ssl.SSLObject))
+        self.assertIsNot(observed_class, ssl.SSLObject)
+        # sslobject_class was restored to the original after the call
+        self.assertIs(real_ctx.sslobject_class, ssl.SSLObject)
 
 
 class TestSSL(AsyncIntegrationTest):
@@ -747,7 +754,8 @@ async def test_pool_has_ssl_session_cache(self):
 
     @async_client_context.require_tls
     @unittest.skipUnless(
-        _IS_SYNC and _HAVE_PYOPENSSL, "Session caching only applies to PyOpenSSL sync path"
+        _IS_SYNC and _HAVE_PYOPENSSL,
+        "Sync stdlib ssl may return None for session on TLS 1.3; test limited to PyOpenSSL",
     )
     async def test_tls_session_cached_after_connect(self):
         await self.client.admin.command("ping")
diff --git a/test/test_ssl.py b/test/test_ssl.py
index 20a21958d2..39fda75859 100644
--- a/test/test_ssl.py
+++ b/test/test_ssl.py
@@ -170,28 +170,35 @@ def test_tls_session_reused_on_second_connection(self):
         _, kwargs = mock_ssl_context.wrap_socket.call_args
         self.assertIs(kwargs.get("session"), fake_session)
 
-    @unittest.skipUnless(
-        not _IS_SYNC and sys.version_info >= (3, 11),
-        "Async session injection requires Python 3.11+",
-    )
-    def test_async_tls_session_injected_into_sslobj(self):
-        """Cached TLS session is set on SSLObject before the handshake on Python 3.11+."""
-        import asyncio.sslproto as _sslproto
+    @unittest.skipUnless(not _IS_SYNC, "Tests async sslobject_class injection path only")
+    def test_async_tls_session_injected_via_sslobject_class(self):
+        """_SessionSSLContext sets sslobject_class on the real context for the wrap_bio() call."""
+        import ssl
         import unittest.mock as mock
 
-        from pymongo.pool_shared import _make_session_ssl_protocol, _SSLSessionCache
+        from pymongo.pool_shared import _SessionSSLContext
 
         fake_session = mock.MagicMock()
-        patched_cls = _make_session_ssl_protocol(fake_session)
+        real_ctx = mock.MagicMock()
+        real_ctx.sslobject_class = ssl.SSLObject
+        wrapped = _SessionSSLContext(real_ctx, fake_session)
+
+        observed_class = None
+
+        def capture_class(*args, **kwargs):
+            nonlocal observed_class
+            observed_class = real_ctx.sslobject_class
+            return mock.MagicMock()
 
-        mock_sslobj = mock.MagicMock()
-        instance = patched_cls.__new__(patched_cls)
-        instance._sslobj = mock_sslobj
-        # Call __init__ via the patched class, bypassing the real SSLProtocol init.
-        with mock.patch.object(_sslproto.SSLProtocol, "__init__", lambda *a, **kw: None):
-            patched_cls.__init__(instance)
+        real_ctx.wrap_bio.side_effect = capture_class
+        wrapped.wrap_bio("incoming", "outgoing", server_side=False)
 
-        self.assertEqual(mock_sslobj.session, fake_session)
+        # sslobject_class was our session-injecting subclass during the call
+        assert observed_class is not None
+        self.assertTrue(issubclass(observed_class, ssl.SSLObject))
+        self.assertIsNot(observed_class, ssl.SSLObject)
+        # sslobject_class was restored to the original after the call
+        self.assertIs(real_ctx.sslobject_class, ssl.SSLObject)
 
 
 class TestSSL(IntegrationTest):
@@ -745,7 +752,8 @@ def test_pool_has_ssl_session_cache(self):
 
     @client_context.require_tls
     @unittest.skipUnless(
-        _IS_SYNC and _HAVE_PYOPENSSL, "Session caching only applies to PyOpenSSL sync path"
+        _IS_SYNC and _HAVE_PYOPENSSL,
+        "Sync stdlib ssl may return None for session on TLS 1.3; test limited to PyOpenSSL",
     )
     def test_tls_session_cached_after_connect(self):
         self.client.admin.command("ping")

From eebb7c97423e7211e99c29d9ebad2fe68a0586d4 Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Thu, 11 Jun 2026 06:31:37 -0500
Subject: [PATCH 12/16] PYTHON-5272 Simplify async session injection to direct
 sslobject_class assignment

Drop the _SessionSSLContext wrapper entirely. On Python 3.11+
SSLProtocol.__init__ calls wrap_bio() synchronously before the first
event-loop yield, so setting ssl_context.sslobject_class directly is
race-free. A per-connection ssl.SSLObject subclass captures the cached
session in a closure and sets self.session after __init__. Session
injection is skipped silently on Python < 3.11 and for PyOpenSSL
contexts (which don't have sslobject_class).
---
 pymongo/pool_shared.py        | 69 ++++++++++-------------------------
 test/asynchronous/test_ssl.py | 41 +++++++++++----------
 test/test_ssl.py              | 41 +++++++++++----------
 3 files changed, 64 insertions(+), 87 deletions(-)

diff --git a/pymongo/pool_shared.py b/pymongo/pool_shared.py
index 99cf2002cb..9d35c82748 100644
--- a/pymongo/pool_shared.py
+++ b/pymongo/pool_shared.py
@@ -73,48 +73,6 @@ def _get_ssl_session(ssl_sock: Any) -> Optional[Any]:
     return getattr(ssl_sock, "session", None)
 
 
-# asyncio's create_connection does not support TLS session resumption natively.
-# https://github.com/python/cpython/issues/79152 tracks this; a patch was submitted
-# in 2018 but never merged, and the issue is now closed.
-# We work around this by wrapping the SSLContext to intercept wrap_bio(), which
-# asyncio always calls before the TLS handshake regardless of Python version.
-
-
-class _SessionSSLContext:
-    """Wraps an SSLContext to inject a cached TLS session via sslobject_class.
-
-    ssl.SSLContext.sslobject_class (Python 3.7+) controls the type returned by
-    wrap_bio().  We create a per-connection subclass of ssl.SSLObject that sets
-    self.session after __init__, then temporarily apply it inside wrap_bio().
-    The patch-and-restore is race-free because wrap_bio() is synchronous.
-    """
-
-    __slots__ = ("_ctx", "_session_class")
-
-    def __init__(self, ctx: Any, session: Any) -> None:
-        self._ctx = ctx
-        base = ctx.sslobject_class
-        _session = session  # captured by inner class; avoids any name clash with outer self
-
-        class _SessionSSLObject(base):  # type: ignore[misc,valid-type]
-            def __init__(self, *args: Any, **kwargs: Any) -> None:
-                super().__init__(*args, **kwargs)
-                self.session = _session
-
-        self._session_class = _SessionSSLObject
-
-    def wrap_bio(self, *args: Any, **kwargs: Any) -> Any:
-        orig = self._ctx.sslobject_class
-        self._ctx.sslobject_class = self._session_class
-        try:
-            return self._ctx.wrap_bio(*args, **kwargs)
-        finally:
-            self._ctx.sslobject_class = orig
-
-    def __getattr__(self, name: str) -> Any:
-        return getattr(self._ctx, name)
-
-
 try:
     from fcntl import F_GETFD, F_SETFD, FD_CLOEXEC, fcntl
 
@@ -389,12 +347,25 @@ async def _configured_protocol_interface(
         )
 
     host = address[0]
-    session = ssl_session_cache.get() if ssl_session_cache is not None else None
-    # Wrap the SSL context to inject the cached session via wrap_bio(), which
-    # asyncio calls before the TLS handshake on all supported Python versions.
-    effective_ssl_context = (
-        _SessionSSLContext(ssl_context, session) if session is not None else ssl_context
-    )
+    # asyncio does not support TLS session resumption natively (cpython#79152,
+    # closed without a fix).  On Python 3.11+ SSLProtocol.__init__ calls
+    # wrap_bio() synchronously before the first event-loop yield, so setting
+    # sslobject_class here is race-free; we skip injection on older versions.
+    if (
+        ssl_session_cache is not None
+        and sys.version_info >= (3, 11)
+        and isinstance(ssl_context, ssl.SSLContext)
+    ):
+        session = ssl_session_cache.get()
+        if session is not None:
+            _session = session
+
+            class _SessionSSLObject(ssl.SSLObject):
+                def __init__(self, *args: Any, **kwargs: Any) -> None:
+                    super().__init__(*args, **kwargs)
+                    self.session = _session
+
+            ssl_context.sslobject_class = _SessionSSLObject
     try:
         # We have to pass hostname / ip address to wrap_socket
         # to use SSLContext.check_hostname.
@@ -402,7 +373,7 @@ async def _configured_protocol_interface(
             lambda: PyMongoProtocol(timeout=timeout),
             sock=sock,
             server_hostname=host,
-            ssl=effective_ssl_context,
+            ssl=ssl_context,
         )
     except _CertificateError:
         # Raise _CertificateError directly like we do after match_hostname
diff --git a/test/asynchronous/test_ssl.py b/test/asynchronous/test_ssl.py
index ad0fa8e7cb..49e2b4b555 100644
--- a/test/asynchronous/test_ssl.py
+++ b/test/asynchronous/test_ssl.py
@@ -170,35 +170,38 @@ def test_tls_session_reused_on_second_connection(self):
         _, kwargs = mock_ssl_context.wrap_socket.call_args
         self.assertIs(kwargs.get("session"), fake_session)
 
-    @unittest.skipUnless(not _IS_SYNC, "Tests async sslobject_class injection path only")
+    @unittest.skipUnless(
+        not _IS_SYNC and sys.version_info >= (3, 11),
+        "Tests async sslobject_class injection (Python 3.11+ only)",
+    )
     def test_async_tls_session_injected_via_sslobject_class(self):
-        """_SessionSSLContext sets sslobject_class on the real context for the wrap_bio() call."""
+        """On Python 3.11+, a cached session is injected by setting sslobject_class."""
         import ssl
         import unittest.mock as mock
 
-        from pymongo.pool_shared import _SessionSSLContext
+        from pymongo.pool_shared import _SSLSessionCache
 
         fake_session = mock.MagicMock()
-        real_ctx = mock.MagicMock()
-        real_ctx.sslobject_class = ssl.SSLObject
-        wrapped = _SessionSSLContext(real_ctx, fake_session)
+        cache = _SSLSessionCache()
+        cache.set(fake_session)
+
+        real_ctx = ssl.create_default_context()
+        self.assertIs(real_ctx.sslobject_class, ssl.SSLObject)
 
-        observed_class = None
+        # Simulate what _configured_protocol_interface does
+        session = cache.get()
+        assert session is not None
+        _session = session
 
-        def capture_class(*args, **kwargs):
-            nonlocal observed_class
-            observed_class = real_ctx.sslobject_class
-            return mock.MagicMock()
+        class _SessionSSLObject(ssl.SSLObject):
+            def __init__(self, *args, **kwargs):
+                super().__init__(*args, **kwargs)
+                self.session = _session
 
-        real_ctx.wrap_bio.side_effect = capture_class
-        wrapped.wrap_bio("incoming", "outgoing", server_side=False)
+        real_ctx.sslobject_class = _SessionSSLObject
 
-        # sslobject_class was our session-injecting subclass during the call
-        assert observed_class is not None
-        self.assertTrue(issubclass(observed_class, ssl.SSLObject))
-        self.assertIsNot(observed_class, ssl.SSLObject)
-        # sslobject_class was restored to the original after the call
-        self.assertIs(real_ctx.sslobject_class, ssl.SSLObject)
+        self.assertIs(real_ctx.sslobject_class, _SessionSSLObject)
+        self.assertTrue(issubclass(real_ctx.sslobject_class, ssl.SSLObject))
 
 
 class TestSSL(AsyncIntegrationTest):
diff --git a/test/test_ssl.py b/test/test_ssl.py
index 39fda75859..3f0b48ec12 100644
--- a/test/test_ssl.py
+++ b/test/test_ssl.py
@@ -170,35 +170,38 @@ def test_tls_session_reused_on_second_connection(self):
         _, kwargs = mock_ssl_context.wrap_socket.call_args
         self.assertIs(kwargs.get("session"), fake_session)
 
-    @unittest.skipUnless(not _IS_SYNC, "Tests async sslobject_class injection path only")
+    @unittest.skipUnless(
+        not _IS_SYNC and sys.version_info >= (3, 11),
+        "Tests async sslobject_class injection (Python 3.11+ only)",
+    )
     def test_async_tls_session_injected_via_sslobject_class(self):
-        """_SessionSSLContext sets sslobject_class on the real context for the wrap_bio() call."""
+        """On Python 3.11+, a cached session is injected by setting sslobject_class."""
         import ssl
         import unittest.mock as mock
 
-        from pymongo.pool_shared import _SessionSSLContext
+        from pymongo.pool_shared import _SSLSessionCache
 
         fake_session = mock.MagicMock()
-        real_ctx = mock.MagicMock()
-        real_ctx.sslobject_class = ssl.SSLObject
-        wrapped = _SessionSSLContext(real_ctx, fake_session)
+        cache = _SSLSessionCache()
+        cache.set(fake_session)
+
+        real_ctx = ssl.create_default_context()
+        self.assertIs(real_ctx.sslobject_class, ssl.SSLObject)
 
-        observed_class = None
+        # Simulate what _configured_socket_interface does
+        session = cache.get()
+        assert session is not None
+        _session = session
 
-        def capture_class(*args, **kwargs):
-            nonlocal observed_class
-            observed_class = real_ctx.sslobject_class
-            return mock.MagicMock()
+        class _SessionSSLObject(ssl.SSLObject):
+            def __init__(self, *args, **kwargs):
+                super().__init__(*args, **kwargs)
+                self.session = _session
 
-        real_ctx.wrap_bio.side_effect = capture_class
-        wrapped.wrap_bio("incoming", "outgoing", server_side=False)
+        real_ctx.sslobject_class = _SessionSSLObject
 
-        # sslobject_class was our session-injecting subclass during the call
-        assert observed_class is not None
-        self.assertTrue(issubclass(observed_class, ssl.SSLObject))
-        self.assertIsNot(observed_class, ssl.SSLObject)
-        # sslobject_class was restored to the original after the call
-        self.assertIs(real_ctx.sslobject_class, ssl.SSLObject)
+        self.assertIs(real_ctx.sslobject_class, _SessionSSLObject)
+        self.assertTrue(issubclass(real_ctx.sslobject_class, ssl.SSLObject))
 
 
 class TestSSL(IntegrationTest):

From 18f64e4eb9d8e085f1caa38b9e896bcfc2a7a81a Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Thu, 11 Jun 2026 06:41:54 -0500
Subject: [PATCH 13/16] PYTHON-5272 Note async session injection is Python
 3.11+ only; drop unnecessary isinstance guard

---
 doc/changelog.rst      |  2 ++
 pymongo/pool_shared.py | 11 ++++-------
 2 files changed, 6 insertions(+), 7 deletions(-)

diff --git a/doc/changelog.rst b/doc/changelog.rst
index 60c87c43e1..585ea045c8 100644
--- a/doc/changelog.rst
+++ b/doc/changelog.rst
@@ -6,6 +6,8 @@ Changes in Version 4.18.0
 
 - Improved TLS connection performance by reusing TLS sessions across connections
   to the same server, avoiding a full handshake on each new connection.
+  Session resumption is supported on all Python versions for synchronous clients
+  and on Python 3.11+ for async clients.
 
 Changes in Version 4.17.0 (2026/04/20)
 --------------------------------------
diff --git a/pymongo/pool_shared.py b/pymongo/pool_shared.py
index 9d35c82748..d0d04a14c3 100644
--- a/pymongo/pool_shared.py
+++ b/pymongo/pool_shared.py
@@ -350,12 +350,9 @@ async def _configured_protocol_interface(
     # asyncio does not support TLS session resumption natively (cpython#79152,
     # closed without a fix).  On Python 3.11+ SSLProtocol.__init__ calls
     # wrap_bio() synchronously before the first event-loop yield, so setting
-    # sslobject_class here is race-free; we skip injection on older versions.
-    if (
-        ssl_session_cache is not None
-        and sys.version_info >= (3, 11)
-        and isinstance(ssl_context, ssl.SSLContext)
-    ):
+    # sslobject_class is race-free.  Session injection is skipped on older
+    # Python versions.  (The async path always uses stdlib ssl, never PyOpenSSL.)
+    if ssl_session_cache is not None and sys.version_info >= (3, 11):
         session = ssl_session_cache.get()
         if session is not None:
             _session = session
@@ -365,7 +362,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
                     super().__init__(*args, **kwargs)
                     self.session = _session
 
-            ssl_context.sslobject_class = _SessionSSLObject
+            ssl_context.sslobject_class = _SessionSSLObject  # type: ignore[attr-defined]
     try:
         # We have to pass hostname / ip address to wrap_socket
         # to use SSLContext.check_hostname.

From e99c4e3ecdba6b5d43c9c9977924ff819fdcf2e5 Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Thu, 11 Jun 2026 07:39:59 -0500
Subject: [PATCH 14/16] PYTHON-5272 Replace _SSLSessionCache class with a plain
 list[Any]

A single-element list provides an atomic mutable slot under the GIL
without needing a custom class or a threading.Lock.  The pool
initialises it as [None]; readers use cache[0] and writers assign
cache[0] = session.
---
 pymongo/asynchronous/pool.py  |  5 ++---
 pymongo/pool_shared.py        | 31 ++++++-------------------------
 pymongo/synchronous/pool.py   |  5 ++---
 test/asynchronous/test_ssl.py | 35 +++++++++++++----------------------
 test/test_ssl.py              | 35 +++++++++++++----------------------
 5 files changed, 36 insertions(+), 75 deletions(-)

diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py
index f2b6df2f3e..e8bea0980a 100644
--- a/pymongo/asynchronous/pool.py
+++ b/pymongo/asynchronous/pool.py
@@ -85,7 +85,6 @@
     _CancellationContext,
     _configured_protocol_interface,
     _raise_connection_failure,
-    _SSLSessionCache,
 )
 from pymongo.read_preferences import ReadPreference
 from pymongo.server_api import _add_to_command
@@ -755,8 +754,8 @@ def __init__(
         self._pending = 0
         self._max_connecting = self.opts.max_connecting
         self._client_id = client_id
-        self._ssl_session_cache: Optional[_SSLSessionCache] = (
-            _SSLSessionCache() if self.opts._ssl_context is not None else None
+        self._ssl_session_cache: Optional[list[Any]] = (
+            [None] if self.opts._ssl_context is not None else None
         )
         # Log before publishing event to prevent potential listener preemption in tests
         if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
diff --git a/pymongo/pool_shared.py b/pymongo/pool_shared.py
index d0d04a14c3..6a98141067 100644
--- a/pymongo/pool_shared.py
+++ b/pymongo/pool_shared.py
@@ -20,7 +20,6 @@
 import socket
 import ssl
 import sys
-import threading
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -48,24 +47,6 @@
     from pymongo.typings import _Address
 
 
-class _SSLSessionCache:
-    """Thread-safe cache for a single TLS session per pool, enabling session resumption."""
-
-    __slots__ = ("_session", "_lock")
-
-    def __init__(self) -> None:
-        self._session: Optional[Any] = None
-        self._lock = threading.Lock()
-
-    def get(self) -> Optional[Any]:
-        with self._lock:
-            return self._session
-
-    def set(self, session: Any) -> None:
-        with self._lock:
-            self._session = session
-
-
 def _get_ssl_session(ssl_sock: Any) -> Optional[Any]:
     """Return the TLS session from an SSL socket, handling both PyOpenSSL and stdlib ssl."""
     if hasattr(ssl_sock, "get_session"):
@@ -327,7 +308,7 @@ async def _async_configured_socket(
 async def _configured_protocol_interface(
     address: _Address,
     options: PoolOptions,
-    ssl_session_cache: Optional[_SSLSessionCache] = None,
+    ssl_session_cache: Optional[list[Any]] = None,
 ) -> AsyncNetworkingInterface:
     """Given (host, port) and PoolOptions, return a configured AsyncNetworkingInterface.
 
@@ -353,7 +334,7 @@ async def _configured_protocol_interface(
     # sslobject_class is race-free.  Session injection is skipped on older
     # Python versions.  (The async path always uses stdlib ssl, never PyOpenSSL.)
     if ssl_session_cache is not None and sys.version_info >= (3, 11):
-        session = ssl_session_cache.get()
+        session = ssl_session_cache[0]
         if session is not None:
             _session = session
 
@@ -399,7 +380,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         if ssl_obj is not None:
             new_session = ssl_obj.session
             if new_session is not None:
-                ssl_session_cache.set(new_session)
+                ssl_session_cache[0] = new_session
 
     return AsyncNetworkingInterface((transport, protocol))
 
@@ -526,7 +507,7 @@ def _configured_socket(address: _Address, options: PoolOptions) -> Union[socket.
 def _configured_socket_interface(
     address: _Address,
     options: PoolOptions,
-    ssl_session_cache: Optional[_SSLSessionCache] = None,
+    ssl_session_cache: Optional[list[Any]] = None,
 ) -> NetworkingInterface:
     """Given (host, port) and PoolOptions, return a NetworkingInterface wrapping a configured socket.
 
@@ -542,7 +523,7 @@ def _configured_socket_interface(
         return NetworkingInterface(sock)
 
     host = address[0]
-    session = ssl_session_cache.get() if ssl_session_cache is not None else None
+    session = ssl_session_cache[0] if ssl_session_cache is not None else None
     try:
         # We have to pass hostname / ip address to wrap_socket
         # to use SSLContext.check_hostname.
@@ -576,7 +557,7 @@ def _configured_socket_interface(
     if ssl_session_cache is not None:
         new_session = _get_ssl_session(ssl_sock)
         if new_session is not None:
-            ssl_session_cache.set(new_session)
+            ssl_session_cache[0] = new_session
 
     ssl_sock.settimeout(options.socket_timeout)
     return NetworkingInterface(ssl_sock)
diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py
index 8e7dacef19..467faaf20c 100644
--- a/pymongo/synchronous/pool.py
+++ b/pymongo/synchronous/pool.py
@@ -82,7 +82,6 @@
     _CancellationContext,
     _configured_socket_interface,
     _raise_connection_failure,
-    _SSLSessionCache,
 )
 from pymongo.read_preferences import ReadPreference
 from pymongo.server_api import _add_to_command
@@ -753,8 +752,8 @@ def __init__(
         self._pending = 0
         self._max_connecting = self.opts.max_connecting
         self._client_id = client_id
-        self._ssl_session_cache: Optional[_SSLSessionCache] = (
-            _SSLSessionCache() if self.opts._ssl_context is not None else None
+        self._ssl_session_cache: Optional[list[Any]] = (
+            [None] if self.opts._ssl_context is not None else None
         )
         # Log before publishing event to prevent potential listener preemption in tests
         if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
diff --git a/test/asynchronous/test_ssl.py b/test/asynchronous/test_ssl.py
index 49e2b4b555..fb444c85d3 100644
--- a/test/asynchronous/test_ssl.py
+++ b/test/asynchronous/test_ssl.py
@@ -129,25 +129,22 @@ def test_use_pyopenssl_when_available(self):
         self.assertTrue(HAVE_PYSSL)
 
     def test_ssl_session_cache(self):
-        from pymongo.pool_shared import _SSLSessionCache
-
-        cache = _SSLSessionCache()
-        self.assertIsNone(cache.get())
-        cache.set("session")
-        self.assertEqual(cache.get(), "session")
-        cache.set("new_session")
-        self.assertEqual(cache.get(), "new_session")
+        cache: list = [None]
+        self.assertIsNone(cache[0])
+        cache[0] = "session"
+        self.assertEqual(cache[0], "session")
+        cache[0] = "new_session"
+        self.assertEqual(cache[0], "new_session")
 
     @unittest.skipUnless(_IS_SYNC, "Tests sync wrap_socket path only")
     def test_tls_session_reused_on_second_connection(self):
         """Cached TLS session is passed to wrap_socket on subsequent connections."""
         import unittest.mock as mock
 
-        from pymongo.pool_shared import _configured_socket_interface, _SSLSessionCache
+        from pymongo.pool_shared import _configured_socket_interface
 
         fake_session = object()
-        cache = _SSLSessionCache()
-        cache.set(fake_session)
+        cache: list = [fake_session]
 
         fake_ssl_sock = mock.MagicMock()
         fake_ssl_sock.getpeercert.return_value = {}
@@ -177,19 +174,15 @@ def test_tls_session_reused_on_second_connection(self):
     def test_async_tls_session_injected_via_sslobject_class(self):
         """On Python 3.11+, a cached session is injected by setting sslobject_class."""
         import ssl
-        import unittest.mock as mock
-
-        from pymongo.pool_shared import _SSLSessionCache
 
-        fake_session = mock.MagicMock()
-        cache = _SSLSessionCache()
-        cache.set(fake_session)
+        fake_session = object()
+        cache: list = [fake_session]
 
         real_ctx = ssl.create_default_context()
         self.assertIs(real_ctx.sslobject_class, ssl.SSLObject)
 
         # Simulate what _configured_protocol_interface does
-        session = cache.get()
+        session = cache[0]
         assert session is not None
         _session = session
 
@@ -750,10 +743,8 @@ async def test_pyopenssl_ignored_in_async(self):
 
     @async_client_context.require_tls
     async def test_pool_has_ssl_session_cache(self):
-        from pymongo.pool_shared import _SSLSessionCache
-
         pool = list(self.client._topology._servers.values())[0].pool
-        self.assertIsInstance(pool._ssl_session_cache, _SSLSessionCache)
+        self.assertIsInstance(pool._ssl_session_cache, list)
 
     @async_client_context.require_tls
     @unittest.skipUnless(
@@ -764,7 +755,7 @@ async def test_tls_session_cached_after_connect(self):
         await self.client.admin.command("ping")
         pool = list(self.client._topology._servers.values())[0].pool
         self.assertIsNotNone(pool._ssl_session_cache)
-        self.assertIsNotNone(pool._ssl_session_cache.get())
+        self.assertIsNotNone(pool._ssl_session_cache[0])
 
 
 if __name__ == "__main__":
diff --git a/test/test_ssl.py b/test/test_ssl.py
index 3f0b48ec12..936a8684fa 100644
--- a/test/test_ssl.py
+++ b/test/test_ssl.py
@@ -129,25 +129,22 @@ def test_use_pyopenssl_when_available(self):
         self.assertTrue(HAVE_PYSSL)
 
     def test_ssl_session_cache(self):
-        from pymongo.pool_shared import _SSLSessionCache
-
-        cache = _SSLSessionCache()
-        self.assertIsNone(cache.get())
-        cache.set("session")
-        self.assertEqual(cache.get(), "session")
-        cache.set("new_session")
-        self.assertEqual(cache.get(), "new_session")
+        cache: list = [None]
+        self.assertIsNone(cache[0])
+        cache[0] = "session"
+        self.assertEqual(cache[0], "session")
+        cache[0] = "new_session"
+        self.assertEqual(cache[0], "new_session")
 
     @unittest.skipUnless(_IS_SYNC, "Tests sync wrap_socket path only")
     def test_tls_session_reused_on_second_connection(self):
         """Cached TLS session is passed to wrap_socket on subsequent connections."""
         import unittest.mock as mock
 
-        from pymongo.pool_shared import _configured_socket_interface, _SSLSessionCache
+        from pymongo.pool_shared import _configured_socket_interface
 
         fake_session = object()
-        cache = _SSLSessionCache()
-        cache.set(fake_session)
+        cache: list = [fake_session]
 
         fake_ssl_sock = mock.MagicMock()
         fake_ssl_sock.getpeercert.return_value = {}
@@ -177,19 +174,15 @@ def test_tls_session_reused_on_second_connection(self):
     def test_async_tls_session_injected_via_sslobject_class(self):
         """On Python 3.11+, a cached session is injected by setting sslobject_class."""
         import ssl
-        import unittest.mock as mock
-
-        from pymongo.pool_shared import _SSLSessionCache
 
-        fake_session = mock.MagicMock()
-        cache = _SSLSessionCache()
-        cache.set(fake_session)
+        fake_session = object()
+        cache: list = [fake_session]
 
         real_ctx = ssl.create_default_context()
         self.assertIs(real_ctx.sslobject_class, ssl.SSLObject)
 
         # Simulate what _configured_socket_interface does
-        session = cache.get()
+        session = cache[0]
         assert session is not None
         _session = session
 
@@ -748,10 +741,8 @@ def test_pyopenssl_ignored_in_async(self):
 
     @client_context.require_tls
     def test_pool_has_ssl_session_cache(self):
-        from pymongo.pool_shared import _SSLSessionCache
-
         pool = list(self.client._topology._servers.values())[0].pool
-        self.assertIsInstance(pool._ssl_session_cache, _SSLSessionCache)
+        self.assertIsInstance(pool._ssl_session_cache, list)
 
     @client_context.require_tls
     @unittest.skipUnless(
@@ -762,7 +753,7 @@ def test_tls_session_cached_after_connect(self):
         self.client.admin.command("ping")
         pool = list(self.client._topology._servers.values())[0].pool
         self.assertIsNotNone(pool._ssl_session_cache)
-        self.assertIsNotNone(pool._ssl_session_cache.get())
+        self.assertIsNotNone(pool._ssl_session_cache[0])
 
 
 if __name__ == "__main__":

From 897e40a51236801f395d7d237abdc14f2f338072 Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Thu, 11 Jun 2026 13:25:16 -0500
Subject: [PATCH 15/16] PYTHON-5272 Add unit tests covering _get_ssl_session
 and async session-save path

---
 test/asynchronous/test_ssl.py | 73 +++++++++++++++++++++++++++++++++++
 test/test_ssl.py              | 71 ++++++++++++++++++++++++++++++++++
 2 files changed, 144 insertions(+)

diff --git a/test/asynchronous/test_ssl.py b/test/asynchronous/test_ssl.py
index fb444c85d3..6c8c284016 100644
--- a/test/asynchronous/test_ssl.py
+++ b/test/asynchronous/test_ssl.py
@@ -167,6 +167,29 @@ def test_tls_session_reused_on_second_connection(self):
         _, kwargs = mock_ssl_context.wrap_socket.call_args
         self.assertIs(kwargs.get("session"), fake_session)
 
+    def test_get_ssl_session_pyopenssl_style(self):
+        """_get_ssl_session uses get_session() when available (PyOpenSSL path)."""
+        import unittest.mock as mock
+
+        from pymongo.pool_shared import _get_ssl_session
+
+        fake_session = object()
+        conn = mock.MagicMock()
+        conn.get_session.return_value = fake_session
+        self.assertIs(_get_ssl_session(conn), fake_session)
+        conn.get_session.assert_called_once()
+
+    def test_get_ssl_session_stdlib_style(self):
+        """_get_ssl_session falls back to .session attribute (stdlib ssl path)."""
+        from pymongo.pool_shared import _get_ssl_session
+
+        fake_session = object()
+
+        class FakeSSLSock:
+            session = fake_session
+
+        self.assertIs(_get_ssl_session(FakeSSLSock()), fake_session)
+
     @unittest.skipUnless(
         not _IS_SYNC and sys.version_info >= (3, 11),
         "Tests async sslobject_class injection (Python 3.11+ only)",
@@ -196,6 +219,56 @@ def __init__(self, *args, **kwargs):
         self.assertIs(real_ctx.sslobject_class, _SessionSSLObject)
         self.assertTrue(issubclass(real_ctx.sslobject_class, ssl.SSLObject))
 
+    @unittest.skipUnless(not _IS_SYNC, "Tests async _configured_protocol_interface only")
+    def test_async_configured_protocol_saves_session_to_cache(self):
+        """After a successful TLS connection the session is stored in the cache."""
+        import asyncio
+        import ssl
+        import unittest.mock as mock
+
+        from pymongo.pool_shared import _configured_protocol_interface
+
+        fake_session = object()
+        cache: list = [None]
+
+        mock_ssl_obj = mock.MagicMock()
+        mock_ssl_obj.session = fake_session
+
+        mock_transport = mock.MagicMock()
+        mock_transport.get_extra_info.side_effect = lambda key: (
+            mock_ssl_obj if key == "ssl_object" else None
+        )
+
+        real_ctx = ssl.create_default_context()
+        real_ctx.check_hostname = False
+        real_ctx.verify_mode = ssl.CERT_NONE
+
+        mock_opts = mock.MagicMock()
+        mock_opts._ssl_context = real_ctx
+        mock_opts.socket_timeout = None
+        mock_opts.tls_allow_invalid_hostnames = True
+
+        mock_loop = mock.MagicMock()
+        mock_loop.create_connection = mock.AsyncMock(
+            return_value=(mock_transport, mock.MagicMock())
+        )
+
+        async def run():
+            with (
+                mock.patch(
+                    "pymongo.pool_shared._async_create_connection",
+                    new=mock.AsyncMock(return_value=mock.MagicMock()),
+                ),
+                mock.patch(
+                    "pymongo.pool_shared.asyncio.get_running_loop",
+                    return_value=mock_loop,
+                ),
+            ):
+                await _configured_protocol_interface(("localhost", 27017), mock_opts, cache)
+
+        asyncio.run(run())
+        self.assertIs(cache[0], fake_session)
+
 
 class TestSSL(AsyncIntegrationTest):
     saved_port: int
diff --git a/test/test_ssl.py b/test/test_ssl.py
index 936a8684fa..58d1407d83 100644
--- a/test/test_ssl.py
+++ b/test/test_ssl.py
@@ -167,6 +167,29 @@ def test_tls_session_reused_on_second_connection(self):
         _, kwargs = mock_ssl_context.wrap_socket.call_args
         self.assertIs(kwargs.get("session"), fake_session)
 
+    def test_get_ssl_session_pyopenssl_style(self):
+        """_get_ssl_session uses get_session() when available (PyOpenSSL path)."""
+        import unittest.mock as mock
+
+        from pymongo.pool_shared import _get_ssl_session
+
+        fake_session = object()
+        conn = mock.MagicMock()
+        conn.get_session.return_value = fake_session
+        self.assertIs(_get_ssl_session(conn), fake_session)
+        conn.get_session.assert_called_once()
+
+    def test_get_ssl_session_stdlib_style(self):
+        """_get_ssl_session falls back to .session attribute (stdlib ssl path)."""
+        from pymongo.pool_shared import _get_ssl_session
+
+        fake_session = object()
+
+        class FakeSSLSock:
+            session = fake_session
+
+        self.assertIs(_get_ssl_session(FakeSSLSock()), fake_session)
+
     @unittest.skipUnless(
         not _IS_SYNC and sys.version_info >= (3, 11),
         "Tests async sslobject_class injection (Python 3.11+ only)",
@@ -196,6 +219,54 @@ def __init__(self, *args, **kwargs):
         self.assertIs(real_ctx.sslobject_class, _SessionSSLObject)
         self.assertTrue(issubclass(real_ctx.sslobject_class, ssl.SSLObject))
 
+    @unittest.skipUnless(not _IS_SYNC, "Tests async _configured_socket_interface only")
+    def test_async_configured_protocol_saves_session_to_cache(self):
+        """After a successful TLS connection the session is stored in the cache."""
+        import asyncio
+        import ssl
+        import unittest.mock as mock
+
+        from pymongo.pool_shared import _configured_socket_interface
+
+        fake_session = object()
+        cache: list = [None]
+
+        mock_ssl_obj = mock.MagicMock()
+        mock_ssl_obj.session = fake_session
+
+        mock_transport = mock.MagicMock()
+        mock_transport.get_extra_info.side_effect = lambda key: (
+            mock_ssl_obj if key == "ssl_object" else None
+        )
+
+        real_ctx = ssl.create_default_context()
+        real_ctx.check_hostname = False
+        real_ctx.verify_mode = ssl.CERT_NONE
+
+        mock_opts = mock.MagicMock()
+        mock_opts._ssl_context = real_ctx
+        mock_opts.socket_timeout = None
+        mock_opts.tls_allow_invalid_hostnames = True
+
+        mock_loop = mock.MagicMock()
+        mock_loop.create_connection = mock.Mock(return_value=(mock_transport, mock.MagicMock()))
+
+        def run():
+            with (
+                mock.patch(
+                    "pymongo.pool_shared._create_connection",
+                    new=mock.SyncMock(return_value=mock.MagicMock()),
+                ),
+                mock.patch(
+                    "pymongo.pool_shared.asyncio.get_running_loop",
+                    return_value=mock_loop,
+                ),
+            ):
+                _configured_socket_interface(("localhost", 27017), mock_opts, cache)
+
+        asyncio.run(run())
+        self.assertIs(cache[0], fake_session)
+
 
 class TestSSL(IntegrationTest):
     saved_port: int

From e5a371cc52f990729af65a415390fe4e68c5896a Mon Sep 17 00:00:00 2001
From: Steven Silvester <steve.silvester@mongodb.com>
Date: Thu, 11 Jun 2026 14:13:31 -0500
Subject: [PATCH 16/16] PYTHON-5272 Add coverage for async TLS session
 injection and save paths

Add three unit tests for _configured_protocol_interface to cover the
previously-missing branches:

- test_async_configured_protocol_injects_session_via_sslobject_class:
  starts the cache with a pre-populated session so the if-session-is-not-None
  branch runs; exercises the _SessionSSLObject.__init__ body via wrap_bio
  with a patched ssl.SSLObject.session setter.
- test_async_configured_protocol_no_cache: passes ssl_session_cache=None to
  cover the False branches of both the injection guard and the save guard.
- test_async_configured_protocol_new_session_is_none: returns None from
  ssl_obj.session to cover the False branch of if-new_session-is-not-None.
---
 test/asynchronous/test_ssl.py | 151 ++++++++++++++++++++++++++++++++++
 test/test_ssl.py              | 145 ++++++++++++++++++++++++++++++++
 2 files changed, 296 insertions(+)

diff --git a/test/asynchronous/test_ssl.py b/test/asynchronous/test_ssl.py
index 6c8c284016..5a50f54729 100644
--- a/test/asynchronous/test_ssl.py
+++ b/test/asynchronous/test_ssl.py
@@ -269,6 +269,157 @@ async def run():
         asyncio.run(run())
         self.assertIs(cache[0], fake_session)
 
+    @unittest.skipUnless(
+        not _IS_SYNC and sys.version_info >= (3, 11),
+        "Tests async session injection on Python 3.11+",
+    )
+    def test_async_configured_protocol_injects_session_via_sslobject_class(self):
+        """When the cache has a session, sslobject_class is set and its __init__ body runs."""
+        import asyncio
+        import ssl
+        import unittest.mock as mock
+
+        from pymongo.pool_shared import _configured_protocol_interface
+
+        initial_session = object()
+        cache: list = [initial_session]
+
+        mock_transport = mock.MagicMock()
+        mock_transport.get_extra_info.return_value = None  # no ssl_object → save block skips
+
+        real_ctx = ssl.create_default_context()
+        real_ctx.check_hostname = False
+        real_ctx.verify_mode = ssl.CERT_NONE
+
+        mock_opts = mock.MagicMock()
+        mock_opts._ssl_context = real_ctx
+        mock_opts.socket_timeout = None
+        mock_opts.tls_allow_invalid_hostnames = True
+
+        mock_loop = mock.MagicMock()
+        mock_loop.create_connection = mock.AsyncMock(
+            return_value=(mock_transport, mock.MagicMock())
+        )
+
+        async def run():
+            with (
+                mock.patch(
+                    "pymongo.pool_shared._async_create_connection",
+                    new=mock.AsyncMock(return_value=mock.MagicMock()),
+                ),
+                mock.patch(
+                    "pymongo.pool_shared.asyncio.get_running_loop",
+                    return_value=mock_loop,
+                ),
+            ):
+                await _configured_protocol_interface(("localhost", 27017), mock_opts, cache)
+
+        asyncio.run(run())
+
+        session_cls = real_ctx.sslobject_class  # type: ignore[attr-defined]
+        self.assertIsNot(session_cls, ssl.SSLObject)
+        self.assertTrue(issubclass(session_cls, ssl.SSLObject))
+
+        # Exercise the __init__ body (super().__init__ + self.session = _session) by
+        # calling wrap_bio, patching the session setter to accept non-SSLSession objects.
+        incoming = ssl.MemoryBIO()
+        outgoing = ssl.MemoryBIO()
+        no_op_session = property(lambda s: None, lambda s, v: None)
+        with mock.patch.object(ssl.SSLObject, "session", no_op_session):
+            ssl_obj = real_ctx.wrap_bio(incoming, outgoing, server_side=False)
+        self.assertIsInstance(ssl_obj, ssl.SSLObject)
+
+    @unittest.skipUnless(not _IS_SYNC, "Tests async _configured_protocol_interface only")
+    def test_async_configured_protocol_no_cache(self):
+        """When ssl_session_cache is None, no injection or save occurs."""
+        import asyncio
+        import ssl
+        import unittest.mock as mock
+
+        from pymongo.pool_shared import _configured_protocol_interface
+
+        real_ctx = ssl.create_default_context()
+        real_ctx.check_hostname = False
+        real_ctx.verify_mode = ssl.CERT_NONE
+
+        mock_opts = mock.MagicMock()
+        mock_opts._ssl_context = real_ctx
+        mock_opts.socket_timeout = None
+        mock_opts.tls_allow_invalid_hostnames = True
+
+        mock_transport = mock.MagicMock()
+        mock_transport.get_extra_info.return_value = None
+
+        mock_loop = mock.MagicMock()
+        mock_loop.create_connection = mock.AsyncMock(
+            return_value=(mock_transport, mock.MagicMock())
+        )
+
+        async def run():
+            with (
+                mock.patch(
+                    "pymongo.pool_shared._async_create_connection",
+                    new=mock.AsyncMock(return_value=mock.MagicMock()),
+                ),
+                mock.patch(
+                    "pymongo.pool_shared.asyncio.get_running_loop",
+                    return_value=mock_loop,
+                ),
+            ):
+                await _configured_protocol_interface(("localhost", 27017), mock_opts, None)
+
+        asyncio.run(run())
+        self.assertIs(real_ctx.sslobject_class, ssl.SSLObject)  # type: ignore[attr-defined]
+
+    @unittest.skipUnless(not _IS_SYNC, "Tests async _configured_protocol_interface only")
+    def test_async_configured_protocol_new_session_is_none(self):
+        """When ssl_object.session is None after connect, the cache is not updated."""
+        import asyncio
+        import ssl
+        import unittest.mock as mock
+
+        from pymongo.pool_shared import _configured_protocol_interface
+
+        cache: list = [None]
+
+        mock_ssl_obj = mock.MagicMock()
+        mock_ssl_obj.session = None
+
+        mock_transport = mock.MagicMock()
+        mock_transport.get_extra_info.side_effect = lambda key: (
+            mock_ssl_obj if key == "ssl_object" else None
+        )
+
+        real_ctx = ssl.create_default_context()
+        real_ctx.check_hostname = False
+        real_ctx.verify_mode = ssl.CERT_NONE
+
+        mock_opts = mock.MagicMock()
+        mock_opts._ssl_context = real_ctx
+        mock_opts.socket_timeout = None
+        mock_opts.tls_allow_invalid_hostnames = True
+
+        mock_loop = mock.MagicMock()
+        mock_loop.create_connection = mock.AsyncMock(
+            return_value=(mock_transport, mock.MagicMock())
+        )
+
+        async def run():
+            with (
+                mock.patch(
+                    "pymongo.pool_shared._async_create_connection",
+                    new=mock.AsyncMock(return_value=mock.MagicMock()),
+                ),
+                mock.patch(
+                    "pymongo.pool_shared.asyncio.get_running_loop",
+                    return_value=mock_loop,
+                ),
+            ):
+                await _configured_protocol_interface(("localhost", 27017), mock_opts, cache)
+
+        asyncio.run(run())
+        self.assertIsNone(cache[0])
+
 
 class TestSSL(AsyncIntegrationTest):
     saved_port: int
diff --git a/test/test_ssl.py b/test/test_ssl.py
index 58d1407d83..f5070a7e39 100644
--- a/test/test_ssl.py
+++ b/test/test_ssl.py
@@ -267,6 +267,151 @@ def run():
         asyncio.run(run())
         self.assertIs(cache[0], fake_session)
 
+    @unittest.skipUnless(
+        not _IS_SYNC and sys.version_info >= (3, 11),
+        "Tests async session injection on Python 3.11+",
+    )
+    def test_async_configured_protocol_injects_session_via_sslobject_class(self):
+        """When the cache has a session, sslobject_class is set and its __init__ body runs."""
+        import asyncio
+        import ssl
+        import unittest.mock as mock
+
+        from pymongo.pool_shared import _configured_socket_interface
+
+        initial_session = object()
+        cache: list = [initial_session]
+
+        mock_transport = mock.MagicMock()
+        mock_transport.get_extra_info.return_value = None  # no ssl_object → save block skips
+
+        real_ctx = ssl.create_default_context()
+        real_ctx.check_hostname = False
+        real_ctx.verify_mode = ssl.CERT_NONE
+
+        mock_opts = mock.MagicMock()
+        mock_opts._ssl_context = real_ctx
+        mock_opts.socket_timeout = None
+        mock_opts.tls_allow_invalid_hostnames = True
+
+        mock_loop = mock.MagicMock()
+        mock_loop.create_connection = mock.Mock(return_value=(mock_transport, mock.MagicMock()))
+
+        def run():
+            with (
+                mock.patch(
+                    "pymongo.pool_shared._create_connection",
+                    new=mock.SyncMock(return_value=mock.MagicMock()),
+                ),
+                mock.patch(
+                    "pymongo.pool_shared.asyncio.get_running_loop",
+                    return_value=mock_loop,
+                ),
+            ):
+                _configured_socket_interface(("localhost", 27017), mock_opts, cache)
+
+        asyncio.run(run())
+
+        session_cls = real_ctx.sslobject_class  # type: ignore[attr-defined]
+        self.assertIsNot(session_cls, ssl.SSLObject)
+        self.assertTrue(issubclass(session_cls, ssl.SSLObject))
+
+        # Exercise the __init__ body (super().__init__ + self.session = _session) by
+        # calling wrap_bio, patching the session setter to accept non-SSLSession objects.
+        incoming = ssl.MemoryBIO()
+        outgoing = ssl.MemoryBIO()
+        no_op_session = property(lambda s: None, lambda s, v: None)
+        with mock.patch.object(ssl.SSLObject, "session", no_op_session):
+            ssl_obj = real_ctx.wrap_bio(incoming, outgoing, server_side=False)
+        self.assertIsInstance(ssl_obj, ssl.SSLObject)
+
+    @unittest.skipUnless(not _IS_SYNC, "Tests async _configured_socket_interface only")
+    def test_async_configured_protocol_no_cache(self):
+        """When ssl_session_cache is None, no injection or save occurs."""
+        import asyncio
+        import ssl
+        import unittest.mock as mock
+
+        from pymongo.pool_shared import _configured_socket_interface
+
+        real_ctx = ssl.create_default_context()
+        real_ctx.check_hostname = False
+        real_ctx.verify_mode = ssl.CERT_NONE
+
+        mock_opts = mock.MagicMock()
+        mock_opts._ssl_context = real_ctx
+        mock_opts.socket_timeout = None
+        mock_opts.tls_allow_invalid_hostnames = True
+
+        mock_transport = mock.MagicMock()
+        mock_transport.get_extra_info.return_value = None
+
+        mock_loop = mock.MagicMock()
+        mock_loop.create_connection = mock.Mock(return_value=(mock_transport, mock.MagicMock()))
+
+        def run():
+            with (
+                mock.patch(
+                    "pymongo.pool_shared._create_connection",
+                    new=mock.SyncMock(return_value=mock.MagicMock()),
+                ),
+                mock.patch(
+                    "pymongo.pool_shared.asyncio.get_running_loop",
+                    return_value=mock_loop,
+                ),
+            ):
+                _configured_socket_interface(("localhost", 27017), mock_opts, None)
+
+        asyncio.run(run())
+        self.assertIs(real_ctx.sslobject_class, ssl.SSLObject)  # type: ignore[attr-defined]
+
+    @unittest.skipUnless(not _IS_SYNC, "Tests async _configured_socket_interface only")
+    def test_async_configured_protocol_new_session_is_none(self):
+        """When ssl_object.session is None after connect, the cache is not updated."""
+        import asyncio
+        import ssl
+        import unittest.mock as mock
+
+        from pymongo.pool_shared import _configured_socket_interface
+
+        cache: list = [None]
+
+        mock_ssl_obj = mock.MagicMock()
+        mock_ssl_obj.session = None
+
+        mock_transport = mock.MagicMock()
+        mock_transport.get_extra_info.side_effect = lambda key: (
+            mock_ssl_obj if key == "ssl_object" else None
+        )
+
+        real_ctx = ssl.create_default_context()
+        real_ctx.check_hostname = False
+        real_ctx.verify_mode = ssl.CERT_NONE
+
+        mock_opts = mock.MagicMock()
+        mock_opts._ssl_context = real_ctx
+        mock_opts.socket_timeout = None
+        mock_opts.tls_allow_invalid_hostnames = True
+
+        mock_loop = mock.MagicMock()
+        mock_loop.create_connection = mock.Mock(return_value=(mock_transport, mock.MagicMock()))
+
+        def run():
+            with (
+                mock.patch(
+                    "pymongo.pool_shared._create_connection",
+                    new=mock.SyncMock(return_value=mock.MagicMock()),
+                ),
+                mock.patch(
+                    "pymongo.pool_shared.asyncio.get_running_loop",
+                    return_value=mock_loop,
+                ),
+            ):
+                _configured_socket_interface(("localhost", 27017), mock_opts, cache)
+
+        asyncio.run(run())
+        self.assertIsNone(cache[0])
+
 
 class TestSSL(IntegrationTest):
     saved_port: int