# -*- 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 = ( "\n" f"\n" f"\n" f"{timestamp}\n" f"\n" "" ) 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