b5021ee07d
SIGN_ERROR根因是私钥与序列号不是同一套证书；扫描匹配 apiclient_cert.pem。 新增 manage.py wx_v3_pay_diag 探针验签。 Co-authored-by: Cursor <cursoragent@cursor.com>
141 lines
4.2 KiB
Python
141 lines
4.2 KiB
Python
import logging
|
||
import re
|
||
import time
|
||
import hashlib
|
||
import base64
|
||
import json
|
||
import os
|
||
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
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _get_v3_config(club_id=None):
|
||
from jituan.services.wechat_pay import get_wechat_v3_config
|
||
return get_wechat_v3_config(club_id)
|
||
|
||
|
||
def load_private_key(club_id=None):
|
||
"""加载私钥（支持 PKCS#1 和 PKCS#8）。"""
|
||
cfg = _get_v3_config(club_id)
|
||
key_path = cfg['PRIVATE_KEY_PATH']
|
||
with open(key_path, 'rb') as f:
|
||
key_data = f.read()
|
||
|
||
return serialization.load_pem_private_key(
|
||
key_data,
|
||
password=None,
|
||
backend=default_backend(),
|
||
)
|
||
|
||
|
||
def build_authorization(method, url, body, club_id=None):
|
||
"""生成 V3 接口的 Authorization 头。"""
|
||
cfg = _get_v3_config(club_id)
|
||
private_key = load_private_key(club_id)
|
||
mchid = cfg['MCHID']
|
||
serial_no = (cfg.get('CERT_SERIAL_NO') or '').strip().upper()
|
||
logger.warning(
|
||
'[WX_V3_SIGN] club=%s mchid=%s serial=%s key_path=%s',
|
||
club_id, mchid, serial_no, cfg.get('PRIVATE_KEY_PATH'),
|
||
)
|
||
|
||
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'
|
||
|
||
signature = private_key.sign(
|
||
message.encode('utf-8'),
|
||
padding.PKCS1v15(),
|
||
hashes.SHA256(),
|
||
)
|
||
signature_b64 = base64.b64encode(signature).decode('utf-8')
|
||
|
||
return (
|
||
f'WECHATPAY2-SHA256-RSA2048 mchid="{mchid}",nonce_str="{nonce}",'
|
||
f'timestamp="{timestamp}",serial_no="{serial_no}",signature="{signature_b64}"'
|
||
)
|
||
|
||
|
||
def verify_wechat_sign(headers, body, club_id=None):
|
||
"""验证微信回调签名（需提前下载平台证书）。"""
|
||
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
|
||
|
||
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:
|
||
return False
|
||
except (ValueError, TypeError):
|
||
return False
|
||
|
||
cfg = _get_v3_config(club_id)
|
||
message = '\n'.join([timestamp, nonce, body]) + '\n'
|
||
cert_path = f"{cfg['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 resolve_callback_club_id(headers, body):
|
||
"""
|
||
多商户回调：依次用各俱乐部平台证书验签，返回匹配的 club_id。
|
||
验签失败返回 None。
|
||
"""
|
||
from jituan.constants import CLUB_ID_DEFAULT
|
||
from jituan.models import Club
|
||
|
||
if verify_wechat_sign(headers, body, club_id=CLUB_ID_DEFAULT):
|
||
return CLUB_ID_DEFAULT
|
||
|
||
for club in Club.query.filter(status=1).exclude(club_id=CLUB_ID_DEFAULT):
|
||
if verify_wechat_sign(headers, body, club_id=club.club_id):
|
||
return club.club_id
|
||
return None
|
||
|
||
|
||
def decrypt_callback_ciphertext(associated_data, nonce, ciphertext, club_id=None):
|
||
"""解密回调数据（使用 AES-GCM）。"""
|
||
cfg = _get_v3_config(club_id)
|
||
api_v3_key = cfg['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')
|