# 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 time import hashlib import base64 import json from urllib.parse import urlparse from cryptography.hazmat.primitives import serialization, hashes from cryptography.hazmat.primitives.asymmetric import padding from cryptography.hazmat.backends import default_backend from Crypto.Cipher import AES from django.conf import settings import os def load_private_key(): """加载私钥（支持 PKCS#1 和 PKCS#8）""" key_path = settings.WECHAT_PAY_V3_CONFIG['PRIVATE_KEY_PATH'] with open(key_path, 'rb') as f: key_data = f.read() # cryptography 自动识别格式 private_key = serialization.load_pem_private_key( key_data, password=None, backend=default_backend() ) return private_key def build_authorization(method, url, body): """生成V3接口的Authorization头""" private_key = load_private_key() mchid = settings.WECHAT_PAY_V3_CONFIG['MCHID'] serial_no = settings.WECHAT_PAY_V3_CONFIG['CERT_SERIAL_NO'] 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' # 使用 cryptography 签名 signature = private_key.sign( message.encode('utf-8'), padding.PKCS1v15(), hashes.SHA256() ) signature_b64 = base64.b64encode(signature).decode('utf-8') 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): """验证微信回调签名（需提前下载平台证书）""" # 这部分使用 rsa 库验签，也可以改用 cryptography，但 rsa 已足够 import rsa 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" if not os.path.exists(cert_path): return False with open(cert_path, 'rb') as f: pub_key = rsa.PublicKey.load_pkcs1_openssl_pem(f.read()) 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): """解密回调数据（使用 AES-GCM）""" 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) 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')