WIP

Author: kislyuk

setup.py

@@ -19,6 +19,7 @@
         "cryptography >= 3.4.8",  # Set to the version in Ubuntu 22.04 due to features we need from cryptography 3.1
         "pyOpenSSL >= 17.5.0",
         "certifi >= 2018.1.18",
+        "tsp-client >= 0.1.3",
     ],
     extras_require={
         "tests": [

signxml/util/__init__.py

@@ -12,8 +12,11 @@
 from dataclasses import dataclass
 from typing import Any, List, Optional
 
+import certifi
 from cryptography.hazmat.primitives import hashes, hmac
 from lxml.etree import QName
+from OpenSSL.crypto import Error as OpenSSLCryptoError
+from OpenSSL.crypto import X509Store, X509StoreContext, X509StoreContextError
 
 from ..exceptions import InvalidCertificate, RedundantCert, SignXMLException
 
@@ -207,9 +210,6 @@ def p_sha1(client_b64_bytes, server_b64_bytes):
 
 
 def _add_cert_to_store(store, cert):
-    from OpenSSL.crypto import Error as OpenSSLCryptoError
-    from OpenSSL.crypto import X509StoreContext, X509StoreContextError
-
     try:
         X509StoreContext(store, cert).verify_certificate()
     except X509StoreContextError as e:
@@ -233,22 +233,21 @@ def verify_x509_cert_chain(cert_chain, ca_pem_file=None, ca_path=None):
     No ordering is implied by the above constraints"
     """
     # TODO: migrate to Cryptography (pending cert validation support) or https://github.com/wbond/certvalidator
-    from OpenSSL import SSL
-
-    context = SSL.Context(SSL.TLSv1_METHOD)
+    x509_store = X509Store()
     if ca_pem_file is None and ca_path is None:
-        import certifi
-
         ca_pem_file = certifi.where()
-    context.load_verify_locations(ensure_bytes(ca_pem_file, none_ok=True), capath=ca_path)
-    store = context.get_cert_store()
+    x509_store.load_locations(cafile=ca_pem_file, capath=ca_path)
+
+    # FIXME: use X509StoreContext(store=x509_store, certificate=cert, chain=cert_chain).get_verified_chain()
+    # This requires identifying the signing cert out-of-band
+
     certs = list(reversed(cert_chain))
     end_of_chain = None
     last_error: Exception = SignXMLException("Invalid certificate chain")
     while len(certs) > 0:
         for cert in certs:
             try:
-                end_of_chain = _add_cert_to_store(store, cert)
+                end_of_chain = _add_cert_to_store(x509_store, cert)
                 certs.remove(cert)
                 break
             except RedundantCert:

signxml/verifier.py

@@ -375,6 +375,7 @@ def verify(
                         raise InvalidInput(msg)
                 else:
                     cert_chain = [load_certificate(FILETYPE_PEM, add_pem_header(cert)) for cert in certs]
+                # FIXME: switch to wbondcrypto cert chain verify
                 signing_cert = verify_x509_cert_chain(cert_chain, ca_pem_file=ca_pem_file, ca_path=ca_path)
             elif isinstance(self.x509_cert, X509):
                 signing_cert = self.x509_cert
@@ -389,6 +390,8 @@ def verify(
 
             try:
                 digest_alg_name = str(digest_algorithm_implementations[signature_alg].name)
+                # FIXME: confirm the specified signature algorithm matches the certificate's public key
+                # FIXME: switch to cryptography verify
                 openssl_verify(signing_cert, raw_signature, signed_info_c14n, digest_alg_name)
             except OpenSSLCryptoError as e:
                 try:

signxml/xades/xades.py

@@ -29,6 +29,7 @@
 
 from lxml.etree import SubElement, _Element
 from OpenSSL.crypto import FILETYPE_ASN1, FILETYPE_PEM, X509, dump_certificate, load_certificate
+from tsp_client import TSPVerifier
 
 from .. import SignatureConfiguration, VerifyResult, XMLSignatureProcessor, XMLSigner, XMLVerifier
 from ..algorithms import DigestAlgorithm
@@ -264,9 +265,31 @@ class XAdESVerifier(XAdESProcessor, XMLVerifier):
     """
 
     # TODO: document/support SignatureTimeStamp / timestamp attestation
+    # TODO: allow setting required attributes, including timestamp
     # SignatureTimeStamp is required by certain profiles but is an unsigned property
-    def _verify_signing_time(self, verify_result: VerifyResult):
-        pass
+
+    def _verify_signing_time(self, verify_result: VerifyResult, all_verify_results: List[VerifyResult]):
+        """
+        The Implicit mechanism (see clause 5.1.4.4.1) shall be used for generating this qualifying property.
+        The input to the computation of the message imprint shall be the result of processing all the ds:Reference
+        elements within the ds:SignedInfo except the one referencing the SignedProperties element, in their order of
+        appearance, as follows:
+        1) process the retrieved ds:Reference element according to the reference-processing model of XMLDSIG [1]
+        clause 4.4.3.2;
+        2) if the result is a XML node set, canonicalize it as specified in clause 4.5; and
+        3) concatenate the resulting octets to those resulting from previously processed ds:Reference elements in
+        ds:SignedInfo.
+        """
+        ts_path = "xades:SignedDataObjectProperties/xades:AllDataObjectsTimeStamp/xades:EncapsulatedTimeStamp"
+        if verify_result.signed_xml is None:
+            return
+        all_data_objs_ts = verify_result.signed_xml.find(ts_path, namespaces=namespaces)
+        if all_data_objs_ts is None:
+            return
+        print("Will verify", all_data_objs_ts.text)
+        ts = b64decode(all_data_objs_ts.text)  # type: ignore
+        tsp_message = b"".join(r.signed_data for r in all_verify_results if r != verify_result)
+        TSPVerifier().verify(ts, message=tsp_message)
 
     def _verify_cert_digest(self, signing_cert_node, expect_cert):
         for cert in self._findall(signing_cert_node, "xades:Cert"):
@@ -320,8 +343,8 @@ def _verify_signature_policy(self, verify_result: VerifyResult, expect_signature
             if b64decode(digest_value.text) != b64decode(expect_signature_policy.DigestValue):
                 raise InvalidInput("Digest mismatch for signature policy hash")
 
-    def _verify_signed_properties(self, verify_result):
-        self._verify_signing_time(verify_result)
+    def _verify_signed_properties(self, verify_result, *, all_verify_results):
+        self._verify_signing_time(verify_result, all_verify_results=all_verify_results)
         self._verify_cert_digests(verify_result)
         if self.expect_signature_policy:
             self._verify_signature_policy(
@@ -364,7 +387,8 @@ def verify(  # type: ignore
                 continue
             if verify_result.signed_xml.tag == xades_tag("SignedProperties"):
                 verify_results[i] = XAdESVerifyResult(  # type: ignore
-                    *astuple(verify_result), signed_properties=self._verify_signed_properties(verify_result)
+                    *astuple(verify_result),
+                    signed_properties=self._verify_signed_properties(verify_result, all_verify_results=verify_results),
                 )
                 break
         else: