Allow customizing the digest and signature method during wsse signing. Fixes #616

Author: apollo13

src/zeep/wsse/signature.py

@@ -45,16 +45,27 @@ def _make_verify_key(cert_data):
 class MemorySignature(object):
     """Sign given SOAP envelope with WSSE sig using given key and cert."""
 
-    def __init__(self, key_data, cert_data, password=None):
+    def __init__(
+        self,
+        key_data,
+        cert_data,
+        password=None,
+        signature_method=None,
+        digest_method=None,
+    ):
         check_xmlsec_import()
 
         self.key_data = key_data
         self.cert_data = cert_data
         self.password = password
+        self.digest_method = digest_method
+        self.signature_method = signature_method
 
     def apply(self, envelope, headers):
         key = _make_sign_key(self.key_data, self.cert_data, self.password)
-        _sign_envelope_with_key(envelope, key)
+        _sign_envelope_with_key(
+            envelope, key, self.signature_method, self.digest_method
+        )
         return envelope, headers
 
     def verify(self, envelope):
@@ -66,9 +77,20 @@ def verify(self, envelope):
 class Signature(MemorySignature):
     """Sign given SOAP envelope with WSSE sig using given key file and cert file."""
 
-    def __init__(self, key_file, certfile, password=None):
+    def __init__(
+        self,
+        key_file,
+        certfile,
+        password=None,
+        signature_method=None,
+        digest_method=None,
+    ):
         super(Signature, self).__init__(
-            _read_file(key_file), _read_file(certfile), password
+            _read_file(key_file),
+            _read_file(certfile),
+            password,
+            signature_method,
+            digest_method,
         )
 
 
@@ -79,7 +101,9 @@ class BinarySignature(Signature):
 
     def apply(self, envelope, headers):
         key = _make_sign_key(self.key_data, self.cert_data, self.password)
-        _sign_envelope_with_key_binary(envelope, key)
+        _sign_envelope_with_key_binary(
+            envelope, key, self.signature_method, self.digest_method
+        )
         return envelope, headers
 
 
@@ -92,7 +116,14 @@ def check_xmlsec_import():
         )
 
 
-def sign_envelope(envelope, keyfile, certfile, password=None):
+def sign_envelope(
+    envelope,
+    keyfile,
+    certfile,
+    password=None,
+    signature_method=None,
+    digest_method=None,
+):
     """Sign given SOAP envelope with WSSE sig using given key and cert.
 
     Sign the wsu:Timestamp node in the wsse:Security header and the soap:Body;
@@ -182,16 +213,18 @@ def sign_envelope(envelope, keyfile, certfile, password=None):
     """
     # Load the signing key and certificate.
     key = _make_sign_key(_read_file(keyfile), _read_file(certfile), password)
-    return _sign_envelope_with_key(envelope, key)
+    return _sign_envelope_with_key(envelope, key, signature_method, digest_method)
 
 
-def _signature_prepare(envelope, key):
+def _signature_prepare(envelope, key, signature_method, digest_method):
     """Prepare envelope and sign."""
     soap_env = detect_soap_env(envelope)
 
     # Create the Signature node.
     signature = xmlsec.template.create(
-        envelope, xmlsec.Transform.EXCL_C14N, xmlsec.Transform.RSA_SHA1
+        envelope,
+        xmlsec.Transform.EXCL_C14N,
+        signature_method or xmlsec.Transform.RSA_SHA1,
     )
 
     # Add a KeyInfo node with X509Data child to the Signature. XMLSec will fill
@@ -208,7 +241,7 @@ def _signature_prepare(envelope, key):
     # Perform the actual signing.
     ctx = xmlsec.SignatureContext()
     ctx.key = key
-    _sign_node(ctx, signature, envelope.find(QName(soap_env, "Body")))
+    _sign_node(ctx, signature, envelope.find(QName(soap_env, "Body")), digest_method)
     timestamp = security.find(QName(ns.WSU, "Timestamp"))
     if timestamp != None:
         _sign_node(ctx, signature, timestamp)
@@ -222,13 +255,17 @@ def _signature_prepare(envelope, key):
     return security, sec_token_ref, x509_data
 
 
-def _sign_envelope_with_key(envelope, key):
-    _, sec_token_ref, x509_data = _signature_prepare(envelope, key)
+def _sign_envelope_with_key(envelope, key, signature_method, digest_method):
+    _, sec_token_ref, x509_data = _signature_prepare(
+        envelope, key, signature_method, digest_method
+    )
     sec_token_ref.append(x509_data)
 
 
-def _sign_envelope_with_key_binary(envelope, key):
-    security, sec_token_ref, x509_data = _signature_prepare(envelope, key)
+def _sign_envelope_with_key_binary(envelope, key, signature_method, digest_method):
+    security, sec_token_ref, x509_data = _signature_prepare(
+        envelope, key, signature_method, digest_method
+    )
     ref = etree.SubElement(
         sec_token_ref,
         QName(ns.WSSE, "Reference"),
@@ -297,7 +334,7 @@ def _verify_envelope_with_key(envelope, key):
         raise SignatureVerificationFailed()
 
 
-def _sign_node(ctx, signature, target):
+def _sign_node(ctx, signature, target, digest_method=None):
     """Add sig for ``target`` in ``signature`` node, using ``ctx`` context.
 
     Doesn't actually perform the signing; ``ctx.sign(signature)`` should be
@@ -320,7 +357,7 @@ def _sign_node(ctx, signature, target):
 
     # Add reference to signature with URI attribute pointing to that ID.
     ref = xmlsec.template.add_reference(
-        signature, xmlsec.Transform.SHA1, uri="#" + node_id
+        signature, digest_method or xmlsec.Transform.SHA1, uri="#" + node_id
     )
     # This is an XML normalization transform which will be performed on the
     # target node contents before signing. This ensures that changes to

tests/test_wsse_signature.py

@@ -2,22 +2,30 @@
 import sys
 
 import pytest
-from lxml.etree import QName
 from lxml import etree
+from lxml.etree import QName
+
 from tests.utils import load_xml
 from zeep import ns, wsse
 from zeep.exceptions import SignatureVerificationFailed
 from zeep.wsse import signature
 from zeep.wsse.signature import xmlsec as xmlsec_installed
 
-DS_NS = "http://www.w3.org/2000/09/xmldsig#"
-
-
 KEY_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), "cert_valid.pem")
 KEY_FILE_PW = os.path.join(
     os.path.dirname(os.path.realpath(__file__)), "cert_valid_pw.pem"
 )
 
+SIGNATURE_METHODS_TESTDATA = (
+    ("RSA_SHA1", "http://www.w3.org/2000/09/xmldsig#rsa-sha1"),
+    ("RSA_SHA256", "http://www.w3.org/2001/04/xmldsig-more#rsa-sha256"),
+)
+
+DIGEST_METHODS_TESTDATA = (
+    ("SHA1", "http://www.w3.org/2000/09/xmldsig#sha1"),
+    ("SHA256", "http://www.w3.org/2001/04/xmlenc#sha256"),
+)
+
 
 skip_if_no_xmlsec = pytest.mark.skipif(
     sys.platform == "win32", reason="does not run on windows"
@@ -58,8 +66,15 @@ def test_sign_timestamp_if_present():
     signature.sign_envelope(envelope, KEY_FILE, KEY_FILE)
     signature.verify_envelope(envelope, KEY_FILE)
 
+
 @skip_if_no_xmlsec
-def test_sign():
+@pytest.mark.parametrize("digest_method,expected_digest_href", DIGEST_METHODS_TESTDATA)
+@pytest.mark.parametrize(
+    "signature_method,expected_signature_href", SIGNATURE_METHODS_TESTDATA
+)
+def test_sign(
+    digest_method, signature_method, expected_digest_href, expected_signature_href
+):
     envelope = load_xml(
         """
         <soapenv:Envelope
@@ -77,9 +92,24 @@ def test_sign():
     """
     )
 
-    signature.sign_envelope(envelope, KEY_FILE, KEY_FILE)
+    signature.sign_envelope(
+        envelope,
+        KEY_FILE,
+        KEY_FILE,
+        signature_method=getattr(xmlsec_installed.Transform, signature_method),
+        digest_method=getattr(xmlsec_installed.Transform, digest_method),
+    )
     signature.verify_envelope(envelope, KEY_FILE)
 
+    digests = envelope.xpath("//ds:DigestMethod", namespaces={"ds": ns.DS})
+    assert len(digests)
+    for digest in digests:
+        assert digest.get("Algorithm") == expected_digest_href
+    signatures = envelope.xpath("//ds:SignatureMethod", namespaces={"ds": ns.DS})
+    assert len(signatures)
+    for sig in signatures:
+        assert sig.get("Algorithm") == expected_signature_href
+
 
 @skip_if_no_xmlsec
 def test_sign_pw():
@@ -158,7 +188,13 @@ def test_signature():
 
 
 @pytest.mark.skipif(sys.platform == "win32", reason="does not run on windows")
-def test_signature_binary():
+@pytest.mark.parametrize("digest_method,expected_digest_href", DIGEST_METHODS_TESTDATA)
+@pytest.mark.parametrize(
+    "signature_method,expected_signature_href", SIGNATURE_METHODS_TESTDATA
+)
+def test_signature_binary(
+    digest_method, signature_method, expected_digest_href, expected_signature_href
+):
     envelope = load_xml(
         """
         <soapenv:Envelope
@@ -176,7 +212,13 @@ def test_signature_binary():
     """
     )
 
-    plugin = wsse.BinarySignature(KEY_FILE_PW, KEY_FILE_PW, "geheim")
+    plugin = wsse.BinarySignature(
+        KEY_FILE_PW,
+        KEY_FILE_PW,
+        "geheim",
+        signature_method=getattr(xmlsec_installed.Transform, signature_method),
+        digest_method=getattr(xmlsec_installed.Transform, digest_method),
+    )
     envelope, headers = plugin.apply(envelope, {})
     plugin.verify(envelope)
     # Test the reference
@@ -190,3 +232,12 @@ def test_signature_binary():
         namespaces={"soapenv": ns.SOAP_ENV_11, "wsse": ns.WSSE, "ds": ns.DS},
     )[0]
     assert "#" + bintok.attrib[QName(ns.WSU, "Id")] == ref.attrib["URI"]
+
+    digests = envelope.xpath("//ds:DigestMethod", namespaces={"ds": ns.DS})
+    assert len(digests)
+    for digest in digests:
+        assert digest.get("Algorithm") == expected_digest_href
+    signatures = envelope.xpath("//ds:SignatureMethod", namespaces={"ds": ns.DS})
+    assert len(signatures)
+    for sig in signatures:
+        assert sig.get("Algorithm") == expected_signature_href