import base64 import hashlib import json import os import re import secrets from datetime import datetime from pathlib import Path from typing import Literal, Optional from zoneinfo import ZoneInfo import django import orjson import pydash import requests from Crypto.Cipher import AES from Crypto.Hash import SHA256 from Crypto.PublicKey import RSA from Crypto.PublicKey.RSA import RsaKey from Crypto.Signature import pkcs1_15 from Crypto.Util.Padding import unpad from dicttoxml import dicttoxml from django.conf import settings from pydantic import BaseModel, Field, SecretStr from requests import Response from sixGuoDjango.settings import base_dir_name os.environ.setdefault(key="DJANGO_SETTINGS_MODULE", value=f"{base_dir_name}.settings") django.setup() class sixWechatPay六果泡泡支付(BaseModel): mchid: str = Field(default="1700383419", alias="商户号", description="微信支付商户号") appid: str = Field(default="wxf612e825377328f0", alias="商户号", description="微信支付商户号") api_v3_key: str = Field(default="YWMGAIPTJTWZN4YB3YUTHWK34K9U9M1W", alias="APIv3密钥", description="用于回调解密") api_v2_key: str = Field(default="OXRNKGGUYHGNJSSIHKFGVMAAOHRPXJAT", alias="APIv2密钥", description="仅老接口需要") serial_no商户证书的序列号: str = Field(default="530160D54E8F56726F506DEE40D7FA49A558E8E0", description="商户证书的序列号") # ---------- 小程序密钥 ---------- mini_program_app_secret: SecretStr = Field(default="8691f774e3c0c5a876b21a3e84109ef0", alias="小程序AppSecret") # ---------- 商户API密钥对（用于签名请求） ---------- merchant_private_key_pem: str = Field(default=Path(settings.BASE_DIR).joinpath("wechat_微信支付证书/江苏六果科技有限公司支付证书/apiclient_key.pem").read_text(), alias="商户私钥PEM", description="apiclient_key.pem 内容") merchant_public_key_pem: str = Field(default=Path(settings.BASE_DIR).joinpath("wechat_微信支付证书/江苏六果科技有限公司支付证书/微信支付公钥.pem").read_text(), alias="商户公钥PEM", description="apiclient_cert.pem 内容（很少用到）") # ---------- 微信支付公钥（用于验签和加密敏感字段） ---------- 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="发请求签名时需要的请求体") pay_methon方法: Literal["GET", "POST", "PATCH"] = Field(default="POST", description="发请求签名时需要的请求体") def model_post_init(self, __context): pass 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) Wechatpay_Timestamp = pydash.get(response_headers, "Wechatpay-Timestamp") Wechatpay_Nonce = pydash.get(response_headers, "Wechatpay-Nonce") Wechatpay_Signa = pydash.get(response_headers, "Wechatpay-Signature") 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. 验证签名 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 # 签名验证失败 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创建成功的签名串 = "" # 创建签名请求参数 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 = "" 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): 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" # 构造一个哈希对象 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.payment.微信支付公钥ID}) return headers def 返回前端支付数据(self, prepay_id: str, wechat_appid: str): """ 小程序调起支付 通过JSAPI下单接口获取到发起支付的必要参数prepay_id，然后使用微信支付提供的小程序方法调起小程序支付。 @param prepay_ids @type prepay_id: @return: @rtype: """ utc_unix_timestamp = int(datetime.now(tz=ZoneInfo("utc")).timestamp()) random_string = secrets.token_hex(nbytes=30) # 随机字符串 # 组装微信官方要求的格式; qian_mian_chuan = f"{wechat_appid}\n{utc_unix_timestamp}\n{random_string}\n{prepay_id}\n" # # 使用组装好地签名串,进行sha256 哈希生成哈希对象 hash_obj = SHA256.new(qian_mian_chuan.encode()) # 生成商户私钥对象 # private_key = RSA.import_key( wechat_config.wechat私钥 ) private_key = RSA.import_key(self.merchant_private_key_pem) # 生成签名 signature = pkcs1_15.new(private_key).sign(hash_obj) # 将生成的签名串,转换为base格式 encoded_signature = base64.b64encode(signature).decode() return { "appId": self.appid, "timeStamp": str(utc_unix_timestamp), "nonceStr": random_string, "package": prepay_id, "signType": "RSA", "paySign": encoded_signature, } def getpay_jsapi发起支付(self, ): headers = self.gouzhao构造签名串() # 发送请求 response: Response = requests.post(url=self.pay_uri, json=self.pay_请求体, headers=headers) prepay_id = pydash.get(obj=response.json(), path="prepay_id", default=None) # 验证请求后返回的数据是否为微信官方所发送; 验证签名结果 = self.验证签名串(response_data=response.json(), response_headers=response.headers) if 验证签名结果: prepay_id = f"prepay_id={prepay_id}" # 调用方法生成返回给前端的对象 return self.返回前端支付数据(prepay_id=prepay_id, wechat_appid=self.appid, ) # 调用 返回前端支付数据()方法生成返回前端的支付字典 else: raise "验证不通过" def modegmc解密(self, nonce: str, ciphertext: str, associated_data: str, ) -> dict: """ 支付成功或者退款回调时需要解密参数,微信返回的响应体.解密后可以获取支付成功或者退款的详细的数据; @param nonce: // 加密使用的随机串初始化向量 @type nonce: @param ciphertext: // Base64编码后的密文(需要解密) @type ciphertext: (需要解密) @param associated_data: // 附加数据包（可能为空） @type associated_data: @return: @rtype: """ nonce_bytes = nonce.encode(encoding="utf-8") ad_bytes = associated_data.encode(encoding="utf-8") encrypted_data = base64.b64decode(ciphertext) cipher = AES.new(key=self.api_v3_key.encode(), mode=AES.MODE_GCM, nonce=nonce_bytes) # 处理附加关联数据(AAD) - 微信支付中通常是"transaction" # 这部分数据在加密过程中被验证但未被加密 # 必须与加密时使用的附加数据完全一致 cipher.update(assoc_data=ad_bytes) auth_tag = encrypted_data[-16:] # 提取认证标签（最后16 字节） # 提取实际加密内容 ciphertext_bytes = encrypted_data[:-16] # 提取实际加密内容 decrypted_data = cipher.decrypt_and_verify(ciphertext=ciphertext_bytes, received_mac_tag=auth_tag) # 回调解密数据: dict = json.loads(decrypted_data.decode()) 回调解密数据: dict = orjson.loads(decrypted_data.decode(encoding="utf-8")) return 回调解密数据 def callBack回调解密(self, headers, response_body: dict) -> dict: """ 执行微信支付回调方法(获取返回的参数数据) @param request: @type request: """ self.验证签名串(response_headers=headers, response_data=response_body) ciphertext = pydash.get(obj=response_body, path="resource.ciphertext", default=None) associated_data = pydash.get(obj=response_body, path="resource.associated_data", default=None) nonce = pydash.get(obj=response_body, path="resource.nonce", default=None) wei_xin_zhi_fu_hui_diao_data = self.modegmc解密( nonce=nonce, ciphertext=ciphertext, associated_data=associated_data, ) return wei_xin_zhi_fu_hui_diao_data def getpay_navite发起二维码支付(self, ): """ 二维码支付,可以适用于 网站 的用户扫码支付 """ headers = self.gouzhao构造签名串() response = requests.post(url=self.pay_uri, headers=headers, json=self.pay_请求体) 生成的支付链接 = response.json() return 生成的支付链接 def v2签名返回XML(self, ): """v2版本的签名：将扫码支付的body转换为微信支付所要求的数据类型""" sorted_dict = dict(pydash.sort_by(collection=self.pay_请求体.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) 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 def 解密群号(self, encryptedData, iv, sessionKey): """ """ try: session_key_bytes = base64.b64decode(sessionKey) encrypted_data_bytes = base64.b64decode(encryptedData) iv_bytes = base64.b64decode(iv) cipher = AES.new(key=session_key_bytes, mode=AES.MODE_CBC, iv=iv_bytes) # 解密并去除填充 decrypted_data = unpad(cipher.decrypt(encrypted_data_bytes), AES.block_size).decode() data_dict = orjson.loads(decrypted_data) openGId = pydash.get(obj=data_dict, path="openGId") timestamp = pydash.get(obj=data_dict, path="watermark.timestamp") appid = pydash.get(obj=data_dict, path="watermark.appid") return openGId, timestamp, appid except Exception as e: raise Exception(f"解密失败: {str(e)}") def 发起退款(self, ): pass uri = f"https://api.mch.weixin.qq.com/v3/refund/domestic/refunds" heardes = self.gouzhao构造签名串() print(heardes) response_data = requests.post(url=self.pay_uri, json=self.pay_请求体, headers=heardes) print(response_data.json()) if __name__ == "__main__": pass