Python - 带标头和时间戳的 Zeep SOAP 请求

Question

我正在尝试使用 Zeep 从 wsdl url

获取响应

需要用户名、密码、随机数 (PasswordDigest)、时间戳和标头。 在 SoapUI 中一切正常，但我无法使其与 Zeep 一起工作，我从服务器收到错误：

故障：验证消息时遇到安全错误

我猜问题来自标题，但我找不到添加它们的正确方法。

以下是 Web 服务文档提供的请求示例：

<soapenv:Envelope xmlns:soapenv=***schemas.xmlsoap.org/soap/envelope/
xmlns:mes="***economie.fgov.be/KBOpub/webservices/v1/messages"
xmlns:dat="***economie.fgov.be/KBOpub/webservices/v1/datamodel">
    <soapenv:Header>
        <wsse:Security>
            <wsu:Timestamp>
                <wsu:Created>2009-09-07T11:27:10.748Z</wsu:Created>
                <wsu:Expires>2009-09-07T11:32:10.748Z</wsu:Expires>
            </wsu:Timestamp>
            <wsse:UsernameToken>
                <wsse:Username>userid</wsse:Username>
                <wsse:Password>x3+DQlYgeVm3BAkobZivkDJ13zo=</wsse:Password>
                <wsse:Nonce>ENp2ha7j2Ar9cvWQeUybTQ==</wsse:Nonce>
                <wsu:Created>2009-09-07T11:27:10.716Z</wsu:Created>
            </wsse:UsernameToken>
       </wsse:Security>
       <mes:RequestContext>
           <mes:Id>c1576d0a-e762-40fe-abf9-ec3f2102650b</mes:Id>
           <mes:Language>fr</mes:Language>
       </mes:RequestContext>
    </soapenv:Header>
    <soapenv:Corps>
        <mes:ReadEnterpriseRequest>
            <dat:EnterpriseNumber>0314595348</dat:EnterpriseNumber>
        </mes:ReadEnterpriseRequest>
    </soapenv:Corps>
</soapenv:Envelope>

我根据其他问题的答案所尝试的：

from zeep.wsse.username import UsernameToken
import datetime
from zeep.wsse.utils import WSU
from zeep.xsd import Element, ComplexType, String, Sequence

wsdl_url = '***bopub.economie.fgov.be/kbopubws110000/services/wsKBOPub?wsdl'

username = '***'
password = '******'

timestamp_token = WSU.Timestamp()
today_datetime = datetime.datetime.today()
expires_datetime = today_datetime + datetime.timedelta(minutes=5)
timestamp_elements = [
        WSU.Created(today_datetime.strftime("%Y-%m-%dT%H:%M:%S.%fZ")),
        WSU.Expires(expires_datetime.strftime("%Y-%m-%dT%H:%M:%S.%fZ"))]

timestamp_token.extend(timestamp_elements)
user_name_token = UsernameToken(username, password, timestamp_token=timestamp_token, use_digest=True)
client = Client(wsdl_url, wsse=user_name_token)
client.set_ns_prefix('mes', "***economie.fgov.be/kbopub/webservices/v1/messages")

headerQ = Element('{***schemas.xmlsoap.org/soap/envelope.xsd}Header', ComplexType([
  Element('{***economie.fgov.be/kbopub/webservices/v1/messages.xsd}RequestContext', ComplexType(
    Sequence([
      Element('{***economie.fgov.be/kbopub/webservices/v1/messages.xsd}Id', String()),
      Element('{***economie.fgov.be/kbopub/webservices/v1/messages.xsd}Language', String())
    ]))
  )
]))
header_value = headerQ({'Id': 'whatever', 'Language': 'fr'})
client.set_default_soapheaders([header_value])

client.service.ReadEnterprise(EnterpriseNumber='0314595348')

我总是从服务器收到相同的安全错误

谢谢

Answer 1

我能够找到自己问题的解决方案，我将解释我是如何找到它的，因为我希望它可以帮助别人。

我首先检查了 SoapUI 中的 http 日志，看看请求应该是什么样子：

<soapenv:Envelope xmlns:dat="http://economie.fgov.be/kbopub/webservices/v1/datamodel" xmlns:mes="http://economie.fgov.be/kbopub/webservices/v1/messages" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
   <soapenv:Header><wsse:Security xmlns:wsse="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd" xmlns:wsu="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd"><wsu:Timestamp wsu:Id="TS-D7D646D96AD83C8414171093156745142"><wsu:Created>2024-03-20T10:46:07.451Z</wsu:Created><wsu:Expires>2024-03-20T10:51:07.451Z</wsu:Expires></wsu:Timestamp><wsse:UsernameToken wsu:Id="UsernameToken-D7D646D96AD83C8414171093156745141"><wsse:Username>***</wsse:Username><wsse:Password Type="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-username-token-profile-1.0#PasswordDigest">***</wsse:Password><wsse:Nonce EncodingType="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-soap-message-security-1.0#Base64Binary">DjBcs/i453gVH3io3+uUCw==</wsse:Nonce><wsu:Created>2024-03-20T10:46:07.451Z</wsu:Created></wsse:UsernameToken></wsse:Security>
      <mes:RequestContext>
         <mes:Id>whatever</mes:Id>
         <mes:Language>fr</mes:Language>
      </mes:RequestContext>
   </soapenv:Header>
   <soapenv:Body>
      <mes:ReadEnterpriseRequest>
         <dat:EnterpriseNumber>0422118165</dat:EnterpriseNumber>
      </mes:ReadEnterpriseRequest>
   </soapenv:Body>
</soapenv:Envelope>

然后我与使用 HistoryPlugin 发送的内容进行了比较：

from zeep.plugins import HistoryPlugin

history = HistoryPlugin()

[...]

client = Client(wsdl_url, wsse=user_name_token, plugins=[history])

from lxml import etree

for hist in [history.last_sent, history.last_received]:
    print(etree.tostring(hist["envelope"], encoding="unicode", pretty_print=True))

并且可以轻松查看需要更改的内容。 最终我的代码如下所示：

from zeep import Client
from zeep.wsse.username import UsernameToken
import datetime
from zeep.wsse.utils import WSU
from zeep.xsd import Element, ComplexType, String, Sequence
from zeep.plugins import HistoryPlugin

history = HistoryPlugin()

# WSDL URL
wsdl_url = 'https://kbopub.economie.fgov.be/kbopubws110000/services/wsKBOPub?wsdl'

# Username and password for authentication
username = ***
password = ***

timestamp_token = WSU.Timestamp()
today_datetime = datetime.datetime.now(datetime.timezone.utc)
expires_datetime = today_datetime + datetime.timedelta(minutes=5)
timestamp_elements = [
        WSU.Created((today_datetime.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3])+'Z'),
        WSU.Expires((expires_datetime.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3])+'Z')]

timestamp_token.extend(timestamp_elements)
user_name_token = UsernameToken(username, password, timestamp_token=timestamp_token, use_digest=True)
client = Client(wsdl_url, wsse=user_name_token, plugins=[history])
client.set_ns_prefix('ns1', "http://economie.fgov.be/kbopub/webservices/v1/datamodel")
client.set_ns_prefix('ns0', "http://economie.fgov.be/kbopub/webservices/v1/messages")
client.set_ns_prefix('soap-env', "http://schemas.xmlsoap.org/soap/envelope/")

# Creating a RequestContext SOAP header
headerQ = Element('{http://economie.fgov.be/kbopub/webservices/v1/messages}RequestContext',
                  ComplexType([Sequence([
                      Element('{http://economie.fgov.be/kbopub/webservices/v1/messages}Id', String()),
                      Element('{http://economie.fgov.be/kbopub/webservices/v1/messages}Language', String())
                  ])]))

# Setting the RequestContext SOAP header value
header_value = headerQ(Id='whatever', Language='fr')
client.set_default_soapheaders([header_value])

print(client.service.ReadEnterprise(EnterpriseNumber=i))