223 lines
10 KiB
Python
223 lines
10 KiB
Python
"""
|
||
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)
|