Repository: pyca/pyopenssl
Pull request #948 (2020-10-09 15:14:16)

 Allow using additional untrusted certificates for chain building in X509StoreContext
The additional certificates provided in the new `chain` parameter will be
untrusted but may be used to build the chain.

This makes it easier to validate a certificate against a store which
contains only root ca certificates, and the intermediates come from e.g.
the same untrusted source as the certificate to be verified.

This PR is based on the work done by @akgood for PR #473. 


Patch:
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index e5f08d2..2ba1f7f 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -24,6 +24,9 @@ Deprecations:
 Changes:
 ^^^^^^^^
 
+- Added a new optional ``chain`` parameter to ``OpenSSL.crypto.X509StoreContext()``
+  where additional untrusted certificates can be specified to help chain building.
+  `#948 <https://github.com/pyca/pyopenssl/pull/948>`_
 - Added ``OpenSSL.crypto.X509Store.load_locations`` to set trusted
   certificate file bundles and/or directories for verification.
   `#943 <https://github.com/pyca/pyopenssl/pull/943>`_
diff --git a/src/OpenSSL/crypto.py b/src/OpenSSL/crypto.py
index 880c2b3..f89a28f 100644
--- a/src/OpenSSL/crypto.py
+++ b/src/OpenSSL/crypto.py
@@ -1750,22 +1750,54 @@ class X509StoreContext(object):
         collected.
     :ivar _store: See the ``store`` ``__init__`` parameter.
     :ivar _cert: See the ``certificate`` ``__init__`` parameter.
+    :ivar _chain: See the ``chain`` ``__init__`` parameter.
     :param X509Store store: The certificates which will be trusted for the
         purposes of any verifications.
     :param X509 certificate: The certificate to be verified.
+    :param chain: List of untrusted certificates that may be used for building
+        the certificate chain. May be ``None``.
+    :type chain: :class:`list` of :class:`X509`
     """
 
-    def __init__(self, store, certificate):
+    def __init__(self, store, certificate, chain=None):
         store_ctx = _lib.X509_STORE_CTX_new()
         self._store_ctx = _ffi.gc(store_ctx, _lib.X509_STORE_CTX_free)
         self._store = store
         self._cert = certificate
-        self._chain = _ffi.NULL
+        self._chain = self._build_certificate_stack(chain)
         # Make the store context available for use after instantiating this
         # class by initializing it now. Per testing, subsequent calls to
         # :meth:`_init` have no adverse affect.
         self._init()
 
+    @staticmethod
+    def _build_certificate_stack(certificates):
+        def cleanup(s):
+            # Equivalent to sk_X509_pop_free, but we don't
+            # currently have a CFFI binding for that available
+            for i in range(_lib.sk_X509_num(s)):
+                x = _lib.sk_X509_value(s, i)
+                _lib.X509_free(x)
+            _lib.sk_X509_free(s)
+
+        if certificates is None or len(certificates) == 0:
+            return _ffi.NULL
+
+        stack = _lib.sk_X509_new_null()
+        _openssl_assert(stack != _ffi.NULL)
+        stack = _ffi.gc(stack, cleanup)
+
+        for cert in certificates:
+            if not isinstance(cert, X509):
+                raise TypeError("One of the elements is not an X509 instance")
+
+            _openssl_assert(_lib.X509_up_ref(cert._x509) > 0)
+            if _lib.sk_X509_push(stack, cert._x509) <= 0:
+                _lib.X509_free(cert._x509)
+                _raise_current_error()
+
+        return stack
+
     def _init(self):
         """
         Set up the store context for a subsequent verification operation.


Test patch:
diff --git a/tests/test_crypto.py b/tests/test_crypto.py
index 820a4b2..815bd8b 100644
--- a/tests/test_crypto.py
+++ b/tests/test_crypto.py
@@ -3824,6 +3824,145 @@ class TestX509StoreContext(object):
         assert store_ctx.verify_certificate() is None
         assert store_ctx.verify_certificate() is None
 
+    @pytest.mark.parametrize(
+        "root_cert, chain, verified_cert",
+        [
+            pytest.param(
+                root_cert,
+                [intermediate_cert],
+                intermediate_server_cert,
+                id="intermediate in chain",
+            ),
+            pytest.param(
+                root_cert,
+                [],
+                intermediate_cert,
+                id="empty chain",
+            ),
+            pytest.param(
+                root_cert,
+                [root_cert, intermediate_server_cert, intermediate_cert],
+                intermediate_server_cert,
+                id="extra certs in chain",
+            ),
+        ],
+    )
+    def test_verify_success_with_chain(self, root_cert, chain, verified_cert):
+        store = X509Store()
+        store.add_cert(root_cert)
+        store_ctx = X509StoreContext(store, verified_cert, chain=chain)
+        assert store_ctx.verify_certificate() is None
+
+    def test_valid_untrusted_chain_reuse(self):
+        """
+        `verify_certificate` using an untrusted chain can be called multiple
+        times with the same ``X509StoreContext`` instance to produce the same
+        result.
+        """
+        store = X509Store()
+        store.add_cert(self.root_cert)
+        chain = [self.intermediate_cert]
+
+        store_ctx = X509StoreContext(
+            store, self.intermediate_server_cert, chain=chain
+        )
+        assert store_ctx.verify_certificate() is None
+        assert store_ctx.verify_certificate() is None
+
+    def test_chain_reference(self):
+        """
+        ``X509StoreContext`` properly keeps references to the untrusted chain
+        certificates.
+        """
+        store = X509Store()
+        store.add_cert(self.root_cert)
+        chain = [load_certificate(FILETYPE_PEM, intermediate_cert_pem)]
+
+        store_ctx = X509StoreContext(
+            store, self.intermediate_server_cert, chain=chain
+        )
+
+        del chain
+        assert store_ctx.verify_certificate() is None
+
+    @pytest.mark.parametrize(
+        "root_cert, chain, verified_cert",
+        [
+            pytest.param(
+                root_cert,
+                [],
+                intermediate_server_cert,
+                id="intermediate missing",
+            ),
+            pytest.param(
+                None,
+                [intermediate_cert],
+                intermediate_server_cert,
+                id="no trusted root",
+            ),
+            pytest.param(
+                None,
+                [root_cert, intermediate_cert],
+                intermediate_server_cert,
+                id="untrusted root, full chain is available",
+            ),
+            pytest.param(
+                intermediate_cert,
+                [root_cert, intermediate_cert],
+                intermediate_server_cert,
+                id="untrusted root, intermediate is trusted and in chain",
+            ),
+        ],
+    )
+    def test_verify_fail_with_chain(self, root_cert, chain, verified_cert):
+        store = X509Store()
+        if root_cert:
+            store.add_cert(root_cert)
+
+        store_ctx = X509StoreContext(store, verified_cert, chain=chain)
+
+        with pytest.raises(X509StoreContextError):
+            store_ctx.verify_certificate()
+
+    @pytest.mark.parametrize(
+        "chain, expected_error",
+        [
+            pytest.param(
+                [intermediate_cert, "This is not a certificate"],
+                TypeError,
+                id="non-certificate in chain",
+            ),
+            pytest.param(
+                42,
+                TypeError,
+                id="non-list chain",
+            ),
+        ],
+    )
+    def test_untrusted_chain_wrong_args(self, chain, expected_error):
+        """
+        Creating ``X509StoreContext`` with wrong chain raises an exception.
+        """
+        store = X509Store()
+        store.add_cert(self.root_cert)
+
+        with pytest.raises(expected_error):
+            X509StoreContext(store, self.intermediate_server_cert, chain=chain)
+
+    def test_failure_building_untrusted_chain_raises(self, monkeypatch):
+        """
+        Creating ``X509StoreContext`` raises ``OpenSSL.crypto.Error`` when
+        the underlying lib fails to add the certificate to the stack.
+        """
+        monkeypatch.setattr(_lib, "sk_X509_push", lambda _stack, _x509: -1)
+
+        store = X509Store()
+        store.add_cert(self.root_cert)
+        chain = [self.intermediate_cert]
+
+        with pytest.raises(Error):
+            X509StoreContext(store, self.intermediate_server_cert, chain=chain)
+
     def test_trusted_self_signed(self):
         """
         `verify_certificate` returns ``None`` when called with a self-signed