Refactoring verify_cert
Apply the changes that we've been talking about in
https://github.com/pyca/pyopenssl/pull/155 regarding the placement of
verify_cert, viz., moving verify_cert from top level of crypto into
X509StoreContext.
This makes the pyOpenSSL API slightly different than the OpenSSL API,
but the plan will be to add back a verify_cert to the top level that is
nice to use.
diff --git a/OpenSSL/crypto.py b/OpenSSL/crypto.py
index b6951c0..395f273 100644
--- a/OpenSSL/crypto.py
+++ b/OpenSSL/crypto.py
@@ -32,35 +32,7 @@
"""
-
-def _exception_from_context_error(exception_type, store_ctx):
- """
- Convert a :py:func:`OpenSSL.crypto.verify_cert` failure into a Python
- exception.
-
- When a call to native OpenSSL X509_verify_cert fails, additonal information
- about the failure can be obtained from the store context.
- """
-
- errors = [
- _lib.X509_STORE_CTX_get_error(store_ctx._store_ctx),
- _lib.X509_STORE_CTX_get_error_depth(store_ctx._store_ctx),
- _native(_ffi.string(_lib.X509_verify_cert_error_string(
- _lib.X509_STORE_CTX_get_error(store_ctx._store_ctx)))),
- ]
- _x509 = _lib.X509_STORE_CTX_get_current_cert(store_ctx._store_ctx)
- if _x509 != _ffi.NULL:
- _cert = _lib.X509_dup(_x509)
- pycert = X509.__new__(X509)
- pycert._x509 = _ffi.gc(_cert, _lib.X509_free)
- e = exception_type(errors)
- e.certificate = pycert
- raise e
-
-
-
_raise_current_error = partial(_exception_from_error_queue, Error)
-_raise_context_error = partial(_exception_from_context_error, Error)
@@ -1387,6 +1359,19 @@
X509StoreType = X509Store
+class X509StoreContextError(Exception):
+ """
+ An error occurred while verifying a certificate using
+ `OpenSSL.X509StoreContext.verify_certificate`.
+
+ :param certificate: The certificate which caused verificate failure.
+ :type cert: :class:`X509`
+
+ """
+ def __init__(self, message, certificate):
+ super(X509StoreContextError, self).__init__(message)
+ self.certificate = certificate
+
class X509StoreContext(object):
"""
@@ -1440,6 +1425,42 @@
_lib.X509_STORE_CTX_cleanup(self._store_ctx)
+ def _exception_from_context(self):
+ """
+ Convert an OpenSSL native context error failure into a Python
+ exception.
+
+ When a call to native OpenSSL X509_verify_cert fails, additonal information
+ about the failure can be obtained from the store context.
+ """
+ errors = [
+ _lib.X509_STORE_CTX_get_error(self._store_ctx),
+ _lib.X509_STORE_CTX_get_error_depth(self._store_ctx),
+ _native(_ffi.string(_lib.X509_verify_cert_error_string(
+ _lib.X509_STORE_CTX_get_error(self._store_ctx)))),
+ ]
+ _x509 = _lib.X509_STORE_CTX_get_current_cert(self._store_ctx)
+ if _x509 != _ffi.NULL:
+ _cert = _lib.X509_dup(_x509)
+ pycert = X509.__new__(X509)
+ pycert._x509 = _ffi.gc(_cert, _lib.X509_free)
+ return X509StoreContextError(errors, pycert)
+
+
+ def verify_certificate(self):
+ """
+ Verify a certificate in a context.
+
+ :param store_ctx: The :py:class:`X509StoreContext` to verify.
+ :raises: Error
+ """
+ self._init()
+ ret = _lib.X509_verify_cert(self._store_ctx)
+ self._cleanup()
+ if ret <= 0:
+ raise self._exception_from_context()
+
+
def load_certificate(type, buffer):
"""
@@ -2388,20 +2409,6 @@
_raise_current_error()
-def verify_cert(store_ctx):
- """
- Verify a certificate in a context.
-
- :param store_ctx: The :py:class:`X509StoreContext` to verify.
- :raises: Error
- """
- store_ctx._init()
- ret = _lib.X509_verify_cert(store_ctx._store_ctx)
- store_ctx._cleanup()
- if ret <= 0:
- _raise_context_error(store_ctx)
-
-
def load_crl(type, buffer):
"""
Load a certificate revocation list from a buffer
diff --git a/OpenSSL/test/test_crypto.py b/OpenSSL/test/test_crypto.py
index fc6350c..0aac1e5 100644
--- a/OpenSSL/test/test_crypto.py
+++ b/OpenSSL/test/test_crypto.py
@@ -17,7 +17,7 @@
from OpenSSL.crypto import TYPE_RSA, TYPE_DSA, Error, PKey, PKeyType
from OpenSSL.crypto import X509, X509Type, X509Name, X509NameType
-from OpenSSL.crypto import X509Store, X509StoreType, X509StoreContext
+from OpenSSL.crypto import X509Store, X509StoreType, X509StoreContext, X509StoreContextError
from OpenSSL.crypto import X509Req, X509ReqType
from OpenSSL.crypto import X509Extension, X509ExtensionType
from OpenSSL.crypto import load_certificate, load_privatekey
@@ -29,7 +29,7 @@
from OpenSSL.crypto import CRL, Revoked, load_crl
from OpenSSL.crypto import NetscapeSPKI, NetscapeSPKIType
from OpenSSL.crypto import (
- sign, verify, verify_cert, get_elliptic_curve, get_elliptic_curves)
+ sign, verify, get_elliptic_curve, get_elliptic_curves)
from OpenSSL.test.util import EqualityTestsMixin, TestCase
from OpenSSL._util import native, lib
@@ -3175,9 +3175,9 @@
-class VerifyCertTests(TestCase):
+class X509StoreContextTests(TestCase):
"""
- Tests for :py:obj:`OpenSSL.crypto.verify_cert`.
+ Tests for :py:obj:`OpenSSL.crypto.X509StoreContext`.
"""
root_cert = load_certificate(FILETYPE_PEM, root_cert_pem)
intermediate_cert = load_certificate(FILETYPE_PEM, intermediate_cert_pem)
@@ -3185,74 +3185,74 @@
def test_valid(self):
"""
- :py:obj:`verify_cert` returns ``None`` when called with a certificate
+ :py:obj:`verify_certificate` returns ``None`` when called with a certificate
and valid chain.
"""
store = X509Store()
store.add_cert(self.root_cert)
store.add_cert(self.intermediate_cert)
store_ctx = X509StoreContext(store, self.intermediate_server_cert)
- self.assertEqual(verify_cert(store_ctx), None)
+ self.assertEqual(store_ctx.verify_certificate(), None)
def test_reuse(self):
"""
- :py:obj:`verify_cert` can be called multiple times with the same
+ :py:obj:`verify_certificate` can be called multiple times with the same
``X509StoreContext`` instance to produce the same result.
"""
store = X509Store()
store.add_cert(self.root_cert)
store.add_cert(self.intermediate_cert)
store_ctx = X509StoreContext(store, self.intermediate_server_cert)
- self.assertEqual(verify_cert(store_ctx), None)
- self.assertEqual(verify_cert(store_ctx), None)
+ self.assertEqual(store_ctx.verify_certificate(), None)
+ self.assertEqual(store_ctx.verify_certificate(), None)
def test_trusted_self_signed(self):
"""
- :py:obj:`verify_cert` returns ``None`` when called with a self-signed
+ :py:obj:`verify_certificate` returns ``None`` when called with a self-signed
certificate and itself in the chain.
"""
store = X509Store()
store.add_cert(self.root_cert)
store_ctx = X509StoreContext(store, self.root_cert)
- self.assertEqual(verify_cert(store_ctx), None)
+ self.assertEqual(store_ctx.verify_certificate(), None)
def test_untrusted_self_signed(self):
"""
- :py:obj:`verify_cert` raises error when a self-signed certificate is
+ :py:obj:`verify_certificate` raises error when a self-signed certificate is
verified without itself in the chain.
"""
store = X509Store()
store_ctx = X509StoreContext(store, self.root_cert)
- e = self.assertRaises(Error, verify_cert, store_ctx)
+ e = self.assertRaises(X509StoreContextError, store_ctx.verify_certificate)
self.assertEqual(e.args[0][2], 'self signed certificate')
self.assertEqual(e.certificate.get_subject().CN, 'Testing Root CA')
def test_invalid_chain_no_root(self):
"""
- :py:obj:`verify_cert` raises error when a root certificate is missing
+ :py:obj:`verify_certificate` raises error when a root certificate is missing
from the chain.
"""
store = X509Store()
store.add_cert(self.intermediate_cert)
store_ctx = X509StoreContext(store, self.intermediate_server_cert)
- e = self.assertRaises(Error, verify_cert, store_ctx)
+ e = self.assertRaises(X509StoreContextError, store_ctx.verify_certificate)
self.assertEqual(e.args[0][2], 'unable to get issuer certificate')
self.assertEqual(e.certificate.get_subject().CN, 'intermediate')
def test_invalid_chain_no_intermediate(self):
"""
- :py:obj:`verify_cert` raises error when an intermediate certificate is
+ :py:obj:`verify_certificate` raises error when an intermediate certificate is
missing from the chain.
"""
store = X509Store()
store.add_cert(self.root_cert)
store_ctx = X509StoreContext(store, self.intermediate_server_cert)
- e = self.assertRaises(Error, verify_cert, store_ctx)
+ e = self.assertRaises(X509StoreContextError, store_ctx.verify_certificate)
self.assertEqual(e.args[0][2], 'unable to get local issuer certificate')
self.assertEqual(e.certificate.get_subject().CN, 'intermediate-service')