112 lines
4.5 KiB
Python
112 lines
4.5 KiB
Python
"""微信支付渠道（V3）：App 下单（服务端签名） + 回调验签/AES-GCM 解密。
|
||
|
||
密钥未配置（开发环境）时：下单返回 dev_mock 占位，回调一律拒绝入账。
|
||
生产接入需配置：WECHAT_APP_ID / WECHAT_MCHID / WECHAT_API_V3_KEY / WECHAT_PRIVATE_KEY（商户 API 私钥）
|
||
/ WECHAT_CERT_SERIAL_NO（商户证书序列号）/ WECHAT_PLATFORM_CERT（微信支付平台证书 PEM）/ WECHAT_NOTIFY_URL。
|
||
"""
|
||
import json
|
||
import time
|
||
import uuid
|
||
|
||
from app.config import settings
|
||
from app.payment.base import PaymentConfigError, rsa2_sign, rsa2_verify, wechat_decrypt_resource
|
||
|
||
APP_PATH = "/v3/pay/transactions/app"
|
||
APP_PACKAGE = "Sign=WXPay"
|
||
DEFAULT_SUBJECT = "积分充值"
|
||
HTTP_TIMEOUT = 10
|
||
|
||
|
||
def is_configured() -> bool:
|
||
return bool(
|
||
settings.wechat_mchid and settings.wechat_app_id
|
||
and settings.wechat_api_v3_key and settings.wechat_private_key
|
||
)
|
||
|
||
|
||
def _authorization(method: str, url_path: str, body: str) -> str:
|
||
"""构造 V3 请求签名的 Authorization 头（WECHATPAY2-SHA256-RSA2048）"""
|
||
timestamp = str(int(time.time()))
|
||
nonce = uuid.uuid4().hex
|
||
message = f"{method}\n{url_path}\n{timestamp}\n{nonce}\n{body}\n"
|
||
signature = rsa2_sign(message, settings.wechat_private_key)
|
||
return (
|
||
f'WECHATPAY2-SHA256-RSA2048 mchid="{settings.wechat_mchid}",'
|
||
f'nonce_str="{nonce}",signature="{signature}",'
|
||
f'timestamp="{timestamp}",serial_no="{settings.wechat_cert_serial_no}"'
|
||
)
|
||
|
||
|
||
async def _create_prepay_id(order_no: str, amount_fen: int) -> str:
|
||
"""V3 下单（App 支付）→ prepay_id"""
|
||
import httpx # 延迟导入：未配置渠道时无需依赖
|
||
|
||
body = json.dumps({
|
||
"appid": settings.wechat_app_id,
|
||
"mchid": settings.wechat_mchid,
|
||
"description": DEFAULT_SUBJECT,
|
||
"out_trade_no": order_no,
|
||
"notify_url": settings.wechat_notify_url,
|
||
"amount": {"total": amount_fen, "currency": "CNY"},
|
||
}, ensure_ascii=False, separators=(",", ":"))
|
||
|
||
headers = {
|
||
"Authorization": _authorization("POST", APP_PATH, body),
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json",
|
||
"User-Agent": "video-server/1.0",
|
||
}
|
||
async with httpx.AsyncClient(base_url=settings.wechat_api_base, timeout=HTTP_TIMEOUT) as client:
|
||
resp = await client.post(APP_PATH, content=body.encode("utf-8"), headers=headers)
|
||
if resp.status_code != 200:
|
||
raise PaymentConfigError(f"微信下单失败: {resp.status_code} {resp.text[:200]}")
|
||
prepay_id = (resp.json() or {}).get("prepay_id") or ""
|
||
if not prepay_id:
|
||
raise PaymentConfigError("微信下单未返回 prepay_id")
|
||
return prepay_id
|
||
|
||
|
||
async def create_app_pay_params(order_no: str, amount_fen: int) -> dict:
|
||
"""生成 App SDK 调起参数（未配置密钥时返回 dev 占位）"""
|
||
if not is_configured():
|
||
return {"dev_mock": True, "prepayid": f"mock-{order_no}"}
|
||
|
||
prepay_id = await _create_prepay_id(order_no, amount_fen)
|
||
# App 调起需二次签名（不同于接口签名串）
|
||
timestamp = str(int(time.time()))
|
||
noncestr = uuid.uuid4().hex
|
||
sign_str = (
|
||
f"{settings.wechat_app_id}\n{settings.wechat_mchid}\n{prepay_id}\n"
|
||
f"{APP_PACKAGE}\n{noncestr}\n{timestamp}\n"
|
||
)
|
||
return {
|
||
"appid": settings.wechat_app_id,
|
||
"partnerid": settings.wechat_mchid,
|
||
"prepayid": prepay_id,
|
||
"package": APP_PACKAGE,
|
||
"noncestr": noncestr,
|
||
"timestamp": timestamp,
|
||
"sign": rsa2_sign(sign_str, settings.wechat_private_key),
|
||
}
|
||
|
||
|
||
def verify_callback(headers: dict, body: bytes) -> dict | None:
|
||
"""V3 回调验签 + AES-GCM 解密，成功返回明文报文 dict，失败返回 None（未配置一律 None）"""
|
||
if not is_configured() or not settings.wechat_platform_cert:
|
||
return None
|
||
timestamp = headers.get("wechatpay-timestamp", "")
|
||
nonce = headers.get("wechatpay-nonce", "")
|
||
signature = headers.get("wechatpay-signature", "")
|
||
if not (timestamp and nonce and signature):
|
||
return None
|
||
# 验签串：timestamp\nnonce\nbody\n（body 为原始报文，勿用解析后结果）
|
||
message = f"{timestamp}\n{nonce}\n{body.decode('utf-8')}\n"
|
||
if not rsa2_verify(message, signature, settings.wechat_platform_cert):
|
||
return None
|
||
try:
|
||
payload = json.loads(body.decode("utf-8"))
|
||
except Exception: # noqa: BLE001
|
||
return None
|
||
resource = payload.get("resource") or {}
|
||
return wechat_decrypt_resource(resource, settings.wechat_api_v3_key)
|