141 lines
5.5 KiB
Python
141 lines
5.5 KiB
Python
"""
|
||
企微回调加解密（WXBizMsgCrypt 语义）。
|
||
|
||
本文件职责：URL 验签、密文解密、回包加密；密钥只从 Settings/环境变量读入。
|
||
禁止：把 Token/AESKey 写进源码或注释；禁止在本文件发业务消息。
|
||
算法依据：企业微信官方回调加解密文档（AES-CBC + SHA1 签名）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import hashlib
|
||
import logging
|
||
import os
|
||
import struct
|
||
import time
|
||
|
||
from cryptography.hazmat.backends import default_backend
|
||
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class WeComCryptoError(Exception):
|
||
"""验签或加解密失败；回调应拒绝入队。"""
|
||
|
||
|
||
class WeComCrypto:
|
||
"""
|
||
企微消息加解密器。
|
||
|
||
参数：
|
||
token：回调 Token（WECOM_CALLBACK_TOKEN）
|
||
encoding_aes_key：43 字符 EncodingAESKey（WECOM_ENCODING_AES_KEY）
|
||
corp_id：企业 CorpId（WECOM_CORP_ID），解密后 receiveid 须匹配
|
||
线程：无共享可变状态，可多线程复用同一实例。
|
||
"""
|
||
|
||
def __init__(self, token: str, encoding_aes_key: str, corp_id: str) -> None:
|
||
if not token or not encoding_aes_key or not corp_id:
|
||
raise WeComCryptoError("企微加解密三件套未配置完整（token/aes_key/corp_id）")
|
||
self._token = token
|
||
self._corp_id = corp_id
|
||
# EncodingAESKey 官方为 43 字符，补 '=' 后 base64 得到 32 字节 AES Key
|
||
try:
|
||
self._aes_key = base64.b64decode(encoding_aes_key + "=")
|
||
except Exception as exc: # noqa: BLE001
|
||
raise WeComCryptoError("EncodingAESKey 非法") from exc
|
||
if len(self._aes_key) != 32:
|
||
raise WeComCryptoError("EncodingAESKey 解码后长度必须为 32 字节")
|
||
|
||
def verify_signature(self, msg_signature: str, timestamp: str, nonce: str, encrypt: str) -> None:
|
||
"""
|
||
校验 msg_signature = sha1(sort(token, timestamp, nonce, encrypt))。
|
||
|
||
失败抛 WeComCryptoError；成功无返回值。
|
||
"""
|
||
expected = self._sha1_signature(timestamp, nonce, encrypt)
|
||
if expected != msg_signature:
|
||
raise WeComCryptoError("msg_signature 校验失败")
|
||
|
||
def decrypt(self, encrypt_b64: str) -> str:
|
||
"""
|
||
解密 Encrypt 字段，返回明文 XML/JSON 字符串。
|
||
|
||
副作用：无外部 I/O。receiveid 必须等于 corp_id，否则视为串包攻击。
|
||
"""
|
||
try:
|
||
cipher_bytes = base64.b64decode(encrypt_b64)
|
||
except Exception as exc: # noqa: BLE001
|
||
raise WeComCryptoError("Encrypt base64 解码失败") from exc
|
||
|
||
iv = self._aes_key[:16]
|
||
decryptor = Cipher(
|
||
algorithms.AES(self._aes_key),
|
||
modes.CBC(iv),
|
||
backend=default_backend(),
|
||
).decryptor()
|
||
plain = decryptor.update(cipher_bytes) + decryptor.finalize()
|
||
pad = plain[-1]
|
||
if isinstance(pad, str): # pragma: no cover
|
||
pad = ord(pad)
|
||
content = plain[:-pad]
|
||
# 16 字节随机 + 4 字节网络序长度 + msg + receiveid
|
||
msg_len = struct.unpack("!I", content[16:20])[0]
|
||
xml = content[20 : 20 + msg_len].decode("utf-8")
|
||
from_receive_id = content[20 + msg_len :].decode("utf-8")
|
||
if from_receive_id != self._corp_id:
|
||
raise WeComCryptoError("解密 receiveid 与 CorpId 不匹配")
|
||
return xml
|
||
|
||
def encrypt(self, plain: str) -> str:
|
||
"""
|
||
加密明文，返回 base64 Encrypt，供被动回包使用。
|
||
|
||
为何仍实现：URL 验证只需解密；部分场景被动回复需加密（本迭代主动出站走 API）。
|
||
"""
|
||
random16 = os.urandom(16)
|
||
text = plain.encode("utf-8")
|
||
corp = self._corp_id.encode("utf-8")
|
||
body = random16 + struct.pack("!I", len(text)) + text + corp
|
||
# PKCS#7 填充到 32 字节块
|
||
amount = 32 - (len(body) % 32)
|
||
body = body + bytes([amount] * amount)
|
||
iv = self._aes_key[:16]
|
||
encryptor = Cipher(
|
||
algorithms.AES(self._aes_key),
|
||
modes.CBC(iv),
|
||
backend=default_backend(),
|
||
).encryptor()
|
||
cipher = encryptor.update(body) + encryptor.finalize()
|
||
return base64.b64encode(cipher).decode("utf-8")
|
||
|
||
def verify_url(self, msg_signature: str, timestamp: str, nonce: str, echostr: str) -> str:
|
||
"""
|
||
处理企微「保存 URL」验证：验签并解密 echostr，返回明文给 HTTP 响应体。
|
||
"""
|
||
self.verify_signature(msg_signature, timestamp, nonce, echostr)
|
||
return self.decrypt(echostr)
|
||
|
||
def build_encrypted_response(self, plain: str, nonce: str | None = None) -> str:
|
||
"""
|
||
构造加密被动回复 XML（当前 echo 主路径走主动发消息，此方法备用）。
|
||
"""
|
||
nonce = nonce or str(int(time.time()))
|
||
timestamp = str(int(time.time()))
|
||
encrypt = self.encrypt(plain)
|
||
signature = self._sha1_signature(timestamp, nonce, encrypt)
|
||
return (
|
||
"<xml>"
|
||
f"<Encrypt><![CDATA[{encrypt}]]></Encrypt>"
|
||
f"<MsgSignature><![CDATA[{signature}]]></MsgSignature>"
|
||
f"<TimeStamp>{timestamp}</TimeStamp>"
|
||
f"<Nonce><![CDATA[{nonce}]]></Nonce>"
|
||
"</xml>"
|
||
)
|
||
|
||
def _sha1_signature(self, timestamp: str, nonce: str, encrypt: str) -> str:
|
||
items = sorted([self._token, timestamp, nonce, encrypt])
|
||
return hashlib.sha1("".join(items).encode("utf-8")).hexdigest()
|