XingQue/Django
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: V3签名从私钥配对证书自动取序列号，xq忽略club错误私钥路径

Browse Source 
SIGN_ERROR根因是私钥与序列号不是同一套证书；扫描匹配 apiclient_cert.pem。
新增 manage.py wx_v3_pay_diag 探针验签。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
XingQue
2026-07-10 22:31:14 +08:00
parent 2e419b5b6e
commit b5021ee07d
3 changed files with 182 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

93
jituan/management/commands/wx_v3_pay_diag.py Normal file
View File

@@ -0,0 +1,93 @@
"""诊断微信 V3 商户证书/私钥/序列号是否匹配。用法: python manage.py wx_v3_pay_diag [--club xq]"""
import os
import requests
from django.core.management.base import BaseCommand
from jituan.constants import CLUB_ID_DEFAULT
from jituan.services.wechat_pay import (
XQ_WX_PAY,
get_wechat_v3_config,
resolve_merchant_cert_serial,
_load_pem_private_key,
_public_keys_match,
)
from utils.wechat_v3 import build_authorization
class Command(BaseCommand):
help = '诊断微信 V3 提现签名：私钥、证书序列号、商户号是否一致'
def add_arguments(self, parser):
parser.add_argument('--club', default=CLUB_ID_DEFAULT, help='俱乐部 ID，默认 xq')
def handle(self, *args, **options):
club_id = (options.get('club') or CLUB_ID_DEFAULT).strip()
cfg = get_wechat_v3_config(club_id)
self.stdout.write('=== 微信 V3 配置诊断 ===')
self.stdout.write(f"club_id : {cfg.get('club_id')}")
self.stdout.write(f"APPID : {cfg.get('APPID')}")
self.stdout.write(f"MCHID : {cfg.get('MCHID')} (期望 {XQ_WX_PAY['mch_id']})")
self.stdout.write(f"CERT_SERIAL : {cfg.get('CERT_SERIAL_NO')}")
self.stdout.write(f"API_V3_KEY : len={len(cfg.get('API_V3_KEY') or '')}")
key_path = cfg.get('PRIVATE_KEY_PATH') or ''
self.stdout.write(f"PRIVATE_KEY : {key_path}")
self.stdout.write(f"key_exists : {os.path.isfile(key_path)}")
cert_dir = os.path.dirname(os.path.abspath(key_path)) if key_path else ''
self.stdout.write(f"cert_dir : {cert_dir}")
if cert_dir and os.path.isdir(cert_dir):
self.stdout.write('cert_dir files:')
for name in sorted(os.listdir(cert_dir)):
if name.lower().endswith('.pem'):
self.stdout.write(f' - {name}')
serial_from_match = resolve_merchant_cert_serial(key_path, '')
self.stdout.write(f"matched_serial: {serial_from_match or '(未匹配到证书)'}")
if key_path and os.path.isfile(key_path):
try:
private_key = _load_pem_private_key(key_path)
self.stdout.write('private_key : 加载成功')
cert_path = key_path.replace('apiclient_key.pem', 'apiclient_cert.pem')
if os.path.isfile(cert_path):
from cryptography import x509
from cryptography.hazmat.backends import default_backend
with open(cert_path, 'rb') as f:
cert = x509.load_pem_x509_certificate(f.read(), default_backend())
matched = _public_keys_match(private_key, cert.public_key())
self.stdout.write(
f"apiclient_cert.pem 与私钥匹配: {matched} "
f"serial={format(cert.serial_number, 'X').upper()}"
)
else:
self.stdout.write(self.style.ERROR(
f'缺少 apiclient_cert.pem（与 apiclient_key.pem 同目录）: {cert_path}'
))
except Exception as exc:
self.stdout.write(self.style.ERROR(f'私钥/证书检查失败: {exc}'))
# 试探请求：查一个不存在的单号，仅验证签名是否通过（404=签名OK，401 SIGN_ERROR=签名失败）
url = 'https://api.mch.weixin.qq.com/v3/fund-app/mch-transfer/transfer-bills/out-bill-no/WX_V3_DIAG_TEST'
auth = build_authorization('GET', url, '', club_id=club_id)
try:
resp = requests.get(
url,
headers={
'Authorization': auth,
'Accept': 'application/json',
'User-Agent': 'Mozilla/5.0',
},
timeout=10,
)
self.stdout.write(f"\n探针 HTTP {resp.status_code}: {resp.text[:300]}")
if resp.status_code == 401:
self.stdout.write(self.style.ERROR(
'签名仍失败：请确认服务器 apiclient_key.pem / apiclient_cert.pem '
'是商户号 1747846572 在微信商户平台下载的那一对'
))
elif resp.status_code in (404, 200):
self.stdout.write(self.style.SUCCESS('签名已通过（404/200 表示微信接受了商户签名）'))
except requests.RequestException as exc:
self.stdout.write(self.style.ERROR(f'探针请求失败: {exc}'))

101
jituan/services/wechat_pay.py
View File

@@ -41,34 +41,93 @@ def _pick_club_field(club, cfg_json, *field_names):
return ''
def _load_pem_private_key(private_key_path):
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
with open(private_key_path, 'rb') as f:
return serialization.load_pem_private_key(
f.read(), password=None, backend=default_backend(),
)
def _public_keys_match(private_key, cert_public_key):
try:
return private_key.public_key().public_numbers() == cert_public_key.public_numbers()
except Exception:
return False
def resolve_merchant_cert_serial(private_key_path, configured_serial=''):
"""从 apiclient_cert.pem 读取与私钥匹配的证书序列号（避免手填错误）。"""
"""
扫描私钥同目录下所有 PEM，找到与 apiclient_key.pem 公钥匹配的商户证书，
返回其序列号。手填/数据库里的序列号不可信，必须以私钥配对证书为准。
"""
configured = (configured_serial or '').strip().upper()
if not private_key_path:
logger.error('[WX_V3_CONFIG] 未配置私钥路径')
return configured
cert_dir = os.path.dirname(private_key_path)
if not os.path.isfile(private_key_path):
logger.error('[WX_V3_CONFIG] 私钥文件不存在 path=%s', private_key_path)
return configured
try:
private_key = _load_pem_private_key(private_key_path)
except Exception as exc:
logger.error('[WX_V3_CONFIG] 私钥加载失败 path=%s err=%s', private_key_path, exc)
return configured
cert_dir = os.path.dirname(os.path.abspath(private_key_path))
candidates = [
private_key_path.replace('apiclient_key.pem', 'apiclient_cert.pem'),
os.path.join(cert_dir, 'apiclient_cert.pem'),
os.path.join(cert_dir, 'wechatpay_cert.pem'),
os.path.join(cert_dir, 'cert.pem'),
]
try:
for fname in sorted(os.listdir(cert_dir)):
if not fname.lower().endswith('.pem'):
continue
if 'key' in fname.lower():
continue
candidates.append(os.path.join(cert_dir, fname))
except OSError as exc:
logger.warning('[WX_V3_CONFIG] 扫描证书目录失败 dir=%s err=%s', cert_dir, exc)
seen = set()
from cryptography import x509
from cryptography.hazmat.backends import default_backend
for path in candidates:
if not path or path in seen:
path = os.path.abspath(path)
if path in seen or not os.path.isfile(path):
continue
seen.add(path)
if not os.path.isfile(path):
continue
try:
from cryptography import x509
from cryptography.hazmat.backends import default_backend
if os.path.samefile(path, private_key_path):
continue
except OSError:
pass
try:
with open(path, 'rb') as f:
cert = x509.load_pem_x509_certificate(f.read(), default_backend())
data = f.read()
if b'PRIVATE KEY' in data:
continue
cert = x509.load_pem_x509_certificate(data, default_backend())
if not _public_keys_match(private_key, cert.public_key()):
continue
serial = format(cert.serial_number, 'X').upper()
logger.warning('[WX_V3_CONFIG] 从证书文件读取序列号 path=%s serial=%s', path, serial)
logger.warning(
'[WX_V3_CONFIG] 私钥与证书已匹配 cert=%s serial=%s (configured=%s)',
path, serial, configured or '(none)',
)
return serial
except Exception as exc:
logger.warning('[WX_V3_CONFIG] 读取证书序列号失败 path=%s err=%s', path, exc)
logger.debug('[WX_V3_CONFIG] 跳过证书 path=%s err=%s', path, exc)
logger.error(
'[WX_V3_CONFIG] 未找到与私钥匹配的 apiclient_cert.pem dir=%s key=%s configured=%s',
cert_dir, private_key_path, configured,
)
return configured
@@ -315,14 +374,15 @@ def get_wechat_v3_config(club_id=None):
mch_id = _pick_club_field(club, cfg_json, 'mch_id') or xq.get('mch_id', '')
api_v3_key = _pick_club_field(club, cfg_json, 'api_v3_key') or xq.get('api_v3_key', '')
cert_serial = (
_pick_club_field(club, cfg_json, 'cert_serial_no')
or xq.get('cert_serial_no', '')
)
private_key_path = (
_pick_club_field(club, cfg_json, 'private_key_path')
or path_defaults['PRIVATE_KEY_PATH']
)
# 星阙私钥路径固定用 settings（club 表可能填了旧商户 1746545364 的证书路径）
private_key_path = path_defaults['PRIVATE_KEY_PATH']
if cid != CLUB_ID_DEFAULT:
private_key_path = (
_pick_club_field(club, cfg_json, 'private_key_path')
or private_key_path
)
platform_cert_dir = (
_pick_club_field(club, cfg_json, 'platform_cert_dir')
or path_defaults['PLATFORM_CERT_DIR']
@@ -336,7 +396,10 @@ def get_wechat_v3_config(club_id=None):
if not appid and cid == CLUB_ID_DEFAULT:
appid = xq.get('wx_appid', '')
cert_serial = resolve_merchant_cert_serial(private_key_path, cert_serial)
# 序列号必须从与私钥配对的 apiclient_cert.pem 读取，禁止盲用数据库/写死值
cert_serial = resolve_merchant_cert_serial(private_key_path, '')
if not cert_serial:
cert_serial = xq.get('cert_serial_no', '')
cfg = {
'APPID': appid,

7
utils/wechat_v3.py
View File

@@ -1,3 +1,4 @@
import logging
import re
import time
import hashlib
@@ -13,6 +14,8 @@ 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
@@ -39,6 +42,10 @@ def build_authorization(method, url, body, club_id=None):
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