Use enum for c14n methods

Also continue type hints and add a test case for inclusivenamespaces prefixlist

Author: kislyuk

signxml/__init__.py

@@ -1,7 +1,7 @@
 # isort: skip_file
 from .signer import XMLSigner  # noqa:F401
 from .verifier import XMLVerifier, VerifyResult  # noqa:F401
-from .algorithms import DigestAlgorithm, SignatureMethod, SignatureType  # noqa:F401
+from .algorithms import DigestAlgorithm, SignatureMethod, CanonicalizationMethod, SignatureType  # noqa:F401
 from .exceptions import InvalidCertificate, InvalidDigest, InvalidInput, InvalidSignature  # noqa:F401
 from .processor import XMLSignatureProcessor  # noqa:F401
 from .util import SigningSettings, namespaces  # noqa:F401

signxml/algorithms.py

@@ -99,6 +99,16 @@ class SignatureMethod(FragmentLookupMixin, InvalidInputErrorMixin, Enum):
     EDDSA_ED448 = "http://www.w3.org/2021/04/xmldsig-more#eddsa-ed448"
 
 
+class CanonicalizationMethod(InvalidInputErrorMixin, Enum):
+    CANONICAL_XML_1_0 = "http://www.w3.org/TR/2001/REC-xml-c14n-20010315"
+    CANONICAL_XML_1_0_WITH_COMMENTS = "http://www.w3.org/TR/2001/REC-xml-c14n-20010315#WithComments"
+    CANONICAL_XML_1_1 = "http://www.w3.org/2006/12/xmlc14n11#"
+    CANONICAL_XML_1_1_DEPRECATED_URI = "http://www.w3.org/2006/12/xml-c14n11"
+    CANONICAL_XML_1_1_WITH_COMMENTS = "http://www.w3.org/2006/12/xmlc14n11#WithComments"
+    EXCLUSIVE_XML_CANONICALIZATION_1_0 = "http://www.w3.org/2001/10/xml-exc-c14n#"
+    EXCLUSIVE_XML_CANONICALIZATION_1_0_WITH_COMMENTS = "http://www.w3.org/2001/10/xml-exc-c14n#WithComments"
+
+
 digest_algorithm_implementations = {
     DigestAlgorithm.SHA1: hashes.SHA1,
     DigestAlgorithm.SHA224: hashes.SHA224,

signxml/processor.py

@@ -6,7 +6,7 @@
 from cryptography.hazmat.primitives.hashes import Hash
 from lxml import etree
 
-from .algorithms import DigestAlgorithm, digest_algorithm_implementations
+from .algorithms import CanonicalizationMethod, DigestAlgorithm, digest_algorithm_implementations
 from .exceptions import InvalidInput
 from .util import namespaces
 
@@ -75,15 +75,6 @@ class XMLSignatureProcessor(XMLProcessor):
     }
     known_ecdsa_curve_oids = {ec().name: oid for oid, ec in known_ecdsa_curves.items()}  # type: ignore
 
-    known_c14n_algorithms = {
-        "http://www.w3.org/TR/2001/REC-xml-c14n-20010315",
-        "http://www.w3.org/TR/2001/REC-xml-c14n-20010315#WithComments",
-        "http://www.w3.org/2001/10/xml-exc-c14n#",
-        "http://www.w3.org/2001/10/xml-exc-c14n#WithComments",
-        "http://www.w3.org/2006/12/xml-c14n11",
-        "http://www.w3.org/2006/12/xml-c14n11#WithComments",
-    }
-    default_c14n_algorithm = "http://www.w3.org/2006/12/xml-c14n11"
     excise_empty_xmlns_declarations = False
 
     id_attributes: Tuple[str, ...] = ("Id", "ID", "id", "xml:id")
@@ -104,7 +95,7 @@ def _find(self, element, query, require=True, anywhere=False):
             result = element.find(namespace + ":" + query, namespaces=namespaces)
 
         if require and result is None:
-            raise InvalidInput("Expected to find XML element {} in {}".format(query, element.tag))
+            raise InvalidInput(f"Expected to find XML element {query} in {element.tag}")
         return result
 
     def _findall(self, element, query, anywhere=False):
@@ -116,12 +107,12 @@ def _findall(self, element, query, anywhere=False):
         else:
             return element.findall(namespace + ":" + query, namespaces=namespaces)
 
-    def _c14n(self, nodes, algorithm, inclusive_ns_prefixes=None):
+    def _c14n(self, nodes, algorithm: CanonicalizationMethod, inclusive_ns_prefixes=None):
         exclusive, with_comments = False, False
 
-        if algorithm.startswith("http://www.w3.org/2001/10/xml-exc-c14n#"):
+        if algorithm.value.startswith("http://www.w3.org/2001/10/xml-exc-c14n#"):
             exclusive = True
-        if algorithm.endswith("#WithComments"):
+        if algorithm.value.endswith("#WithComments"):
             with_comments = True
 
         if not isinstance(nodes, list):
@@ -152,17 +143,17 @@ def _resolve_reference(self, doc_root, reference, uri_resolver=None):
             # doc_root.xpath(uri.lstrip("#"))[0]
         elif uri.startswith("#"):
             for id_attribute in self.id_attributes:
-                xpath_query = "//*[@*[local-name() = '{}']=$uri]".format(id_attribute)
+                xpath_query = f"//*[@*[local-name() = '{id_attribute}']=$uri]"
                 results = doc_root.xpath(xpath_query, uri=uri.lstrip("#"))
                 if len(results) > 1:
-                    raise InvalidInput("Ambiguous reference URI {} resolved to {} nodes".format(uri, len(results)))
+                    raise InvalidInput(f"Ambiguous reference URI {uri} resolved to {len(results)} nodes")
                 elif len(results) == 1:
                     return results[0]
-            raise InvalidInput("Unable to resolve reference URI: {}".format(uri))
+            raise InvalidInput(f"Unable to resolve reference URI: {uri}")
         else:
             if uri_resolver is None:
-                raise InvalidInput("External URI dereferencing is not configured: {}".format(uri))
+                raise InvalidInput(f"External URI dereferencing is not configured: {uri}")
             result = uri_resolver(uri)
             if result is None:
-                raise InvalidInput("Unable to resolve reference URI: {}".format(uri))
+                raise InvalidInput(f"Unable to resolve reference URI: {uri}")
             return result

signxml/signer.py

@@ -1,14 +1,20 @@
 from base64 import b64encode
-from typing import Union
+from typing import List, Optional, Union
 
 from cryptography.hazmat.primitives.asymmetric import ec, utils
 from cryptography.hazmat.primitives.asymmetric.padding import PKCS1v15
 from cryptography.hazmat.primitives.hmac import HMAC
 from cryptography.hazmat.primitives.serialization import load_pem_private_key
-from lxml.etree import Element, SubElement
+from lxml.etree import Element, SubElement, _Element
 from OpenSSL.crypto import FILETYPE_PEM, dump_certificate
 
-from .algorithms import DigestAlgorithm, SignatureMethod, SignatureType, digest_algorithm_implementations
+from .algorithms import (
+    CanonicalizationMethod,
+    DigestAlgorithm,
+    SignatureMethod,
+    SignatureType,
+    digest_algorithm_implementations,
+)
 from .exceptions import InvalidInput
 from .processor import XMLSignatureProcessor
 from .util import (
@@ -50,10 +56,10 @@ def __init__(
         method: SignatureType = SignatureType.enveloped,
         signature_algorithm: Union[SignatureMethod, str] = SignatureMethod.RSA_SHA256,
         digest_algorithm: Union[DigestAlgorithm, str] = DigestAlgorithm.SHA256,
-        c14n_algorithm=XMLSignatureProcessor.default_c14n_algorithm,
+        c14n_algorithm=CanonicalizationMethod.CANONICAL_XML_1_1,
     ):
         if method is None or method not in SignatureType:
-            raise InvalidInput("Unknown signature method {}".format(method))
+            raise InvalidInput(f"Unknown signature method {method}")
         self.signature_type = method
         if isinstance(signature_algorithm, str) and "#" not in signature_algorithm:
             self.sign_alg = SignatureMethod.from_fragment(signature_algorithm)
@@ -63,8 +69,7 @@ def __init__(
             self.digest_alg = DigestAlgorithm.from_fragment(digest_algorithm)
         else:
             self.digest_alg = DigestAlgorithm(digest_algorithm)
-        assert c14n_algorithm in self.known_c14n_algorithms
-        self.c14n_alg = c14n_algorithm
+        self.c14n_alg = CanonicalizationMethod(c14n_algorithm)
         self.namespaces = dict(ds=namespaces.ds)
         self._parser = None
         self.signature_annotators = [self._add_key_info]
@@ -73,15 +78,15 @@ def sign(
         self,
         data,
         key=None,
-        passphrase=None,
+        passphrase: Optional[bytes] = None,
         cert=None,
-        reference_uri=None,
-        key_name=None,
-        key_info=None,
-        id_attribute=None,
-        always_add_key_value=False,
-        payload_inclusive_ns_prefixes=frozenset(),
-        signature_inclusive_ns_prefixes=frozenset(),
+        reference_uri: Optional[Union[str, List[str]]] = None,
+        key_name: Optional[str] = None,
+        key_info: Optional[_Element] = None,
+        id_attribute: Optional[str] = None,
+        always_add_key_value: bool = False,
+        payload_inclusive_ns_prefixes: Optional[List[str]] = None,
+        signature_inclusive_ns_prefixes: Optional[List[str]] = None,
         signature_properties=None,
     ):
         """
@@ -100,7 +105,6 @@ def sign(
             :py:class:`cryptography.hazmat.primitives.interfaces.DSAPrivateKey`, or
             :py:class:`cryptography.hazmat.primitives.interfaces.EllipticCurvePrivateKey` object
         :param passphrase: Passphrase to use to decrypt the key, if any.
-        :type passphrase: string
         :param cert:
             X.509 certificate to use for signing. This should be a string containing a PEM-formatted certificate, or an
             array of strings or OpenSSL.crypto.X509 objects containing the certificate and a chain of intermediate
@@ -110,30 +114,24 @@ def sign(
             Custom reference URI or list of reference URIs to incorporate into the signature. When ``method`` is set to
             ``detached`` or ``enveloped``, reference URIs are set to this value and only the referenced elements are
             signed.
-        :type reference_uri: string or list
         :param key_name: Add a KeyName element in the KeyInfo element that may be used by the signer to communicate a
             key identifier to the recipient. Typically, KeyName contains an identifier related to the key pair used to
             sign the message.
-        :type key_name: string
         :param key_info:
             A custom KeyInfo element to insert in the signature. Use this to supply ``<wsse:SecurityTokenReference>``
             or other custom key references. An example value can be found here:
             https://github.com/XML-Security/signxml/blob/master/test/wsse_keyinfo.xml
-        :type key_info: :py:class:`lxml.etree.Element`
         :param id_attribute:
             Name of the attribute whose value ``URI`` refers to. By default, SignXML will search for "Id", then "ID".
-        :type id_attribute: string
         :param always_add_key_value:
             Write the key value to the KeyInfo element even if a X509 certificate is present. Use of this parameter
             is discouraged, as it introduces an ambiguity and a security hazard. The public key used to sign the
             document is already encoded in the certificate (which is in X509Data), so the verifier must either ignore
             KeyValue or make sure it matches what's in the certificate. This parameter is provided for compatibility
             purposes only.
-        :type always_add_key_value: boolean
         :param payload_inclusive_ns_prefixes:
             Provide a list of XML namespace prefixes whose declarations should be preserved when canonicalizing the
             content referenced by the signature (**InclusiveNamespaces PrefixList**).
-        :type inclusive_ns_prefixes: list
         :param signature_inclusive_ns_prefixes:
             Provide a list of XML namespace prefixes whose declarations should be preserved when canonicalizing the
             signature itself (**InclusiveNamespaces PrefixList**).
@@ -161,9 +159,9 @@ def sign(
             cert_chain = cert
 
         if isinstance(reference_uri, (str, bytes)):
-            reference_uris = [reference_uri]
+            input_reference_uris = [reference_uri]
         else:
-            reference_uris = reference_uri
+            input_reference_uris = reference_uri  # type: ignore
 
         signing_settings = SigningSettings(
             key=None,
@@ -181,7 +179,7 @@ def sign(
             else:
                 signing_settings.key = key
 
-        sig_root, doc_root, c14n_inputs, reference_uris = self._unpack(data, reference_uris)
+        sig_root, doc_root, c14n_inputs, reference_uris = self._unpack(data, input_reference_uris)
 
         if self.signature_type == SignatureType.detached and signature_properties is not None:
             reference_uris.append("#prop")
@@ -324,7 +322,7 @@ def _unpack(self, data, reference_uris):
 
     def _build_sig(self, sig_root, reference_uris, c14n_inputs, sig_insp, payload_insp):
         signed_info = SubElement(sig_root, ds_tag("SignedInfo"), nsmap=self.namespaces)
-        sig_c14n_method = SubElement(signed_info, ds_tag("CanonicalizationMethod"), Algorithm=self.c14n_alg)
+        sig_c14n_method = SubElement(signed_info, ds_tag("CanonicalizationMethod"), Algorithm=self.c14n_alg.value)
         if sig_insp:
             SubElement(sig_c14n_method, ec_tag("InclusiveNamespaces"), PrefixList=" ".join(sig_insp))
 
@@ -334,9 +332,9 @@ def _build_sig(self, sig_root, reference_uris, c14n_inputs, sig_insp, payload_in
             transforms = SubElement(reference, ds_tag("Transforms"))
             if self.signature_type == SignatureType.enveloped:
                 SubElement(transforms, ds_tag("Transform"), Algorithm=namespaces.ds + "enveloped-signature")
-                SubElement(transforms, ds_tag("Transform"), Algorithm=self.c14n_alg)
+                SubElement(transforms, ds_tag("Transform"), Algorithm=self.c14n_alg.value)
             else:
-                c14n_xform = SubElement(transforms, ds_tag("Transform"), Algorithm=self.c14n_alg)
+                c14n_xform = SubElement(transforms, ds_tag("Transform"), Algorithm=self.c14n_alg.value)
                 if payload_insp:
                     SubElement(c14n_xform, ec_tag("InclusiveNamespaces"), PrefixList=" ".join(payload_insp))
 
@@ -356,8 +354,8 @@ def _build_signature_properties(self, signature_properties):
             signature_property = Element(
                 ds_tag("SignatureProperty"),
                 attrib={
-                    "Id": el.attrib.pop("Id", "sigprop{}".format(i)),
-                    "Target": el.attrib.pop("Target", "#sigproptarget{}".format(i)),
+                    "Id": el.attrib.pop("Id", f"sigprop{i}"),
+                    "Target": el.attrib.pop("Target", f"#sigproptarget{i}"),
                 },
             )
             signature_property.append(el)

signxml/util/__init__.py

@@ -61,7 +61,7 @@ def xades141_tag(tag):
 @dataclass
 class SigningSettings:
     key: Any
-    key_name: str
+    key_name: Optional[str]
     key_info: Any
     always_add_key_value: bool
     cert_chain: Optional[List]

signxml/verifier.py

@@ -12,8 +12,8 @@
 from OpenSSL.crypto import load_certificate
 from OpenSSL.crypto import verify as openssl_verify
 
-from .algorithms import DigestAlgorithm, SignatureMethod, digest_algorithm_implementations
-from .exceptions import InvalidCertificate, InvalidDigest, InvalidInput, InvalidSignature  # noqa
+from .algorithms import CanonicalizationMethod, DigestAlgorithm, SignatureMethod, digest_algorithm_implementations
+from .exceptions import InvalidCertificate, InvalidDigest, InvalidInput, InvalidSignature
 from .processor import XMLSignatureProcessor
 from .util import (
     _remove_sig,
@@ -59,10 +59,15 @@ def _get_signature(self, root):
             return self._find(root, "Signature", anywhere=True)
 
     def _verify_signature_with_pubkey(
-        self, signed_info_c14n, raw_signature, key_value, der_encoded_key_value, signature_alg
-    ):
+        self,
+        signed_info_c14n: bytes,
+        raw_signature: bytes,
+        key_value: etree._Element,
+        der_encoded_key_value: Optional[etree._Element],
+        signature_alg: SignatureMethod,
+    ) -> None:
         if der_encoded_key_value is not None:
-            key = load_der_public_key(b64decode(der_encoded_key_value.text))
+            key = load_der_public_key(b64decode(der_encoded_key_value.text))  # type: ignore
 
         digest_algorithm_implementation = digest_algorithm_implementations[signature_alg]()
         if signature_alg.name.startswith("ECDSA_"):
@@ -94,7 +99,7 @@ def _verify_signature_with_pubkey(
             elif not isinstance(key, dsa.DSAPublicKey):
                 raise InvalidInput("DER encoded key value does not match specified signature algorithm")
             # TODO: supply meaningful key_size_bits for signature length assertion
-            dss_signature = self._encode_dss_signature(raw_signature, len(raw_signature) * 8 / 2)
+            dss_signature = self._encode_dss_signature(raw_signature, len(raw_signature) * 8 // 2)
             key.verify(dss_signature, data=signed_info_c14n, algorithm=digest_algorithm_implementation)
         elif signature_alg.name.startswith("RSA_"):
             if key_value is not None:
@@ -113,7 +118,7 @@ def _verify_signature_with_pubkey(
         else:
             raise NotImplementedError()
 
-    def _encode_dss_signature(self, raw_signature, key_size_bits):
+    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
         if len(raw_signature) != want_raw_signature_len:
             raise InvalidSignature(
@@ -131,7 +136,7 @@ def _get_inclusive_ns_prefixes(self, transform_node):
         else:
             return inclusive_namespaces.get("PrefixList").split(" ")
 
-    def _apply_transforms(self, payload, transforms_node, signature, c14n_algorithm):
+    def _apply_transforms(self, payload, transforms_node, signature, c14n_algorithm: CanonicalizationMethod):
         transforms, c14n_applied = [], False
         if transforms_node is not None:
             transforms = self._findall(transforms_node, "Transform")
@@ -146,10 +151,15 @@ def _apply_transforms(self, payload, transforms_node, signature, c14n_algorithm)
 
         for transform in transforms:
             algorithm = transform.get("Algorithm")
-            if algorithm in self.known_c14n_algorithms:
-                inclusive_ns_prefixes = self._get_inclusive_ns_prefixes(transform)
-                payload = self._c14n(payload, algorithm=algorithm, inclusive_ns_prefixes=inclusive_ns_prefixes)
-                c14n_applied = True
+            try:
+                c14n_algorithm_from_transform = CanonicalizationMethod(algorithm)
+            except ValueError:
+                continue
+            inclusive_ns_prefixes = self._get_inclusive_ns_prefixes(transform)
+            payload = self._c14n(
+                payload, algorithm=c14n_algorithm_from_transform, inclusive_ns_prefixes=inclusive_ns_prefixes
+            )
+            c14n_applied = True
 
         if not c14n_applied and not isinstance(payload, (str, bytes)):
             payload = self._c14n(payload, algorithm=c14n_algorithm)
@@ -172,7 +182,7 @@ def verify(
         id_attribute: Optional[str] = None,
         expect_references: Union[int, bool] = 1,
         ignore_ambiguous_key_info: bool = False,
-    ) -> List[VerifyResult]:
+    ) -> Union[VerifyResult, List[VerifyResult]]:
         """
         Verify the XML signature supplied in the data and return a list of **VerifyResult** data structures
         representing the data signed by the signature, or raise an exception if the signature is not valid. By default,
@@ -274,7 +284,7 @@ def verify(
 
         signed_info = self._find(signature, "SignedInfo")
         c14n_method = self._find(signed_info, "CanonicalizationMethod")
-        c14n_algorithm = c14n_method.get("Algorithm")
+        c14n_algorithm = CanonicalizationMethod(c14n_method.get("Algorithm"))
         inclusive_ns_prefixes = self._get_inclusive_ns_prefixes(c14n_method)
         signature_method = self._find(signed_info, "SignatureMethod")
         signature_value = self._find(signature, "SignatureValue")
@@ -331,7 +341,7 @@ def verify(
                     lib, func, reason = e.args[0][0]
                 except Exception:
                     reason = e
-                raise InvalidSignature("Signature verification failed: {}".format(reason))
+                raise InvalidSignature(f"Signature verification failed: {reason}")
 
             # If both X509Data and KeyValue are present, match one against the other and raise an error on mismatch
             if key_value is not None:
@@ -408,7 +418,7 @@ def verify(
             msg = "Expected to find {} references, but found {}"
             raise InvalidSignature(msg.format(expect_references, len(verify_results)))
 
-        return verify_results
+        return verify_results if expect_references > 1 else verify_results[0]
 
     def validate_schema(self, signature):
         last_exception = None