使用 JwT 令牌身份验证对 HTTP Rest API 的 Python 发布请求会生成重复的帖子

Posted 2023-03-10

【中文标题】使用 JwT 令牌身份验证对 HTTP Rest API 的 Python 发布请求会生成重复的帖子

【英文标题】：Python Post Requests to HTTP RestAPI with JwToken authentication generates duplicate posts

【发布时间】：2018-01-22 00:26:41

【问题描述】：

我一直在编写一个 API 程序来测试发布到具有 JwToken 身份验证的 http RestAPI。在这种情况下，它用于患者管理系统，我正在生成一个约会。 API 业务规则不允许同时进行重复预订。

我正在使用 python 3.5.3 虚拟环境（在 pycharm IDE 中编写）并使用 Pytest 框架运行我的测试。 也使用 PyJWT 1.5.2。 , Requests 2.18.3, simplejson 3.11.1 和 urllib3 1.22 已安装（我假设 Requests 正在使用 urllib3）。我使用的是 simplejson.dumps 而不是普通的 json.dumps，因为虚拟环境没有那个库，而且我在添加它时遇到了麻烦。据我所知，simplejson 具有与转储过程相同的功能。

使用下面的代码，我发现我可以运行 requests.post 调用以成功传递 Json 数据有效负载并生成一个帖子，但随后它似乎执行了第二个帖子，这会产生 409 冲突错误。我可以访问相应 api 服务器上的日志数据库，并且可以看到它实际上已尝试发布两次，但我无法弄清楚为什么会发生这种情况，我认为请求库中有一些东西被调用了两次.可能是由于我发布的 Json。

输出如下所示：

https://targerserver.url.com.au/API/Core/v2.1/appointment/
200
'statusMessages': [], 'appointment': 'startDateTime': '2017-08-15T11:00:00 +10:00', 'appointmentReferenceNumber': '39960337', 'notes': '', 'clients': ['clientId': 'abeff2be-ce6e-4324-9b57-e28ab7967b6c'], 'status': 'Booked', 'locationId': 'd8d4fe7c-765a-46a3-a389-54ce298a27e9', 'notifyPractitioner': False, 'endDateTime': '2017-08-15T11:30:00 +10:00', 'subject': 'Jim Beam ', 'appointmentId': '08b37ce3-25e1-4e2a-9bb7-9ec2d716f83b', 'practitioner': 'practitionerId': 'a630f4ad-8b4b-4e06-8cee-7db56ba8b9bf'
collected 1 item

test_PMSAPI_availability.py https://targerserver.url.com.au/API/Core/v2.1/appointment/
409

我的 Json 需要一个对象（它是一个字典）以及另一个字段的键列表（其中有一个条目），我想知道请求库是否没有处理这个问题。这是 json 的样例

payload_str = "startDateTime":"2017-08-15T11:00+10:00","endDateTime":"2017-08-15T11:30+10:00","practitioner": "practitionerId":"a630f4ad-8b4b-4e06-8cee-7db56ba8b9bf", "locationId":"d8d4fe7c-765a-46a3-a389-54ce298a27e9","clients":["clientId":"abeff2be-ce6e-4324-9b57-e28ab7967b6c"]

我有类似的代码可用于同一系统上的 Get 调用，但发布 Json 似乎确实有问题。我们有其他工具可以调用相同的 API 端点，它们似乎没有这个问题。日志表明我提供的 JSON 数据与来自其他工具的具有相同数据的数据相同。

我从输出中可以看到，在初始响应中收到了成功的 200 代码，但是如果查询 response.status_code 它已变成 409 响应。我还尝试不对响应做任何事情，以防导致请求重新查询并产生冲突。

我的代码如下所示：

import jwt
import _datetime
import requests
import simplejson
from requests.packages.urllib3.exceptions import InsecureRequestWarning
from string import Template

def app_undertest_credentials(keyname):
    app_credentials = 'app_consumer_id': 'MyTestApp',
                       'app_consumer_secret': 'where my secret goes',
                       'app_access_token': 'where my access token goes',
                       'base_url': 'https://targerserver.url.com.au'
                       

    return app_credentials.get(keyname)
def end_points_dict(keynameStr, versionStr):
    end_points = 'location': '/API/Core/$version/location/',
                  'practitioner': '/API/Core/$version/practitioner/',
                  'availabilityslot': '/API/Core/$version/AvailabilitySlot/',
                  'client': '/API/Core/$version/client/',
                  'healthfundproviderlist': '/API/Core/$version/healthfundproviderlist/',
                  'timezone': '/API/Core/$version/timezone/',
                  'clientgroup': '/API/Core/$version/clientgroup/',
                  'appointment': '/API/Core/$version/appointment/'
                  
    lower_keynameStr = keynameStr.lower()
    url_extension_no_version = Template(end_points.get(lower_keynameStr))
    url_extension_with_version = url_extension_no_version.safe_substitute(version=versionStr)
    return url_extension_with_version

def test_api_appointment_post():
    # Set Client app credentials
    app_consumer_id = app_undertest_credentials('app_consumer_id')
    app_consumer_secret = app_undertest_credentials('app_consumer_secret')
    app_access_token = app_undertest_credentials('app_access_token')
    base_url = app_undertest_credentials('base_url')
    end_point_url_sfx_str = end_points_dict('Appointment', 'v2.1')
    httpmethod = 'POST'

    # Create dictionary for json post payload
    data_payload = 'startDateTime':'2017-08-15T11:00+10:00',
                'endDateTime':'2017-08-15T11:30+10:00',
                'practitioner': 'practitionerId':'a630f4ad-8b4b-4e06-8cee-7db56ba8b9bf',
                'locationId': 'd8d4fe7c-765a-46a3-a389-54ce298a27e9',
                'clients': ['clientId':'abeff2be-ce6e-4324-9b57-e28ab7967b6c']

    # Create claims dictionary payload for generation of JwToken
    claims = 
        'iss': 'http://myappsdomain.com.au',
        'aud': 'https://targetservers.domain.com.au',
        'nbf': _datetime.datetime.utcnow(),
        'exp': _datetime.datetime.utcnow() + _datetime.timedelta(seconds=60),
        'consumerId': app_consumer_id,
        'accessToken': app_access_token,
        'url': base_url + end_point_url_sfx_str,
        'httpMethod': http_method
    

    #create jwtoken and then convert to string
    encoded_jwt_byte = jwt.encode(claim_payload, app_consumer_secret, algorithm='HS256')
    jwt_str = encoded_jwt_byte.decode()

    #Create authentication header
    headers = 'Authorization': 'JwToken' + ' ' + jwt_str, 'content-type': 'application/json'

    uri = base_url + end_point_url_sfx_str
    response = requests.post(uri, headers=headers, json=datapayload)
    print(response.status)
    print(response.json())
    response.close()

我正在考虑使用 wireshark 来确定我的呼叫实际发送的内容，但我怀疑这只会告诉我呼叫被发送了两次

【问题讨论】：

Edit 你的问题是你正在使用哪个 Python 模块和版本，python_jwt 或 pyjwt？

我在其他模块版本中添加了使用 PyJWT 1.5.2。 , 请求 2.18.3、simplejson 3.11.1 和 urllib3 1.22

【参考方案1】：

好的，找到了我重复发布问题的原因。这是我自己愚蠢的错。在屏幕底部的 python 文件的底部，我调用了测试函数并且没有注意到这一行（因此我错过了在上面的代码中发布它）。它将测试功能置于一个完全重复的循环中。 .... :( 如此愚蠢的新手错误。 感谢 stovfl 关于处理编码的 JwToken 的建议。

事后看来，我在 *** 上发现的每篇关于 API 重复帖子的帖子都是因为用户代码中有一个循环，我只是找不到它。

【参考方案2】：

评论：但是我需要一个 base64 编码的字符串来发送 api 请求。

关于来源

segments.append(base64url_encode(signature))
    return base64.urlsafe_b64encode(input).replace(b'=', b'')
return b'.'.join(segments)

全部为 base64 并返回为 bytes。所以你应该没问题使用

jwt_str = str(encoded_jwt_byte)

---

抱歉，不能使用PyJWT。 尝试使用python_jwt，它按预期工作。

使用 Python：3.4.2 测试 - 请求：2.11.1

你真的需要先encode(...，然后再decode()吗？

#create jwtoken and then convert to string
encoded_jwt_byte = jwt.encode(claim_payload, app_consumer_secret, algorithm='HS256')
jwt_str = encoded_jwt_byte.decode()

jwt_str 会不会和claim_payload 一样？

【讨论】：

由于某种原因，当我使用 jwt.encode 创建 jwtoken 时，我的系统会创建字节码，尽管文档说它应该返回一个字符串。例如它看起来像 b'eyJ0eXAiOiJKV1QiLCJhbGciOiJIU.... 但我需要一个 base64 编码的字符串来发送 api 请求。通过再次解码，它会创建一个 base64 编码的字符串，我可以在 jwt.io 进行验证，并且似乎适用于我的 get 请求。如果您有任何建议，我会收到一个字节字符串而不是 base64 编码的字符串，我会感兴趣

刚刚意识到我可以使用 jwt_str = str(encoded_jwt_byte,'utf-8') 而不是解码，当它没有解码的秘密时，它似乎在做同样的事情。

@Roochiedoor：更新了我的答案