marco
5271d22d24
- GET 接入验证（校验 signature → 原样返回 echostr），供公众平台配置服务器 URL - POST 收消息（安全模式：msg_signature 验签 + AES-256-CBC 解密），图片消息用 PicUrl 落盘 - config 加 WX_MP_TOKEN / WX_MP_AES_KEY + wx_mp_callback_configured - 识别平台订单/pending标记/openid↔device 绑定留后续 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
70 lines
2.7 KiB
Python
70 lines
2.7 KiB
Python
"""微信服务号消息接收回调的验签与解密(安全模式)。
|
|
|
|
服务号"服务器配置"选安全模式后:
|
|
- URL 接入验证(GET): sha1(sort(token, timestamp, nonce)) == signature → 原样返回 echostr
|
|
- 消息(POST): body 里 <Encrypt> 是密文;
|
|
msg_signature = sha1(sort(token, timestamp, nonce, encrypt))
|
|
密文 AES-256-CBC 解出: random(16B) + msg_len(4B big-endian) + msg + appid, 去 PKCS7(块=32) padding
|
|
|
|
只做验签 + 解密(收消息)。被动回复(加密)MVP 不需要 —— 收到后走客服消息异步回执。
|
|
密钥/口令来自 settings(WX_MP_TOKEN / WX_MP_AES_KEY / WX_MP_APPID),本模块只做纯算法、不读配置。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
|
|
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
|
|
|
|
def verify_url_signature(token: str, timestamp: str, nonce: str, signature: str) -> bool:
|
|
"""GET 接入验证:token/timestamp/nonce 三者字典序排序拼接后 sha1。"""
|
|
return _consteq(_sha1(token, timestamp, nonce), signature)
|
|
|
|
|
|
def verify_msg_signature(
|
|
token: str, timestamp: str, nonce: str, encrypt: str, msg_signature: str
|
|
) -> bool:
|
|
"""POST 消息验签:四者(含密文 encrypt)字典序排序拼接后 sha1。"""
|
|
return _consteq(_sha1(token, timestamp, nonce, encrypt), msg_signature)
|
|
|
|
|
|
def decrypt_message(aes_key_b64: str, expected_appid: str, encrypt_b64: str) -> str:
|
|
"""解密 <Encrypt> 密文, 返回明文消息 XML。appid 不符抛 ValueError。
|
|
|
|
aes_key_b64: EncodingAESKey(43 位, 不含结尾 '='), 补 '=' 后 base64 解出 32 字节 AES-256 key。
|
|
"""
|
|
aes_key = base64.b64decode(aes_key_b64 + "=") # 43 → 32 bytes
|
|
iv = aes_key[:16]
|
|
decryptor = Cipher(algorithms.AES(aes_key), modes.CBC(iv)).decryptor()
|
|
plain = decryptor.update(base64.b64decode(encrypt_b64)) + decryptor.finalize()
|
|
plain = _pkcs7_unpad(plain)
|
|
|
|
# random(16) + msg_len(4, big-endian) + msg(msg_len) + from_appid
|
|
content = plain[16:]
|
|
msg_len = int.from_bytes(content[:4], "big")
|
|
msg = content[4 : 4 + msg_len]
|
|
from_appid = content[4 + msg_len :].decode("utf-8")
|
|
if expected_appid and from_appid != expected_appid:
|
|
raise ValueError(f"appid mismatch: {from_appid!r} != {expected_appid!r}")
|
|
return msg.decode("utf-8")
|
|
|
|
|
|
def _sha1(*parts: str) -> str:
|
|
return hashlib.sha1("".join(sorted(parts)).encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _consteq(a: str, b: str) -> bool:
|
|
return hmac.compare_digest(a, b)
|
|
|
|
|
|
def _pkcs7_unpad(data: bytes) -> bytes:
|
|
"""微信用块大小 32 的 PKCS7,末字节即 padding 长度(1..32)。越界则原样返回(容错)。"""
|
|
if not data:
|
|
return data
|
|
pad = data[-1]
|
|
if pad < 1 or pad > 32:
|
|
return data
|
|
return data[:-pad]
|