Allow accessing a connection's verfied certificate chain (#894)

* Allow accessing a connection's verfied certificate chain

Add X509StoreContext.get_verified_chain using X509_STORE_CTX_get1_chain.
Add Connection.get_verified_chain using SSL_get0_verified_chain if
available (ie OpenSSL 1.1+) and X509StoreContext.get_verified_chain
otherwise.
Fixes #740.

* TLSv1_METHOD -> SSLv23_METHOD

* Use X509_up_ref instead of X509_dup

* Add _openssl_assert where appropriate

* SSL_get_peer_cert_chain should not be null

* Reformat with black

* Fix <OpenSSL.crypto.X509 object at 0x7fdbb59e8050> != <OpenSSL.crypto.X509 object at 0x7fdbb59daad0>

* Add Changelog entry

* Remove _add_chain

1 files changed, 54 insertions, 7 deletions

diff --git a/src/OpenSSL/SSL.py b/src/OpenSSL/SSL.py
index 8a54994..29e489a 100644
--- a/src/OpenSSL/SSL.py
+++ b/src/OpenSSL/SSL.py
@@ -28,6 +28,7 @@ from OpenSSL.crypto import (
     X509Name,
     X509,
     X509Store,
+    X509StoreContext,
 )
 
 __all__ = [
@@ -2126,6 +2127,22 @@ class Connection(object):
             return X509._from_raw_x509_ptr(cert)
         return None
 
+    @staticmethod
+    def _cert_stack_to_list(cert_stack):
+        """
+        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)
+        return result
+
     def get_peer_cert_chain(self):
         """
         Retrieve the other side's certificate (if any)
@@ -2137,13 +2154,43 @@ class Connection(object):
         if cert_stack == _ffi.NULL:
             return None
 
-        result = []
-        for i in range(_lib.sk_X509_num(cert_stack)):
-            # TODO could incref instead of dup here
-            cert = _lib.X509_dup(_lib.sk_X509_value(cert_stack, i))
-            pycert = X509._from_raw_x509_ptr(cert)
-            result.append(pycert)
-        return result
+        return self._cert_stack_to_list(cert_stack)
+
+    def get_verified_chain(self):
+        """
+        Retrieve the verified certificate chain of the peer including the
+        peer's end entity certificate. It must be called after a session has
+        been successfully established. If peer verification was not successful
+        the chain may be incomplete, invalid, or None.
+
+        :return: A list of X509 instances giving the peer's verified
+                 certificate chain, or None if it does not have one.
+
+        .. versionadded:: 20.0
+        """
+        if hasattr(_lib, "SSL_get0_verified_chain"):
+            # OpenSSL 1.1+
+            cert_stack = _lib.SSL_get0_verified_chain(self._ssl)
+            if cert_stack == _ffi.NULL:
+                return None
+
+            return self._cert_stack_to_list(cert_stack)
+
+        pycert = self.get_peer_certificate()
+        if pycert is None:
+            return None
+
+        # Should never be NULL because the peer presented a certificate.
+        cert_stack = _lib.SSL_get_peer_cert_chain(self._ssl)
+        _openssl_assert(cert_stack != _ffi.NULL)
+
+        pystore = self._context.get_cert_store()
+        if pystore is None:
+            return None
+
+        pystorectx = X509StoreContext(pystore, pycert)
+        pystorectx._chain = cert_stack
+        return pystorectx.get_verified_chain()
 
     def want_read(self):
         """