92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
import base64
|
||
import hashlib
|
||
import os
|
||
import struct
|
||
from typing import Tuple
|
||
|
||
from Crypto.Cipher import AES
|
||
|
||
|
||
BLOCK_SIZE = 32 # PKCS5/PKCS7 padding 使用 32 字节块大小
|
||
|
||
|
||
def pkcs5_unpadding(data: bytes) -> bytes:
|
||
pad = data[-1]
|
||
if pad < 1 or pad > 32:
|
||
return data
|
||
return data[:-pad]
|
||
|
||
|
||
def pkcs5_padding(data: bytes) -> bytes:
|
||
amount_to_pad = BLOCK_SIZE - (len(data) % BLOCK_SIZE)
|
||
return data + bytes([amount_to_pad]) * amount_to_pad
|
||
|
||
|
||
def get_aes_key_and_iv(encoding_aes_key: str) -> Tuple[bytes, bytes]:
|
||
"""
|
||
encoding_aes_key 是平台给的 43 位字符串，需要先补一个 '=' 再 base64 解码。
|
||
得到 32 字节 key，前 16 字节为 iv。
|
||
"""
|
||
aes_key = base64.b64decode(encoding_aes_key + "=")
|
||
if len(aes_key) != 32:
|
||
raise ValueError("encodingAESKey 无效，解码后长度不是 32 字节")
|
||
iv = aes_key[:16]
|
||
return aes_key, iv
|
||
|
||
|
||
def aes_decrypt_base64(cipher_b64: str, encoding_aes_key: str) -> str:
|
||
aes_key, iv = get_aes_key_and_iv(encoding_aes_key)
|
||
# 回调密文偶尔会带换行/空白，先规整再补齐 base64 padding。
|
||
compact = "".join(str(cipher_b64 or "").split())
|
||
compact += "=" * (-len(compact) % 4)
|
||
cipher_bytes = base64.b64decode(compact)
|
||
cipher = AES.new(aes_key, AES.MODE_CBC, iv)
|
||
decrypted = cipher.decrypt(cipher_bytes)
|
||
decrypted = pkcs5_unpadding(decrypted)
|
||
# 兼容两种格式：
|
||
# 1) 微信消息体：16B随机串 + 4B长度 + 明文 + appid
|
||
# 2) 纯文本：直接是明文 JSON/XML
|
||
try:
|
||
if len(decrypted) < 20:
|
||
raise ValueError("payload too short")
|
||
content = decrypted[16:]
|
||
msg_len = struct.unpack("!I", content[:4])[0]
|
||
msg_start = 4
|
||
msg_end = msg_start + msg_len
|
||
if msg_end > len(content):
|
||
raise ValueError("msg length out of range")
|
||
msg = content[msg_start:msg_end]
|
||
text = msg.decode("utf-8")
|
||
if text:
|
||
return text
|
||
except Exception:
|
||
pass
|
||
return decrypted.decode("utf-8")
|
||
|
||
|
||
def aes_encrypt_base64(plaintext: str, encoding_aes_key: str, app_id: str | None = None) -> str:
|
||
aes_key, iv = get_aes_key_and_iv(encoding_aes_key)
|
||
raw = plaintext.encode("utf-8")
|
||
if app_id:
|
||
# 微信消息体格式：16B随机串 + 4B网络序长度 + 明文 + appid
|
||
data = os.urandom(16) + struct.pack("!I", len(raw)) + raw + app_id.encode("utf-8")
|
||
else:
|
||
# 纯明文格式
|
||
data = raw
|
||
padded = pkcs5_padding(data)
|
||
cipher = AES.new(aes_key, AES.MODE_CBC, iv)
|
||
encrypted = cipher.encrypt(padded)
|
||
return base64.b64encode(encrypted).decode("utf-8")
|
||
|
||
|
||
def calc_signature(token: str, timestamp: int, skill_name: str, intent_name: str, query: str) -> str:
|
||
"""
|
||
文档签名算法：
|
||
str = token + time.Now.Unix + skill_name + intent_name + query
|
||
signature = MD5(str) 小写 hex
|
||
参考：https://developers.weixin.qq.com/doc/aispeech/confapi/thirdapi/thirdapi.html
|
||
"""
|
||
s = f"{token}{timestamp}{skill_name}{intent_name}{query}"
|
||
return hashlib.md5(s.encode("utf-8")).hexdigest()
|
||
|