Allow using additional untrusted certificates for chain building in X509StoreContext (#948)

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.

Co-authored-by: Sandor Oroszi <[email protected]>

Author: orosam

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>`_

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.

tests/test_crypto.py

@@ -3824,6 +3824,145 @@ def test_reuse(self):
         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