Merge commit from fork

* Ensure HMAC is used when hmac_key is set

* Fix types

* Clarify comment

Author: kislyuk

signxml/verifier.py

@@ -1,14 +1,15 @@
 from base64 import b64decode
 from dataclasses import dataclass, replace
-from typing import Callable, FrozenSet, List, Optional, Union
+from typing import Callable, FrozenSet, List, Optional, Union, Tuple
 from warnings import warn
 
 import cryptography.exceptions
 from cryptography import x509
 from cryptography.hazmat.primitives.asymmetric import dsa, ec, rsa, utils
 from cryptography.hazmat.primitives.asymmetric.padding import MGF1, PSS, AsymmetricPadding, PKCS1v15
 from cryptography.hazmat.primitives.hmac import HMAC
-from cryptography.hazmat.primitives.serialization import load_der_public_key
+from cryptography.hazmat.primitives.serialization import load_der_public_key, Encoding, PublicFormat
+
 from lxml import etree
 
 from .algorithms import (
@@ -27,7 +28,6 @@
     bits_to_bytes_unit,
     bytes_to_long,
     ds_tag,
-    ensure_bytes,
     namespaces,
 )
 
@@ -42,6 +42,11 @@ class SignatureConfiguration:
     """
     If ``True``, a valid X.509 certificate-based signature with an established chain of trust is required to
     pass validation. If ``False``, other types of valid signatures (e.g. HMAC or RSA public key) are accepted.
+
+    Ensure you establish trust: to specify an HMAC shared secret that will be required to be used by the signature, use
+    the ``hmac_key`` parameter. If ``require_x509`` is ``False`` and no ``hmac_key`` is set, any valid XML signature
+    satisfying the other SignatureConfiguration parameters will be accepted, regardless of the key used. You must then
+    establish trust by examining the ``VerifyResult`` contents, in particular asserting on the value of ``signature_key``.
     """
 
     location: str = ".//"
@@ -62,13 +67,15 @@ class SignatureConfiguration:
     signature_methods: FrozenSet[SignatureMethod] = frozenset(sm for sm in SignatureMethod if "SHA1" not in sm.name)
     """
     Set of acceptable signature methods (signature algorithms). Any signature generated using an algorithm not listed
-    here will fail verification.
+    here will fail verification. It is recommended that you restrict this set to only methods expected to be used in
+    signatures handled by your application.
     """
 
     digest_algorithms: FrozenSet[DigestAlgorithm] = frozenset(da for da in DigestAlgorithm if "SHA1" not in da.name)
     """
     Set of acceptable digest algorithms. Any signature or reference transform generated using an algorithm not listed
-    here will cause verification to fail.
+    here will cause verification to fail. It is recommended that you restrict this set to only methods expected to be
+    used in signatures handled by your application.
     """
 
     ignore_ambiguous_key_info: bool = False
@@ -86,7 +93,8 @@ class SignatureConfiguration:
 class VerifyResult:
     """
     This is a dataclass representing structured data returned by :func:`signxml.XMLVerifier.verify`. The results of a
-    verification contain the signed bytes, the parsed signed XML, and the parsed signature XML. Example usage:
+    verification contain the signed bytes, the parsed signed XML, the parsed signature XML, and the key that was used
+    to verify the signature. Example usage:
 
         verified_data = signxml.XMLVerifier().verify(input_data).signed_xml
     """
@@ -100,6 +108,9 @@ class VerifyResult:
     signature_xml: etree._Element
     "The signature element parsed as XML"
 
+    signature_key: bytes
+    "The cryptographic key that was used to verify the signature, in PEM format (for asymmetric keys) or raw secret bytes (for HMAC keys)"
+
 
 class XMLVerifier(XMLSignatureProcessor):
     """
@@ -123,7 +134,7 @@ def _verify_signature_with_pubkey(
         key_value: Optional[etree._Element] = None,
         der_encoded_key_value: Optional[etree._Element] = None,
         signing_certificate: Optional[x509.Certificate] = None,
-    ) -> bytes:
+    ) -> Tuple[bytes, bytes | dsa.DSAPublicKey | rsa.RSAPublicKey | ec.EllipticCurvePublicKey]:
         if der_encoded_key_value is not None:
             assert der_encoded_key_value.text is not None
             key = load_der_public_key(b64decode(der_encoded_key_value.text))
@@ -177,7 +188,7 @@ def _verify_signature_with_pubkey(
             key.verify(raw_signature, data=signed_info_c14n, padding=padding, algorithm=digest_alg_impl)
         else:
             raise InvalidInput(f"Unsupported signature algorithm {signature_alg}")
-        return signed_info_c14n
+        return signed_info_c14n, key
 
     def _encode_dss_signature(self, raw_signature: bytes, key_size_bits: int) -> bytes:
         want_raw_signature_len = bits_to_bytes_unit(key_size_bits) * 2
@@ -279,7 +290,7 @@ def verify(
         cert_subject_name: Optional[str] = None,
         cert_resolver: Optional[Callable] = None,
         ca_pem_file: Optional[Union[str, bytes]] = None,
-        hmac_key: Optional[str] = None,
+        hmac_key: Optional[bytes] = None,
         validate_schema: bool = True,
         parser=None,
         uri_resolver: Optional[Callable] = None,
@@ -366,6 +377,15 @@ def verify(
         :raises: :class:`signxml.exceptions.InvalidSignature`
         """
         self.hmac_key = hmac_key
+        if hmac_key is not None:
+            if expect_config.signature_methods == SignatureConfiguration().signature_methods:  # default value
+                expect_config = replace(
+                    expect_config,
+                    signature_methods=frozenset(sm for sm in SignatureMethod if sm.name.startswith("HMAC")),
+                )
+            elif any(not sm.name.startswith("HMAC") for sm in expect_config.signature_methods):
+                raise InvalidInput("When hmac_key is set, all expected signature methods must use HMAC")
+
         self.config = expect_config
         if deprecated_kwargs:
             self.config = replace(expect_config, **deprecated_kwargs)
@@ -452,7 +472,7 @@ def verify(
                     raise InvalidSignature("Certificate subject common name mismatch")
 
             try:
-                verified_signed_info_c14n = self._verify_signature_with_pubkey(
+                verified_signed_info_c14n, key_used = self._verify_signature_with_pubkey(
                     signed_info_c14n=signed_info_c14n,
                     raw_signature=raw_signature,
                     signing_certificate=signing_cert,
@@ -471,18 +491,19 @@ def verify(
             if self.hmac_key is None:
                 raise InvalidInput('Parameter "hmac_key" is required when verifying a HMAC signature')
 
-            signer = HMAC(key=ensure_bytes(self.hmac_key), algorithm=digest_algorithm_implementations[signature_alg]())
+            signer = HMAC(key=self.hmac_key, algorithm=digest_algorithm_implementations[signature_alg]())
             signer.update(signed_info_c14n)
             try:
                 signer.verify(raw_signature)
                 verified_signed_info_c14n = signed_info_c14n
             except cryptography.exceptions.InvalidSignature:
                 raise InvalidSignature("Signature mismatch (HMAC)")
+            key_used = self.hmac_key
         else:
             if key_value is None and der_encoded_key_value is None:
                 raise InvalidInput("Expected to find either KeyValue or X509Data XML element in KeyInfo")
 
-            verified_signed_info_c14n = self._verify_signature_with_pubkey(
+            verified_signed_info_c14n, key_used = self._verify_signature_with_pubkey(
                 signed_info_c14n=signed_info_c14n,
                 raw_signature=raw_signature,
                 key_value=key_value,
@@ -493,15 +514,17 @@ def verify(
         verified_signed_info = self._fromstring(verified_signed_info_c14n)
         verify_results: List[VerifyResult] = []
         for idx, reference in enumerate(self._findall(verified_signed_info, "Reference")):
-            verify_results.append(self._verify_reference(reference, idx, root, uri_resolver, c14n_algorithm, signature))
+            verify_results.append(
+                self._verify_reference(reference, idx, root, uri_resolver, c14n_algorithm, signature, key_used)
+            )
 
         if type(self.config.expect_references) is int and len(verify_results) != self.config.expect_references:
             msg = "Expected to find {} references, but found {}"
             raise InvalidSignature(msg.format(self.config.expect_references, len(verify_results)))
 
         return verify_results if self.config.expect_references > 1 else verify_results[0]
 
-    def _verify_reference(self, reference, index, root, uri_resolver, c14n_algorithm, signature):
+    def _verify_reference(self, reference, index, root, uri_resolver, c14n_algorithm, signature, signature_key_used):
         copied_root = self._fromstring(self._tostring(root))
         copied_signature_ref = self._get_signature(copied_root)
         transforms = self._find(reference, "Transforms", require=False)
@@ -522,7 +545,12 @@ def _verify_reference(self, reference, index, root, uri_resolver, c14n_algorithm
             payload_c14n_xml = self._fromstring(payload_c14n)
         except etree.XMLSyntaxError:
             payload_c14n_xml = None
-        return VerifyResult(payload_c14n, payload_c14n_xml, signature)
+
+        if isinstance(signature_key_used, bytes):
+            signature_key = signature_key_used
+        else:
+            signature_key = signature_key_used.public_bytes(Encoding.PEM, PublicFormat.SubjectPublicKeyInfo)
+        return VerifyResult(payload_c14n, payload_c14n_xml, signature, signature_key=signature_key)
 
     def validate_schema(self, signature):
         last_exception = None

test/test.py

@@ -53,6 +53,9 @@ def check_deprecated_methods(self):
 
 
 sha1_ok = SignatureConfiguration(signature_methods=list(SignatureMethod), digest_algorithms=list(DigestAlgorithm))
+hmac_only = SignatureConfiguration(
+    signature_methods=[sm for sm in SignatureMethod if "HMAC" in sm.name], digest_algorithms=list(DigestAlgorithm)
+)
 xades_sha1_ok = XAdESSignatureConfiguration(
     signature_methods=list(SignatureMethod), digest_algorithms=list(DigestAlgorithm)
 )
@@ -162,7 +165,10 @@ def test_case(case):
                 )
                 # print(etree.tostring(signed))
                 hmac_key = self.keys["hmac"] if sig_alg_type == "hmac" else None
-                verify_kwargs = dict(require_x509=False, hmac_key=hmac_key, validate_schema=True, expect_config=sha1_ok)
+                expect_config = hmac_only if sig_alg_type == "hmac" else sha1_ok
+                verify_kwargs = dict(
+                    require_x509=False, hmac_key=hmac_key, validate_schema=True, expect_config=expect_config
+                )
 
                 if method == methods.detached:
 
@@ -199,7 +205,7 @@ def resolver(uri):
                         signed_data,
                         hmac_key=hmac_key,
                         uri_resolver=verify_kwargs.get("uri_resolver"),
-                        expect_config=sha1_ok,
+                        expect_config=expect_config,
                     )
 
                 if method != methods.detached:
@@ -287,10 +293,10 @@ def get_x509_cert(**kwargs):
                     XMLVerifier().verify(
                         sig,
                         require_x509=False,
-                        hmac_key="testkey",
+                        hmac_key=b"testkey" if "hmac" in signature_file else None,
                         validate_schema=True,
                         cert_resolver=get_x509_cert if "x509digest" in signature_file else None,
-                        expect_config=sha1_ok,
+                        expect_config=hmac_only if "hmac" in signature_file else sha1_ok,
                     )
                     sig.decode("utf-8")
                 except Exception as e:
@@ -346,6 +352,11 @@ def cert_resolver(x509_issuer_name, x509_serial_number, x509_digest):
         signature_files += glob(os.path.join(interop_dir, "pyXMLSecurity", "*.xml"))
         for signature_file in signature_files:
             print("Verifying", signature_file)
+            hmac_key = None
+            expect_config = sha1_ok
+            if "hmac" in signature_file:
+                hmac_key = b"test" if "phaos" in signature_file else b"secret"
+                expect_config = hmac_only
             with open(signature_file, "rb") as fh:
                 try:
                     sig = fh.read()
@@ -354,13 +365,13 @@ def cert_resolver(x509_issuer_name, x509_serial_number, x509_digest):
                     verifier.verify(
                         sig,
                         require_x509=False,
-                        hmac_key="test" if "phaos" in signature_file else "secret",
+                        hmac_key=hmac_key,
                         validate_schema=True,
                         uri_resolver=resolver,
                         x509_cert=get_x509_cert(signature_file),
                         cert_resolver=cert_resolver if "issuer-serial" in signature_file else None,
                         ca_pem_file=get_ca_pem_file(signature_file),
-                        expect_config=sha1_ok,
+                        expect_config=expect_config,
                     )
                     decoded_sig = sig.decode("utf-8")
                     if "HMACOutputLength" in decoded_sig or "bad" in signature_file or "expired" in signature_file: