six_005
This commit is contained in:
doudou201703
@@ -6,13 +6,14 @@ import re
|
||||
import secrets
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Literal, Optional
|
||||
from typing import Literal
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
import django
|
||||
import orjson
|
||||
import pydash
|
||||
import requests
|
||||
import xmltodict
|
||||
from Crypto.Cipher import AES
|
||||
from Crypto.Hash import SHA256
|
||||
from Crypto.PublicKey import RSA
|
||||
@@ -46,59 +47,102 @@ class sixWechatPay六果泡泡支付(BaseModel):
|
||||
# ---------- 微信支付公钥（用于验签和加密敏感字段） ----------
|
||||
wechatpay_public_key_id: str = Field(default="PUB_KEY_ID_0117003834192026080500181856002000", alias="微信支付公钥ID")
|
||||
wechatpay_public_key_pem: str = Field(default=Path(settings.BASE_DIR).joinpath("wechat_微信支付证书/江苏六果科技有限公司支付证书/微信支付公钥.pem").read_text(), alias="微信支付公钥PEM", description="pub_key.pem 内容")
|
||||
|
||||
# ---------- 平台证书（灰度过渡期备用） ----------
|
||||
platform_cert_pem: Optional[
|
||||
str] = Field(default=Path(settings.BASE_DIR).joinpath("wechat_微信支付证书/江苏六果科技有限公司支付证书/sixgou_ping_tai_zheng_shu_cert.pem").read_text(), alias="平台证书PEM", description="wechatpay.pem 内容")
|
||||
pay_uri: str | None = Field(None, )
|
||||
pay_请求体: dict | bytes | str | None = Field(None, description="发请求签名时需要的请求体")
|
||||
platform_cert_pem: str = Field(default=Path(settings.BASE_DIR).joinpath("wechat_微信支付证书/江苏六果科技有限公司支付证书/sixgou_ping_tai_zheng_shu_cert.pem").read_text(), alias="平台证书PEM",
|
||||
description="wechatpay.pem 内容")
|
||||
platform_cert_pem_id: str = Field(default="11D762828E947104028F2D045DE78FA806D0D4F2", description="平台证书的证收序列号ID")
|
||||
pay_uri: str | None = Field(default=None, description="请求支付时需要的微信的url地址")
|
||||
pay_请求体: dict = Field(default=None, description="发请求签名时需要的请求体")
|
||||
pay_methon方法: Literal["GET", "POST", "PATCH"] = Field(default="POST", description="发请求签名时需要的请求体")
|
||||
pay_Headers: dict = Field(default={}, description="生成的初始化请求头")
|
||||
|
||||
def model_post_init(self, __context):
|
||||
pass
|
||||
""""""
|
||||
if self.pay_uri is None or self.pay_uri == "":
|
||||
return None
|
||||
url_data = re.search(r'/v3.*', string=self.pay_uri).group()
|
||||
utc_unix_timestamp = int(datetime.now(tz=ZoneInfo("utc")).timestamp())
|
||||
random_str = secrets.token_hex(nbytes=16) # 随机字符串
|
||||
if isinstance(self.pay_请求体, str):
|
||||
body_inner = self.pay_请求体
|
||||
elif isinstance(self.pay_请求体, dict):
|
||||
body_inner = json.dumps(self.pay_请求体)
|
||||
elif self.pay_请求体 is None:
|
||||
body_inner = ""
|
||||
elif isinstance(self.pay_请求体, bytes):
|
||||
body_inner = self.pay_请求体.decode()
|
||||
else:
|
||||
raise TypeError(f"不支持的请求体类型: {type(self.pay_请求体)}")
|
||||
pay_signaturestring创建成功的签名串 = f"{self.pay_methon方法}\n{url_data}\n{utc_unix_timestamp}\n{random_str}\n{body_inner}\n"
|
||||
# 构造一个哈希对象
|
||||
hash_obj = SHA256.new(pay_signaturestring创建成功的签名串.encode(encoding="utf-8"))
|
||||
# 加载商户私钥
|
||||
# 调用了 RSA.import_key() 方法，它会从字符串或文件中导入一个 PEM 格式的 RSA 私钥，并返回一个 RsaKey 类型的对象
|
||||
private_key: RsaKey = RSA.import_key(self.merchant_private_key_pem) # 用私钥来加密
|
||||
# 生成签名 使用商户的 私钥进行签名
|
||||
signature = pkcs1_15.new(rsa_key=private_key).sign(msg_hash=hash_obj)
|
||||
# 对签名进行Base64编码，符合微信支付API要求
|
||||
encoded_signature = base64.b64encode(s=signature).decode(encoding="utf-8")
|
||||
# 组装成微信官方需要的数据;
|
||||
data = {
|
||||
"mchid": self.mchid,
|
||||
"nonce_str": random_str,
|
||||
"serial_no": self.serial_no商户证书的序列号,
|
||||
"timestamp": utc_unix_timestamp,
|
||||
"signature": encoded_signature,
|
||||
} # 组装好的字典数据;
|
||||
ftr = "WECHATPAY2-SHA256-RSA2048"
|
||||
s_with_space = ftr.ljust(len(ftr) + 1) # 将字符串长度 +1（填充空格）
|
||||
str_list = pydash.join(array=[f'{key}="{value}"' for key, value in data.items()], separator=",")
|
||||
# 生成请求头
|
||||
Authorization = f"{s_with_space}{str_list}"
|
||||
headers = {"Content-Type": "application/json", "Authorization": Authorization, }
|
||||
headers.update({"Wechatpay-Serial": self.wechatpay_public_key_id})
|
||||
self.pay_Headers = headers
|
||||
return None
|
||||
|
||||
def 验证签名串(self, response_headers, response_data: dict, ):
|
||||
"""
|
||||
@param response_headers:
|
||||
@type response_headers:
|
||||
@param response_data:
|
||||
@type response_data:
|
||||
@return:
|
||||
@rtype:
|
||||
"""
|
||||
public_key = RSA.import_key(self.platform_cert_pem)
|
||||
""" 注释 """
|
||||
public_key = RSA.import_key(self.wechatpay_public_key_pem) # 在这里需要使用最新的微信支付公钥证书
|
||||
Wechatpay_Timestamp = pydash.get(response_headers, "Wechatpay-Timestamp")
|
||||
Wechatpay_Nonce = pydash.get(response_headers, "Wechatpay-Nonce")
|
||||
Wechatpay_Signa = pydash.get(response_headers, "Wechatpay-Signature")
|
||||
Wechatpay_Serial = pydash.get(response_headers, "Wechatpay-Serial")
|
||||
response_data_str = orjson.dumps(response_data, ).decode(encoding="utf-8")
|
||||
|
||||
string_to_verify = f"{Wechatpay_Timestamp}\n{Wechatpay_Nonce}\n{response_data_str}\n" # 构造验签名串
|
||||
hash_obj = SHA256.new(string_to_verify.encode())
|
||||
# 6. 验证签名
|
||||
if Wechatpay_Serial == self.platform_cert_pem_id:
|
||||
public_key = RSA.import_key(self.platform_cert_pem)
|
||||
# 6. 验证签名
|
||||
if Wechatpay_Serial == self.wechatpay_public_key_id:
|
||||
public_key = RSA.import_key(self.wechatpay_public_key_pem) # 在这里需要使用最新的微信支付公钥证书
|
||||
try:
|
||||
pkcs1_15.new(public_key).verify(msg_hash=hash_obj, signature=base64.b64decode(Wechatpay_Signa))
|
||||
return True # 签名验证成功
|
||||
except (ValueError, TypeError) as e:
|
||||
raise e # 签名验证失败
|
||||
print(e, "捕获了签名串验证不通过的错误") # 签名验证失败
|
||||
raise "捕获了签名串验证不通过的错误"
|
||||
|
||||
def gouzhao构造签名串(self, ):
|
||||
"""
|
||||
|
||||
"""
|
||||
"""注释"""
|
||||
url_data = re.search(r'/v3.*', string=self.pay_uri).group()
|
||||
utc_unix_timestamp = int(datetime.now(tz=ZoneInfo("utc")).timestamp())
|
||||
random_str = secrets.token_hex(nbytes=16) # 随机字符串
|
||||
pay_signaturestring创建成功的签名串 = ""
|
||||
# 创建签名请求参数
|
||||
# 统一处理请求体，得到 body_inner 字符串
|
||||
if isinstance(self.pay_请求体, str):
|
||||
pay_signaturestring创建成功的签名串 = f"{self.pay_methon方法}\n{url_data}\n{utc_unix_timestamp}\n{random_str}\n{self.pay_请求体}\n"
|
||||
if isinstance(self.pay_请求体, dict):
|
||||
body_inner = json.dumps(obj=self.pay_请求体)
|
||||
pay_signaturestring创建成功的签名串 = f"{self.pay_methon方法}\n{url_data}\n{utc_unix_timestamp}\n{random_str}\n{body_inner}\n"
|
||||
if self.pay_请求体 is None:
|
||||
body_inner = self.pay_请求体
|
||||
elif isinstance(self.pay_请求体, dict):
|
||||
body_inner = json.dumps(self.pay_请求体)
|
||||
elif self.pay_请求体 is None:
|
||||
body_inner = ""
|
||||
pay_signaturestring创建成功的签名串 = f"{self.pay_methon方法}\n{url_data}\n{utc_unix_timestamp}\n{random_str}\n{body_inner}\n"
|
||||
if isinstance(self.pay_请求体, bytes):
|
||||
elif isinstance(self.pay_请求体, bytes):
|
||||
body_inner = self.pay_请求体.decode()
|
||||
pay_signaturestring创建成功的签名串 = f"{self.pay_methon方法}\n{url_data}\n{utc_unix_timestamp}\n{random_str}\n{body_inner}\n"
|
||||
else:
|
||||
raise TypeError(f"不支持的请求体类型: {type(self.pay_请求体)}")
|
||||
pay_signaturestring创建成功的签名串 = f"{self.pay_methon方法}\n{url_data}\n{utc_unix_timestamp}\n{random_str}\n{body_inner}\n"
|
||||
|
||||
# 构造一个哈希对象
|
||||
hash_obj = SHA256.new(pay_signaturestring创建成功的签名串.encode(encoding="utf-8"))
|
||||
@@ -160,9 +204,8 @@ class sixWechatPay六果泡泡支付(BaseModel):
|
||||
}
|
||||
|
||||
def getpay_jsapi发起支付(self, ):
|
||||
headers = self.gouzhao构造签名串()
|
||||
# 发送请求
|
||||
response: Response = requests.post(url=self.pay_uri, json=self.pay_请求体, headers=headers)
|
||||
response: Response = requests.post(url=self.pay_uri, json=self.pay_请求体, headers=self.pay_Headers)
|
||||
prepay_id = pydash.get(obj=response.json(), path="prepay_id", default=None)
|
||||
# 验证请求后返回的数据是否为微信官方所发送;
|
||||
验证签名结果 = self.验证签名串(response_data=response.json(), response_headers=response.headers)
|
||||
@@ -225,21 +268,29 @@ class sixWechatPay六果泡泡支付(BaseModel):
|
||||
生成的支付链接 = response.json()
|
||||
return 生成的支付链接
|
||||
|
||||
def v2签名返回XML(self, ):
|
||||
def v2签名返回XML(self, sign_data):
|
||||
"""v2版本的签名：将扫码支付的body转换为微信支付所要求的数据类型"""
|
||||
sorted_dict = dict(pydash.sort_by(collection=self.pay_请求体.items(), iteratee=0))
|
||||
sorted_dict = dict(pydash.sort_by(collection=sign_data.items(), iteratee=0))
|
||||
# 拼接字符串: (将密钥添加到最末尾)
|
||||
params_list_str = pydash.join(array=[f"{k}={v}" for k, v in sorted_dict.items()], separator="&") + "&key=" + self.api_v2_key
|
||||
# 得到params_str就是需要 参与 签名的字符串
|
||||
signature = hashlib.md5(string=params_list_str.encode()).hexdigest().upper() # 使用MD5加密模式
|
||||
update后的数据 = pydash.set_(obj=self.pay_请求体, path="sign", value=signature)
|
||||
update后的数据 = pydash.set_(obj=sign_data, path="sign", value=signature)
|
||||
xml = dicttoxml(obj=update后的数据, custom_root="xml", xml_declaration=False, attr_type=False, cdata=True)
|
||||
print(type(xml))
|
||||
headers = self.gouzhao构造签名串()
|
||||
response = requests.post(url=self.pay_uri, data=xml, headers=headers)
|
||||
print(response.text)
|
||||
print(response.json())
|
||||
return
|
||||
return xml
|
||||
|
||||
def getpay_code扫用户付款码(self, body_data, ) -> dict:
|
||||
""" """
|
||||
xml = self.v2签名返回XML(sign_data=body_data)
|
||||
fu_kuan_url = "https://api.mch.weixin.qq.com/pay/micropay"
|
||||
res = requests.post(url=fu_kuan_url, data=xml)
|
||||
# 返回的数据 是 ISO-8859-1 类型
|
||||
responseStr = res.text.encode("ISO-8859-1")
|
||||
# 将请求得到的数据 转换为字典格式;
|
||||
xml_dict = xmltodict.parse(responseStr.decode())
|
||||
# 得到请求结果并进行解密
|
||||
print(xml_dict)
|
||||
return xml_dict
|
||||
|
||||
def 解密群号(self, encryptedData, iv, sessionKey):
|
||||
"""
|
||||
|
||||
Reference in New Issue
Block a user