如何用Python Zeep为SOAP请求WS-Security头的wsa元素签名

我正在对接一个要求同时支持WS-Addressing和WS-Security的SOAP数据服务，目前已实现请求体（Body）和时间戳（Timestamp）的签名，但还需要为wsa:To、wsa:MessageID和wsa:Action元素添加签名，使用的是Python Zeep模块。

现有辅助类代码 #

我从Zeep的GitHub Issues中找到两个实用类：

from datetime import datetime, timedelta
from zeep.wsse.signature import BinarySignature
from zeep.wsse.utils import utils
from zeep.plugin import Plugin

class BinarySignatureTimestamp(BinarySignature):
    def apply(self, envelope, headers):
        security = utils.get_security_header(envelope)

        created = datetime.utcnow()
        expired = created + timedelta(seconds=1 * 60)

        timestamp = utils.WSU('Timestamp')
        timestamp.append(utils.WSU('Created', created.replace(microsecond=0).isoformat()+'Z'))
        timestamp.append(utils.WSU('Expires', expired.replace(microsecond=0).isoformat()+'Z'))

        security.append(timestamp)

        super().apply(envelope, headers)
        return envelope, headers

# 移除原始WSA元素，避免与WsAddressingPlugin生成的元素重复
class RemoveWSA(Plugin):
    def egress(self, envelope, http_headers, operation, binding_options):
        env = envelope.find('{http://schemas.xmlsoap.org/soap/envelope/}Header')  
        for child in env:
            if child.tag.startswith('{http://www.w3.org/2005/08/addressing}'):
                env.remove(child)
        return envelope, http_headers

当前客户端实现 #

from requests import Session
from zeep import Client, Settings
from zeep.transports import Transport
from zeep.cache import SqliteCache
from zeep.wsse.addressing import WsAddressingPlugin

session = Session()
session.cert = ('client.pem', 'privkey.pem')
transport = Transport(session=session, cache=SqliteCache())
settings = Settings(strict=False, xml_huge_tree=True)

client = Client(
    'http://schemas.kvk.nl/contracts/kvk/dataservice/catalogus/2015/02/KVK-KvKDataservice.wsdl', 
    settings=settings,
    transport=transport,
    plugins=[RemoveWSA(), WsAddressingPlugin(address_url='http://es.kvk.nl/kvk-DataservicePP/2015/02')],
    wsse=BinarySignatureTimestamp(
        "privkey.pem", 
        "client.pem",
        "certificate_password"
    )
)

# 替换WSDL中的默认请求地址
client.service._binding_options["address"] = 'https://webservices.preprod.kvk.nl/postbus1'

request_data = {"klantreferentie": "", "kvkNummer": "90003128"}
client.service.ophalenInschrijving(**request_data)

需求说明 #

需要为wsa:To、wsa:MessageID和wsa:Action元素生成如下格式的签名块：

<ds:Reference URI="#id-094973EF5ECC133BCB1679398962130734">
  <ds:Transforms>
    <ds:Transform Algorithm="http://www.w3.org/2001/10/xml-exc-c14n#">
      <ec:InclusiveNamespaces PrefixList="ns wsa" xmlns:ec="http://www.w3.org/2001/10/xml-exc-c14n#"/>
    </ds:Transform>
  </ds:Transforms>
  <ds:DigestMethod Algorithm="http://www.w3.org/2000/09/xmldsig#sha1"/>
  <ds:DigestValue>EPWWh6fOlEYHPoQVXaHYC6Ty6GQ=</ds:DigestValue>
</ds:Reference>

解决方案 #

要实现WSA元素的签名，需要修改BinarySignatureTimestamp类，扩展签名逻辑：

1. 修改签名类代码

from datetime import datetime, timedelta
import uuid
from zeep.wsse.signature import BinarySignature
from zeep.wsse.utils import get_security_header, WSU
from zeep.plugin import Plugin

class BinarySignatureTimestamp(BinarySignature):
    def apply(self, envelope, headers):
        # 添加Timestamp元素
        security = get_security_header(envelope)
        created = datetime.utcnow()
        expired = created + timedelta(seconds=60)
        timestamp = WSU('Timestamp')
        timestamp.append(WSU('Created', created.replace(microsecond=0).isoformat()+'Z'))
        timestamp.append(WSU('Expires', expired.replace(microsecond=0).isoformat()+'Z'))
        security.append(timestamp)

        # 定位SOAP Header和WSA元素
        soap_header = envelope.find('{http://schemas.xmlsoap.org/soap/envelope/}Header')
        wsa_ns = 'http://www.w3.org/2005/08/addressing'
        wsu_ns = 'http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd'
        
        # 目标WSA元素列表
        wsa_elements = [
            soap_header.find(f'{{{wsa_ns}}}To'),
            soap_header.find(f'{{{wsa_ns}}}MessageID'),
            soap_header.find(f'{{{wsa_ns}}}Action')
        ]

        # 给每个WSA元素添加唯一wsu:Id属性（签名引用需要）
        for elem in wsa_elements:
            if elem is not None:
                elem_id = f'id-{uuid.uuid4().hex.upper()}'
                elem.set(f'{{{wsu_ns}}}Id', elem_id)

        # 调用父类方法完成签名
        return super().apply(envelope, headers)

    def _sign_elements(self, envelope):
        # 获取默认签名元素（Body + Timestamp）
        signed_elements = super()._sign_elements(envelope)
        
        # 添加WSA元素到签名列表
        soap_header = envelope.find('{http://schemas.xmlsoap.org/soap/envelope/}Header')
        wsa_ns = 'http://www.w3.org/2005/08/addressing'
        wsa_elements = [
            soap_header.find(f'{{{wsa_ns}}}To'),
            soap_header.find(f'{{{wsa_ns}}}MessageID'),
            soap_header.find(f'{{{wsa_ns}}}Action')
        ]
        
        # 过滤不存在的元素，加入签名列表
        signed_elements.extend([elem for elem in wsa_elements if elem is not None])
        return signed_elements

2. 关键说明

• 添加wsu:Id属性：签名需要通过URI引用目标元素，因此必须给每个WSA元素添加带命名空间的wsu:Id属性，生成唯一ID值。
• 扩展签名元素列表：重写_sign_elements方法，将WSA元素加入Zeep默认的签名元素集合（Body和Timestamp）。
• 命名空间处理：Zeep会自动处理XML规范化（xml-exc-c14n#）及命名空间前缀包含逻辑，生成符合要求的<ec:InclusiveNamespaces>节点，确保与服务端要求一致。

3. 验证签名结果

发送请求后，可通过抓包工具查看SOAP消息的ds:Signature块，确认包含三个WSA元素的ds:Reference节点，且每个节点的URI对应元素的wsu:Id值。

内容的提问来源于stack exchange，提问作者KSchouten