Repository: pyca/pyopenssl
Pull request #1353 (2024-08-09 21:40:10)

Allow Connection.get_peer_cert_chain to return cryptography certificates
refs https://github.com/pyca/pyopenssl/issues/1321


Patch:
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 81e1b22..c6cb471 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -21,7 +21,7 @@ Deprecations:
 Changes:
 ^^^^^^^^
 
-* ``OpenSSL.SSL.Connection.get_certificate`` and ``OpenSSL.SSL.Connection.get_peer_certificate`` now take an ``as_cryptography`` keyword-argument. When ``True`` is passed then a ``cryptography.x509.Certificate`` is returned, instead of an ``OpenSSL.crypto.X509``. In the future, passing ``False`` (the default) will be deprecated.
+* ``OpenSSL.SSL.Connection.get_certificate``, ``OpenSSL.SSL.Connection.get_peer_certificate``, and ``OpenSSL.SSL.Connection.get_peer_cert_chain`` now take an ``as_cryptography`` keyword-argument. When ``True`` is passed then ``cryptography.x509.Certificate`` are returned, instead of ``OpenSSL.crypto.X509``. In the future, passing ``False`` (the default) will be deprecated.
 
 
 24.2.1 (2024-07-20)
diff --git a/src/OpenSSL/SSL.py b/src/OpenSSL/SSL.py
index b420880..0b944ca 100644
--- a/src/OpenSSL/SSL.py
+++ b/src/OpenSSL/SSL.py
@@ -2781,10 +2781,48 @@ class Connection:
             result.append(pycert)
         return result
 
-    def get_peer_cert_chain(self) -> list[X509] | None:
+    @staticmethod
+    def _cert_stack_to_cryptography_list(
+        cert_stack: Any,
+    ) -> list[x509.Certificate]:
+        """
+        Internal helper to convert a STACK_OF(X509) to a list of X509
+        instances.
+        """
+        result = []
+        for i in range(_lib.sk_X509_num(cert_stack)):
+            cert = _lib.sk_X509_value(cert_stack, i)
+            _openssl_assert(cert != _ffi.NULL)
+            res = _lib.X509_up_ref(cert)
+            _openssl_assert(res >= 1)
+            pycert = X509._from_raw_x509_ptr(cert)
+            result.append(pycert.to_cryptography())
+        return result
+
+    @typing.overload
+    def get_peer_cert_chain(
+        self, *, as_cryptography: typing.Literal[True]
+    ) -> list[x509.Certificate] | None:
+        pass
+
+    @typing.overload
+    def get_peer_cert_chain(
+        self, *, as_cryptography: typing.Literal[False] = False
+    ) -> list[X509] | None:
+        pass
+
+    def get_peer_cert_chain(
+        self,
+        *,
+        as_cryptography: typing.Literal[True] | typing.Literal[False] = False,
+    ) -> list[X509] | list[x509.Certificate] | None:
         """
         Retrieve the other side's certificate (if any)
 
+        :param bool as_cryptography: Controls whether a list of
+            ``cryptography.x509.Certificate`` or ``OpenSSL.crypto.X509``
+            object should be returned.
+
         :return: A list of X509 instances giving the peer's certificate chain,
                  or None if it does not have one.
         """
@@ -2792,6 +2830,8 @@ class Connection:
         if cert_stack == _ffi.NULL:
             return None
 
+        if as_cryptography:
+            return self._cert_stack_to_cryptography_list(cert_stack)
         return self._cert_stack_to_list(cert_stack)
 
     def get_verified_chain(self) -> list[X509] | None:


Test patch:
diff --git a/tests/test_ssl.py b/tests/test_ssl.py
index caf7b8f..cd21b27 100644
--- a/tests/test_ssl.py
+++ b/tests/test_ssl.py
@@ -2612,6 +2612,21 @@ class TestConnection:
         assert "Intermediate Certificate" == chain[1].get_subject().CN
         assert "Authority Certificate" == chain[2].get_subject().CN
 
+        cryptography_chain = client.get_peer_cert_chain(as_cryptography=True)
+        assert len(cryptography_chain) == 3
+        assert (
+            cryptography_chain[0].subject.rfc4514_string()
+            == "CN=Server Certificate"
+        )
+        assert (
+            cryptography_chain[1].subject.rfc4514_string()
+            == "CN=Intermediate Certificate"
+        )
+        assert (
+            cryptography_chain[2].subject.rfc4514_string()
+            == "CN=Authority Certificate"
+        )
+
     def test_get_peer_cert_chain_none(self):
         """
         `Connection.get_peer_cert_chain` returns `None` if the peer sends