nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-10-31 04:14:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(wechat): preserve encryption errors and type crypto contracts

Browse Source
This commit is contained in:
jxxghp
2026-10-17 01:58:42 +08:00
parent 451f5b0f30
commit ffbb3a73fe
2 changed files with 174 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
app/adapters/external/wechat.py vendored
View File

@@ -15,6 +15,7 @@ import socket
import struct
import time
import xml.etree.cElementTree as ET
from typing import NoReturn, Optional, cast
from Crypto.Cipher import AES
@@ -41,10 +42,11 @@ WXBizMsgCrypt_GenReturnXml_Error = -40011
class FormatException(Exception):
"""企业微信编码密钥的格式错误。"""
pass
def throw_exception(message, exception_class=FormatException):
def throw_exception(message: str, exception_class: type[Exception] = FormatException) -> NoReturn:
"""my define raise exception function"""
raise exception_class(message)
@@ -53,7 +55,7 @@ class SHA1:
"""计算企业微信的消息签名接口"""
@staticmethod
def getSHA1(token, timestamp, nonce, encrypt):
def getSHA1(token: Optional[str], timestamp: str | int | None, nonce: Optional[str], encrypt: Optional[str]) -> tuple[int, Optional[str]]:
"""用SHA1算法生成安全签名
@param token: 票据
@param timestamp: 时间戳
@@ -65,7 +67,8 @@ class SHA1:
sortlist = [token, timestamp, nonce, encrypt]
sortlist.sort()
sha = hashlib.sha1()
sha.update("".join(sortlist).encode())
# 其余三项只能为 str 或 None；排序成功已排除 None 和整数时间戳。
sha.update("".join(cast(list[str], sortlist)).encode())
return WXBizMsgCrypt_OK, sha.hexdigest()
except Exception as e:
logger = logging.getLogger()
@@ -85,7 +88,7 @@ class XMLParse:
</xml>"""
@staticmethod
def extract(xmltext):
def extract(xmltext: str | bytes) -> tuple[int, Optional[str]]:
"""提取出xml数据包中的加密消息
@param xmltext: 待提取的xml字符串
@return: 提取出的加密消息字符串
@@ -93,13 +96,15 @@ class XMLParse:
try:
xml_tree = ET.fromstring(xmltext)
encrypt = xml_tree.find("Encrypt")
if encrypt is None:
raise AttributeError("'NoneType' object has no attribute 'text'")
return WXBizMsgCrypt_OK, encrypt.text
except Exception as e:
logger = logging.getLogger()
logger.error(e)
return WXBizMsgCrypt_ParseXml_Error, None
def generate(self, encrypt, signature, timestamp, nonce):
def generate(self, encrypt: str, signature: Optional[str], timestamp: str, nonce: str) -> str:
"""生成xml消息
@param encrypt: 加密后的消息密文
@param signature: 安全签名
@@ -122,7 +127,7 @@ class PKCS7Encoder:
block_size = 32
def encode(self, text):
def encode(self, text: bytes) -> bytes:
""" 对需要加密的明文进行填充补位
@param text: 需要进行填充补位操作的明文
@return: 补齐明文字符串
@@ -137,7 +142,7 @@ class PKCS7Encoder:
return text + (pad * amount_to_pad).encode()
@staticmethod
def decode(decrypted):
def decode(decrypted: str) -> str:
"""删除解密后明文的补位字符
@param decrypted: 解密后的明文
@return: 删除补位字符后的明文
@@ -151,30 +156,34 @@ class PKCS7Encoder:
class Prpcrypt(object):
"""提供接收和推送给企业微信消息的加解密接口"""
def __init__(self, key):
def __init__(self, key: bytes) -> None:
"""保存已解码的 AES 密钥与 CBC 模式。"""
# self.key = base64.b64decode(key+"=")
self.key = key
# 设置加解密模式为AES的CBC模式
self.mode = AES.MODE_CBC
def encrypt(self, text, receiveid):
def encrypt(self, text: str, receiveid: Optional[str]) -> tuple[int, Optional[bytes]]:
"""对明文进行加密
@param text: 需要加密的明文
@param receiveid: receiveid
@return: 加密得到的字符串
"""
# 16位随机字符串添加到明文开头
text = text.encode()
text = self.get_random_str() + struct.pack("I", socket.htonl(len(text))) + text + receiveid.encode()
plaintext = text.encode()
plaintext = self.get_random_str() + struct.pack("I", socket.htonl(len(plaintext))) + plaintext
if receiveid is None:
raise AttributeError("'NoneType' object has no attribute 'encode'")
plaintext += receiveid.encode()
# 使用自定义的填充方式对明文进行补位填充
pkcs7 = PKCS7Encoder()
text = pkcs7.encode(text)
plaintext = pkcs7.encode(plaintext)
# 加密
cryptor = AES.new(self.key, self.mode, self.key[:16])
try:
ciphertext = cryptor.encrypt(text)
ciphertext = cryptor.encrypt(plaintext)
# 使用BASE64对加密后的字符串进行编码
return WXBizMsgCrypt_OK, base64.b64encode(ciphertext)
except Exception as e:
@@ -182,7 +191,7 @@ class Prpcrypt(object):
logger.error(e)
return WXBizMsgCrypt_EncryptAES_Error, None
def decrypt(self, text, receiveid):
def decrypt(self, text: str | bytes, receiveid: Optional[str]) -> tuple[int, Optional[bytes]]:
"""对解密后的明文进行补位删除
@param text: 密文
@param receiveid: receiveid
@@ -191,7 +200,7 @@ class Prpcrypt(object):
try:
cryptor = AES.new(self.key, self.mode, self.key[:16])
# 使用BASE64对密文进行解码，然后AES-CBC解密
plain_text = cryptor.decrypt(base64.b64decode(text))
plain_text: bytes = cryptor.decrypt(base64.b64decode(text))
except Exception as e:
logger = logging.getLogger()
logger.error(e)
@@ -216,7 +225,7 @@ class Prpcrypt(object):
return 0, xml_content
@staticmethod
def get_random_str():
def get_random_str() -> bytes:
""" 随机生成16位字符串
@return: 16位字符串
"""
@@ -224,9 +233,13 @@ class Prpcrypt(object):
class WXBizMsgCrypt(object):
"""验证企业微信回调并封装加解密错误码。"""
# 构造函数
def __init__(self, sToken, sEncodingAESKey, sReceiveId):
def __init__(self, sToken: Optional[str], sEncodingAESKey: Optional[str], sReceiveId: Optional[str]) -> None:
"""解码并验证 AES 密钥，保留配置错误的格式异常。"""
try:
if sEncodingAESKey is None:
raise TypeError("unsupported operand type(s) for +: 'NoneType' and 'str'")
self.key = base64.b64decode(sEncodingAESKey + "=")
assert len(self.key) == 32
except Exception as err:
@@ -244,7 +257,8 @@ class WXBizMsgCrypt(object):
# @param sReplyEchoStr: 解密之后的echostr，当return返回0时有效
# @return：成功0，失败返回对应的错误码
def VerifyURL(self, sMsgSignature, sTimeStamp, sNonce, sEchoStr):
def VerifyURL(self, sMsgSignature: str, sTimeStamp: str | int, sNonce: str, sEchoStr: str) -> tuple[int, Optional[bytes]]:
"""校验 URL 签名后返回解密的验证内容及原始错误码。"""
sha1 = SHA1()
ret, signature = sha1.getSHA1(self.m_sToken, sTimeStamp, sNonce, sEchoStr)
if ret != 0:
@@ -255,7 +269,8 @@ class WXBizMsgCrypt(object):
ret, sReplyEchoStr = pc.decrypt(sEchoStr, self.m_sReceiveId)
return ret, sReplyEchoStr
def EncryptMsg(self, sReplyMsg, sNonce, timestamp=None):
def EncryptMsg(self, sReplyMsg: str, sNonce: str, timestamp: Optional[str] = None) -> tuple[int, Optional[str]]:
"""加密回复并生成签名 XML，加密失败直接返回对应错误码。"""
# 将企业回复用户的消息加密打包
# @param sReplyMsg: 企业号待回复用户的消息，xml格式的字符串
# @param sTimeStamp: 时间戳，可以自己生成，也可以用URL参数的timestamp,如为None则自动用当前时间
@@ -263,10 +278,11 @@ class WXBizMsgCrypt(object):
# sEncryptMsg: 加密后的可以直接回复用户的密文，包括msg_signature, timestamp, nonce, encrypt的xml格式的字符串,
# return：成功0，sEncryptMsg,失败返回对应的错误码None
pc = Prpcrypt(self.key)
ret, encrypt = pc.encrypt(sReplyMsg, self.m_sReceiveId)
encrypt = encrypt.decode('utf8')
ret, encrypted = pc.encrypt(sReplyMsg, self.m_sReceiveId)
if ret != 0:
return ret, None
# encrypt 的成功分支只返回 base64 bytes，失败时不可对 None 解码。
encrypt = cast(bytes, encrypted).decode('utf8')
if timestamp is None:
timestamp = str(int(time.time()))
# 生成安全签名
@@ -277,7 +293,8 @@ class WXBizMsgCrypt(object):
xmlParse = XMLParse()
return ret, xmlParse.generate(encrypt, signature, timestamp, sNonce)
def DecryptMsg(self, sPostData, sMsgSignature, sTimeStamp, sNonce):
def DecryptMsg(self, sPostData: str | bytes, sMsgSignature: str, sTimeStamp: str, sNonce: str) -> tuple[int, Optional[bytes]]:
"""先验证 XML 与消息签名，再按协议解密并检查接收方。"""
# 检验消息的真实性，并且获取解密后的明文
# @param sMsgSignature: 签名串，对应URL参数的msg_signature
# @param sTimeStamp: 时间戳，对应URL参数的timestamp
@@ -297,5 +314,6 @@ class WXBizMsgCrypt(object):
if not signature == sMsgSignature:
return WXBizMsgCrypt_ValidateSignature_Error, None
pc = Prpcrypt(self.key)
ret, xml_content = pc.decrypt(encrypt, self.m_sReceiveId)
# SHA1 成功已证明密文为字符串；空字段会在签名计算阶段返回错误。
ret, xml_content = pc.decrypt(cast(str, encrypt), self.m_sReceiveId)
return ret, xml_content

133
tests/test_wechat_crypto_contracts.py Normal file
View File

@@ -0,0 +1,133 @@
"""企业微信加密协议与失败返回值的离线回归。"""
import base64
import hashlib
import struct
import xml.etree.ElementTree as ET
from unittest.mock import Mock
import pytest
from Crypto.Cipher import AES
from app.adapters.external import wechat
KEY = bytes(range(32))
ENCODING_KEY = base64.b64encode(KEY).decode().rstrip("=")
RANDOM_PREFIX = b"0123456789abcdef"
MESSAGE = "<xml><Content><![CDATA[中文消息]]></Content></xml>"
@pytest.fixture
def crypt(monkeypatch):
"""固定协议随机前缀，使用测试密钥而非真实企业凭据。"""
monkeypatch.setattr(wechat.Prpcrypt, "get_random_str", staticmethod(lambda: RANDOM_PREFIX))
return wechat.WXBizMsgCrypt("token", ENCODING_KEY, "corp-id")
def test_encrypt_failure_returns_original_error_code(crypt, monkeypatch):
"""AES 加密失败应返回约定错误码，不能对 None 解码而抛出额外异常。"""
cipher = Mock(encrypt=Mock(side_effect=ValueError("encryption failed")))
monkeypatch.setattr(wechat.AES, "new", Mock(return_value=cipher))
assert crypt.EncryptMsg(MESSAGE, "nonce", "1700000000") == (
wechat.WXBizMsgCrypt_EncryptAES_Error, None,
)
def test_encrypt_matches_independent_wire_vector_and_roundtrips(crypt):
"""独立构造网络字节序、32 字节填充和 SHA1，校验密文及两个解密入口。"""
message = MESSAGE.encode()
plaintext = RANDOM_PREFIX + struct.pack("!I", len(message)) + message + b"corp-id"
padding = 32 - len(plaintext) % 32
expected = base64.b64encode(AES.new(KEY, AES.MODE_CBC, KEY[:16]).encrypt(
plaintext + bytes([padding]) * padding,
)).decode()
signature = hashlib.sha1("".join(sorted(["token", "1700000000", "nonce", expected])).encode()).hexdigest()
code, payload = crypt.EncryptMsg(MESSAGE, "nonce", "1700000000")
assert code == 0
root = ET.fromstring(payload)
assert root.findtext("Encrypt") == expected
assert root.findtext("MsgSignature") == signature
assert root.findtext("TimeStamp") == "1700000000"
assert root.findtext("Nonce") == "nonce"
assert crypt.DecryptMsg(payload, signature, "1700000000", "nonce") == (0, message)
assert crypt.VerifyURL(signature, "1700000000", "nonce", expected) == (0, message)
def test_default_timestamp_remains_integer_seconds(crypt, monkeypatch):
"""缺省时间戳继续按整数秒生成。"""
monkeypatch.setattr(wechat.time, "time", lambda: 1700000000.75)
code, payload = crypt.EncryptMsg(MESSAGE, "nonce")
assert code == 0
assert ET.fromstring(payload).findtext("TimeStamp") == "1700000000"
def test_integer_verification_timestamp_keeps_signature_failure(crypt):
"""验证入口收到整数时间戳仍返回原签名错误，不隐式转成可通过验证的字符串。"""
assert crypt.VerifyURL("signature", 1700000000, "nonce", "cipher") == (
wechat.WXBizMsgCrypt_ComputeSignature_Error, None,
)
def test_missing_recipient_keeps_encrypt_exception_and_decrypt_error(crypt):
"""缺失接收方时，加密的异常和解密的接收方错误均保持兼容。"""
missing = wechat.WXBizMsgCrypt("token", ENCODING_KEY, None)
with pytest.raises(AttributeError, match="has no attribute 'encode'"):
missing.EncryptMsg(MESSAGE, "nonce", "1700000000")
_, payload = crypt.EncryptMsg(MESSAGE, "nonce", "1700000000")
signature = ET.fromstring(payload).findtext("MsgSignature")
assert missing.DecryptMsg(payload, signature, "1700000000", "nonce") == (
wechat.WXBizMsgCrypt_ValidateCorpid_Error, None,
)
@pytest.mark.parametrize("xml, expected", [
("broken", wechat.WXBizMsgCrypt_ParseXml_Error),
("<xml/>", wechat.WXBizMsgCrypt_ParseXml_Error),
("<xml><Encrypt/></xml>", wechat.WXBizMsgCrypt_ComputeSignature_Error),
])
def test_invalid_xml_preserves_failure_code(crypt, xml, expected):
"""损坏 XML、缺失字段和空字段保持原有的不同错误码。"""
assert crypt.DecryptMsg(xml, "signature", "1700000000", "nonce") == (expected, None)
def test_signature_and_recipient_mismatch_preserve_failures(crypt):
"""签名错误在解密前拒绝，接收方错误在解密后返回原错误码。"""
_, payload = crypt.EncryptMsg(MESSAGE, "nonce", "1700000000")
signature = ET.fromstring(payload).findtext("MsgSignature")
assert crypt.DecryptMsg(payload, "bad", "1700000000", "nonce") == (
wechat.WXBizMsgCrypt_ValidateSignature_Error, None,
)
other = wechat.WXBizMsgCrypt("token", ENCODING_KEY, "other-corp")
assert other.DecryptMsg(payload, signature, "1700000000", "nonce") == (
wechat.WXBizMsgCrypt_ValidateCorpid_Error, None,
)
@pytest.mark.parametrize("length", [0, 1, 31, 32, 33, 64])
def test_padding_keeps_legacy_bytes_and_text_contract(length):
"""编码使用 bytes，旧解填充入口继续处理字符串，整块也补齐一块。"""
data = b"a" * length
encoded = wechat.PKCS7Encoder().encode(data)
padding = 32 - length % 32
assert encoded == data + bytes([padding]) * padding
assert wechat.PKCS7Encoder.decode(encoded.decode("latin1")) == data.decode("latin1")
@pytest.mark.parametrize("invalid", [None, "", "invalid-key"])
def test_invalid_encoding_key_keeps_format_exception(invalid):
"""无效密钥仍在构造阶段报告既有格式错误。"""
with pytest.raises(wechat.FormatException, match="EncodingAESKey unvalid"):
wechat.WXBizMsgCrypt("token", invalid, "corp-id")
@pytest.mark.parametrize("encrypted", [None, "payload"])
def test_sha1_keeps_empty_cipher_failure_and_valid_signature(encrypted):
"""空密文继续返回计算失败；有效输入保持原有字典序签名。"""
result = wechat.SHA1.getSHA1("token", "1700000000", "nonce", encrypted)
if encrypted is None:
assert result == (wechat.WXBizMsgCrypt_ComputeSignature_Error, None)
else:
expected = hashlib.sha1("".join(sorted(["token", "1700000000", "nonce", encrypted])).encode()).hexdigest()
assert result == (0, expected)