236 lines
8.0 KiB
Python
236 lines
8.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
企业微信消息加解密工具
|
|
基于企业微信官方加解密方案实现（AES-CBC-256 + SHA1签名）
|
|
参考文档: https://developer.work.weixin.qq.com/document/path/90968
|
|
"""
|
|
|
|
import base64
|
|
import hashlib
|
|
import random
|
|
import struct
|
|
import time
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
from cryptography.hazmat.backends import default_backend
|
|
|
|
import logging
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
class WXBizMsgCryptError(Exception):
|
|
"""企业微信加解密异常"""
|
|
pass
|
|
|
|
|
|
class PKCS7Encoder:
|
|
"""PKCS7 填充/去除"""
|
|
block_size = 32
|
|
|
|
@classmethod
|
|
def encode(cls, text):
|
|
"""对需要加密的明文进行填充补位"""
|
|
text_length = len(text)
|
|
amount_to_pad = cls.block_size - (text_length % cls.block_size)
|
|
if amount_to_pad == 0:
|
|
amount_to_pad = cls.block_size
|
|
pad = chr(amount_to_pad).encode()
|
|
return text + pad * amount_to_pad
|
|
|
|
@classmethod
|
|
def decode(cls, decrypted):
|
|
"""去除填充补位"""
|
|
pad = decrypted[-1]
|
|
if pad < 1 or pad > cls.block_size:
|
|
pad = 0
|
|
return decrypted[:-pad]
|
|
|
|
|
|
class WXBizMsgCrypt:
|
|
"""
|
|
企业微信消息加解密类
|
|
|
|
用法:
|
|
crypt = WXBizMsgCrypt(token, encoding_aes_key, corpid)
|
|
# URL验证
|
|
echostr_plain = crypt.verify_url(msg_signature, timestamp, nonce, echostr)
|
|
# 解密消息
|
|
xml_content = crypt.decrypt_msg(msg_signature, timestamp, nonce, post_data)
|
|
# 加密消息
|
|
encrypted_xml = crypt.encrypt_msg(reply_msg, nonce, timestamp)
|
|
"""
|
|
|
|
def __init__(self, token, encoding_aes_key, corpid):
|
|
"""
|
|
:param token: 接收消息服务器配置的Token
|
|
:param encoding_aes_key: 43位的EncodingAESKey
|
|
:param corpid: 企业ID
|
|
"""
|
|
self.token = token
|
|
self.corpid = corpid
|
|
try:
|
|
self.aes_key = base64.b64decode(encoding_aes_key + "=")
|
|
except Exception as e:
|
|
raise WXBizMsgCryptError(f"EncodingAESKey 无效: {e}")
|
|
if len(self.aes_key) != 32:
|
|
raise WXBizMsgCryptError("EncodingAESKey 解码后长度应为32字节")
|
|
|
|
def _generate_signature(self, timestamp, nonce, encrypt):
|
|
"""生成消息签名"""
|
|
sort_list = sorted([self.token, timestamp, nonce, encrypt])
|
|
sha1 = hashlib.sha1()
|
|
sha1.update("".join(sort_list).encode('utf-8'))
|
|
return sha1.hexdigest()
|
|
|
|
def _encrypt(self, text):
|
|
"""对明文进行加密"""
|
|
# 16字节随机字符串
|
|
random_str = ''.join(random.choices(
|
|
'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789', k=16
|
|
)).encode('utf-8')
|
|
text = text.encode('utf-8')
|
|
# 拼接: random(16) + msg_len(4) + msg + corpid
|
|
content = random_str + struct.pack("!I", len(text)) + text + self.corpid.encode('utf-8')
|
|
# PKCS7填充
|
|
content = PKCS7Encoder.encode(content)
|
|
# AES-CBC加密
|
|
cipher = Cipher(algorithms.AES(self.aes_key), modes.CBC(self.aes_key[:16]), backend=default_backend())
|
|
encryptor = cipher.encryptor()
|
|
encrypted = encryptor.update(content) + encryptor.finalize()
|
|
return base64.b64encode(encrypted).decode('utf-8')
|
|
|
|
def _decrypt(self, encrypted):
|
|
"""对密文进行解密"""
|
|
try:
|
|
cipher = Cipher(algorithms.AES(self.aes_key), modes.CBC(self.aes_key[:16]), backend=default_backend())
|
|
decryptor = cipher.decryptor()
|
|
decrypted = decryptor.update(base64.b64decode(encrypted)) + decryptor.finalize()
|
|
except Exception as e:
|
|
raise WXBizMsgCryptError(f"AES解密失败: {e}")
|
|
|
|
# 去除PKCS7填充
|
|
decrypted = PKCS7Encoder.decode(decrypted)
|
|
try:
|
|
# 去掉16字节随机字符串
|
|
# 取4字节消息长度
|
|
msg_len = struct.unpack("!I", decrypted[16:20])[0]
|
|
# 提取消息内容
|
|
msg = decrypted[20:20 + msg_len]
|
|
# 提取corpid
|
|
from_corpid = decrypted[20 + msg_len:]
|
|
except Exception as e:
|
|
raise WXBizMsgCryptError(f"消息格式解析失败: {e}")
|
|
|
|
# 验证corpid
|
|
if from_corpid.decode('utf-8') != self.corpid:
|
|
raise WXBizMsgCryptError(
|
|
f"CorpID不匹配: 期望 {self.corpid}, 实际 {from_corpid.decode('utf-8')}"
|
|
)
|
|
return msg.decode('utf-8')
|
|
|
|
def verify_url(self, msg_signature, timestamp, nonce, echostr):
|
|
"""
|
|
验证URL有效性（企业微信配置回调地址时的验证请求）
|
|
|
|
:param msg_signature: 签名串
|
|
:param timestamp: 时间戳
|
|
:param nonce: 随机串
|
|
:param echostr: 加密的随机串（需解密后返回）
|
|
:return: 解密后的echostr明文
|
|
"""
|
|
# 验证签名
|
|
signature = self._generate_signature(timestamp, nonce, echostr)
|
|
if signature != msg_signature:
|
|
raise WXBizMsgCryptError("URL验证签名不匹配")
|
|
# 解密echostr
|
|
return self._decrypt(echostr)
|
|
|
|
def decrypt_msg(self, msg_signature, timestamp, nonce, post_data):
|
|
"""
|
|
解密企业微信推送的加密消息
|
|
|
|
:param msg_signature: 签名串
|
|
:param timestamp: 时间戳
|
|
:param nonce: 随机串
|
|
:param post_data: POST请求的XML原文
|
|
:return: 解密后的XML明文
|
|
"""
|
|
try:
|
|
xml_tree = ET.fromstring(post_data)
|
|
encrypt = xml_tree.find("Encrypt").text
|
|
except Exception as e:
|
|
raise WXBizMsgCryptError(f"XML解析失败: {e}")
|
|
|
|
# 验证签名
|
|
signature = self._generate_signature(timestamp, nonce, encrypt)
|
|
if signature != msg_signature:
|
|
raise WXBizMsgCryptError("消息签名验证失败")
|
|
|
|
# 解密
|
|
return self._decrypt(encrypt)
|
|
|
|
def encrypt_msg(self, reply_msg, nonce, timestamp=None):
|
|
"""
|
|
加密回复消息（被动回复时使用）
|
|
|
|
:param reply_msg: 回复消息XML明文
|
|
:param nonce: 随机串
|
|
:param timestamp: 时间戳（可选，默认当前时间）
|
|
:return: 加密后的XML字符串
|
|
"""
|
|
if timestamp is None:
|
|
timestamp = str(int(time.time()))
|
|
|
|
encrypt = self._encrypt(reply_msg)
|
|
signature = self._generate_signature(timestamp, nonce, encrypt)
|
|
|
|
resp_xml = (
|
|
"<xml>\n"
|
|
f"<Encrypt><![CDATA[{encrypt}]]></Encrypt>\n"
|
|
f"<MsgSignature><![CDATA[{signature}]]></MsgSignature>\n"
|
|
f"<TimeStamp>{timestamp}</TimeStamp>\n"
|
|
f"<Nonce><![CDATA[{nonce}]]></Nonce>\n"
|
|
"</xml>"
|
|
)
|
|
return resp_xml
|
|
|
|
|
|
def parse_wecom_message(xml_text):
|
|
"""
|
|
解析企业微信推送的消息XML，提取关键字段
|
|
|
|
:param xml_text: 解密后的XML明文
|
|
:return: dict 包含 msg_type, from_user, content, msg_id 等
|
|
"""
|
|
try:
|
|
root = ET.fromstring(xml_text)
|
|
except ET.ParseError as e:
|
|
_logger.error(f"消息XML解析失败: {e}")
|
|
return None
|
|
|
|
result = {
|
|
'to_user': root.findtext('ToUserName', ''),
|
|
'from_user': root.findtext('FromUserName', ''),
|
|
'create_time': root.findtext('CreateTime', ''),
|
|
'msg_type': root.findtext('MsgType', ''),
|
|
'msg_id': root.findtext('MsgId', ''),
|
|
'agent_id': root.findtext('AgentID', ''),
|
|
}
|
|
|
|
msg_type = result['msg_type']
|
|
if msg_type == 'text':
|
|
result['content'] = root.findtext('Content', '')
|
|
elif msg_type == 'image':
|
|
result['pic_url'] = root.findtext('PicUrl', '')
|
|
result['media_id'] = root.findtext('MediaId', '')
|
|
elif msg_type == 'voice':
|
|
result['media_id'] = root.findtext('MediaId', '')
|
|
result['format'] = root.findtext('Format', '')
|
|
elif msg_type == 'event':
|
|
result['event'] = root.findtext('Event', '')
|
|
result['event_key'] = root.findtext('EventKey', '')
|
|
|
|
return result
|