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

修复多商户轮换死锁，并加固回调解密与证书校验。

Browse Source 
余额不足等明确错误在查单宽限 PENDING 时会卡死不换号；现确认无单后切换，停用商户仍可解回调、请求带 notify_url。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
XingQue
2026-07-14 03:28:42 +08:00
parent ca4b52217f
commit 537e538528
3 changed files with 121 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
utils/wechat_mch_service.py
View File

@@ -45,11 +45,29 @@ def row_to_wx_cfg(row) -> Dict:
}
def wx_cfg_is_complete(cfg: Optional[Dict]) -> bool:
def wx_cfg_is_complete(cfg: Optional[Dict], *, require_key_file: bool = True) -> bool:
if not cfg:
return False
required = ('MCHID', 'APPID', 'API_V3_KEY', 'PRIVATE_KEY_PATH', 'CERT_SERIAL_NO', 'TRANSFER_SCENE_ID')
return all(str(cfg.get(k) or '').strip() for k in required)
if not all(str(cfg.get(k) or '').strip() for k in required):
return False
if require_key_file:
pk = str(cfg.get('PRIVATE_KEY_PATH') or '').strip()
if not os.path.isfile(pk):
logger.warning('商户私钥文件不存在已跳过 mch_id=%s path=%s', cfg.get('MCHID'), pk)
return False
return True
def _rows_to_complete_cfgs(rows) -> List[Dict]:
result = []
for row in rows:
cfg = row_to_wx_cfg(row)
if wx_cfg_is_complete(cfg):
result.append(cfg)
else:
logger.warning('商户配置不完整已跳过 mch_id=%s id=%s', row.mch_id, row.id)
return result
def list_enabled_wx_cfgs() -> List[Dict]:
@@ -59,13 +77,7 @@ def list_enabled_wx_cfgs() -> List[Dict]:
rows = list(
WechatPayMchConfig.objects.filter(enabled=True).order_by('priority', 'id')
)
result = []
for row in rows:
cfg = row_to_wx_cfg(row)
if wx_cfg_is_complete(cfg):
result.append(cfg)
else:
logger.warning('商户配置不完整已跳过 mch_id=%s id=%s', row.mch_id, row.id)
result = _rows_to_complete_cfgs(rows)
if result:
return result
@@ -76,7 +88,29 @@ def list_enabled_wx_cfgs() -> List[Dict]:
return []
def list_wx_cfgs_for_callback() -> List[Dict]:
"""
回调验签/解密：包含已停用商户。
解密只需 API_V3_KEY；验签缺平台公钥时会自动跳过该户。
"""
from peizhi.models import WechatPayMchConfig
rows = list(WechatPayMchConfig.objects.all().order_by('priority', 'id'))
result = []
for row in rows:
cfg = row_to_wx_cfg(row)
if str(cfg.get('MCHID') or '').strip() and str(cfg.get('API_V3_KEY') or '').strip():
result.append(cfg)
if result:
return result
fb = _settings_fallback_cfg()
if str(fb.get('MCHID') or '').strip() and str(fb.get('API_V3_KEY') or '').strip():
return [fb]
return []
def get_wx_cfg_by_mch_id(mch_id: str) -> Optional[Dict]:
"""按 mch_id 取配置（含停用）；查单需可签名，故要求私钥文件存在。"""
mch_id = (mch_id or '').strip()
if not mch_id:
return None

8
utils/wechat_v3.py
View File

@@ -102,8 +102,8 @@ def verify_wechat_sign(headers, body, wx_cfg=None):
if wx_cfg:
cfgs = [wx_cfg]
else:
from utils.wechat_mch_service import list_enabled_wx_cfgs
cfgs = list_enabled_wx_cfgs()
from utils.wechat_mch_service import list_wx_cfgs_for_callback
cfgs = list_wx_cfgs_for_callback()
for cfg in cfgs:
cert_path = _resolve_platform_pub_path(cfg, serial=serial)
@@ -139,10 +139,10 @@ def decrypt_callback_with_merchants(associated_data, nonce, ciphertext):
仅对启用商户密钥做 AES 尝试（通常很少），成功后返回 (plaintext, wx_cfg)。
业务侧再用 out_bill_no 反查记录上的 mch_id，对齐一致性。
"""
from utils.wechat_mch_service import list_enabled_wx_cfgs
from utils.wechat_mch_service import list_wx_cfgs_for_callback
last_err = None
for cfg in list_enabled_wx_cfgs():
for cfg in list_wx_cfgs_for_callback():
try:
text = decrypt_callback_ciphertext(associated_data, nonce, ciphertext, wx_cfg=cfg)
return text, cfg

78
yonghu/tixian_shenhe_views.py
View File

@@ -29,6 +29,7 @@ from .tixian_shenhe_services import (
ensure_shenhe_collect_completed,
get_audit_for_collect,
handle_post_transfer_failure,
query_wechat_transfer_bill,
reconcile_shenhe_wechat_bills,
reject_audit_and_refund,
release_collect_quota_for_record,
@@ -37,6 +38,7 @@ from .tixian_shenhe_services import (
shijidaozhang_exceeds_wechat_limit,
validate_collect_eligibility,
wechat_limit_collect_message,
_sync_record_from_wx_query,
)
logger = logging.getLogger('yonghu.tixian_shenhe')
@@ -65,15 +67,16 @@ def _is_wx_balance_insufficient(err_code='', err_msg=''):
def _is_wx_switchable_error(err_code='', err_msg=''):
"""
可立刻换下一商户重试的错误（本笔尚未进入 WAIT_USER_CONFIRM）。
仅白名单，避免误伤导致重复开单。
仅白名单，避免误伤导致重复开单。不含裸词 LIMIT，防误匹配。
"""
if _is_wx_balance_insufficient(err_code, err_msg):
return True
blob = f'{err_code} {err_msg}'.upper()
keywords = (
'FREQUENCY_LIMITED', 'RATE_LIMIT', 'QUOTA', 'LIMIT',
'FREQUENCY_LIMITED', 'RATE_LIMIT', 'QUOTA_EXCEEDED', 'QUOTA',
'日限额', '超出限额', '超过限额', '限额', '次数超限',
'ACCOUNT_ERROR', 'NO_AUTH', 'PERM', '商户号异常', '账户异常',
'ACCOUNT_ERROR', 'NO_AUTH', 'PERMISSION', '商户号异常', '账户异常',
'CERT_ERROR', 'PRIVATE_KEY', 'SIGN_ERROR', '商户证书', '私钥',
)
return any(k in blob for k in keywords)
@@ -120,7 +123,7 @@ def _build_wx_transfer_body(wx_cfg, tixian_id, openid, yonghuid, leixing, shijid
transfer_scene_report_infos = [
{'info_type': '采购商品名称', 'info_content': '平台服务'},
]
return {
body = {
'appid': wx_cfg['APPID'],
'out_bill_no': tixian_id,
'transfer_scene_id': transfer_scene_id,
@@ -129,6 +132,10 @@ def _build_wx_transfer_body(wx_cfg, tixian_id, openid, yonghuid, leixing, shijid
'transfer_remark': f'兄弟们 收款前记得录屏发快手',
'transfer_scene_report_infos': transfer_scene_report_infos,
}
notify_url = str(wx_cfg.get('NOTIFY_URL') or '').strip()
if notify_url:
body['notify_url'] = notify_url
return body
def _call_wechat_transfer(body, wx_cfg):
@@ -363,6 +370,28 @@ def process_audit_collect(request):
'code': 12,
'msg': '收款请求已提交，系统正在核对状态，请稍后再试，切勿重复点击',
})
except (OSError, IOError, ValueError) as e:
# 证书/私钥本地问题：未真正发到微信，可换下一商户
logger.error(
'微信打款本地配置异常: bill=%s mch=%s err=%s',
tixian_id, mch_id, e, exc_info=True,
)
err_code, err_msg = 'CERT_ERROR', f'商户证书/私钥异常: {e}'
last_user_msg = err_msg
last_was_balance = False
_mark_auto_record_failed(tixian_id, f'[{mch_id}] {err_msg}')
if mch_idx < len(mch_cfgs) - 1:
logger.warning(
'商户本地配置失败，切换下一商户 shenhe=%s from=%s idx=%s/%s',
shenhe_danhao, mch_id, mch_idx + 1, len(mch_cfgs),
)
continue
if quota_reserved:
release_collect_quota_for_record(
TixianAutoRecord.objects.get(tixian_id=tixian_id),
)
quota_reserved = False
return Response({'code': 99, 'msg': '微信商户配置异常，请联系管理员'})
if resp.status_code == 200 and wx_result.get('state') == 'WAIT_USER_CONFIRM':
package_info = wx_result.get('package_info')
@@ -408,6 +437,47 @@ def process_audit_collect(request):
'data': {'already_completed': True, 'tixianjilu_id': tixianjilu_id_val},
})
if post_action == RECONCILE_PENDING:
# 关键漏洞修复：微信已返回明确可切换业务错误时，查单宽限会得到 PENDING(not_found)，
# 若不处理会死卡「请稍后再试」永不换号。确认查单仍 not_found 后才换；查单不可用则仍禁换。
can_try_switch = (
_is_wx_switchable_error(err_code, err_msg)
and mch_idx < len(mch_cfgs) - 1
)
if can_try_switch:
wx_q = query_wechat_transfer_bill(tixian_id, wx_cfg=wx_cfg)
if wx_q is None:
return Response({
'code': 12,
'msg': '收款请求已提交，系统正在核对状态，请稍后再试，切勿重复点击',
})
if wx_q.get('_not_found'):
user_msg = err_msg or '微信收款发起失败，请稍后重试'
last_user_msg = user_msg
last_was_balance = _is_wx_balance_insufficient(err_code, err_msg)
_mark_auto_record_failed(tixian_id, f'[{mch_id}] {user_msg}')
logger.warning(
'商户转账失败(查单无单)，切换下一商户 shenhe=%s from=%s idx=%s/%s',
shenhe_danhao, mch_id, mch_idx + 1, len(mch_cfgs),
)
continue
# 微信侧其实已有单：按查单结果走，禁止盲换
ar = TixianAutoRecord.objects.get(tixian_id=tixian_id)
sync_action, sync_payload = _sync_record_from_wx_query(ar, wx_q)
if sync_action == RECONCILE_WAIT_CONFIRM:
if isinstance(sync_payload, dict):
sync_payload.setdefault('mch_id', mch_id)
sync_payload.setdefault('tixianjilu_id', tixianjilu_id_val)
quota_resp = _verify_collect_quota_or_response(user_main, audit, shenhe_danhao)
if quota_resp:
return quota_resp
return _build_collect_success_response(sync_payload)
if sync_action == RECONCILE_COMPLETED:
ensure_shenhe_collect_completed(shenhe_danhao)
return Response({
'code': 0,
'msg': '提现已成功到账',
'data': {'already_completed': True, 'tixianjilu_id': tixianjilu_id_val},
})
pending_msg = (
post_payload.get('msg', post_payload)
if isinstance(post_payload, dict) else post_payload