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
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 @@
X509Name,
X509,
X509Store,
+ X509StoreContext,
)
__all__ = [
@@ -2126,6 +2127,22 @@
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 @@
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):
"""