PYTHON-5849 Fix OCSP and pyOpenSSL context compatibility with pyOpenSSL 26.2.0

pymongo/ocsp_support.py

@@ -328,22 +328,15 @@ def _ocsp_callback(conn: Connection, ocsp_bytes: bytes, user_data: Optional[_Cal
     """Callback for use with OpenSSL.SSL.Context.set_ocsp_client_callback."""
     # always pass in user_data but OpenSSL requires it be optional
     assert user_data
-    pycert = conn.get_peer_certificate()
-    if pycert is None:
+    cert = conn.get_peer_certificate(as_cryptography=True)
+    if cert is None:
         _LOGGER.debug("No peer cert?")
         return False
-    cert = pycert.to_cryptography()
-    # Use the verified chain when available (pyopenssl>=20.0).
-    if hasattr(conn, "get_verified_chain"):
-        pychain = conn.get_verified_chain()
-        trusted_ca_certs = None
-    else:
-        pychain = conn.get_peer_cert_chain()
-        trusted_ca_certs = user_data.trusted_ca_certs
-    if not pychain:
+    chain = conn.get_verified_chain(as_cryptography=True)
+    trusted_ca_certs = None
+    if not chain:
         _LOGGER.debug("No peer cert chain?")
         return False
-    chain = [cer.to_cryptography() for cer in pychain]
     issuer = _get_issuer_cert(cert, chain, trusted_ca_certs)
     must_staple = False
     # https://tools.ietf.org/html/rfc7633#section-4.2.3.1

pymongo/pyopenssl_context.py

@@ -201,13 +201,16 @@ class SSLContext:
     context.
     """
 
-    __slots__ = ("_protocol", "_ctx", "_callback_data", "_check_hostname")
+    __slots__ = ("_protocol", "_ctx", "_callback_data", "_check_hostname", "_options")
 
     def __init__(self, protocol: int):
         self._protocol = protocol
         self._ctx = _SSL.Context(self._protocol)
         self._callback_data = _CallbackData()
         self._check_hostname = True
+        # Cache options to avoid calling set_options() after a Connection is
+        # created (pyOpenSSL >= 26.2.0 disallows mutations on a used Context).
+        self._options = self._ctx.set_options(0)
         # OCSP
         # XXX: Find a better place to do this someday, since this is client
         # side configuration and wrap_socket tries to support both client and
@@ -231,24 +234,7 @@ def __get_verify_mode(self) -> VerifyMode:
 
     def __set_verify_mode(self, value: VerifyMode) -> None:
         """Setter for verify_mode."""
-
-        def _cb(
-            _connobj: _SSL.Connection,
-            _x509obj: _crypto.X509,
-            _errnum: int,
-            _errdepth: int,
-            retcode: int,
-        ) -> bool:
-            # It seems we don't need to do anything here. Twisted doesn't,
-            # and OpenSSL's SSL_CTX_set_verify let's you pass NULL
-            # for the callback option. It's weird that PyOpenSSL requires
-            # this.
-            # This is optional in pyopenssl >= 20 and can be removed once minimum
-            # supported version is bumped
-            # See: pyopenssl.org/en/latest/changelog.html#id47
-            return bool(retcode)
-
-        self._ctx.set_verify(_VERIFY_MAP[value], _cb)
+        self._ctx.set_verify(_VERIFY_MAP[value], None)
 
     verify_mode = property(__get_verify_mode, __set_verify_mode)
 
@@ -271,16 +257,14 @@ def __set_check_ocsp_endpoint(self, value: bool) -> None:
     check_ocsp_endpoint = property(__get_check_ocsp_endpoint, __set_check_ocsp_endpoint)
 
     def __get_options(self) -> int:
-        # Calling set_options adds the option to the existing bitmask and
-        # returns the new bitmask.
-        # https://www.pyopenssl.org/en/stable/api/ssl.html#OpenSSL.SSL.Context.set_options
-        return self._ctx.set_options(0)
+        return self._options
 
     def __set_options(self, value: int) -> None:
         # Explicitly convert to int, since newer CPython versions
         # use enum.IntFlag for options. The values are the same
         # regardless of implementation.
-        self._ctx.set_options(int(value))
+        self._options = int(value)
+        self._ctx.set_options(self._options)
 
     options = property(__get_options, __set_options)
 

requirements/ocsp.txt

@@ -1,7 +1,8 @@
 # PyOpenSSL 17.0.0 introduced support for OCSP. 17.1.0 introduced
 # a related feature we need. 17.2.0 fixes a bug
 # in set_default_verify_paths we should really avoid.
-# service_identity 18.1.0 introduced support for IP addr matching.
+# service_identity 24.2.0 dropped use of X509.get_extension() which was
+# removed in pyOpenSSL 26.2.0.
 # Fallback to certifi on Windows if we can't load CA certs from the system
 # store and just use certifi on macOS.
 # pyopenssl, cryptography, and service_identity must be set in tandem.
@@ -10,4 +11,4 @@ certifi>=2023.7.22;os.name=='nt' or sys_platform=='darwin'
 pyopenssl>=26.2.0
 requests>=2.23.0,<3.0
 cryptography>=42.0.0
-service_identity>=23.1.0
+service_identity>=24.2.0

test/test_pyopenssl_context.py

@@ -22,17 +22,19 @@
 
 import ssl
 import sys
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 sys.path[0:0] = [""]
 
 from test import unittest
 
 try:
     from pymongo import pyopenssl_context as _ctx_module
+    from pymongo.ocsp_support import _ocsp_callback
     from pymongo.pyopenssl_context import (
         PROTOCOL_SSLv23,
         SSLContext,
+        _CallbackData,
         _is_ip_address,
         _ragged_eof,
     )
@@ -267,5 +269,44 @@ def test_delegates_to_ctx(self):
         mock_lvl.assert_called_once_with("/tmp/ca.pem", None)
 
 
+# ---------------------------------------------------------------------------
+# _ocsp_callback — early-exit branches
+# ---------------------------------------------------------------------------
+
+
+class TestOcspCallback(unittest.TestCase):
+    """Unit tests for _ocsp_callback using a mocked SSL Connection."""
+
+    def _make_callback_data(self):
+        return _CallbackData()
+
+    def _make_conn(self, *, peer_cert, chain):
+        """Return a mock Connection whose certificate methods return the given values."""
+        conn = MagicMock()
+        conn.get_peer_certificate.return_value = peer_cert
+        conn.get_verified_chain.return_value = chain
+        return conn
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_returns_false_when_peer_cert_is_none(self):
+        conn = self._make_conn(peer_cert=None, chain=None)
+        result = _ocsp_callback(conn, b"", self._make_callback_data())
+        self.assertFalse(result)
+        conn.get_peer_certificate.assert_called_once_with(as_cryptography=True)
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_returns_false_when_chain_is_none(self):
+        conn = self._make_conn(peer_cert=MagicMock(), chain=None)
+        result = _ocsp_callback(conn, b"", self._make_callback_data())
+        self.assertFalse(result)
+        conn.get_verified_chain.assert_called_once_with(as_cryptography=True)
+
+    @unittest.skipUnless(_HAVE_PYOPENSSL, "PyOpenSSL is not available.")
+    def test_returns_false_when_chain_is_empty(self):
+        conn = self._make_conn(peer_cert=MagicMock(), chain=[])
+        result = _ocsp_callback(conn, b"", self._make_callback_data())
+        self.assertFalse(result)
+
+
 if __name__ == "__main__":
     unittest.main()