diff --git a/utils/wechat_v3.py b/utils/wechat_v3.py index 6211e6e..cee86ce 100644 --- a/utils/wechat_v3.py +++ b/utils/wechat_v3.py @@ -10,6 +10,7 @@ import os import time from urllib.parse import urlparse +from cryptography import x509 from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import padding @@ -18,6 +19,13 @@ from Crypto.Cipher import AES logger = logging.getLogger('utils.wechat_v3') +def _load_platform_public_key(pem_bytes): + """加载微信平台证书或公钥 PEM(兼容 CERTIFICATE / PUBLIC KEY / RSA PUBLIC KEY)。""" + if b'BEGIN CERTIFICATE' in pem_bytes: + return x509.load_pem_x509_certificate(pem_bytes, default_backend()).public_key() + return serialization.load_pem_public_key(pem_bytes, backend=default_backend()) + + def _resolve_cfg(wx_cfg=None): if wx_cfg: return wx_cfg @@ -85,9 +93,7 @@ def _resolve_platform_pub_path(wx_cfg, serial=None): def verify_wechat_sign(headers, body, wx_cfg=None): - """验证微信回调签名。可指定商户;未指定则尝试启用商户列表。""" - import rsa - + """验证微信回调签名(用 cryptography,不依赖 rsa 包)。可指定商户或试全部。""" serial = headers.get('Wechatpay-Serial') signature = headers.get('Wechatpay-Signature') timestamp = headers.get('Wechatpay-Timestamp') @@ -111,10 +117,16 @@ def verify_wechat_sign(headers, body, wx_cfg=None): continue try: with open(cert_path, 'rb') as f: - pub_key = rsa.PublicKey.load_pkcs1_openssl_pem(f.read()) - rsa.verify(message.encode('utf-8'), signature_bytes, pub_key) + pub_key = _load_platform_public_key(f.read()) + pub_key.verify( + signature_bytes, + message.encode('utf-8'), + padding.PKCS1v15(), + hashes.SHA256(), + ) return True - except Exception: + except Exception as e: + logger.debug('回调验签未通过 path=%s err=%s', cert_path, e) continue return False