From 28690bbff50b61c3d5930bc76ae7d247363bdb1f Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Thu, 9 Jul 2026 16:54:41 -0500 Subject: [PATCH 01/10] PYTHON-5930 Fix Windows SSL_CERT_FILE/SSL_CERT_DIR trust store handling Honor SSL_CERT_FILE/SSL_CERT_DIR as the exclusive trust source across platforms instead of merging them with the OS/certifi certificate store. --- doc/changelog.rst | 3 +++ pymongo/ssl_support.py | 12 +++++++++++- 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/doc/changelog.rst b/doc/changelog.rst index b3e1c75319..f00484e970 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -8,6 +8,9 @@ Changes in Version 4.18.0 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. +- Fixed a bug on Windows, and on macOS when using PyOpenSSL, where + ``SSL_CERT_FILE``/``SSL_CERT_DIR`` were merged with, rather than replacing, + the OS/certifi certificate store. Changes in Version 4.17.0 (2026/04/20) -------------------------------------- diff --git a/pymongo/ssl_support.py b/pymongo/ssl_support.py index 65196ef945..0ec35e2763 100644 --- a/pymongo/ssl_support.py +++ b/pymongo/ssl_support.py @@ -16,6 +16,7 @@ from __future__ import annotations +import os import types import warnings from typing import Any, Optional, Union @@ -133,7 +134,16 @@ def get_ssl_context( if ca_certs is not None: ctx.load_verify_locations(ca_certs) elif verify_mode != CERT_NONE: - ctx.load_default_certs() + cert_file = os.environ.get("SSL_CERT_FILE") + cert_dir = os.environ.get("SSL_CERT_DIR") + if cert_file or cert_dir: + # On Windows, load_default_certs() merges the OS certificate store + # with SSL_CERT_FILE/SSL_CERT_DIR, unlike other platforms where these + # variables are the sole source of trust. Honor them exclusively for + # consistent cross-platform behavior. + ctx.load_verify_locations(cafile=cert_file, capath=cert_dir) + else: + ctx.load_default_certs() ctx.verify_mode = verify_mode return ctx From 6ed98a74615e6b21fcb3afccabec6adc9f1f51b2 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Thu, 9 Jul 2026 17:32:55 -0500 Subject: [PATCH 02/10] PYTHON-5930 Scope the SSL_CERT_FILE/SSL_CERT_DIR fix to Windows/macOS+PyOpenSSL Only bypass load_default_certs() where it actually merges an OS/certifi store (win32, and darwin with PyOpenSSL). Everywhere else, load_default_certs() already honors these env vars correctly, including falling back to the platform default for whichever of the two is left unset -- calling load_verify_locations() unconditionally dropped that fallback, silently narrowing the trust store e.g. on Linux when only one var was set. Adds regression tests covering both branches. --- pymongo/ssl_support.py | 16 +++++++++++----- test/asynchronous/test_ssl.py | 36 +++++++++++++++++++++++++++++++++++ test/test_ssl.py | 36 +++++++++++++++++++++++++++++++++++ 3 files changed, 83 insertions(+), 5 deletions(-) diff --git a/pymongo/ssl_support.py b/pymongo/ssl_support.py index 0ec35e2763..837005e57c 100644 --- a/pymongo/ssl_support.py +++ b/pymongo/ssl_support.py @@ -17,6 +17,7 @@ from __future__ import annotations import os +import sys import types import warnings from typing import Any, Optional, Union @@ -136,11 +137,16 @@ def get_ssl_context( elif verify_mode != CERT_NONE: cert_file = os.environ.get("SSL_CERT_FILE") cert_dir = os.environ.get("SSL_CERT_DIR") - if cert_file or cert_dir: - # On Windows, load_default_certs() merges the OS certificate store - # with SSL_CERT_FILE/SSL_CERT_DIR, unlike other platforms where these - # variables are the sole source of trust. Honor them exclusively for - # consistent cross-platform behavior. + # On Windows, and on macOS when using PyOpenSSL, load_default_certs() + # merges the OS/certifi certificate store with SSL_CERT_FILE/SSL_CERT_DIR + # instead of letting them replace it. Elsewhere, load_default_certs() + # already honors these variables correctly -- including falling back to + # the platform default for whichever of the two is left unset -- so only + # bypass it where the unwanted merge actually happens. + merges_os_store = sys.platform == "win32" or ( + ssl.IS_PYOPENSSL and sys.platform == "darwin" + ) + if (cert_file or cert_dir) and merges_os_store: ctx.load_verify_locations(cafile=cert_file, capath=cert_dir) else: ctx.load_default_certs() diff --git a/test/asynchronous/test_ssl.py b/test/asynchronous/test_ssl.py index 5944d44ff2..586fece5a3 100644 --- a/test/asynchronous/test_ssl.py +++ b/test/asynchronous/test_ssl.py @@ -136,6 +136,42 @@ def test_config_ssl(self): def test_use_pyopenssl_when_available(self): self.assertTrue(HAVE_PYSSL) + def test_ssl_cert_file_env_var_preserves_default_fallback_off_windows(self): + # PYTHON-5930: on platforms/backends where load_default_certs() doesn't + # merge in an OS/certifi store (e.g. stdlib ssl on Linux/macOS), setting + # only SSL_CERT_FILE must still fall back to the platform's default CA + # directory for SSL_CERT_DIR, matching OpenSSL's own semantics. Bypassing + # load_default_certs() unconditionally would silently drop that fallback. + env = dict(os.environ) + env.pop("SSL_CERT_DIR", None) + env["SSL_CERT_FILE"] = CA_PEM + with ( + mock.patch.dict(os.environ, env, clear=True), + mock.patch.object(sys, "platform", "linux"), + mock.patch("ssl.SSLContext.load_default_certs") as mock_default, + mock.patch("ssl.SSLContext.load_verify_locations") as mock_verify, + ): + get_ssl_context(None, None, None, None, False, False, False, False) + mock_default.assert_called_once() + mock_verify.assert_not_called() + + def test_ssl_cert_file_env_var_bypasses_default_certs_on_windows(self): + # PYTHON-5930: on win32 (and macOS+PyOpenSSL), load_default_certs() merges + # the OS/certifi store with SSL_CERT_FILE/SSL_CERT_DIR, so it must be + # bypassed in favor of loading the env vars exclusively. + env = dict(os.environ) + env.pop("SSL_CERT_DIR", None) + env["SSL_CERT_FILE"] = CA_PEM + with ( + mock.patch.dict(os.environ, env, clear=True), + mock.patch.object(sys, "platform", "win32"), + mock.patch("ssl.SSLContext.load_default_certs") as mock_default, + mock.patch("ssl.SSLContext.load_verify_locations") as mock_verify, + ): + get_ssl_context(None, None, None, None, False, False, False, False) + mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) + mock_default.assert_not_called() + def test_ssl_session_cache(self): cache: list = [None] self.assertIsNone(cache[0]) diff --git a/test/test_ssl.py b/test/test_ssl.py index 58e8c7e189..d16da02179 100644 --- a/test/test_ssl.py +++ b/test/test_ssl.py @@ -135,6 +135,42 @@ def test_config_ssl(self): def test_use_pyopenssl_when_available(self): self.assertTrue(HAVE_PYSSL) + def test_ssl_cert_file_env_var_preserves_default_fallback_off_windows(self): + # PYTHON-5930: on platforms/backends where load_default_certs() doesn't + # merge in an OS/certifi store (e.g. stdlib ssl on Linux/macOS), setting + # only SSL_CERT_FILE must still fall back to the platform's default CA + # directory for SSL_CERT_DIR, matching OpenSSL's own semantics. Bypassing + # load_default_certs() unconditionally would silently drop that fallback. + env = dict(os.environ) + env.pop("SSL_CERT_DIR", None) + env["SSL_CERT_FILE"] = CA_PEM + with ( + mock.patch.dict(os.environ, env, clear=True), + mock.patch.object(sys, "platform", "linux"), + mock.patch("ssl.SSLContext.load_default_certs") as mock_default, + mock.patch("ssl.SSLContext.load_verify_locations") as mock_verify, + ): + get_ssl_context(None, None, None, None, False, False, False, False) + mock_default.assert_called_once() + mock_verify.assert_not_called() + + def test_ssl_cert_file_env_var_bypasses_default_certs_on_windows(self): + # PYTHON-5930: on win32 (and macOS+PyOpenSSL), load_default_certs() merges + # the OS/certifi store with SSL_CERT_FILE/SSL_CERT_DIR, so it must be + # bypassed in favor of loading the env vars exclusively. + env = dict(os.environ) + env.pop("SSL_CERT_DIR", None) + env["SSL_CERT_FILE"] = CA_PEM + with ( + mock.patch.dict(os.environ, env, clear=True), + mock.patch.object(sys, "platform", "win32"), + mock.patch("ssl.SSLContext.load_default_certs") as mock_default, + mock.patch("ssl.SSLContext.load_verify_locations") as mock_verify, + ): + get_ssl_context(None, None, None, None, False, False, False, False) + mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) + mock_default.assert_not_called() + def test_ssl_session_cache(self): cache: list = [None] self.assertIsNone(cache[0]) From d0c962b9b8cdf8bd3eb1a3e49cd3428eda442ad8 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Thu, 9 Jul 2026 17:39:13 -0500 Subject: [PATCH 03/10] PYTHON-5930 Fix pyopenssl-macos Evergreen variant running on the wrong host create_pyopenssl_variants() never passed host=host to create_variant(), unlike create_encryption_variants(). The win64 case worked by accident via create_variant()'s display-name-based win64 special case, but pyopenssl-macos silently fell back to the rhel87-small default and never actually ran on macOS, so it couldn't validate macOS-specific PyOpenSSL behavior. --- .evergreen/generated_configs/variants.yml | 2 +- .evergreen/scripts/generate_config.py | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/.evergreen/generated_configs/variants.yml b/.evergreen/generated_configs/variants.yml index b167508539..6c3dddb798 100644 --- a/.evergreen/generated_configs/variants.yml +++ b/.evergreen/generated_configs/variants.yml @@ -467,7 +467,7 @@ buildvariants: - name: .test-standard !.pypy .async .replica_set-noauth-ssl display_name: PyOpenSSL macOS run_on: - - rhel87-small + - macos-14 batchtime: 1440 expansions: SUB_TEST_NAME: pyopenssl diff --git a/.evergreen/scripts/generate_config.py b/.evergreen/scripts/generate_config.py index 6dfa014a16..e156bbda0a 100644 --- a/.evergreen/scripts/generate_config.py +++ b/.evergreen/scripts/generate_config.py @@ -251,6 +251,7 @@ def create_pyopenssl_variants(): create_variant( tasks, display_name, + host=host, expansions=expansions, batchtime=batchtime, ) From b6ffcf6c15327b57a8e5e6cd2b2e237abcfb9c4e Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 10 Jul 2026 06:15:59 -0500 Subject: [PATCH 04/10] PYTHON-5930 Trim comment in get_ssl_context --- pymongo/ssl_support.py | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/pymongo/ssl_support.py b/pymongo/ssl_support.py index 837005e57c..afab22921f 100644 --- a/pymongo/ssl_support.py +++ b/pymongo/ssl_support.py @@ -137,12 +137,8 @@ def get_ssl_context( elif verify_mode != CERT_NONE: cert_file = os.environ.get("SSL_CERT_FILE") cert_dir = os.environ.get("SSL_CERT_DIR") - # On Windows, and on macOS when using PyOpenSSL, load_default_certs() - # merges the OS/certifi certificate store with SSL_CERT_FILE/SSL_CERT_DIR - # instead of letting them replace it. Elsewhere, load_default_certs() - # already honors these variables correctly -- including falling back to - # the platform default for whichever of the two is left unset -- so only - # bypass it where the unwanted merge actually happens. + # load_default_certs() wrongly merges in the OS/certifi store on + # Windows, and on macOS with PyOpenSSL; bypass it only there. merges_os_store = sys.platform == "win32" or ( ssl.IS_PYOPENSSL and sys.platform == "darwin" ) From f99e779b59533a5383039cc13f0a19463810f82c Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 10 Jul 2026 06:30:48 -0500 Subject: [PATCH 05/10] PYTHON-5930 Add regression test for macOS+PyOpenSSL bypass branch Addresses review feedback: the darwin+PyOpenSSL bypass path had no direct test coverage, only the win32 bypass and the non-Windows fallback were exercised. --- test/asynchronous/test_ssl.py | 17 +++++++++++++++++ test/test_ssl.py | 17 +++++++++++++++++ 2 files changed, 34 insertions(+) diff --git a/test/asynchronous/test_ssl.py b/test/asynchronous/test_ssl.py index 586fece5a3..0faca24cb3 100644 --- a/test/asynchronous/test_ssl.py +++ b/test/asynchronous/test_ssl.py @@ -172,6 +172,23 @@ def test_ssl_cert_file_env_var_bypasses_default_certs_on_windows(self): mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) mock_default.assert_not_called() + @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.") + def test_ssl_cert_file_env_var_bypasses_default_certs_on_macos_pyopenssl(self): + # PYTHON-5930: on macOS with PyOpenSSL, load_default_certs() merges in + # certifi certs, so it must be bypassed just like on win32. + env = dict(os.environ) + env.pop("SSL_CERT_DIR", None) + env["SSL_CERT_FILE"] = CA_PEM + with ( + mock.patch.dict(os.environ, env, clear=True), + mock.patch.object(sys, "platform", "darwin"), + mock.patch("pymongo.pyopenssl_context.SSLContext.load_default_certs") as mock_default, + mock.patch("pymongo.pyopenssl_context.SSLContext.load_verify_locations") as mock_verify, + ): + get_ssl_context(None, None, None, None, False, False, False, True) + mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) + mock_default.assert_not_called() + def test_ssl_session_cache(self): cache: list = [None] self.assertIsNone(cache[0]) diff --git a/test/test_ssl.py b/test/test_ssl.py index d16da02179..3a2c314327 100644 --- a/test/test_ssl.py +++ b/test/test_ssl.py @@ -171,6 +171,23 @@ def test_ssl_cert_file_env_var_bypasses_default_certs_on_windows(self): mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) mock_default.assert_not_called() + @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.") + def test_ssl_cert_file_env_var_bypasses_default_certs_on_macos_pyopenssl(self): + # PYTHON-5930: on macOS with PyOpenSSL, load_default_certs() merges in + # certifi certs, so it must be bypassed just like on win32. + env = dict(os.environ) + env.pop("SSL_CERT_DIR", None) + env["SSL_CERT_FILE"] = CA_PEM + with ( + mock.patch.dict(os.environ, env, clear=True), + mock.patch.object(sys, "platform", "darwin"), + mock.patch("pymongo.pyopenssl_context.SSLContext.load_default_certs") as mock_default, + mock.patch("pymongo.pyopenssl_context.SSLContext.load_verify_locations") as mock_verify, + ): + get_ssl_context(None, None, None, None, False, False, False, True) + mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) + mock_default.assert_not_called() + def test_ssl_session_cache(self): cache: list = [None] self.assertIsNone(cache[0]) From e3724fb531442be52a57f2c614cba948dad581e5 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 10 Jul 2026 08:54:46 -0500 Subject: [PATCH 06/10] Update pymongo/ssl_support.py Co-authored-by: Noah Stapp --- pymongo/ssl_support.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pymongo/ssl_support.py b/pymongo/ssl_support.py index afab22921f..31d254306d 100644 --- a/pymongo/ssl_support.py +++ b/pymongo/ssl_support.py @@ -138,7 +138,7 @@ def get_ssl_context( cert_file = os.environ.get("SSL_CERT_FILE") cert_dir = os.environ.get("SSL_CERT_DIR") # load_default_certs() wrongly merges in the OS/certifi store on - # Windows, and on macOS with PyOpenSSL; bypass it only there. + # Windows and on macOS with PyOpenSSL merges_os_store = sys.platform == "win32" or ( ssl.IS_PYOPENSSL and sys.platform == "darwin" ) From b7ad901d87a45fc19a59f9b18e22b8003b580b5f Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 10 Jul 2026 09:08:40 -0500 Subject: [PATCH 07/10] PYTHON-5930 Clarify SSL_CERT_FILE regression test names and comments Addresses review feedback: rename the Linux fallback test to name the platform it actually covers, and drop the redundant macOS+PyOpenSSL mention from the Windows test's comment since a dedicated test now covers that case. --- test/asynchronous/test_ssl.py | 16 +++++++--------- test/test_ssl.py | 16 +++++++--------- 2 files changed, 14 insertions(+), 18 deletions(-) diff --git a/test/asynchronous/test_ssl.py b/test/asynchronous/test_ssl.py index 0faca24cb3..b61b64d7f7 100644 --- a/test/asynchronous/test_ssl.py +++ b/test/asynchronous/test_ssl.py @@ -136,12 +136,10 @@ def test_config_ssl(self): def test_use_pyopenssl_when_available(self): self.assertTrue(HAVE_PYSSL) - def test_ssl_cert_file_env_var_preserves_default_fallback_off_windows(self): - # PYTHON-5930: on platforms/backends where load_default_certs() doesn't - # merge in an OS/certifi store (e.g. stdlib ssl on Linux/macOS), setting - # only SSL_CERT_FILE must still fall back to the platform's default CA - # directory for SSL_CERT_DIR, matching OpenSSL's own semantics. Bypassing - # load_default_certs() unconditionally would silently drop that fallback. + def test_ssl_cert_file_env_var_uses_default_certs_on_linux(self): + # PYTHON-5930: on Linux, load_default_certs() already honors SSL_CERT_FILE + # correctly (unlike Windows/macOS+PyOpenSSL), so it must still be called + # instead of bypassed. env = dict(os.environ) env.pop("SSL_CERT_DIR", None) env["SSL_CERT_FILE"] = CA_PEM @@ -156,9 +154,9 @@ def test_ssl_cert_file_env_var_preserves_default_fallback_off_windows(self): mock_verify.assert_not_called() def test_ssl_cert_file_env_var_bypasses_default_certs_on_windows(self): - # PYTHON-5930: on win32 (and macOS+PyOpenSSL), load_default_certs() merges - # the OS/certifi store with SSL_CERT_FILE/SSL_CERT_DIR, so it must be - # bypassed in favor of loading the env vars exclusively. + # PYTHON-5930: on win32, load_default_certs() merges the OS certificate + # store with SSL_CERT_FILE/SSL_CERT_DIR, so it must be bypassed in favor + # of loading the env vars exclusively. env = dict(os.environ) env.pop("SSL_CERT_DIR", None) env["SSL_CERT_FILE"] = CA_PEM diff --git a/test/test_ssl.py b/test/test_ssl.py index 3a2c314327..4fd3701693 100644 --- a/test/test_ssl.py +++ b/test/test_ssl.py @@ -135,12 +135,10 @@ def test_config_ssl(self): def test_use_pyopenssl_when_available(self): self.assertTrue(HAVE_PYSSL) - def test_ssl_cert_file_env_var_preserves_default_fallback_off_windows(self): - # PYTHON-5930: on platforms/backends where load_default_certs() doesn't - # merge in an OS/certifi store (e.g. stdlib ssl on Linux/macOS), setting - # only SSL_CERT_FILE must still fall back to the platform's default CA - # directory for SSL_CERT_DIR, matching OpenSSL's own semantics. Bypassing - # load_default_certs() unconditionally would silently drop that fallback. + def test_ssl_cert_file_env_var_uses_default_certs_on_linux(self): + # PYTHON-5930: on Linux, load_default_certs() already honors SSL_CERT_FILE + # correctly (unlike Windows/macOS+PyOpenSSL), so it must still be called + # instead of bypassed. env = dict(os.environ) env.pop("SSL_CERT_DIR", None) env["SSL_CERT_FILE"] = CA_PEM @@ -155,9 +153,9 @@ def test_ssl_cert_file_env_var_preserves_default_fallback_off_windows(self): mock_verify.assert_not_called() def test_ssl_cert_file_env_var_bypasses_default_certs_on_windows(self): - # PYTHON-5930: on win32 (and macOS+PyOpenSSL), load_default_certs() merges - # the OS/certifi store with SSL_CERT_FILE/SSL_CERT_DIR, so it must be - # bypassed in favor of loading the env vars exclusively. + # PYTHON-5930: on win32, load_default_certs() merges the OS certificate + # store with SSL_CERT_FILE/SSL_CERT_DIR, so it must be bypassed in favor + # of loading the env vars exclusively. env = dict(os.environ) env.pop("SSL_CERT_DIR", None) env["SSL_CERT_FILE"] = CA_PEM From 9e55d86e6971779f78e736043a423f6eeaffc002 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 10 Jul 2026 10:22:52 -0500 Subject: [PATCH 08/10] PYTHON-5930 Move TestClientSSL to a shared, non-mirrored test file Addresses review feedback: the tests in TestClientSSL are pure unit tests (mocked or manual asyncio.run()) with no dependency on the async/sync test class machinery, so they were needlessly duplicated by synchro. Moved them to test/test_ssl_support.py, following the existing test_pyopenssl_context.py precedent for standalone test files that aren't generated/mirrored. --- test/asynchronous/test_ssl.py | 366 +----------------------------- test/test_ssl.py | 357 +---------------------------- test/test_ssl_support.py | 410 ++++++++++++++++++++++++++++++++++ 3 files changed, 414 insertions(+), 719 deletions(-) create mode 100644 test/test_ssl_support.py diff --git a/test/asynchronous/test_ssl.py b/test/asynchronous/test_ssl.py index b61b64d7f7..e953aa963f 100644 --- a/test/asynchronous/test_ssl.py +++ b/test/asynchronous/test_ssl.py @@ -16,12 +16,10 @@ from __future__ import annotations -import asyncio import os import pathlib import socket import sys -import unittest.mock as mock sys.path[0:0] = [""] @@ -30,17 +28,11 @@ from pymongo import AsyncMongoClient, ssl_support from pymongo.errors import ConfigurationError, ConnectionFailure, OperationFailure from pymongo.hello import HelloCompat -from pymongo.pool_shared import ( - _configured_protocol_interface, - _configured_socket_interface, - _get_ssl_session, -) -from pymongo.ssl_support import HAVE_PYSSL, HAVE_SSL, _ssl, get_ssl_context +from pymongo.ssl_support import HAVE_PYSSL, HAVE_SSL, get_ssl_context from pymongo.write_concern import WriteConcern from test.asynchronous import ( HAVE_IPADDRESS, AsyncIntegrationTest, - AsyncPyMongoTestCase, SkipTest, async_client_context, connected, @@ -56,7 +48,7 @@ _HAVE_PYOPENSSL = False try: - from pymongo import pyopenssl_context + import pymongo.pyopenssl_context _HAVE_PYOPENSSL = True except ImportError: @@ -91,360 +83,6 @@ # use 'localhost' for the hostname of all hosts. -class TestClientSSL(AsyncPyMongoTestCase): - @unittest.skipIf(HAVE_SSL, "The ssl module is available, can't test what happens without it.") - def test_no_ssl_module(self): - # Explicit - self.assertRaises(ConfigurationError, self.simple_client, ssl=True) - - # Implied - self.assertRaises(ConfigurationError, self.simple_client, tlsCertificateKeyFile=CLIENT_PEM) - - @unittest.skipUnless(HAVE_SSL, "The ssl module is not available.") - @ignore_deprecations - def test_config_ssl(self): - # Tests various ssl configurations - self.assertRaises(ValueError, self.simple_client, ssl="foo") - self.assertRaises( - ConfigurationError, self.simple_client, tls=False, tlsCertificateKeyFile=CLIENT_PEM - ) - self.assertRaises(TypeError, self.simple_client, ssl=0) - self.assertRaises(TypeError, self.simple_client, ssl=5.5) - self.assertRaises(TypeError, self.simple_client, ssl=[]) - - self.assertRaises(IOError, self.simple_client, tlsCertificateKeyFile="NoSuchFile") - self.assertRaises(TypeError, self.simple_client, tlsCertificateKeyFile=True) - self.assertRaises(TypeError, self.simple_client, tlsCertificateKeyFile=[]) - - # Test invalid combinations - self.assertRaises( - ConfigurationError, self.simple_client, tls=False, tlsCertificateKeyFile=CLIENT_PEM - ) - self.assertRaises(ConfigurationError, self.simple_client, tls=False, tlsCAFile=CA_PEM) - self.assertRaises(ConfigurationError, self.simple_client, tls=False, tlsCRLFile=CRL_PEM) - self.assertRaises( - ConfigurationError, self.simple_client, tls=False, tlsAllowInvalidCertificates=False - ) - self.assertRaises( - ConfigurationError, self.simple_client, tls=False, tlsAllowInvalidHostnames=False - ) - self.assertRaises( - ConfigurationError, self.simple_client, tls=False, tlsDisableOCSPEndpointCheck=False - ) - - @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.") - def test_use_pyopenssl_when_available(self): - self.assertTrue(HAVE_PYSSL) - - def test_ssl_cert_file_env_var_uses_default_certs_on_linux(self): - # PYTHON-5930: on Linux, load_default_certs() already honors SSL_CERT_FILE - # correctly (unlike Windows/macOS+PyOpenSSL), so it must still be called - # instead of bypassed. - env = dict(os.environ) - env.pop("SSL_CERT_DIR", None) - env["SSL_CERT_FILE"] = CA_PEM - with ( - mock.patch.dict(os.environ, env, clear=True), - mock.patch.object(sys, "platform", "linux"), - mock.patch("ssl.SSLContext.load_default_certs") as mock_default, - mock.patch("ssl.SSLContext.load_verify_locations") as mock_verify, - ): - get_ssl_context(None, None, None, None, False, False, False, False) - mock_default.assert_called_once() - mock_verify.assert_not_called() - - def test_ssl_cert_file_env_var_bypasses_default_certs_on_windows(self): - # PYTHON-5930: on win32, load_default_certs() merges the OS certificate - # store with SSL_CERT_FILE/SSL_CERT_DIR, so it must be bypassed in favor - # of loading the env vars exclusively. - env = dict(os.environ) - env.pop("SSL_CERT_DIR", None) - env["SSL_CERT_FILE"] = CA_PEM - with ( - mock.patch.dict(os.environ, env, clear=True), - mock.patch.object(sys, "platform", "win32"), - mock.patch("ssl.SSLContext.load_default_certs") as mock_default, - mock.patch("ssl.SSLContext.load_verify_locations") as mock_verify, - ): - get_ssl_context(None, None, None, None, False, False, False, False) - mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) - mock_default.assert_not_called() - - @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.") - def test_ssl_cert_file_env_var_bypasses_default_certs_on_macos_pyopenssl(self): - # PYTHON-5930: on macOS with PyOpenSSL, load_default_certs() merges in - # certifi certs, so it must be bypassed just like on win32. - env = dict(os.environ) - env.pop("SSL_CERT_DIR", None) - env["SSL_CERT_FILE"] = CA_PEM - with ( - mock.patch.dict(os.environ, env, clear=True), - mock.patch.object(sys, "platform", "darwin"), - mock.patch("pymongo.pyopenssl_context.SSLContext.load_default_certs") as mock_default, - mock.patch("pymongo.pyopenssl_context.SSLContext.load_verify_locations") as mock_verify, - ): - get_ssl_context(None, None, None, None, False, False, False, True) - mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) - mock_default.assert_not_called() - - def test_ssl_session_cache(self): - 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.""" - - fake_session = object() - cache: list = [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) - - def test_get_ssl_session_pyopenssl_style(self): - """_get_ssl_session uses get_session() when available (PyOpenSSL path).""" - 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).""" - 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)", - ) - def test_async_tls_session_injected_via_sslobject_class(self): - """On Python 3.11+, a cached session is injected by setting sslobject_class.""" - 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[0] - assert session is not None - _session = session - - class _SessionSSLObject(ssl.SSLObject): - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self.session = _session - - real_ctx.sslobject_class = _SessionSSLObject - - 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.""" - 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) - - @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.""" - 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.""" - 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.""" - 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 4fd3701693..c9954a1724 100644 --- a/test/test_ssl.py +++ b/test/test_ssl.py @@ -16,12 +16,10 @@ from __future__ import annotations -import asyncio import os import pathlib import socket import sys -import unittest.mock as mock sys.path[0:0] = [""] @@ -30,16 +28,11 @@ from pymongo import MongoClient, ssl_support from pymongo.errors import ConfigurationError, ConnectionFailure, OperationFailure from pymongo.hello import HelloCompat -from pymongo.pool_shared import ( - _configured_socket_interface, - _get_ssl_session, -) -from pymongo.ssl_support import HAVE_PYSSL, HAVE_SSL, _ssl, get_ssl_context +from pymongo.ssl_support import HAVE_PYSSL, HAVE_SSL, get_ssl_context from pymongo.write_concern import WriteConcern from test import ( HAVE_IPADDRESS, IntegrationTest, - PyMongoTestCase, SkipTest, client_context, connected, @@ -55,7 +48,7 @@ _HAVE_PYOPENSSL = False try: - from pymongo import pyopenssl_context + import pymongo.pyopenssl_context _HAVE_PYOPENSSL = True except ImportError: @@ -90,352 +83,6 @@ # use 'localhost' for the hostname of all hosts. -class TestClientSSL(PyMongoTestCase): - @unittest.skipIf(HAVE_SSL, "The ssl module is available, can't test what happens without it.") - def test_no_ssl_module(self): - # Explicit - self.assertRaises(ConfigurationError, self.simple_client, ssl=True) - - # Implied - self.assertRaises(ConfigurationError, self.simple_client, tlsCertificateKeyFile=CLIENT_PEM) - - @unittest.skipUnless(HAVE_SSL, "The ssl module is not available.") - @ignore_deprecations - def test_config_ssl(self): - # Tests various ssl configurations - self.assertRaises(ValueError, self.simple_client, ssl="foo") - self.assertRaises( - ConfigurationError, self.simple_client, tls=False, tlsCertificateKeyFile=CLIENT_PEM - ) - self.assertRaises(TypeError, self.simple_client, ssl=0) - self.assertRaises(TypeError, self.simple_client, ssl=5.5) - self.assertRaises(TypeError, self.simple_client, ssl=[]) - - self.assertRaises(IOError, self.simple_client, tlsCertificateKeyFile="NoSuchFile") - self.assertRaises(TypeError, self.simple_client, tlsCertificateKeyFile=True) - self.assertRaises(TypeError, self.simple_client, tlsCertificateKeyFile=[]) - - # Test invalid combinations - self.assertRaises( - ConfigurationError, self.simple_client, tls=False, tlsCertificateKeyFile=CLIENT_PEM - ) - self.assertRaises(ConfigurationError, self.simple_client, tls=False, tlsCAFile=CA_PEM) - self.assertRaises(ConfigurationError, self.simple_client, tls=False, tlsCRLFile=CRL_PEM) - self.assertRaises( - ConfigurationError, self.simple_client, tls=False, tlsAllowInvalidCertificates=False - ) - self.assertRaises( - ConfigurationError, self.simple_client, tls=False, tlsAllowInvalidHostnames=False - ) - self.assertRaises( - ConfigurationError, self.simple_client, tls=False, tlsDisableOCSPEndpointCheck=False - ) - - @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.") - def test_use_pyopenssl_when_available(self): - self.assertTrue(HAVE_PYSSL) - - def test_ssl_cert_file_env_var_uses_default_certs_on_linux(self): - # PYTHON-5930: on Linux, load_default_certs() already honors SSL_CERT_FILE - # correctly (unlike Windows/macOS+PyOpenSSL), so it must still be called - # instead of bypassed. - env = dict(os.environ) - env.pop("SSL_CERT_DIR", None) - env["SSL_CERT_FILE"] = CA_PEM - with ( - mock.patch.dict(os.environ, env, clear=True), - mock.patch.object(sys, "platform", "linux"), - mock.patch("ssl.SSLContext.load_default_certs") as mock_default, - mock.patch("ssl.SSLContext.load_verify_locations") as mock_verify, - ): - get_ssl_context(None, None, None, None, False, False, False, False) - mock_default.assert_called_once() - mock_verify.assert_not_called() - - def test_ssl_cert_file_env_var_bypasses_default_certs_on_windows(self): - # PYTHON-5930: on win32, load_default_certs() merges the OS certificate - # store with SSL_CERT_FILE/SSL_CERT_DIR, so it must be bypassed in favor - # of loading the env vars exclusively. - env = dict(os.environ) - env.pop("SSL_CERT_DIR", None) - env["SSL_CERT_FILE"] = CA_PEM - with ( - mock.patch.dict(os.environ, env, clear=True), - mock.patch.object(sys, "platform", "win32"), - mock.patch("ssl.SSLContext.load_default_certs") as mock_default, - mock.patch("ssl.SSLContext.load_verify_locations") as mock_verify, - ): - get_ssl_context(None, None, None, None, False, False, False, False) - mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) - mock_default.assert_not_called() - - @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.") - def test_ssl_cert_file_env_var_bypasses_default_certs_on_macos_pyopenssl(self): - # PYTHON-5930: on macOS with PyOpenSSL, load_default_certs() merges in - # certifi certs, so it must be bypassed just like on win32. - env = dict(os.environ) - env.pop("SSL_CERT_DIR", None) - env["SSL_CERT_FILE"] = CA_PEM - with ( - mock.patch.dict(os.environ, env, clear=True), - mock.patch.object(sys, "platform", "darwin"), - mock.patch("pymongo.pyopenssl_context.SSLContext.load_default_certs") as mock_default, - mock.patch("pymongo.pyopenssl_context.SSLContext.load_verify_locations") as mock_verify, - ): - get_ssl_context(None, None, None, None, False, False, False, True) - mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) - mock_default.assert_not_called() - - def test_ssl_session_cache(self): - 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.""" - - fake_session = object() - cache: list = [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) - - def test_get_ssl_session_pyopenssl_style(self): - """_get_ssl_session uses get_session() when available (PyOpenSSL path).""" - 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).""" - 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)", - ) - def test_async_tls_session_injected_via_sslobject_class(self): - """On Python 3.11+, a cached session is injected by setting sslobject_class.""" - 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[0] - assert session is not None - _session = session - - class _SessionSSLObject(ssl.SSLObject): - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self.session = _session - - real_ctx.sslobject_class = _SessionSSLObject - - 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.""" - 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) - - @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.""" - 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.""" - 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.""" - 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 diff --git a/test/test_ssl_support.py b/test/test_ssl_support.py new file mode 100644 index 0000000000..2d8e13cae1 --- /dev/null +++ b/test/test_ssl_support.py @@ -0,0 +1,410 @@ +# Copyright 2011-present MongoDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for pymongo/ssl_support.py and the SSL session cache in pool_shared.py. + +These tests are pure unit tests (mocked or in-process) with no async-specific +behavior, so unlike test_ssl.py they are not generated/mirrored for sync and async. +""" + +from __future__ import annotations + +import asyncio +import os +import pathlib +import sys +import unittest.mock as mock + +sys.path[0:0] = [""] + +from pymongo import MongoClient +from pymongo.errors import ConfigurationError +from pymongo.pool_shared import ( + _configured_protocol_interface, + _configured_socket_interface, + _get_ssl_session, +) +from pymongo.ssl_support import HAVE_PYSSL, HAVE_SSL, get_ssl_context +from test import unittest +from test.utils_shared import ignore_deprecations + +_HAVE_PYOPENSSL = False +try: + import pymongo.pyopenssl_context + + _HAVE_PYOPENSSL = True +except ImportError: + pass + +if HAVE_SSL: + import ssl + +CERT_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "certificates") +CLIENT_PEM = os.path.join(CERT_PATH, "client.pem") +CA_PEM = os.path.join(CERT_PATH, "ca.pem") +CRL_PEM = os.path.join(CERT_PATH, "crl.pem") + + +class TestClientSSL(unittest.TestCase): + @unittest.skipIf(HAVE_SSL, "The ssl module is available, can't test what happens without it.") + def test_no_ssl_module(self): + # Explicit + self.assertRaises(ConfigurationError, MongoClient, ssl=True) + + # Implied + self.assertRaises(ConfigurationError, MongoClient, tlsCertificateKeyFile=CLIENT_PEM) + + @unittest.skipUnless(HAVE_SSL, "The ssl module is not available.") + @ignore_deprecations + def test_config_ssl(self): + # Tests various ssl configurations + self.assertRaises(ValueError, MongoClient, ssl="foo") + self.assertRaises( + ConfigurationError, MongoClient, tls=False, tlsCertificateKeyFile=CLIENT_PEM + ) + self.assertRaises(TypeError, MongoClient, ssl=0) + self.assertRaises(TypeError, MongoClient, ssl=5.5) + self.assertRaises(TypeError, MongoClient, ssl=[]) + + self.assertRaises(IOError, MongoClient, tlsCertificateKeyFile="NoSuchFile") + self.assertRaises(TypeError, MongoClient, tlsCertificateKeyFile=True) + self.assertRaises(TypeError, MongoClient, tlsCertificateKeyFile=[]) + + # Test invalid combinations + self.assertRaises( + ConfigurationError, MongoClient, tls=False, tlsCertificateKeyFile=CLIENT_PEM + ) + self.assertRaises(ConfigurationError, MongoClient, tls=False, tlsCAFile=CA_PEM) + self.assertRaises(ConfigurationError, MongoClient, tls=False, tlsCRLFile=CRL_PEM) + self.assertRaises( + ConfigurationError, MongoClient, tls=False, tlsAllowInvalidCertificates=False + ) + self.assertRaises( + ConfigurationError, MongoClient, tls=False, tlsAllowInvalidHostnames=False + ) + self.assertRaises( + ConfigurationError, MongoClient, tls=False, tlsDisableOCSPEndpointCheck=False + ) + + @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.") + def test_use_pyopenssl_when_available(self): + self.assertTrue(HAVE_PYSSL) + + def test_ssl_cert_file_env_var_uses_default_certs_on_linux(self): + # PYTHON-5930: on Linux, load_default_certs() already honors SSL_CERT_FILE + # correctly (unlike Windows/macOS+PyOpenSSL), so it must still be called + # instead of bypassed. + env = dict(os.environ) + env.pop("SSL_CERT_DIR", None) + env["SSL_CERT_FILE"] = CA_PEM + with ( + mock.patch.dict(os.environ, env, clear=True), + mock.patch.object(sys, "platform", "linux"), + mock.patch("ssl.SSLContext.load_default_certs") as mock_default, + mock.patch("ssl.SSLContext.load_verify_locations") as mock_verify, + ): + get_ssl_context(None, None, None, None, False, False, False, False) + mock_default.assert_called_once() + mock_verify.assert_not_called() + + def test_ssl_cert_file_env_var_bypasses_default_certs_on_windows(self): + # PYTHON-5930: on win32, load_default_certs() merges the OS certificate + # store with SSL_CERT_FILE/SSL_CERT_DIR, so it must be bypassed in favor + # of loading the env vars exclusively. + env = dict(os.environ) + env.pop("SSL_CERT_DIR", None) + env["SSL_CERT_FILE"] = CA_PEM + with ( + mock.patch.dict(os.environ, env, clear=True), + mock.patch.object(sys, "platform", "win32"), + mock.patch("ssl.SSLContext.load_default_certs") as mock_default, + mock.patch("ssl.SSLContext.load_verify_locations") as mock_verify, + ): + get_ssl_context(None, None, None, None, False, False, False, False) + mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) + mock_default.assert_not_called() + + @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.") + def test_ssl_cert_file_env_var_bypasses_default_certs_on_macos_pyopenssl(self): + # PYTHON-5930: on macOS with PyOpenSSL, load_default_certs() merges in + # certifi certs, so it must be bypassed just like on win32. + env = dict(os.environ) + env.pop("SSL_CERT_DIR", None) + env["SSL_CERT_FILE"] = CA_PEM + with ( + mock.patch.dict(os.environ, env, clear=True), + mock.patch.object(sys, "platform", "darwin"), + mock.patch("pymongo.pyopenssl_context.SSLContext.load_default_certs") as mock_default, + mock.patch("pymongo.pyopenssl_context.SSLContext.load_verify_locations") as mock_verify, + ): + get_ssl_context(None, None, None, None, False, False, False, True) + mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) + mock_default.assert_not_called() + + def test_ssl_session_cache(self): + 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") + + def test_tls_session_reused_on_second_connection(self): + """Cached TLS session is passed to wrap_socket on subsequent connections.""" + + fake_session = object() + cache: list = [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) + + def test_get_ssl_session_pyopenssl_style(self): + """_get_ssl_session uses get_session() when available (PyOpenSSL path).""" + 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).""" + fake_session = object() + + class FakeSSLSock: + session = fake_session + + self.assertIs(_get_ssl_session(FakeSSLSock()), fake_session) + + @unittest.skipUnless( + sys.version_info >= (3, 11), + "Tests async sslobject_class injection (Python 3.11+ only)", + ) + def test_async_tls_session_injected_via_sslobject_class(self): + """On Python 3.11+, a cached session is injected by setting sslobject_class.""" + 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[0] + assert session is not None + _session = session + + class _SessionSSLObject(ssl.SSLObject): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.session = _session + + real_ctx.sslobject_class = _SessionSSLObject + + self.assertIs(real_ctx.sslobject_class, _SessionSSLObject) + self.assertTrue(issubclass(real_ctx.sslobject_class, ssl.SSLObject)) + + def test_async_configured_protocol_saves_session_to_cache(self): + """After a successful TLS connection the session is stored in the cache.""" + 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) + + @unittest.skipUnless( + 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.""" + 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) + + def test_async_configured_protocol_no_cache(self): + """When ssl_session_cache is None, no injection or save occurs.""" + 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] + + def test_async_configured_protocol_new_session_is_none(self): + """When ssl_object.session is None after connect, the cache is not updated.""" + 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]) + + +if __name__ == "__main__": + unittest.main() From 2552339982e687fceb8f6c9fc6ade728bac48e49 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 10 Jul 2026 12:50:23 -0500 Subject: [PATCH 09/10] Update pymongo/ssl_support.py Co-authored-by: Noah Stapp --- pymongo/ssl_support.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pymongo/ssl_support.py b/pymongo/ssl_support.py index 31d254306d..d505f8df27 100644 --- a/pymongo/ssl_support.py +++ b/pymongo/ssl_support.py @@ -135,8 +135,8 @@ def get_ssl_context( if ca_certs is not None: ctx.load_verify_locations(ca_certs) elif verify_mode != CERT_NONE: - cert_file = os.environ.get("SSL_CERT_FILE") - cert_dir = os.environ.get("SSL_CERT_DIR") + cert_file = os.environ.get("SSL_CERT_FILE") or None + cert_dir = os.environ.get("SSL_CERT_DIR") or None # load_default_certs() wrongly merges in the OS/certifi store on # Windows and on macOS with PyOpenSSL merges_os_store = sys.platform == "win32" or ( From 4b0f06d918928cf7a8248583de1ae77f5961bd06 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 10 Jul 2026 12:53:01 -0500 Subject: [PATCH 10/10] Revert "PYTHON-5930 Move TestClientSSL to a shared, non-mirrored test file" This reverts commit 9e55d86e6971779f78e736043a423f6eeaffc002. --- test/asynchronous/test_ssl.py | 366 +++++++++++++++++++++++++++++- test/test_ssl.py | 357 ++++++++++++++++++++++++++++- test/test_ssl_support.py | 410 ---------------------------------- 3 files changed, 719 insertions(+), 414 deletions(-) delete mode 100644 test/test_ssl_support.py diff --git a/test/asynchronous/test_ssl.py b/test/asynchronous/test_ssl.py index e953aa963f..b61b64d7f7 100644 --- a/test/asynchronous/test_ssl.py +++ b/test/asynchronous/test_ssl.py @@ -16,10 +16,12 @@ from __future__ import annotations +import asyncio import os import pathlib import socket import sys +import unittest.mock as mock sys.path[0:0] = [""] @@ -28,11 +30,17 @@ from pymongo import AsyncMongoClient, ssl_support from pymongo.errors import ConfigurationError, ConnectionFailure, OperationFailure from pymongo.hello import HelloCompat -from pymongo.ssl_support import HAVE_PYSSL, HAVE_SSL, get_ssl_context +from pymongo.pool_shared import ( + _configured_protocol_interface, + _configured_socket_interface, + _get_ssl_session, +) +from pymongo.ssl_support import HAVE_PYSSL, HAVE_SSL, _ssl, get_ssl_context from pymongo.write_concern import WriteConcern from test.asynchronous import ( HAVE_IPADDRESS, AsyncIntegrationTest, + AsyncPyMongoTestCase, SkipTest, async_client_context, connected, @@ -48,7 +56,7 @@ _HAVE_PYOPENSSL = False try: - import pymongo.pyopenssl_context + from pymongo import pyopenssl_context _HAVE_PYOPENSSL = True except ImportError: @@ -83,6 +91,360 @@ # use 'localhost' for the hostname of all hosts. +class TestClientSSL(AsyncPyMongoTestCase): + @unittest.skipIf(HAVE_SSL, "The ssl module is available, can't test what happens without it.") + def test_no_ssl_module(self): + # Explicit + self.assertRaises(ConfigurationError, self.simple_client, ssl=True) + + # Implied + self.assertRaises(ConfigurationError, self.simple_client, tlsCertificateKeyFile=CLIENT_PEM) + + @unittest.skipUnless(HAVE_SSL, "The ssl module is not available.") + @ignore_deprecations + def test_config_ssl(self): + # Tests various ssl configurations + self.assertRaises(ValueError, self.simple_client, ssl="foo") + self.assertRaises( + ConfigurationError, self.simple_client, tls=False, tlsCertificateKeyFile=CLIENT_PEM + ) + self.assertRaises(TypeError, self.simple_client, ssl=0) + self.assertRaises(TypeError, self.simple_client, ssl=5.5) + self.assertRaises(TypeError, self.simple_client, ssl=[]) + + self.assertRaises(IOError, self.simple_client, tlsCertificateKeyFile="NoSuchFile") + self.assertRaises(TypeError, self.simple_client, tlsCertificateKeyFile=True) + self.assertRaises(TypeError, self.simple_client, tlsCertificateKeyFile=[]) + + # Test invalid combinations + self.assertRaises( + ConfigurationError, self.simple_client, tls=False, tlsCertificateKeyFile=CLIENT_PEM + ) + self.assertRaises(ConfigurationError, self.simple_client, tls=False, tlsCAFile=CA_PEM) + self.assertRaises(ConfigurationError, self.simple_client, tls=False, tlsCRLFile=CRL_PEM) + self.assertRaises( + ConfigurationError, self.simple_client, tls=False, tlsAllowInvalidCertificates=False + ) + self.assertRaises( + ConfigurationError, self.simple_client, tls=False, tlsAllowInvalidHostnames=False + ) + self.assertRaises( + ConfigurationError, self.simple_client, tls=False, tlsDisableOCSPEndpointCheck=False + ) + + @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.") + def test_use_pyopenssl_when_available(self): + self.assertTrue(HAVE_PYSSL) + + def test_ssl_cert_file_env_var_uses_default_certs_on_linux(self): + # PYTHON-5930: on Linux, load_default_certs() already honors SSL_CERT_FILE + # correctly (unlike Windows/macOS+PyOpenSSL), so it must still be called + # instead of bypassed. + env = dict(os.environ) + env.pop("SSL_CERT_DIR", None) + env["SSL_CERT_FILE"] = CA_PEM + with ( + mock.patch.dict(os.environ, env, clear=True), + mock.patch.object(sys, "platform", "linux"), + mock.patch("ssl.SSLContext.load_default_certs") as mock_default, + mock.patch("ssl.SSLContext.load_verify_locations") as mock_verify, + ): + get_ssl_context(None, None, None, None, False, False, False, False) + mock_default.assert_called_once() + mock_verify.assert_not_called() + + def test_ssl_cert_file_env_var_bypasses_default_certs_on_windows(self): + # PYTHON-5930: on win32, load_default_certs() merges the OS certificate + # store with SSL_CERT_FILE/SSL_CERT_DIR, so it must be bypassed in favor + # of loading the env vars exclusively. + env = dict(os.environ) + env.pop("SSL_CERT_DIR", None) + env["SSL_CERT_FILE"] = CA_PEM + with ( + mock.patch.dict(os.environ, env, clear=True), + mock.patch.object(sys, "platform", "win32"), + mock.patch("ssl.SSLContext.load_default_certs") as mock_default, + mock.patch("ssl.SSLContext.load_verify_locations") as mock_verify, + ): + get_ssl_context(None, None, None, None, False, False, False, False) + mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) + mock_default.assert_not_called() + + @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.") + def test_ssl_cert_file_env_var_bypasses_default_certs_on_macos_pyopenssl(self): + # PYTHON-5930: on macOS with PyOpenSSL, load_default_certs() merges in + # certifi certs, so it must be bypassed just like on win32. + env = dict(os.environ) + env.pop("SSL_CERT_DIR", None) + env["SSL_CERT_FILE"] = CA_PEM + with ( + mock.patch.dict(os.environ, env, clear=True), + mock.patch.object(sys, "platform", "darwin"), + mock.patch("pymongo.pyopenssl_context.SSLContext.load_default_certs") as mock_default, + mock.patch("pymongo.pyopenssl_context.SSLContext.load_verify_locations") as mock_verify, + ): + get_ssl_context(None, None, None, None, False, False, False, True) + mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) + mock_default.assert_not_called() + + def test_ssl_session_cache(self): + 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.""" + + fake_session = object() + cache: list = [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) + + def test_get_ssl_session_pyopenssl_style(self): + """_get_ssl_session uses get_session() when available (PyOpenSSL path).""" + 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).""" + 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)", + ) + def test_async_tls_session_injected_via_sslobject_class(self): + """On Python 3.11+, a cached session is injected by setting sslobject_class.""" + 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[0] + assert session is not None + _session = session + + class _SessionSSLObject(ssl.SSLObject): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.session = _session + + real_ctx.sslobject_class = _SessionSSLObject + + 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.""" + 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) + + @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.""" + 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.""" + 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.""" + 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 c9954a1724..4fd3701693 100644 --- a/test/test_ssl.py +++ b/test/test_ssl.py @@ -16,10 +16,12 @@ from __future__ import annotations +import asyncio import os import pathlib import socket import sys +import unittest.mock as mock sys.path[0:0] = [""] @@ -28,11 +30,16 @@ from pymongo import MongoClient, ssl_support from pymongo.errors import ConfigurationError, ConnectionFailure, OperationFailure from pymongo.hello import HelloCompat -from pymongo.ssl_support import HAVE_PYSSL, HAVE_SSL, get_ssl_context +from pymongo.pool_shared import ( + _configured_socket_interface, + _get_ssl_session, +) +from pymongo.ssl_support import HAVE_PYSSL, HAVE_SSL, _ssl, get_ssl_context from pymongo.write_concern import WriteConcern from test import ( HAVE_IPADDRESS, IntegrationTest, + PyMongoTestCase, SkipTest, client_context, connected, @@ -48,7 +55,7 @@ _HAVE_PYOPENSSL = False try: - import pymongo.pyopenssl_context + from pymongo import pyopenssl_context _HAVE_PYOPENSSL = True except ImportError: @@ -83,6 +90,352 @@ # use 'localhost' for the hostname of all hosts. +class TestClientSSL(PyMongoTestCase): + @unittest.skipIf(HAVE_SSL, "The ssl module is available, can't test what happens without it.") + def test_no_ssl_module(self): + # Explicit + self.assertRaises(ConfigurationError, self.simple_client, ssl=True) + + # Implied + self.assertRaises(ConfigurationError, self.simple_client, tlsCertificateKeyFile=CLIENT_PEM) + + @unittest.skipUnless(HAVE_SSL, "The ssl module is not available.") + @ignore_deprecations + def test_config_ssl(self): + # Tests various ssl configurations + self.assertRaises(ValueError, self.simple_client, ssl="foo") + self.assertRaises( + ConfigurationError, self.simple_client, tls=False, tlsCertificateKeyFile=CLIENT_PEM + ) + self.assertRaises(TypeError, self.simple_client, ssl=0) + self.assertRaises(TypeError, self.simple_client, ssl=5.5) + self.assertRaises(TypeError, self.simple_client, ssl=[]) + + self.assertRaises(IOError, self.simple_client, tlsCertificateKeyFile="NoSuchFile") + self.assertRaises(TypeError, self.simple_client, tlsCertificateKeyFile=True) + self.assertRaises(TypeError, self.simple_client, tlsCertificateKeyFile=[]) + + # Test invalid combinations + self.assertRaises( + ConfigurationError, self.simple_client, tls=False, tlsCertificateKeyFile=CLIENT_PEM + ) + self.assertRaises(ConfigurationError, self.simple_client, tls=False, tlsCAFile=CA_PEM) + self.assertRaises(ConfigurationError, self.simple_client, tls=False, tlsCRLFile=CRL_PEM) + self.assertRaises( + ConfigurationError, self.simple_client, tls=False, tlsAllowInvalidCertificates=False + ) + self.assertRaises( + ConfigurationError, self.simple_client, tls=False, tlsAllowInvalidHostnames=False + ) + self.assertRaises( + ConfigurationError, self.simple_client, tls=False, tlsDisableOCSPEndpointCheck=False + ) + + @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.") + def test_use_pyopenssl_when_available(self): + self.assertTrue(HAVE_PYSSL) + + def test_ssl_cert_file_env_var_uses_default_certs_on_linux(self): + # PYTHON-5930: on Linux, load_default_certs() already honors SSL_CERT_FILE + # correctly (unlike Windows/macOS+PyOpenSSL), so it must still be called + # instead of bypassed. + env = dict(os.environ) + env.pop("SSL_CERT_DIR", None) + env["SSL_CERT_FILE"] = CA_PEM + with ( + mock.patch.dict(os.environ, env, clear=True), + mock.patch.object(sys, "platform", "linux"), + mock.patch("ssl.SSLContext.load_default_certs") as mock_default, + mock.patch("ssl.SSLContext.load_verify_locations") as mock_verify, + ): + get_ssl_context(None, None, None, None, False, False, False, False) + mock_default.assert_called_once() + mock_verify.assert_not_called() + + def test_ssl_cert_file_env_var_bypasses_default_certs_on_windows(self): + # PYTHON-5930: on win32, load_default_certs() merges the OS certificate + # store with SSL_CERT_FILE/SSL_CERT_DIR, so it must be bypassed in favor + # of loading the env vars exclusively. + env = dict(os.environ) + env.pop("SSL_CERT_DIR", None) + env["SSL_CERT_FILE"] = CA_PEM + with ( + mock.patch.dict(os.environ, env, clear=True), + mock.patch.object(sys, "platform", "win32"), + mock.patch("ssl.SSLContext.load_default_certs") as mock_default, + mock.patch("ssl.SSLContext.load_verify_locations") as mock_verify, + ): + get_ssl_context(None, None, None, None, False, False, False, False) + mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) + mock_default.assert_not_called() + + @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.") + def test_ssl_cert_file_env_var_bypasses_default_certs_on_macos_pyopenssl(self): + # PYTHON-5930: on macOS with PyOpenSSL, load_default_certs() merges in + # certifi certs, so it must be bypassed just like on win32. + env = dict(os.environ) + env.pop("SSL_CERT_DIR", None) + env["SSL_CERT_FILE"] = CA_PEM + with ( + mock.patch.dict(os.environ, env, clear=True), + mock.patch.object(sys, "platform", "darwin"), + mock.patch("pymongo.pyopenssl_context.SSLContext.load_default_certs") as mock_default, + mock.patch("pymongo.pyopenssl_context.SSLContext.load_verify_locations") as mock_verify, + ): + get_ssl_context(None, None, None, None, False, False, False, True) + mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) + mock_default.assert_not_called() + + def test_ssl_session_cache(self): + 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.""" + + fake_session = object() + cache: list = [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) + + def test_get_ssl_session_pyopenssl_style(self): + """_get_ssl_session uses get_session() when available (PyOpenSSL path).""" + 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).""" + 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)", + ) + def test_async_tls_session_injected_via_sslobject_class(self): + """On Python 3.11+, a cached session is injected by setting sslobject_class.""" + 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[0] + assert session is not None + _session = session + + class _SessionSSLObject(ssl.SSLObject): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.session = _session + + real_ctx.sslobject_class = _SessionSSLObject + + 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.""" + 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) + + @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.""" + 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.""" + 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.""" + 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 diff --git a/test/test_ssl_support.py b/test/test_ssl_support.py deleted file mode 100644 index 2d8e13cae1..0000000000 --- a/test/test_ssl_support.py +++ /dev/null @@ -1,410 +0,0 @@ -# Copyright 2011-present MongoDB, Inc. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""Unit tests for pymongo/ssl_support.py and the SSL session cache in pool_shared.py. - -These tests are pure unit tests (mocked or in-process) with no async-specific -behavior, so unlike test_ssl.py they are not generated/mirrored for sync and async. -""" - -from __future__ import annotations - -import asyncio -import os -import pathlib -import sys -import unittest.mock as mock - -sys.path[0:0] = [""] - -from pymongo import MongoClient -from pymongo.errors import ConfigurationError -from pymongo.pool_shared import ( - _configured_protocol_interface, - _configured_socket_interface, - _get_ssl_session, -) -from pymongo.ssl_support import HAVE_PYSSL, HAVE_SSL, get_ssl_context -from test import unittest -from test.utils_shared import ignore_deprecations - -_HAVE_PYOPENSSL = False -try: - import pymongo.pyopenssl_context - - _HAVE_PYOPENSSL = True -except ImportError: - pass - -if HAVE_SSL: - import ssl - -CERT_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "certificates") -CLIENT_PEM = os.path.join(CERT_PATH, "client.pem") -CA_PEM = os.path.join(CERT_PATH, "ca.pem") -CRL_PEM = os.path.join(CERT_PATH, "crl.pem") - - -class TestClientSSL(unittest.TestCase): - @unittest.skipIf(HAVE_SSL, "The ssl module is available, can't test what happens without it.") - def test_no_ssl_module(self): - # Explicit - self.assertRaises(ConfigurationError, MongoClient, ssl=True) - - # Implied - self.assertRaises(ConfigurationError, MongoClient, tlsCertificateKeyFile=CLIENT_PEM) - - @unittest.skipUnless(HAVE_SSL, "The ssl module is not available.") - @ignore_deprecations - def test_config_ssl(self): - # Tests various ssl configurations - self.assertRaises(ValueError, MongoClient, ssl="foo") - self.assertRaises( - ConfigurationError, MongoClient, tls=False, tlsCertificateKeyFile=CLIENT_PEM - ) - self.assertRaises(TypeError, MongoClient, ssl=0) - self.assertRaises(TypeError, MongoClient, ssl=5.5) - self.assertRaises(TypeError, MongoClient, ssl=[]) - - self.assertRaises(IOError, MongoClient, tlsCertificateKeyFile="NoSuchFile") - self.assertRaises(TypeError, MongoClient, tlsCertificateKeyFile=True) - self.assertRaises(TypeError, MongoClient, tlsCertificateKeyFile=[]) - - # Test invalid combinations - self.assertRaises( - ConfigurationError, MongoClient, tls=False, tlsCertificateKeyFile=CLIENT_PEM - ) - self.assertRaises(ConfigurationError, MongoClient, tls=False, tlsCAFile=CA_PEM) - self.assertRaises(ConfigurationError, MongoClient, tls=False, tlsCRLFile=CRL_PEM) - self.assertRaises( - ConfigurationError, MongoClient, tls=False, tlsAllowInvalidCertificates=False - ) - self.assertRaises( - ConfigurationError, MongoClient, tls=False, tlsAllowInvalidHostnames=False - ) - self.assertRaises( - ConfigurationError, MongoClient, tls=False, tlsDisableOCSPEndpointCheck=False - ) - - @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.") - def test_use_pyopenssl_when_available(self): - self.assertTrue(HAVE_PYSSL) - - def test_ssl_cert_file_env_var_uses_default_certs_on_linux(self): - # PYTHON-5930: on Linux, load_default_certs() already honors SSL_CERT_FILE - # correctly (unlike Windows/macOS+PyOpenSSL), so it must still be called - # instead of bypassed. - env = dict(os.environ) - env.pop("SSL_CERT_DIR", None) - env["SSL_CERT_FILE"] = CA_PEM - with ( - mock.patch.dict(os.environ, env, clear=True), - mock.patch.object(sys, "platform", "linux"), - mock.patch("ssl.SSLContext.load_default_certs") as mock_default, - mock.patch("ssl.SSLContext.load_verify_locations") as mock_verify, - ): - get_ssl_context(None, None, None, None, False, False, False, False) - mock_default.assert_called_once() - mock_verify.assert_not_called() - - def test_ssl_cert_file_env_var_bypasses_default_certs_on_windows(self): - # PYTHON-5930: on win32, load_default_certs() merges the OS certificate - # store with SSL_CERT_FILE/SSL_CERT_DIR, so it must be bypassed in favor - # of loading the env vars exclusively. - env = dict(os.environ) - env.pop("SSL_CERT_DIR", None) - env["SSL_CERT_FILE"] = CA_PEM - with ( - mock.patch.dict(os.environ, env, clear=True), - mock.patch.object(sys, "platform", "win32"), - mock.patch("ssl.SSLContext.load_default_certs") as mock_default, - mock.patch("ssl.SSLContext.load_verify_locations") as mock_verify, - ): - get_ssl_context(None, None, None, None, False, False, False, False) - mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) - mock_default.assert_not_called() - - @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.") - def test_ssl_cert_file_env_var_bypasses_default_certs_on_macos_pyopenssl(self): - # PYTHON-5930: on macOS with PyOpenSSL, load_default_certs() merges in - # certifi certs, so it must be bypassed just like on win32. - env = dict(os.environ) - env.pop("SSL_CERT_DIR", None) - env["SSL_CERT_FILE"] = CA_PEM - with ( - mock.patch.dict(os.environ, env, clear=True), - mock.patch.object(sys, "platform", "darwin"), - mock.patch("pymongo.pyopenssl_context.SSLContext.load_default_certs") as mock_default, - mock.patch("pymongo.pyopenssl_context.SSLContext.load_verify_locations") as mock_verify, - ): - get_ssl_context(None, None, None, None, False, False, False, True) - mock_verify.assert_called_once_with(cafile=CA_PEM, capath=None) - mock_default.assert_not_called() - - def test_ssl_session_cache(self): - 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") - - def test_tls_session_reused_on_second_connection(self): - """Cached TLS session is passed to wrap_socket on subsequent connections.""" - - fake_session = object() - cache: list = [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) - - def test_get_ssl_session_pyopenssl_style(self): - """_get_ssl_session uses get_session() when available (PyOpenSSL path).""" - 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).""" - fake_session = object() - - class FakeSSLSock: - session = fake_session - - self.assertIs(_get_ssl_session(FakeSSLSock()), fake_session) - - @unittest.skipUnless( - sys.version_info >= (3, 11), - "Tests async sslobject_class injection (Python 3.11+ only)", - ) - def test_async_tls_session_injected_via_sslobject_class(self): - """On Python 3.11+, a cached session is injected by setting sslobject_class.""" - 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[0] - assert session is not None - _session = session - - class _SessionSSLObject(ssl.SSLObject): - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self.session = _session - - real_ctx.sslobject_class = _SessionSSLObject - - self.assertIs(real_ctx.sslobject_class, _SessionSSLObject) - self.assertTrue(issubclass(real_ctx.sslobject_class, ssl.SSLObject)) - - def test_async_configured_protocol_saves_session_to_cache(self): - """After a successful TLS connection the session is stored in the cache.""" - 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) - - @unittest.skipUnless( - 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.""" - 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) - - def test_async_configured_protocol_no_cache(self): - """When ssl_session_cache is None, no injection or save occurs.""" - 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] - - def test_async_configured_protocol_new_session_is_none(self): - """When ssl_object.session is None after connect, the cache is not updated.""" - 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]) - - -if __name__ == "__main__": - unittest.main()