PYTHON-5930 Fix SSL_CERT_FILE/SSL_CERT_DIR trust store handling

.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

.evergreen/scripts/generate_config.py

@@ -251,6 +251,7 @@ def create_pyopenssl_variants():
             create_variant(
                 tasks,
                 display_name,
+                host=host,
                 expansions=expansions,
                 batchtime=batchtime,
             )

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

pymongo/ssl_support.py

@@ -16,6 +16,8 @@
 
 from __future__ import annotations
 
+import os
+import sys
 import types
 import warnings
 from typing import Any, Optional, Union
@@ -133,7 +135,17 @@ 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") 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 (
+                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()
         ctx.verify_mode = verify_mode
         return ctx
 

test/asynchronous/test_ssl.py

@@ -136,6 +136,57 @@ def test_config_ssl(self):
     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])

test/test_ssl.py

@@ -135,6 +135,57 @@ def test_config_ssl(self):
     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])