six_001
This commit is contained in:
doudou201703
@@ -0,0 +1,222 @@
|
||||
"""
|
||||
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)
|
||||
Reference in New Issue
Block a user