python3实现微信公众平台（公众号和企业微信）消息被动回复以及加解密

爱被打了一巴掌 2022-04-24 03:02 541阅读 0赞

微信公众平台消息加密被动回复以及加解密，官方提供的是python2示例代码：

微信公众号：[https://mp.weixin.qq.com/wiki?t=resource/res\_main&id=mp1434696670][https_mp.weixin.qq.com_wiki_t_resource_res_main_id_mp1434696670]

企业微信：[https://github.com/sbzhu/weworkapi\_python/tree/master/callback][https_github.com_sbzhu_weworkapi_python_tree_master_callback]

现将上述代码修改成python3版本，且支持中文编码消息：

一、微信公众号：

WXBizMsgCrypt.py

#!/usr/bin/env python
    # -*- encoding:utf-8 -*-
     
    """ 
        python3对公众平台发送给公众账号的消息加解密代码.支持中文.
    """
    # ------------------------------------------------------------------------
     
    import base64
    import string
    import random
    import hashlib
    import time
    import struct
     
    import binascii
    from Crypto.Cipher import AES
    import xml.etree.cElementTree as ET
    import socket
     
    from wx_encry import ierror
     
    """ AES加解密用 pycrypto """
     
     
    class FormatException(Exception):
        pass
     
     
    def throw_exception(message, exception_class=FormatException):
        """my define raise exception function"""
        raise exception_class(message)
     
     
    class SHA1:
        """计算公众平台的消息签名接口"""
     
        def getSHA1(self, token, timestamp, nonce, encrypt):
            """用SHA1算法生成安全签名
            @param token:  票据
            @param timestamp: 时间戳
            @param encrypt: 密文
            @param nonce: 随机字符串
            @return: 安全签名
            """
            try:
                token = token.decode()
                sortlist = [token, timestamp, nonce, encrypt]
                sortlist.sort()
                sha = hashlib.sha1()
                sha.update("".join(sortlist).encode("utf8"))
                return ierror.WXBizMsgCrypt_OK, sha.hexdigest()
            except Exception as e:
                print(e)
                return ierror.WXBizMsgCrypt_ComputeSignature_Error, None
     
     
    class XMLParse(object):
        """提供提取消息格式中的密文及生成回复消息格式的接口"""
     
        # xml消息模板
        AES_TEXT_RESPONSE_TEMPLATE = """<xml>
            <Encrypt><![CDATA[%(msg_encrypt)s]]></Encrypt>
            <MsgSignature><![CDATA[%(msg_signaturet)s]]></MsgSignature>
            <TimeStamp>%(timestamp)s</TimeStamp>
            <Nonce><![CDATA[%(nonce)s]]></Nonce>
            </xml>"""
     
        def extract(self, xmltext):
            """提取出xml数据包中的加密消息
            @param xmltext: 待提取的xml字符串
            @return: 提取出的加密消息字符串
            """
            try:
                xml_tree = ET.fromstring(xmltext)
                encrypt = xml_tree.find("Encrypt")
                touser_name = xml_tree.find("ToUserName")
                return ierror.WXBizMsgCrypt_OK, encrypt.text, touser_name.text
            except Exception as e:
                print(e)
                return ierror.WXBizMsgCrypt_ParseXml_Error, None, None
     
        def generate(self, encrypt, signature, timestamp, nonce):
            """生成xml消息
            @param encrypt: 加密后的消息密文
            @param signature: 安全签名
            @param timestamp: 时间戳
            @param nonce: 随机字符串
            @return: 生成的xml字符串
            """
            resp_dict = {
                'msg_encrypt': encrypt,
                'msg_signaturet': signature,
                'timestamp': timestamp,
                'nonce': nonce,
            }
            resp_xml = self.AES_TEXT_RESPONSE_TEMPLATE % resp_dict
            return resp_xml
     
     
    class PKCS7Encoder(object):
        """提供基于PKCS7算法的加解密接口"""
     
        block_size = 32
     
        def encode(self, text):
            """ 对需要加密的明文进行填充补位
            @param text: 需要进行填充补位操作的明文
            @return: 补齐明文字符串
            """
            text_length = len(text)
            # 计算需要填充的位数
            amount_to_pad = self.block_size - (text_length % self.block_size)
            if amount_to_pad == 0:
                amount_to_pad = self.block_size
            # 获得补位所用的字符
            pad = chr(amount_to_pad).encode()
            return text + pad * amount_to_pad
     
        def decode(self, decrypted):
            """删除解密后明文的补位字符
            @param decrypted: 解密后的明文
            @return: 删除补位字符后的明文
            """
            pad = ord(decrypted[-1])
            if pad < 1 or pad > 32:
                pad = 0
            return decrypted[:-pad]
     
     
    class Prpcrypt(object):
        """提供接收和推送给公众平台消息的加解密接口"""
     
        def __init__(self, key):
            # self.key = base64.b64decode(key+"=")
            self.key = key
            # 设置加解密模式为AES的CBC模式
            self.mode = AES.MODE_CBC
     
        def encrypt(self, text, appid):
            """对明文进行加密
            @param text: 需要加密的明文
            @return: 加密得到的字符串
            """
            # 16位随机字符串添加到明文开头
            len_str = struct.pack("I", socket.htonl(len(text.encode())))
            # text = self.get_random_str() + binascii.b2a_hex(len_str).decode() + text + appid
            text = self.get_random_str() + len_str + text.encode() + appid
            # 使用自定义的填充方式对明文进行补位填充
            pkcs7 = PKCS7Encoder()
            text = pkcs7.encode(text)
            # 加密
            cryptor = AES.new(self.key, self.mode, self.key[:16])
            try:
                ciphertext = cryptor.encrypt(text)
                # 使用BASE64对加密后的字符串进行编码
                return ierror.WXBizMsgCrypt_OK, base64.b64encode(ciphertext).decode('utf8')
            except Exception as e:
                return ierror.WXBizMsgCrypt_EncryptAES_Error, None
     
        def decrypt(self, text, appid):
            """对解密后的明文进行补位删除
            @param text: 密文
            @return: 删除填充补位后的明文
            """
            try:
                cryptor = AES.new(self.key, self.mode, self.key[:16])
                # 使用BASE64对密文进行解码，然后AES-CBC解密
                plain_text = cryptor.decrypt(base64.b64decode(text))
            except Exception as e:
                print(e)
                return ierror.WXBizMsgCrypt_DecryptAES_Error, None
            try:
                # pad = ord(plain_text[-1])
                pad = plain_text[-1]
                # 去掉补位字符串
                # pkcs7 = PKCS7Encoder()
                # plain_text = pkcs7.encode(plain_text)
                # 去除16位随机字符串
                content = plain_text[16:-pad]
                xml_len = socket.ntohl(struct.unpack("I", content[: 4])[0])
                xml_content = content[4: xml_len + 4]
                from_appid = content[xml_len + 4:]
            except Exception as e:
                return ierror.WXBizMsgCrypt_IllegalBuffer, None
            if from_appid != appid:
                return ierror.WXBizMsgCrypt_ValidateAppid_Error, None
            return 0, xml_content.decode()
     
        def get_random_str(self):
            """ 随机生成16位字符串
            @return: 16位字符串
            """
            rule = string.ascii_letters + string.digits
            str = random.sample(rule, 16)
            return "".join(str).encode()
     
     
    class WXBizMsgCrypt(object):
        # 构造函数
        # @param sToken: 公众平台上，开发者设置的Token
        # @param sEncodingAESKey: 公众平台上，开发者设置的EncodingAESKey
        # @param sAppId: 企业号的AppId
        def __init__(self, sToken, sEncodingAESKey, sAppId):
            try:
                self.key = base64.b64decode(sEncodingAESKey + "=")
                assert len(self.key) == 32
            except Exception:
                throw_exception("[error]: EncodingAESKey unvalid !", FormatException)
                # return ierror.WXBizMsgCrypt_IllegalAesKey)
            self.token = sToken.encode()
            self.appid = sAppId.encode()
     
        def EncryptMsg(self, sReplyMsg, sNonce, timestamp=None):
            # 将公众号回复用户的消息加密打包
            # @param sReplyMsg: 企业号待回复用户的消息，xml格式的字符串
            # @param sTimeStamp: 时间戳，可以自己生成，也可以用URL参数的timestamp,如为None则自动用当前时间
            # @param sNonce: 随机串，可以自己生成，也可以用URL参数的nonce
            # sEncryptMsg: 加密后的可以直接回复用户的密文，包括msg_signature, timestamp, nonce, encrypt的xml格式的字符串,
            # return：成功0，sEncryptMsg,失败返回对应的错误码None
            pc = Prpcrypt(self.key)
            ret, encrypt = pc.encrypt(sReplyMsg, self.appid)
            if ret != 0:
                return ret, None
            if timestamp is None:
                timestamp = str(int(time.time()))
            # 生成安全签名
            sha1 = SHA1()
            ret, signature = sha1.getSHA1(self.token, timestamp, sNonce, encrypt)
     
            if ret != 0:
                return ret, None
            xmlParse = XMLParse()
            return ret, xmlParse.generate(encrypt, signature, timestamp, sNonce)
     
        def DecryptMsg(self, sPostData, sMsgSignature, sTimeStamp, sNonce):
            # 检验消息的真实性，并且获取解密后的明文
            # @param sMsgSignature: 签名串，对应URL参数的msg_signature
            # @param sTimeStamp: 时间戳，对应URL参数的timestamp
            # @param sNonce: 随机串，对应URL参数的nonce
            # @param sPostData: 密文，对应POST请求的数据
            #  xml_content: 解密后的原文，当return返回0时有效
            # @return: 成功0，失败返回对应的错误码
            # 验证安全签名
            xmlParse = XMLParse()
            ret, encrypt, touser_name = xmlParse.extract(sPostData)
            if ret != 0:
                return ret, None
            sha1 = SHA1()
            ret, signature = sha1.getSHA1(self.token, sTimeStamp, sNonce, encrypt)
            if ret != 0:
                return ret, None
            if not signature == sMsgSignature:
                return ierror.WXBizMsgCrypt_ValidateSignature_Error, None
            pc = Prpcrypt(self.key)
            ret, xml_content = pc.decrypt(encrypt, self.appid)
            return ret, xml_content

二、现主要针对企业微信的使用做说明：

WXBizMsgCrypt.py代码与上述公众号的大致相同，有些许基于企业微信的参数改动：

"""  python3对企业微信发送给企业后台的消息加解密,支持中文
    
    """
    # ------------------------------------------------------------------------
    
    import base64
    import string
    import random
    import hashlib
    import time
    import struct
    import binascii
    
    from Crypto.Cipher import AES
    import xml.etree.cElementTree as ET
    import socket
    from .ierror import ierror
    
    """
    关于Crypto.Cipher模块，ImportError: No module named 'Crypto'解决方案
    请到官方网站 https://www.dlitz.net/software/pycrypto/ 下载pycrypto。
    下载后，按照README中的“Installation”小节的提示进行pycrypto安装。
    """
    
    
    class FormatException(Exception):
        pass
    
    
    def throw_exception(message, exception_class=FormatException):
        """my define raise exception function"""
        raise exception_class(message)
    
    
    class SHA1:
        """计算企业微信的消息签名接口"""
    
        def getSHA1(self, token, timestamp, nonce, encrypt):
            """用SHA1算法生成安全签名
            @param token:  票据
            @param timestamp: 时间戳
            @param encrypt: 密文
            @param nonce: 随机字符串
            @return: 安全签名
            """
            try:
                sortlist = [token, timestamp, nonce, encrypt]
                sortlist.sort()
                sha = hashlib.sha1()
                sha.update(("".join(sortlist)).encode('utf8'))    # python3
                return ierror.WXBizMsgCrypt_OK, sha.hexdigest()
            except Exception as e:
                print(e)
                return ierror.WXBizMsgCrypt_ComputeSignature_Error, None
    
    
    class XMLParse:
        """提供提取消息格式中的密文及生成回复消息格式的接口"""
    
        # xml消息模板
        AES_TEXT_RESPONSE_TEMPLATE = """<xml>
    <Encrypt><![CDATA[%(msg_encrypt)s]]></Encrypt>
    <MsgSignature><![CDATA[%(msg_signaturet)s]]></MsgSignature>
    <TimeStamp>%(timestamp)s</TimeStamp>
    <Nonce><![CDATA[%(nonce)s]]></Nonce>
    </xml>"""
    
        def extract(self, xmltext):
            """提取出xml数据包中的加密消息
            @param xmltext: 待提取的xml字符串
            @return: 提取出的加密消息字符串
            """
            try:
                xml_tree = ET.fromstring(xmltext)
                encrypt = xml_tree.find("Encrypt")
                return ierror.WXBizMsgCrypt_OK, encrypt.text
            except Exception as e:
                print(e)
                return ierror.WXBizMsgCrypt_ParseXml_Error, None, None
    
        def generate(self, encrypt, signature, timestamp, nonce):
            """生成xml消息
            @param encrypt: 加密后的消息密文
            @param signature: 安全签名
            @param timestamp: 时间戳
            @param nonce: 随机字符串
            @return: 生成的xml字符串
            """
            resp_dict = {
                'msg_encrypt': encrypt,
                'msg_signaturet': signature,
                'timestamp': timestamp,
                'nonce': nonce,
            }
            resp_xml = self.AES_TEXT_RESPONSE_TEMPLATE % resp_dict
            return resp_xml
    
    
    class PKCS7Encoder():
        """提供基于PKCS7算法的加解密接口"""
    
        block_size = 32
    
        def encode(self, text):
            """ 对需要加密的明文进行填充补位
            @param text: 需要进行填充补位操作的明文
            @return: 补齐明文字符串
            """
            text_length = len(text)
            # 计算需要填充的位数
            amount_to_pad = self.block_size - (text_length % self.block_size)
            if amount_to_pad == 0:
                amount_to_pad = self.block_size
            # 获得补位所用的字符
            pad = chr(amount_to_pad).encode()
            return text + pad * amount_to_pad
    
        def decode(self, decrypted):
            """删除解密后明文的补位字符
            @param decrypted: 解密后的明文
            @return: 删除补位字符后的明文
            """
            pad = ord(decrypted[-1])
            if pad < 1 or pad > 32:
                pad = 0
            return decrypted[:-pad]
    
    
    class Prpcrypt(object):
        """提供接收和推送给企业微信消息的加解密接口"""
    
        def __init__(self, key):
    
            # self.key = base64.b64decode(key+"=")
            self.key = key
            # 设置加解密模式为AES的CBC模式
            self.mode = AES.MODE_CBC
    
        def encrypt(self, text, receiveid):
            """对明文进行加密
            @param text: 需要加密的明文
            @return: 加密得到的字符串
            """
            # 16位随机字符串添加到明文开头
            len_str = struct.pack("I", socket.htonl(len(text.encode())))
            # text = self.get_random_str() + struct.pack("I", socket.htonl(len(text))) + text + receiveid
            text = self.get_random_str() + len_str + text.encode() + receiveid.encode()
            # 使用自定义的填充方式对明文进行补位填充
            pkcs7 = PKCS7Encoder()
            text = pkcs7.encode(text)
            # 加密
            cryptor = AES.new(self.key, self.mode, self.key[:16])
            try:
                ciphertext = cryptor.encrypt(text)
                # 使用BASE64对加密后的字符串进行编码
                return ierror.WXBizMsgCrypt_OK, base64.b64encode(ciphertext).decode('utf8')
            except Exception as e:
                print(e)
                return ierror.WXBizMsgCrypt_EncryptAES_Error, None
    
        def decrypt(self, text, receiveid):
            """对解密后的明文进行补位删除
            @param text: 密文
            @return: 删除填充补位后的明文
            """
            try:
                cryptor = AES.new(self.key, self.mode, self.key[:16])
                # 使用BASE64对密文进行解码，然后AES-CBC解密
                # plain_text = cryptor.decrypt(base64.b64decode(text))                   # python2
                plain_text = cryptor.decrypt(base64.b64decode(text)).decode('utf-8')     # python3
            except Exception as e:
                print(e)
                return ierror.WXBizMsgCrypt_DecryptAES_Error, None
            try:
                pad = ord(plain_text[-1])
                # 去掉补位字符串
                # pkcs7 = PKCS7Encoder()
                # plain_text = pkcs7.encode(plain_text)
                # 去除16位随机字符串
                content = plain_text[16:-pad].encode('utf-8')
                xml_len = socket.ntohl(struct.unpack("I", content[: 4])[0])
                xml_content = content[4: xml_len + 4]
                from_receiveid = content[xml_len + 4:].decode()
            except Exception as e:
                print(e)
                return ierror.WXBizMsgCrypt_IllegalBuffer, None
            if from_receiveid != receiveid:
                return ierror.WXBizMsgCrypt_ValidateCorpid_Error, None
            return 0, xml_content.decode()
    
        def get_random_str(self):
            """ 随机生成16位字符串
            @return: 16位字符串
            """
            # rule = string.letters + string.digits           # python2
            rule = string.ascii_letters + string.digits       # python3
            str = random.sample(rule, 16)
            return "".join(str).encode()
    
    
    class WXBizMsgCrypt(object):
        # 构造函数
        def __init__(self, sToken, sEncodingAESKey, sReceiveId):
            try:
                self.key = base64.b64decode(sEncodingAESKey + "=")
                assert len(self.key) == 32
            except:
                throw_exception("[error]: EncodingAESKey unvalid !", FormatException)
                # return ierror.WXBizMsgCrypt_IllegalAesKey,None
            self.m_sToken = sToken
            self.m_sReceiveId = sReceiveId
    
            # 验证URL
            # @param sMsgSignature: 签名串，对应URL参数的msg_signature
            # @param sTimeStamp: 时间戳，对应URL参数的timestamp
            # @param sNonce: 随机串，对应URL参数的nonce
            # @param sEchoStr: 随机串，对应URL参数的echostr
            # @param sReplyEchoStr: 解密之后的echostr，当return返回0时有效
            # @return：成功0，失败返回对应的错误码
    
        def VerifyURL(self, sMsgSignature, sTimeStamp, sNonce, sEchoStr):
            sha1 = SHA1()
            ret, signature = sha1.getSHA1(self.m_sToken, sTimeStamp, sNonce, sEchoStr)
            if ret != 0:
                return ret, None
            if not signature == sMsgSignature:
                return ierror.WXBizMsgCrypt_ValidateSignature_Error, None
            pc = Prpcrypt(self.key)
            ret, sReplyEchoStr = pc.decrypt(sEchoStr, self.m_sReceiveId)
            return ret, sReplyEchoStr
    
        def EncryptMsg(self, sReplyMsg, sNonce, timestamp=None):
            # 将企业回复用户的消息加密打包
            # @param sReplyMsg: 企业号待回复用户的消息，xml格式的字符串
            # @param sTimeStamp: 时间戳，可以自己生成，也可以用URL参数的timestamp,如为None则自动用当前时间
            # @param sNonce: 随机串，可以自己生成，也可以用URL参数的nonce
            # sEncryptMsg: 加密后的可以直接回复用户的密文，包括msg_signature, timestamp, nonce, encrypt的xml格式的字符串,
            # return：成功0，sEncryptMsg,失败返回对应的错误码None
            pc = Prpcrypt(self.key)
            ret, encrypt = pc.encrypt(sReplyMsg, self.m_sReceiveId)
            if ret != 0:
                return ret, None
            if timestamp is None:
                timestamp = str(int(time.time()))
            # 生成安全签名
            sha1 = SHA1()
            ret, signature = sha1.getSHA1(self.m_sToken, timestamp, sNonce, encrypt)
            if ret != 0:
                return ret, None
            xmlParse = XMLParse()
            return ret, xmlParse.generate(encrypt, signature, timestamp, sNonce)
    
        def DecryptMsg(self, sPostData, sMsgSignature, sTimeStamp, sNonce):
            # 检验消息的真实性，并且获取解密后的明文
            # @param sMsgSignature: 签名串，对应URL参数的msg_signature
            # @param sTimeStamp: 时间戳，对应URL参数的timestamp
            # @param sNonce: 随机串，对应URL参数的nonce
            # @param sPostData: 密文，对应POST请求的数据
            #  xml_content: 解密后的原文，当return返回0时有效
            # @return: 成功0，失败返回对应的错误码
            # 验证安全签名
            xmlParse = XMLParse()
            ret, encrypt = xmlParse.extract(sPostData)
            if ret != 0:
                return ret, None
            sha1 = SHA1()
            ret, signature = sha1.getSHA1(self.m_sToken, sTimeStamp, sNonce, encrypt)
            if ret != 0:
                return ret, None
            if not signature == sMsgSignature:
                return ierror.WXBizMsgCrypt_ValidateSignature_Error, None
            pc = Prpcrypt(self.key)
            ret, xml_content = pc.decrypt(encrypt, self.m_sReceiveId)
            return ret, xml_content

完整的消息被动回复以及加解密项目包下载：

[https://download.csdn.net/download/weixin\_41762173/11107750][https_download.csdn.net_download_weixin_41762173_11107750]

[https_mp.weixin.qq.com_wiki_t_resource_res_main_id_mp1434696670]: https://mp.weixin.qq.com/wiki?t=resource/res_main&id=mp1434696670
[https_github.com_sbzhu_weworkapi_python_tree_master_callback]: https://github.com/sbzhu/weworkapi_python/tree/master/callback
[https_download.csdn.net_download_weixin_41762173_11107750]: https://download.csdn.net/download/weixin_41762173/11107750