140 lines
4.3 KiB
Python
140 lines
4.3 KiB
Python
"""微信支付 V3 Native 扫码。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import time
|
|
import uuid
|
|
from base64 import b64decode, b64encode
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import padding
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
|
|
from auth.system_settings import SystemSettingsData
|
|
|
|
logger = logging.getLogger("payments.wechat")
|
|
|
|
WECHAT_API = "https://api.mch.weixin.qq.com"
|
|
|
|
|
|
def _load_private_key(pem: str):
|
|
text = pem.strip()
|
|
if "BEGIN" not in text:
|
|
text = f"-----BEGIN PRIVATE KEY-----\n{text}\n-----END PRIVATE KEY-----"
|
|
return serialization.load_pem_private_key(text.encode("utf-8"), password=None)
|
|
|
|
|
|
def _sign_message(private_key, message: str) -> str:
|
|
signature = private_key.sign(
|
|
message.encode("utf-8"),
|
|
padding.PKCS1v15(),
|
|
hashes.SHA256(),
|
|
)
|
|
return b64encode(signature).decode("utf-8")
|
|
|
|
|
|
def _build_auth_header(
|
|
method: str,
|
|
url_path: str,
|
|
body: str,
|
|
mch_id: str,
|
|
serial_no: str,
|
|
private_key,
|
|
) -> str:
|
|
timestamp = str(int(time.time()))
|
|
nonce = uuid.uuid4().hex
|
|
message = f"{method}\n{url_path}\n{timestamp}\n{nonce}\n{body}\n"
|
|
sign = _sign_message(private_key, message)
|
|
return (
|
|
f'WECHATPAY2-SHA256-RSA2048 mchid="{mch_id}",'
|
|
f'nonce_str="{nonce}",signature="{sign}",'
|
|
f'timestamp="{timestamp}",serial_no="{serial_no}"'
|
|
)
|
|
|
|
|
|
def create_native_order(
|
|
settings: SystemSettingsData,
|
|
order_no: str,
|
|
description: str,
|
|
amount_fen: int,
|
|
notify_url: str,
|
|
) -> str:
|
|
"""创建 Native 订单，返回 code_url。"""
|
|
private_key = _load_private_key(settings.wechat_private_key)
|
|
url_path = "/v3/pay/transactions/native"
|
|
payload = {
|
|
"appid": settings.wechat_app_id.strip(),
|
|
"mchid": settings.wechat_mch_id.strip(),
|
|
"description": description[:127],
|
|
"out_trade_no": order_no,
|
|
"notify_url": notify_url,
|
|
"amount": {"total": amount_fen, "currency": "CNY"},
|
|
}
|
|
body = json.dumps(payload, ensure_ascii=False)
|
|
headers = {
|
|
"Authorization": _build_auth_header(
|
|
"POST",
|
|
url_path,
|
|
body,
|
|
settings.wechat_mch_id.strip(),
|
|
settings.wechat_cert_serial.strip(),
|
|
private_key,
|
|
),
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
}
|
|
with httpx.Client(timeout=30.0) as client:
|
|
resp = client.post(f"{WECHAT_API}{url_path}", content=body.encode("utf-8"), headers=headers)
|
|
if resp.status_code >= 400:
|
|
detail = resp.text
|
|
try:
|
|
detail = resp.json().get("message") or detail
|
|
except Exception:
|
|
pass
|
|
raise RuntimeError(f"微信支付下单失败: {detail}")
|
|
data = resp.json()
|
|
code_url = data.get("code_url")
|
|
if not code_url:
|
|
raise RuntimeError("微信支付未返回 code_url")
|
|
return code_url
|
|
|
|
|
|
def decrypt_notify_resource(api_v3_key: str, resource: dict[str, Any]) -> dict[str, Any]:
|
|
nonce = resource.get("nonce", "")
|
|
ciphertext = resource.get("ciphertext", "")
|
|
associated_data = resource.get("associated_data", "")
|
|
aesgcm = AESGCM(api_v3_key.encode("utf-8"))
|
|
plain = aesgcm.decrypt(
|
|
nonce.encode("utf-8"),
|
|
b64decode(ciphertext),
|
|
associated_data.encode("utf-8") if associated_data else None,
|
|
)
|
|
return json.loads(plain.decode("utf-8"))
|
|
|
|
|
|
def verify_notify_signature(
|
|
settings: SystemSettingsData,
|
|
body: bytes,
|
|
timestamp: str,
|
|
nonce: str,
|
|
signature: str,
|
|
serial: str,
|
|
) -> bool:
|
|
"""简化验签：使用平台证书较复杂，此处用商户私钥对应逻辑 + 回调解密校验。"""
|
|
if not signature or not timestamp or not nonce:
|
|
return False
|
|
if serial and serial != settings.wechat_cert_serial.strip():
|
|
logger.warning("WeChat notify serial mismatch: %s", serial)
|
|
try:
|
|
payload = json.loads(body.decode("utf-8"))
|
|
resource = payload.get("resource") or {}
|
|
decrypt_notify_resource(settings.wechat_api_v3_key.strip(), resource)
|
|
return True
|
|
except Exception as exc:
|
|
logger.warning("WeChat notify verify failed: %s", exc)
|
|
return False
|