Sign elements required by WSDL

Author: tpazderka

src/zeep/wsdl/bindings/soap.py

@@ -95,9 +95,13 @@ def _create(self, operation, args, kwargs, client=None, options=None):
             if client.wsse:
                 if isinstance(client.wsse, list):
                     for wsse in client.wsse:
-                        envelope, http_headers = wsse.apply(envelope, http_headers)
+                        envelope, http_headers = wsse.apply(
+                            envelope, http_headers, operation_obj.binding.signatures
+                        )
                 else:
-                    envelope, http_headers = client.wsse.apply(envelope, http_headers)
+                    envelope, http_headers = client.wsse.apply(
+                        envelope, http_headers, operation_obj.binding.signatures
+                    )
 
         # Add extra http headers from the setings object
         if client.settings.extra_http_headers:

src/zeep/wsdl/definitions.py

@@ -136,9 +136,9 @@ def __init__(self, wsdl, name, port_name):
         self.wsdl = wsdl
         self._operations = {}
         self.signatures = {
-            'header': [],  # Elements of header, that should be signed
-            'body': False,   # If body should be signed
-            'everything': False,  # If every header should be signed
+            "header": [],  # Elements of header, that should be signed
+            "body": False,  # If body should be signed
+            "everything": False,  # If every header should be signed
         }
 
     def resolve(self, definitions: Definition) -> None:

src/zeep/wsdl/wsdl.py

@@ -447,24 +447,28 @@ def parse_binding(
                         continue
 
                     # Begin heuristics for signed parts...
-                    binding_policy = binding.name.localname + '_policy'
-                    signed_parts = doc.xpath('wsp:Policy[@wsu:Id="{}"]//sp:SignedParts'.format(binding_policy),
-                                             namespaces=NSMAP)
+                    binding_policy = binding.name.localname + "_policy"
+                    signed_parts = doc.xpath(
+                        'wsp:Policy[@wsu:Id="{}"]//sp:SignedParts'.format(
+                            binding_policy
+                        ),
+                        namespaces=NSMAP,
+                    )
                     for sign in signed_parts:
                         if len(sign.getchildren()) == 0:
                             # No children, we should sign everything
-                            binding.signatures['body'] = True
-                            binding.signatures['everything'] = True
+                            binding.signatures["body"] = True
+                            binding.signatures["everything"] = True
                             break
 
                         for child in sign.iterchildren():
                             if len(child.items()) > 0:
                                 # Header ...
                                 part = {attr: value for attr, value in child.items()}
-                                binding.signatures['header'].append(part)
-                            elif child.tag.split('}')[-1].lower() == 'body':
+                                binding.signatures["header"].append(part)
+                            elif child.tag.split("}")[-1].lower() == "body":
                                 # Body ...
-                                binding.signatures['body'] = True
+                                binding.signatures["body"] = True
                     logger.debug("Adding binding: %s", binding.name.text)
                     result[binding.name.text] = binding
                     break

src/zeep/wsse/signature.py

@@ -14,6 +14,7 @@
 from zeep import ns
 from zeep.exceptions import SignatureVerificationFailed
 from zeep.utils import detect_soap_env
+from zeep.wsdl.utils import get_or_create_header
 from zeep.wsse.utils import ensure_id, get_security_header
 
 try:
@@ -61,10 +62,10 @@ def __init__(
         self.digest_method = digest_method
         self.signature_method = signature_method
 
-    def apply(self, envelope, headers):
+    def apply(self, envelope, headers, signatures=None):
         key = _make_sign_key(self.key_data, self.cert_data, self.password)
         _sign_envelope_with_key(
-            envelope, key, self.signature_method, self.digest_method
+            envelope, key, self.signature_method, self.digest_method, signatures
         )
         return envelope, headers
 
@@ -99,10 +100,10 @@ class BinarySignature(Signature):
 
     Place the key information into BinarySecurityElement."""
 
-    def apply(self, envelope, headers):
+    def apply(self, envelope, headers, signatures=None):
         key = _make_sign_key(self.key_data, self.cert_data, self.password)
         _sign_envelope_with_key_binary(
-            envelope, key, self.signature_method, self.digest_method
+            envelope, key, self.signature_method, self.digest_method, signatures
         )
         return envelope, headers
 
@@ -123,6 +124,7 @@ def sign_envelope(
     password=None,
     signature_method=None,
     digest_method=None,
+    signatures=None,
 ):
     """Sign given SOAP envelope with WSSE sig using given key and cert.
 
@@ -213,10 +215,12 @@ def sign_envelope(
     """
     # Load the signing key and certificate.
     key = _make_sign_key(_read_file(keyfile), _read_file(certfile), password)
-    return _sign_envelope_with_key(envelope, key, signature_method, digest_method)
+    return _sign_envelope_with_key(
+        envelope, key, signature_method, digest_method, signatures
+    )
 
 
-def _signature_prepare(envelope, key, signature_method, digest_method):
+def _signature_prepare(envelope, key, signature_method, digest_method, signatures=None):
     """Prepare envelope and sign."""
     soap_env = detect_soap_env(envelope)
 
@@ -241,10 +245,29 @@ def _signature_prepare(envelope, key, signature_method, digest_method):
     # Perform the actual signing.
     ctx = xmlsec.SignatureContext()
     ctx.key = key
-    _sign_node(ctx, signature, envelope.find(QName(soap_env, "Body")), digest_method)
+    # Sign default elements if present
     timestamp = security.find(QName(ns.WSU, "Timestamp"))
     if timestamp != None:
         _sign_node(ctx, signature, timestamp, digest_method)
+
+    # Sign extra elements defined in WSDL
+    if signatures is not None:
+        if signatures["body"] or signatures["everything"]:
+            _sign_node(
+                ctx, signature, envelope.find(QName(soap_env, "Body")), digest_method
+            )
+        header = get_or_create_header(envelope)
+        if signatures["everything"]:
+            for node in header.iterchildren():
+                _sign_node(ctx, signature, node, digest_method)
+        else:
+            for node in signatures["header"]:
+                _sign_node(
+                    ctx,
+                    signature,
+                    header.find(QName(node["Namespace"], node["Name"])),
+                    digest_method,
+                )
     ctx.sign(signature)
 
     # Place the X509 data inside a WSSE SecurityTokenReference within
@@ -255,16 +278,20 @@ def _signature_prepare(envelope, key, signature_method, digest_method):
     return security, sec_token_ref, x509_data
 
 
-def _sign_envelope_with_key(envelope, key, signature_method, digest_method):
+def _sign_envelope_with_key(
+    envelope, key, signature_method, digest_method, signatures=None
+):
     _, sec_token_ref, x509_data = _signature_prepare(
-        envelope, key, signature_method, digest_method
+        envelope, key, signature_method, digest_method, signatures=signatures
     )
     sec_token_ref.append(x509_data)
 
 
-def _sign_envelope_with_key_binary(envelope, key, signature_method, digest_method):
+def _sign_envelope_with_key_binary(
+    envelope, key, signature_method, digest_method, signatures=None
+):
     security, sec_token_ref, x509_data = _signature_prepare(
-        envelope, key, signature_method, digest_method
+        envelope, key, signature_method, digest_method, signatures=signatures
     )
     ref = etree.SubElement(
         sec_token_ref,

src/zeep/wsse/username.py

@@ -65,7 +65,7 @@ def __init__(
         self.zulu_timestamp = zulu_timestamp
         self.hash_password = hash_password
 
-    def apply(self, envelope, headers):
+    def apply(self, envelope, headers, operation_obj=None):
         security = utils.get_security_header(envelope)
 
         # The token placeholder might already exists since it is specified in

tests/test_wsse_signature.py

@@ -118,12 +118,15 @@ def test_sign(
     """
     )
 
+    # Force body signature
+    signatures = {"everything": False, "body": True, "header": []}
     signature.sign_envelope(
         envelope,
         KEY_FILE,
         KEY_FILE,
         signature_method=getattr(xmlsec.Transform, signature_method),
         digest_method=getattr(xmlsec.Transform, digest_method),
+        signatures=signatures,
     )
     signature.verify_envelope(envelope, KEY_FILE)
 
@@ -156,7 +159,9 @@ def test_sign_pw():
     """
     )
 
-    signature.sign_envelope(envelope, KEY_FILE_PW, KEY_FILE_PW, "geheim")
+    # Force body signature
+    signatures = {"everything": False, "body": True, "header": []}
+    signature.sign_envelope(envelope, KEY_FILE_PW, KEY_FILE_PW, "geheim", signatures=signatures)
     signature.verify_envelope(envelope, KEY_FILE_PW)
 
 
@@ -179,7 +184,9 @@ def test_verify_error():
     """
     )
 
-    signature.sign_envelope(envelope, KEY_FILE, KEY_FILE)
+    # Force body signature
+    signatures = {"everything": False, "body": True, "header": []}
+    signature.sign_envelope(envelope, KEY_FILE, KEY_FILE, signatures=signatures)
     nsmap = {"tns": "http://tests.python-zeep.org/"}
 
     for elm in envelope.xpath("//tns:Argument", namespaces=nsmap):
@@ -208,8 +215,10 @@ def test_signature():
     """
     )
 
+    # Force body signature
+    signatures = {"everything": False, "body": True, "header": []}
     plugin = wsse.Signature(KEY_FILE_PW, KEY_FILE_PW, "geheim")
-    envelope, headers = plugin.apply(envelope, {})
+    envelope, headers = plugin.apply(envelope, {}, signatures=signatures)
     plugin.verify(envelope)
 
 
@@ -238,14 +247,16 @@ def test_signature_binary(
     """
     )
 
+    # Force body signature
+    signatures = {"everything": False, "body": True, "header": []}
     plugin = wsse.BinarySignature(
         KEY_FILE_PW,
         KEY_FILE_PW,
         "geheim",
         signature_method=getattr(xmlsec.Transform, signature_method),
         digest_method=getattr(xmlsec.Transform, digest_method),
     )
-    envelope, headers = plugin.apply(envelope, {})
+    envelope, headers = plugin.apply(envelope, {}, signatures=signatures)
     plugin.verify(envelope)
     # Test the reference
     bintok = envelope.xpath(