""" Author: xiao 792282@qq.com Date: 2025-08-10 11:36:05 LastEditors: xiao 792282@qq.com LastEditTime: 2025-08-17 13:38:40 FilePath: django_小程序/同城配送工具/同城配送.py Description: 这是默认设置,可以在设置》工具》File Description中进行配置 """ import os import django from sixGuoDjango.settings import BASE_DIR from wechat_token.token工具包 import sixToken os.environ.setdefault(key="DJANGO_SETTINGS_MODULE", value=f"{BASE_DIR.name}.settings") django.setup() import json import time from dataclasses import dataclass, field from pathlib import Path import pydash from django.conf import settings from pydantic import BaseModel import base64 from requests import Response, post import orjson from Crypto.Cipher import AES from Crypto.Hash import SHA256 from Crypto.PublicKey import RSA from Crypto import Random from Crypto.Signature import pss public_key = Path(settings.BASE_DIR).joinpath("wechat_微信支付证书/江苏六果科技有限公司支付证书/public_key.pem").read_text() # public_key = Path("../../../wechat_微信支付证书/江苏六果科技有限公司支付证书/public_key.pem").resolve().read_text() private_key = Path(settings.BASE_DIR).joinpath("wechat_微信支付证书/江苏六果科技有限公司支付证书/private_key.pem").read_text() # private_key = Path("../../../wechat_微信支付证书/江苏六果科技有限公司支付证书/private_key.pem").resolve().read_text() 平台证书文本 = Path(settings.BASE_DIR).joinpath("wechat_微信支付证书/江苏六果科技有限公司支付证书/开放平台证书.pem").read_text() # 平台证书文本 = Path("../../../wechat_微信支付证书/江苏六果科技有限公司支付证书/开放平台证书.pem").resolve().read_text() class postData请求体类型(BaseModel): iv: str data: str authtag: str @dataclass(kw_only=True) class six同城加解密工具: """同城配送类加签,解密""" # add数据: bytes uri_urls: str data_body: dict _timestamp: int = field(init=False) uri_path: str = field(init=False) mini_token: str = field(init=False) add_GCM使用的认证数据_ADD数据: bytes = field(init=False) postData加密后的数据: postData请求体类型 = field(init=False) heardes_请求头数据: dict = field(init=False) aes_key对称密钥 = base64.b64decode("CV/RZev/1A1eg7TOgzvEOy5/to2aM8SE6Ab/R9E3BgI=") aes_str对称密钥 = "CV/RZev/1A1eg7TOgzvEOy5/to2aM8SE6Ab/R9E3BgI=" aes对称密钥编号 = "b02be3855ff9b990b2183505f165317e" rsa非对称密钥编号 = "442e1b521422ec70844a3272e5c857dd" api开放平台证书编号 = "5330e9294308d338a5e145fd03d0ea6a" 平台证书对象 = 平台证书文本 rsa_private = private_key rsa_public = public_key mini_appid = "wxf612e825377328f0" mini_AppSecret = "8691f774e3c0c5a876b21a3e84109ef0" def __post_init__(self): self.mini_token = sixToken.get_mini_token() self._timestamp = int(time.time()) # self.本地解密( ) def 加密(self, ): """ :return: :rtype: """ iv_random_bytes = Random.get_random_bytes(12) iv_base64_12 = base64.b64encode(iv_random_bytes).decode() random_bytes_n = Random.get_random_bytes(16) _n = base64.b64encode(random_bytes_n).decode() # 随机字符串，_n的值 推荐使用16-32 字节非固定长度随机base64字符串 _appid = self.mini_appid # 当前小程序的Appid self.add_GCM使用的认证数据_ADD数据 = (f"{self.uri_urls}|{self.mini_appid}|{str(self._timestamp).strip()}|" f"{self.aes对称密钥编号}").encode(encoding="utf-8") data明文 = { "_n": _n, "_appid": self.mini_appid, "_timestamp": self._timestamp, } data明文格式 = dict(data明文, **self.data_body) data明文格式_bytes = orjson.dumps(data明文格式).decode() cliper = AES.new(key=self.aes_key对称密钥, mode=AES.MODE_GCM, nonce=iv_random_bytes) """ AAD 1：攻击者无法修改 AAD 而不被发现（解密时会验证失败）。 2：确保加密数据与特定请求关联（如 API 路径、AppID 等）。 3: 不加密但受保护：AAD 本身不加密，但它的完整性会被 MAC 验证。 """ cliper.update(assoc_data=self.add_GCM使用的认证数据_ADD数据) # real_ciphertext, real_authTag = cliper.encrypt_and_digest(data明文格式_bytes.encode()) data_base64 = base64.b64encode(real_ciphertext).decode() authtag_base64 = base64.b64encode(real_authTag).decode() self.postData加密后的数据 = postData请求体类型(**{"iv": iv_base64_12, "data": data_base64, "authtag": authtag_base64, }) self.本地解密() # 测试调用一下本地解密,是否有效; print("这里会报错的吗") def 本地解密(self, ): """ 解密功能 :return: :rtype: """ # 将base64字符串解码; iv_bytes = base64.b64decode(self.postData加密后的数据.iv) # 将base64字符串解码; ciphertext = base64.b64decode(self.postData加密后的数据.data) # 将base64字符串解码; cipher = AES.new(self.aes_key对称密钥, AES.MODE_GCM, nonce=iv_bytes) cipher.update(self.add_GCM使用的认证数据_ADD数据) try: padded_data = cipher.decrypt(ciphertext) return except ValueError: raise ValueError("解密出错了") def 对加密数据进行签名(self, ): """ 对返回的加密数据进行解密; """ # 转换为紧凑 JSON 格式 postdata = json.dumps(self.postData加密后的数据.model_dump(), ensure_ascii=False, ) # 构造签名串 sign_str = f"{self.uri_urls}\n{self.mini_appid}\n{self._timestamp}\n{postdata}" # RSA 签名 private_key_obj = RSA.import_key(self.rsa_private) h = SHA256.new(sign_str.encode()) # ✅ 改为 PSS 签名 signature = pss.new(private_key_obj).sign(h) sig = base64.b64encode(signature).decode() self.heardes_请求头数据 = { "Wechatmp-Appid": self.mini_appid, "Wechatmp-TimeStamp": str(self._timestamp), "Wechatmp-Signature": sig, "Content-Type": "application/json", "Accept": "application/json", } def 验证签名串(self, respose: Response): """""" # 获取原始响应文本（避免任何自动转换，我们使用response.content然后手动解码为utf-8） Wechatmp_TimeStamp = pydash.get(obj=respose.headers, path="Wechatmp-TimeStamp", default=None) Wechatmp_Appid = pydash.get(obj=respose.headers, path="Wechatmp-Appid", default=None) Wechatmp_Signature = pydash.get(obj=respose.headers, path="Wechatmp-Signature", default=None) # 2. 获取原始响应体（非常重要） raw_body = respose.text.strip() # 去除首尾空白，确保无额外\n 创建拼接字答串 = f"{self.uri_urls}\n{Wechatmp_Appid}\n{str(Wechatmp_TimeStamp)}\n{raw_body}" signature = base64.b64decode(Wechatmp_Signature) msg_hash = SHA256.new(创建拼接字答串.encode(encoding="utf-8")) public_obj = RSA.import_key(self.平台证书对象) # 创建PSS签名验证器 try: pss.new(rsa_key=public_obj).verify(msg_hash=msg_hash, signature=signature) # type: ignore print("验证成功了") return True except Exception as e: print(f"❌ 签名验证失败: {str(e)}") raise e def 解签(self, respose: Response): Wechatmp_TimeStamp = pydash.get(obj=respose.headers, path="Wechatmp-TimeStamp", default=None) Wechatmp_Appid = pydash.get(obj=respose.headers, path="Wechatmp-Appid", default=None) # Wechatmp_Signature 平台证书签名数据，使用base64编码 self.验证签名串(respose=respose) # ✅ 强制重新构造 AAD，确保与加密端一致 add_data = f"{self.uri_urls}|{Wechatmp_Appid}|{Wechatmp_TimeStamp}|{self.aes对称密钥编号}" iv = pydash.get(obj=respose.json(), path="iv", default=None) authtag = pydash.get(obj=respose.json(), path="authtag", default=None) data = pydash.get(obj=respose.json(), path="data", default=None) iv_bytes = base64.b64decode(iv) ciphertext = base64.b64decode(data) cipher = AES.new(self.aes_key对称密钥, AES.MODE_GCM, nonce=iv_bytes) cipher.update(add_data.encode()) # ✅ 使用新构造的 AAD try: padded_data = cipher.decrypt(ciphertext) # 只能调用decrypt这个方法() deciphertext = padded_data.decode() return deciphertext except ValueError as e: raise ValueError("解密出错了，请检查 AAD、IV、密文或密钥一致性") from e def call(self): """""" self.加密() self.对加密数据进行签名() data = post(url=self.uri_urls, params={"access_token": self.mini_token, }, headers=self.heardes_请求头数据, json=self.postData加密后的数据.model_dump()) print(data.json(), "请求出错了吗") return self.解签(respose=data) if __name__ == "__main__": # six同城加解密工具.对加密数据进行签名( urlpath="https://api.weixin.qq.com/wxa/getuserriskrank", data_body={ } ) # # # six = six同城加解密工具(uri_urls="https://api.weixin.qq.com/cgi-bin/express/intracity/querystore", # data_body={}).call() # print(six) 查询运费请求数据 = { "wx_store_id": "4000000000016308013", "user_lng": "119.893083", "user_lat": "32.474616", "user_address": "江苏省泰州市海陵区城西街道任景路U牌美容店", "user_name": "元宝", "user_phone": "13372551360", "cargo": {"cargo_name": "测试商品", "cargo_type": 3, "cargo_num": 7, "cargo_price": 5000, "cargo_weight": 2500} } res = six同城加解密工具(uri_urls="https://api.weixin.qq.com/cgi-bin/express/intracity/preaddorder", data_body=查询运费请求数据).call() print(res)