回调验签改用 cryptography,去掉对缺失 rsa 包的依赖。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
XingQue
2026-07-14 14:09:43 +08:00
parent bcf304dd37
commit 5c8be5da9e

View File

@@ -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