api_test.py 3.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115
  1. from gmssl import sm2
  2. import requests
  3. import time
  4. import json
  5. import hashlib
  6. import pprint
  7. import datetime
  8. pay_url = "http://127.0.0.1:8848/pay"
  9. search_url = "http://127.0.0.1:8848/search"
  10. # 测试用 秘钥组1 脚本加密用
  11. pk1 = "04865e672fa71ab6e37429fd731800792c1015cf4e936bd01ec0092b7da571fb63cb99cca8104b35243175d3f535784f6127af451bed43b5b929db6b864de3b207"
  12. sk1 = "b73190a4fa4f6786fa36a35c1c16e533943fb5951a40c6b9e2d06b5a19d5c19a"
  13. # 测试用 秘钥组2 脚本解密用
  14. pk2 = "04f87981e38e98c5d400daed3e3dbf0a77e622860923b37ab94622c435401604a2e23f7c2ddb656f0b7735ca1f36cc203e6ac8fe47d6bc15f3739608e81b424c04"
  15. sk2 = "6907ee4516db5e2f04eef3976551f2c8ead793648edf37b82035013527b3ee42"
  16. sinKey = "ZctLTt2tugU2ntZdVgMXssrGuENPH6V7b7ROYN88Msw="
  17. pub_head = {
  18. "appId": "CESHI",
  19. "method": "sd.tovc.recharge",
  20. "signType": "md5",
  21. "sign": "",
  22. "timestamp": "",
  23. "version": "v1.0",
  24. "requestId": str(int(time.time()*1000)),
  25. "bizContent": ""
  26. }
  27. def encrypt(data: str):
  28. en = sm2.CryptSM2(private_key="", public_key=pk1)
  29. r = en.encrypt(data.encode("utf-8"))
  30. # result = bytearray([0x4])
  31. result = bytearray()
  32. result.extend(r) # type:ignore
  33. return result.hex() # type:ignore
  34. def decrypt(data: str):
  35. en = sm2.CryptSM2(private_key=sk2, public_key="")
  36. r = en.decrypt(data.encode("utf-8"))
  37. return r.hex() # type:ignore
  38. def sign_head(head: dict):
  39. keys = list(head.keys())
  40. keys.sort()
  41. ks = []
  42. for k in keys:
  43. v = head[k].strip()
  44. if k != "sign" and v:
  45. ks.append(f"{k}={v}")
  46. ks.append(f"key={sinKey}")
  47. s = "&".join(ks)
  48. m5 = hashlib.md5(s.encode("utf-8"))
  49. sign = m5.hexdigest()
  50. head["sign"] = sign
  51. def test_pay():
  52. pay_body = {
  53. "gameUserId": "13856023633",
  54. "serverType": "tel",
  55. "productNo": "YDHF10",
  56. "spOrderId": str(int(time.time()*1000)),
  57. "source": "SD"
  58. }
  59. pay_body = json.dumps(pay_body, ensure_ascii=False)
  60. head = pub_head.copy()
  61. now = datetime.datetime.now()
  62. head["timestamp"] = f"{now:%Y-%m-%d %H:%M:%S}"
  63. head["requestId"] = str(int(time.time()*1000))
  64. head["bizContent"] = encrypt(pay_body)
  65. sign_head(head)
  66. pprint.pprint(pay_body)
  67. r = requests.post(pay_url, json=head)
  68. result = r.json()
  69. print(result)
  70. order = {
  71. 'code': '0000',
  72. 'msg': 'success',
  73. 'data':'{"resultCode":"000","resultDesc":"OK","data":{"state":"0","orderCash":"","orderId":"1679647728031106994","spOrderId":"1679647727617","gameUserId":"13856023633","productName":"YDHF10"}}',
  74. 'sign': '2B4C3FBD6C65A8A78BCA7EDF71F6C666',
  75. 'requestId': '1679647727617'}
  76. def test_search():
  77. search_body = {
  78. "spOrderId": "1679680499000"
  79. }
  80. pay_body = json.dumps(search_body, ensure_ascii=False)
  81. head = pub_head.copy()
  82. now = datetime.datetime.now()
  83. head["timestamp"] = f"{now:%Y-%m-%d %H:%M:%S}"
  84. head["requestId"] = str(int(time.time()*1000))
  85. head["bizContent"] = encrypt(pay_body)
  86. sign_head(head)
  87. pprint.pprint(head)
  88. r = requests.post(search_url, json=head)
  89. result = r.json()
  90. print(result)
  91. # test_pay()
  92. test_search()
  93. # s = "{\"HEADER\":{\"VERSION\":\"V1.1\",\"TIMESTAMP\":\"2023-03-24 21:31:47\",\"SEQNO\":\"1679664707214\",\"APPID\":\"jtceshi\",\"SECERTKEY\":\"376F93167C475920C274EE682650031D\"},\"MSGBODY\":{\"CONTENT\":{\"ORDERID\":\"1679647728031106994\",\"USER\":\"13856023633\",\"REQDATE\":\"2023-03-24\",\"PACKAGEID\":\"YDHF10\",\"STATUS\":\"4\",\"EXTORDER\":\"1679647727617\",\"MSG\":\"充值失败\",\"CHECKTIME\":\"20230324164950000\",\"PRICE\":\"100000\"},\"RESP\":{\"RCODE\":\"00\",\"RMSG\":\"OK\"}}}"
  94. # pprint.pprint(json.loads(s))