Add OpenSSL.crypto.verify_chain method.
This change adds support for verifying a certificate or a certificate
chain. This implementation uses OpenSSL's underlying X509_STORE_CTX_*
class of functions to accomplish this.
This change also adds an intermediate signing certificate/key and a
service certificate/key signed with the intermediate signing
certificate, to make testing the OpenSSL.crypto.verify_chain method
easier to test. I figured I would add it to the top level module so
other people can use an intermediate signing certificate in their own
tests.
Issue: https://github.com/pyca/pyopenssl/issues/154
diff --git a/OpenSSL/_util.py b/OpenSSL/_util.py
index baeecc6..cf13666 100644
--- a/OpenSSL/_util.py
+++ b/OpenSSL/_util.py
@@ -5,11 +5,25 @@
ffi = binding.ffi
lib = binding.lib
-def exception_from_error_queue(exceptionType):
- def text(charp):
- return native(ffi.string(charp))
+
+
+def text(charp):
+ return native(ffi.string(charp))
+
+
+
+def exception_from_error_queue(exception_type):
+ """
+ Convert an OpenSSL library failure into a Python exception.
+
+ When a call to the native OpenSSL library fails, this is usually signalled
+ by the return value, and an error code is stored in an error queue
+ associated with the current thread. The err library provides functions to
+ obtain these error codes and textual error messages.
+ """
errors = []
+
while True:
error = lib.ERR_get_error()
if error == 0:
@@ -19,7 +33,7 @@
text(lib.ERR_func_error_string(error)),
text(lib.ERR_reason_error_string(error))))
- raise exceptionType(errors)
+ raise exception_type(errors)
diff --git a/OpenSSL/crypto.py b/OpenSSL/crypto.py
index 8d971f2..6c07962 100644
--- a/OpenSSL/crypto.py
+++ b/OpenSSL/crypto.py
@@ -25,13 +25,43 @@
TYPE_DSA = _lib.EVP_PKEY_DSA
+
class Error(Exception):
"""
An error occurred in an `OpenSSL.crypto` API.
"""
+
+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 contained 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)
+
+
def _untested_error(where):
"""
@@ -1357,6 +1387,44 @@
+class X509StoreContext(object):
+ """
+ An X.509 store context.
+
+ A :py:class:`X509StoreContext` is used to verify a certificate in some
+ context in conjunction with :py:func:`verify_cert`. The information
+ encapsulated in this object includes, but is not limited to, a set of
+ trusted certificates, verification parameters and revoked certificates.
+
+ :param store: A :py:class:`X509Store` of trusted certificates.
+ :param cert: An :py:class:`X509` certificate to be validated during a
+ subsequent call to :py:func:`verify_cert`.
+ """
+
+ def __init__(self, store, cert):
+ store_ctx = _lib.X509_STORE_CTX_new()
+ self._store_ctx = _ffi.gc(store_ctx, _lib.X509_STORE_CTX_free)
+ self._store = store
+ self._cert = cert
+
+ def _init(self):
+ """
+ Set up the store context for a subsequent verification operation.
+ """
+ ret = _lib.X509_STORE_CTX_init(self._store_ctx, self._store._store, self._cert._x509, _ffi.NULL)
+ if ret <= 0:
+ _raise_current_error()
+
+ def _cleanup(self):
+ """
+ Internally cleans up the store context.
+
+ The store context can then be reused with a new call to
+ :py:meth:`init`.
+ """
+ _lib.X509_STORE_CTX_cleanup(self._store_ctx)
+
+
def load_certificate(type, buffer):
"""
Load a certificate from a buffer
@@ -2304,6 +2372,19 @@
_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):
"""
diff --git a/OpenSSL/test/test_crypto.py b/OpenSSL/test/test_crypto.py
index f704ac0..8c04938 100644
--- a/OpenSSL/test/test_crypto.py
+++ b/OpenSSL/test/test_crypto.py
@@ -10,6 +10,7 @@
import base64
import os
import re
+import sys
from subprocess import PIPE, Popen
from datetime import datetime, timedelta
@@ -17,7 +18,8 @@
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, X509Req, X509ReqType
+from OpenSSL.crypto import X509Store, X509StoreType, X509StoreContext
+from OpenSSL.crypto import X509Req, X509ReqType
from OpenSSL.crypto import X509Extension, X509ExtensionType
from OpenSSL.crypto import load_certificate, load_privatekey
from OpenSSL.crypto import FILETYPE_PEM, FILETYPE_ASN1, FILETYPE_TEXT
@@ -28,7 +30,7 @@
from OpenSSL.crypto import CRL, Revoked, load_crl
from OpenSSL.crypto import NetscapeSPKI, NetscapeSPKIType
from OpenSSL.crypto import (
- sign, verify, get_elliptic_curve, get_elliptic_curves)
+ sign, verify, verify_cert, get_elliptic_curve, get_elliptic_curves)
from OpenSSL.test.util import EqualityTestsMixin, TestCase
from OpenSSL._util import native, lib
@@ -83,6 +85,40 @@
-----END RSA PRIVATE KEY-----
""")
+intermediate_cert_pem = b("""-----BEGIN CERTIFICATE-----
+MIICVzCCAcCgAwIBAgIRAMPzhm6//0Y/g2pmnHR2C4cwDQYJKoZIhvcNAQENBQAw
+WDELMAkGA1UEBhMCVVMxCzAJBgNVBAgTAklMMRAwDgYDVQQHEwdDaGljYWdvMRAw
+DgYDVQQKEwdUZXN0aW5nMRgwFgYDVQQDEw9UZXN0aW5nIFJvb3QgQ0EwHhcNMTQw
+ODI4MDIwNDA4WhcNMjQwODI1MDIwNDA4WjBmMRUwEwYDVQQDEwxpbnRlcm1lZGlh
+dGUxDDAKBgNVBAoTA29yZzERMA8GA1UECxMIb3JnLXVuaXQxCzAJBgNVBAYTAlVT
+MQswCQYDVQQIEwJDQTESMBAGA1UEBxMJU2FuIERpZWdvMIGfMA0GCSqGSIb3DQEB
+AQUAA4GNADCBiQKBgQDYcEQw5lfbEQRjr5Yy4yxAHGV0b9Al+Lmu7wLHMkZ/ZMmK
+FGIbljbviiD1Nz97Oh2cpB91YwOXOTN2vXHq26S+A5xe8z/QJbBsyghMur88CjdT
+21H2qwMa+r5dCQwEhuGIiZ3KbzB/n4DTMYI5zy4IYPv0pjxShZn4aZTCCK2IUwID
+AQABoxMwETAPBgNVHRMBAf8EBTADAQH/MA0GCSqGSIb3DQEBDQUAA4GBAPIWSkLX
+QRMApOjjyC+tMxumT5e2pMqChHmxobQK4NMdrf2VCx+cRT6EmY8sK3/Xl/X8UBQ+
+9n5zXb1ZwhW/sTWgUvmOceJ4/XVs9FkdWOOn1J0XBch9ZIiFe/s5ASIgG7fUdcUF
+9mAWS6FK2ca3xIh5kIupCXOFa0dPvlw/YUFT
+-----END CERTIFICATE-----
+""")
+
+intermediate_key_pem = b("""-----BEGIN RSA PRIVATE KEY-----
+MIICWwIBAAKBgQDYcEQw5lfbEQRjr5Yy4yxAHGV0b9Al+Lmu7wLHMkZ/ZMmKFGIb
+ljbviiD1Nz97Oh2cpB91YwOXOTN2vXHq26S+A5xe8z/QJbBsyghMur88CjdT21H2
+qwMa+r5dCQwEhuGIiZ3KbzB/n4DTMYI5zy4IYPv0pjxShZn4aZTCCK2IUwIDAQAB
+AoGAfSZVV80pSeOKHTYfbGdNY/jHdU9eFUa/33YWriXU+77EhpIItJjkRRgivIfo
+rhFJpBSGmDLblaqepm8emsXMeH4+2QzOYIf0QGGP6E6scjTt1PLqdqKfVJ1a2REN
+147cujNcmFJb/5VQHHMpaPTgttEjlzuww4+BCDPsVRABWrkCQQD3loH36nLoQTtf
++kQq0T6Bs9/UWkTAGo0ND81ALj0F8Ie1oeZg6RNT96RxZ3aVuFTESTv6/TbjWywO
+wdzlmV1vAkEA38rTJ6PTwaJlw5OttdDzAXGPB9tDmzh9oSi7cHwQQXizYd8MBYx4
+sjHUKD3dCQnb1dxJFhd3BT5HsnkRMbVZXQJAbXduH17ZTzcIOXc9jHDXYiFVZV5D
+52vV0WCbLzVCZc3jMrtSUKa8lPN5EWrdU3UchWybyG0MR5mX8S5lrF4SoQJAIyUD
+DBKaSqpqONCUUx1BTFS9FYrFjzbL4+c1qHCTTPTblt8kUCrDOZjBrKAqeiTmNSum
+/qUot9YUBF8m6BuGsQJATHHmdFy/fG1VLkyBp49CAa8tN3Z5r/CgTznI4DfMTf4C
+NbRHn2UmYlwQBa+L5lg9phewNe8aEwpPyPLoV85U8Q==
+-----END RSA PRIVATE KEY-----
+""")
+
server_cert_pem = b("""-----BEGIN CERTIFICATE-----
MIICKDCCAZGgAwIBAgIJAJn/HpR21r/8MA0GCSqGSIb3DQEBBQUAMFgxCzAJBgNV
BAYTAlVTMQswCQYDVQQIEwJJTDEQMA4GA1UEBxMHQ2hpY2FnbzEQMA4GA1UEChMH
@@ -116,6 +152,40 @@
-----END RSA PRIVATE KEY-----
"""))
+intermediate_server_cert_pem = b("""-----BEGIN CERTIFICATE-----
+MIICWDCCAcGgAwIBAgIRAPQFY9jfskSihdiNSNdt6GswDQYJKoZIhvcNAQENBQAw
+ZjEVMBMGA1UEAxMMaW50ZXJtZWRpYXRlMQwwCgYDVQQKEwNvcmcxETAPBgNVBAsT
+CG9yZy11bml0MQswCQYDVQQGEwJVUzELMAkGA1UECBMCQ0ExEjAQBgNVBAcTCVNh
+biBEaWVnbzAeFw0xNDA4MjgwMjEwNDhaFw0yNDA4MjUwMjEwNDhaMG4xHTAbBgNV
+BAMTFGludGVybWVkaWF0ZS1zZXJ2aWNlMQwwCgYDVQQKEwNvcmcxETAPBgNVBAsT
+CG9yZy11bml0MQswCQYDVQQGEwJVUzELMAkGA1UECBMCQ0ExEjAQBgNVBAcTCVNh
+biBEaWVnbzCBnzANBgkqhkiG9w0BAQEFAAOBjQAwgYkCgYEAqpJZygd+w1faLOr1
+iOAmbBhx5SZWcTCZ/ZjHQTJM7GuPT624QkqsixFghRKdDROwpwnAP7gMRukLqiy4
++kRuGT5OfyGggL95i2xqA+zehjj08lSTlvGHpePJgCyTavIy5+Ljsj4DKnKyuhxm
+biXTRrH83NDgixVkObTEmh/OVK0CAwEAATANBgkqhkiG9w0BAQ0FAAOBgQBa0Npw
+UkzjaYEo1OUE1sTI6Mm4riTIHMak4/nswKh9hYup//WVOlr/RBSBtZ7Q/BwbjobN
+3bfAtV7eSAqBsfxYXyof7G1ALANQERkq3+oyLP1iVt08W1WOUlIMPhdCF/QuCwy6
+x9MJLhUCGLJPM+O2rAPWVD9wCmvq10ALsiH3yA==
+-----END CERTIFICATE-----
+""")
+
+intermediate_server_key_pem = b("""-----BEGIN RSA PRIVATE KEY-----
+MIICXAIBAAKBgQCqklnKB37DV9os6vWI4CZsGHHlJlZxMJn9mMdBMkzsa49PrbhC
+SqyLEWCFEp0NE7CnCcA/uAxG6QuqLLj6RG4ZPk5/IaCAv3mLbGoD7N6GOPTyVJOW
+8Yel48mALJNq8jLn4uOyPgMqcrK6HGZuJdNGsfzc0OCLFWQ5tMSaH85UrQIDAQAB
+AoGAIQ594j5zna3/9WaPsTgnmhlesVctt4AAx/n827DA4ayyuHFlXUuVhtoWR5Pk
+5ezj9mtYW8DyeCegABnsu2vZni/CdvU6uiS1Hv6qM1GyYDm9KWgovIP9rQCDSGaz
+d57IWVGxx7ODFkm3gN5nxnSBOFVHytuW1J7FBRnEsehRroECQQDXHFOv82JuXDcz
+z3+4c74IEURdOHcbycxlppmK9kFqm5lsUdydnnGW+mvwDk0APOB7Wg7vyFyr393e
+dpmBDCzNAkEAyv6tVbTKUYhSjW+QhabJo896/EqQEYUmtMXxk4cQnKeR/Ao84Rkf
+EqD5IykMUfUI0jJU4DGX+gWZ10a7kNbHYQJAVFCuHNFxS4Cpwo0aqtnzKoZaHY/8
+X9ABZfafSHCtw3Op92M+7ikkrOELXdS9KdKyyqbKJAKNEHF3LbOfB44WIQJAA2N4
+9UNNVUsXRbElEnYUS529CdUczo4QdVgQjkvk5RiPAUwSdBd9Q0xYnFOlFwEmIowg
+ipWJWe0aAlP18ZcEQQJBAL+5lekZ/GUdQoZ4HAsN5a9syrzavJ9VvU1KOOPorPZK
+nMRZbbQgP+aSB7yl6K0gaLaZ8XaK0pjxNBh6ASqg9f4=
+-----END RSA PRIVATE KEY-----
+""")
+
client_cert_pem = b("""-----BEGIN CERTIFICATE-----
MIICJjCCAY+gAwIBAgIJAKxpFI5lODkjMA0GCSqGSIb3DQEBBQUAMFgxCzAJBgNV
BAYTAlVTMQswCQYDVQQIEwJJTDEQMA4GA1UEBxMHQ2hpY2FnbzEQMA4GA1UEChMH
@@ -3105,6 +3175,88 @@
self.assertRaises(Error, load_crl, FILETYPE_PEM, b"hello, world")
+class VerifyCertTests(TestCase):
+ """
+ Tests for :py:obj:`OpenSSL.crypto.verify_cert`.
+ """
+ root_cert = load_certificate(FILETYPE_PEM, root_cert_pem)
+ intermediate_cert = load_certificate(FILETYPE_PEM, intermediate_cert_pem)
+ intermediate_server_cert = load_certificate(FILETYPE_PEM, intermediate_server_cert_pem)
+
+ def test_valid(self):
+ """
+ :py:obj:`verify_cert` does nothing 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)
+
+ def test_reuse(self):
+ """
+ :py:obj:`verify_cert` can be called multiple times.
+ """
+ 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)
+
+ def test_trusted_self_signed(self):
+ """
+ :py:obj:`verify_cert` does nothign 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)
+
+ def test_untrusted_self_signed(self):
+ """
+ :py:obj:`verify_cert` raises error when a self-signed certificate is
+ verified without itself in the chain.
+ """
+ store = X509Store()
+ store_ctx = X509StoreContext(store, self.root_cert)
+ try:
+ verify_cert(store_ctx)
+ self.assertTrue(False)
+ except Error as e:
+ self.assertTrue('self signed certificate' in str(e))
+ 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
+ from the chain.
+ """
+ store = X509Store()
+ store.add_cert(self.intermediate_cert)
+ store_ctx = X509StoreContext(store, self.intermediate_server_cert)
+ try:
+ verify_cert(store_ctx)
+ except Error as e:
+ self.assertTrue('unable to get issuer certificate' in str(e))
+ 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
+ missing from the chain.
+ """
+ store = X509Store()
+ store.add_cert(self.root_cert)
+ store_ctx = X509StoreContext(store, self.intermediate_server_cert)
+ try:
+ verify_cert(store_ctx)
+ except Error as e:
+ self.assertTrue('unable to get local issuer certificate' in str(e))
+ self.assertEqual(e.certificate.get_subject().CN, 'intermediate-service')
+
class SignVerifyTests(TestCase):
"""