做了一些有限修正

utils/wechat_v3.py

@@ -1,112 +1,4 @@
-# utils/wechat_v3.py
-'''import time
-import hashlib
-import base64
-import json
-from urllib.parse import urlparse
-
-import rsa
-from Crypto.Cipher import AES
-
-from django.conf import settings
-
-
-def load_private_key():
-    """加载商户私钥（PEM格式）"""
-    with open(settings.WECHAT_PAY_V3_CONFIG['PRIVATE_KEY_PATH'], 'rb') as f:
-        key_data = f.read()
-    return rsa.PrivateKey.load_pkcs1(key_data)
-
-
-def build_authorization(method, url, body):
-    """
-    生成V3接口的Authorization头
-    :param method: GET/POST
-    :param url: 完整URL（含域名和路径）
-    :param body: 请求体JSON字符串
-    :return: Authorization字符串
-    """
-    private_key = load_private_key()
-    mchid = settings.WECHAT_PAY_V3_CONFIG['MCHID']
-    serial_no = settings.WECHAT_PAY_V3_CONFIG['CERT_SERIAL_NO']
-
-    # 提取URL路径（不含query）
-    parsed = urlparse(url)
-    path = parsed.path
-    if parsed.query:
-        path += '?' + parsed.query
-
-    timestamp = str(int(time.time()))
-    nonce = hashlib.md5((timestamp + 'wechat').encode()).hexdigest()[:32]
-
-    # 构造签名串
-    message = '\n'.join([method, path, timestamp, nonce, body]) + '\n'
-
-    # 使用私钥签名（SHA256）
-    signature = rsa.sign(message.encode('utf-8'), private_key, 'SHA-256')
-    signature_b64 = base64.b64encode(signature).decode('utf-8')
-
-    # 组装Authorization
-    auth = f'WECHATPAY2-SHA256-RSA2048 mchid="{mchid}",nonce_str="{nonce}",timestamp="{timestamp}",serial_no="{serial_no}",signature="{signature_b64}"'
-    return auth
-
-
-def verify_wechat_sign(headers, body):
-    """
-    验证微信回调的签名
-    需要提前从微信平台下载证书公钥，存放在 PLATFORM_CERT_DIR 下，文件名 = 序列号.pem
-    :param headers: 请求头
-    :param body: 原始请求体字符串
-    :return: bool
-    """
-    serial = headers.get('Wechatpay-Serial')
-    signature = headers.get('Wechatpay-Signature')
-    timestamp = headers.get('Wechatpay-Timestamp')
-    nonce = headers.get('Wechatpay-Nonce')
-
-    if not all([serial, signature, timestamp, nonce]):
-        return False
-
-    # 构造验签名串
-    message = '\n'.join([timestamp, nonce, body]) + '\n'
-
-    # 读取对应序列号的平台证书公钥
-    cert_path = f"{settings.WECHAT_PAY_V3_CONFIG['PLATFORM_CERT_DIR']}/{serial}.pem"
-    try:
-        with open(cert_path, 'rb') as f:
-            pub_key = rsa.PublicKey.load_pkcs1_openssl_pem(f.read())
-    except FileNotFoundError:
-        # 如果证书不存在，可尝试下载（生产环境建议定时下载）
-        return False
-
-    signature_bytes = base64.b64decode(signature)
-    try:
-        rsa.verify(message.encode('utf-8'), signature_bytes, pub_key)
-        return True
-    except rsa.VerificationError:
-        return False
-
-
-def decrypt_callback_ciphertext(associated_data, nonce, ciphertext):
-    """
-    使用APIv3密钥解密回调中的加密数据
-    :param associated_data: 附加数据
-    :param nonce: 随机串
-    :param ciphertext: 密文（Base64）
-    :return: 解密后的JSON字符串
-    """
-    api_v3_key = settings.WECHAT_PAY_V3_CONFIG['API_V3_KEY'].encode('utf-8')
-    nonce_bytes = nonce.encode('utf-8')
-    ciphertext_bytes = base64.b64decode(ciphertext)
-
-    # AES-GCM解密
-    cipher = AES.new(api_v3_key, AES.MODE_GCM, nonce=nonce_bytes)
-    if associated_data:
-        cipher.update(associated_data.encode('utf-8'))
-    plaintext = cipher.decrypt_and_verify(ciphertext_bytes[:-16], ciphertext_bytes[-16:])
-    return plaintext.decode('utf-8')'''
-
-# utils/wechat_v3.py
+import re
 import time
 import hashlib
 import base64
@@ -178,6 +70,16 @@ def verify_wechat_sign(headers, body):
     if not all([serial, signature, timestamp, nonce]):
         return False
 
+    if not re.match(r'^[A-Za-z0-9-]+$', serial):
+        return False
+
+    try:
+        ts = int(headers.get('Wechatpay-Timestamp', '0'))
+        if abs(time.time() - ts) > 300:  # 5分钟有效期
+            return False
+    except (ValueError, TypeError):
+        return False
+
     message = '\n'.join([timestamp, nonce, body]) + '\n'
     cert_path = f"{settings.WECHAT_PAY_V3_CONFIG['PLATFORM_CERT_DIR']}/{serial}.pem"
     if not os.path.exists(cert_path):
@@ -204,4 +106,4 @@ def decrypt_callback_ciphertext(associated_data, nonce, ciphertext):
     if associated_data:
         cipher.update(associated_data.encode('utf-8'))
     plaintext = cipher.decrypt_and_verify(ciphertext_bytes[:-16], ciphertext_bytes[-16:])
-    return plaintext.decode('utf-8')
+    return plaintext.decode('utf-8')