diff --git a/jituan/management/commands/wx_v3_pay_diag.py b/jituan/management/commands/wx_v3_pay_diag.py new file mode 100644 index 0000000..eca748e --- /dev/null +++ b/jituan/management/commands/wx_v3_pay_diag.py @@ -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}')) diff --git a/jituan/services/wechat_pay.py b/jituan/services/wechat_pay.py index cdd6cfd..d85f9cb 100644 --- a/jituan/services/wechat_pay.py +++ b/jituan/services/wechat_pay.py @@ -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, diff --git a/utils/wechat_v3.py b/utils/wechat_v3.py index 166d0fd..1c3d304 100644 --- a/utils/wechat_v3.py +++ b/utils/wechat_v3.py @@ -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