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.

Author: blink1073

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:

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)

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:

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()

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()