346 lines
12 KiB
Python
346 lines
12 KiB
Python
"""支付宝支付（RSA2）配置、签名、验签与 wap.pay 跳转 URL 生成。
|
||
|
||
沙箱/生产由 settings.ALIPAY_SANDBOX 开关控制，配置存 app_secrets.py。
|
||
多 club 暂回落全局配置（与 wechat_pay 一致），未来再迁移 Club 表。
|
||
"""
|
||
import base64
|
||
import json
|
||
import logging
|
||
import time
|
||
from urllib.parse import quote
|
||
|
||
from django.conf import settings
|
||
|
||
from cryptography.hazmat.primitives import hashes, serialization
|
||
from cryptography.hazmat.primitives.asymmetric import padding
|
||
|
||
from jituan.constants import CLUB_ID_DEFAULT
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ==================== 配置读取 ====================
|
||
|
||
def get_alipay_config(club_id=None):
|
||
"""
|
||
读取支付宝配置。
|
||
根据 settings.ALIPAY_SANDBOX 开关返回沙箱或生产配置。
|
||
club_id 暂未使用（预留多 club 扩展），统一回落全局配置。
|
||
"""
|
||
sandbox = getattr(settings, 'ALIPAY_SANDBOX', True)
|
||
if sandbox:
|
||
cfg = getattr(settings, 'ALIPAY_SANDBOX_CFG', {})
|
||
else:
|
||
cfg = getattr(settings, 'ALIPAY_PROD', {})
|
||
|
||
return {
|
||
'APP_ID': cfg.get('APP_ID', ''),
|
||
'PID': cfg.get('PID', ''),
|
||
'APP_PRIVATE_KEY': cfg.get('APP_PRIVATE_KEY', ''),
|
||
'ALIPAY_PUBLIC_KEY': cfg.get('ALIPAY_PUBLIC_KEY', ''),
|
||
'GATEWAY': cfg.get('GATEWAY', 'https://openapi.alipay.com/gateway.do'),
|
||
'NOTIFY_URL': cfg.get('NOTIFY_URL', ''),
|
||
'RETURN_URL': cfg.get('RETURN_URL', ''),
|
||
'sandbox': sandbox,
|
||
}
|
||
|
||
|
||
# ==================== 密钥加载 ====================
|
||
|
||
def _load_private_key(key_str):
|
||
"""加载 RSA 私钥，兼容 PEM、PKCS#8 DER、PKCS#1、PKCS#8 多种格式。
|
||
|
||
沙箱私钥常为 PKCS#1 格式（裸 base64 前缀 MIIEowIBAAKCAQEA），
|
||
生产私钥常为 PKCS#8 格式（裸 base64 前缀 MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcw），
|
||
此处按顺序尝试多种加载方式以兼容两者。
|
||
"""
|
||
if isinstance(key_str, bytes):
|
||
key_str = key_str.decode('utf-8')
|
||
|
||
key_str = key_str.strip()
|
||
|
||
# 1. 标准 PEM（含 -----BEGIN 头）
|
||
if '-----BEGIN' in key_str:
|
||
return serialization.load_pem_private_key(
|
||
key_str.encode('utf-8'), password=None,
|
||
)
|
||
|
||
raw = key_str.replace('\n', '').replace('\r', '').replace(' ', '')
|
||
|
||
# 2. PKCS#8 DER（裸 base64）
|
||
try:
|
||
der_bytes = base64.b64decode(raw)
|
||
return serialization.load_der_private_key(der_bytes, password=None)
|
||
except Exception:
|
||
pass
|
||
|
||
# 3. PKCS#1 PEM 包装（-----BEGIN RSA PRIVATE KEY-----）
|
||
pem_pkcs1 = _wrap_pem(raw, 'RSA PRIVATE KEY')
|
||
try:
|
||
return serialization.load_pem_private_key(pem_pkcs1.encode('utf-8'), password=None)
|
||
except Exception:
|
||
pass
|
||
|
||
# 4. PKCS#8 PEM 包装（-----BEGIN PRIVATE KEY-----）
|
||
pem_pkcs8 = _wrap_pem(raw, 'PRIVATE KEY')
|
||
return serialization.load_pem_private_key(pem_pkcs8.encode('utf-8'), password=None)
|
||
|
||
|
||
def _load_public_key(key_str):
|
||
"""加载支付宝公钥，兼容 PEM、X.509 DER 及 PEM 包装格式。
|
||
|
||
支付宝公钥标准为 X.509 SubjectPublicKeyInfo，裸 base64 前缀通常为
|
||
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8A，load_der_public_key 原生支持；
|
||
为稳妥起见追加 PEM 包装兜底。
|
||
"""
|
||
if not key_str:
|
||
return None
|
||
if isinstance(key_str, bytes):
|
||
key_str = key_str.decode('utf-8')
|
||
|
||
key_str = key_str.strip()
|
||
|
||
# 1. 标准 PEM（含 -----BEGIN 头）
|
||
if '-----BEGIN' in key_str:
|
||
return serialization.load_pem_public_key(key_str.encode('utf-8'))
|
||
|
||
raw = key_str.replace('\n', '').replace('\r', '').replace(' ', '')
|
||
|
||
# 2. X.509 DER（裸 base64）
|
||
try:
|
||
der_bytes = base64.b64decode(raw)
|
||
return serialization.load_der_public_key(der_bytes)
|
||
except Exception:
|
||
pass
|
||
|
||
# 3. PEM 包装兜底（-----BEGIN PUBLIC KEY-----）
|
||
pem_pub = _wrap_pem(raw, 'PUBLIC KEY')
|
||
return serialization.load_pem_public_key(pem_pub.encode('utf-8'))
|
||
|
||
|
||
def _wrap_pem(raw_b64, label):
|
||
"""将裸 base64 字符串包装为标准 PEM 格式（每行 64 字符）。"""
|
||
lines = [raw_b64[i:i + 64] for i in range(0, len(raw_b64), 64)]
|
||
return f'-----BEGIN {label}-----\n' + '\n'.join(lines) + f'\n-----END {label}-----'
|
||
|
||
|
||
# ==================== 签名 / 验签 ====================
|
||
|
||
def _build_sign_content(params):
|
||
"""构建待签名字符串：按 key 字典序排列，仅排除 sign，空值不参与。
|
||
|
||
注意：sign_type 必须参与签名（支付宝验签字符串包含 sign_type）。
|
||
"""
|
||
sorted_keys = sorted(k for k in params if k != 'sign')
|
||
return '&'.join(f'{k}={params[k]}' for k in sorted_keys if params[k] != '')
|
||
|
||
|
||
def sign_rsa2(content, private_key_str):
|
||
"""RSA2 (SHA256WithRSA) 签名，返回 Base64 编码的签名字符串。"""
|
||
private_key = _load_private_key(private_key_str)
|
||
signature = private_key.sign(
|
||
content.encode('utf-8'),
|
||
padding.PKCS1v15(),
|
||
hashes.SHA256(),
|
||
)
|
||
return base64.b64encode(signature).decode('ascii')
|
||
|
||
|
||
def verify_rsa2(content, sign_b64, public_key_str):
|
||
"""使用支付宝公钥验证 RSA2 签名。"""
|
||
public_key = _load_public_key(public_key_str)
|
||
if public_key is None:
|
||
return False
|
||
try:
|
||
signature = base64.b64decode(sign_b64)
|
||
public_key.verify(
|
||
signature,
|
||
content.encode('utf-8'),
|
||
padding.PKCS1v15(),
|
||
hashes.SHA256(),
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
logger.warning('支付宝验签失败: %s', e)
|
||
return False
|
||
|
||
|
||
def verify_notify_params(params, club_id=None):
|
||
"""
|
||
验证支付宝异步回调签名。
|
||
params 为已解析的字典（来自 POST form 数据）。
|
||
|
||
注意：回调验签与请求签名规则不同！
|
||
- 请求签名：sign_type 参与签名（只排除 sign）
|
||
- 回调验签：sign_type 不参与签名（排除 sign 和 sign_type）
|
||
支付宝官方文档：通知返回参数中除去 sign、sign_type 两个参数外，皆是待验签参数。
|
||
"""
|
||
cfg = get_alipay_config(club_id)
|
||
sign = params.get('sign', '')
|
||
if not sign:
|
||
logger.warning('[ALIPAY_VERIFY_DEBUG] sign 为空')
|
||
return False
|
||
sign_type = params.get('sign_type', '')
|
||
if sign_type and sign_type.upper() != 'RSA2':
|
||
logger.warning(f'[ALIPAY_VERIFY_DEBUG] sign_type 非 RSA2: {sign_type}')
|
||
return False
|
||
# 回调验签：排除 sign 和 sign_type（支付宝官方规则）
|
||
sorted_keys = sorted(k for k in params if k not in ('sign', 'sign_type'))
|
||
sign_content = '&'.join(f'{k}={params[k]}' for k in sorted_keys if params[k] != '')
|
||
logger.warning(f'[ALIPAY_VERIFY_DEBUG] sign_content(前300)={sign_content[:300]}')
|
||
logger.warning(f'[ALIPAY_VERIFY_DEBUG] sign(前50)={sign[:50]}')
|
||
result = verify_rsa2(sign_content, sign, cfg['ALIPAY_PUBLIC_KEY'])
|
||
logger.warning(f'[ALIPAY_VERIFY_DEBUG] verify_rsa2 result={result}')
|
||
return result
|
||
|
||
|
||
# ==================== wap.pay 跳转 URL ====================
|
||
|
||
def build_wap_pay_url(out_trade_no, amount, subject, body='', club_id=None, expire_minutes=30):
|
||
"""
|
||
生成 alipay.trade.wap.pay 跳转 URL（GET 方式）。
|
||
|
||
参数：
|
||
out_trade_no: 商户订单号（对应 Order.OrderID）
|
||
amount: 订单金额（Decimal 或字符串，单位元）
|
||
subject: 订单标题
|
||
body: 订单描述（可选）
|
||
club_id: 预留多 club（暂未使用）
|
||
expire_minutes: 超时关闭时间（分钟）
|
||
|
||
返回：
|
||
完整的支付宝跳转 URL，前端 web-view 直接加载即可。
|
||
"""
|
||
cfg = get_alipay_config(club_id)
|
||
|
||
biz_content = {
|
||
'out_trade_no': out_trade_no,
|
||
'total_amount': str(amount),
|
||
'subject': subject or '订单支付',
|
||
'body': body or '',
|
||
'product_code': 'QUICK_WAP_WAY',
|
||
'timeout_express': f'{expire_minutes}m',
|
||
}
|
||
|
||
params = {
|
||
'app_id': cfg['APP_ID'],
|
||
'method': 'alipay.trade.wap.pay',
|
||
'charset': 'utf-8',
|
||
'sign_type': 'RSA2',
|
||
'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
|
||
'version': '1.0',
|
||
'biz_content': json.dumps(biz_content, separators=(',', ':'), ensure_ascii=False),
|
||
'notify_url': cfg['NOTIFY_URL'],
|
||
'return_url': cfg['RETURN_URL'],
|
||
}
|
||
|
||
# 签名
|
||
sign_content = _build_sign_content(params)
|
||
params['sign'] = sign_rsa2(sign_content, cfg['APP_PRIVATE_KEY'])
|
||
|
||
# 拼接 GET URL
|
||
parts = []
|
||
for k in sorted(params.keys()):
|
||
parts.append(f'{k}={quote(str(params[k]), safe="")}')
|
||
query_string = '&'.join(parts)
|
||
|
||
return f"{cfg['GATEWAY']}?{query_string}"
|
||
|
||
|
||
# ==================== 当面付二维码（precreate）====================
|
||
|
||
def build_precreate_qr(out_trade_no, amount, subject, body='', club_id=None, expire_minutes=30):
|
||
"""
|
||
调用 alipay.trade.precreate 生成当面付二维码链接。
|
||
|
||
与 wap.pay 不同：precreate 是服务器主动 POST 调用，返回 qr_code 链接，
|
||
由商户自行生成二维码展示；用户用支付宝 APP 扫码完成支付。
|
||
适用于微信小程序等无法跳转 alipay:// scheme 的环境。
|
||
|
||
返回:
|
||
dict: {'qr_code': 'https://qr.alipay.com/...', 'out_trade_no': ...}
|
||
失败抛异常。
|
||
"""
|
||
import requests
|
||
|
||
cfg = get_alipay_config(club_id)
|
||
|
||
biz_content = {
|
||
'out_trade_no': out_trade_no,
|
||
'total_amount': str(amount),
|
||
'subject': subject or '订单支付',
|
||
'body': body or '',
|
||
'timeout_express': f'{expire_minutes}m',
|
||
}
|
||
|
||
params = {
|
||
'app_id': cfg['APP_ID'],
|
||
'method': 'alipay.trade.precreate',
|
||
'charset': 'utf-8',
|
||
'sign_type': 'RSA2',
|
||
'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
|
||
'version': '1.0',
|
||
'biz_content': json.dumps(biz_content, separators=(',', ':'), ensure_ascii=False),
|
||
'notify_url': cfg['NOTIFY_URL'],
|
||
}
|
||
|
||
# 签名（sign_type 参与签名，与 wap.pay 一致）
|
||
sign_content = _build_sign_content(params)
|
||
params['sign'] = sign_rsa2(sign_content, cfg['APP_PRIVATE_KEY'])
|
||
|
||
# 服务器主动调用网关
|
||
resp = requests.post(cfg['GATEWAY'], data=params, timeout=10)
|
||
try:
|
||
result = resp.json()
|
||
except Exception:
|
||
raise Exception(f"支付宝 precreate 返回非 JSON: {resp.text[:500]}")
|
||
|
||
biz_resp = result.get('alipay_trade_precreate_response', {})
|
||
code = biz_resp.get('code', '')
|
||
|
||
if code == '10000':
|
||
return {
|
||
'qr_code': biz_resp.get('qr_code', ''),
|
||
'out_trade_no': biz_resp.get('out_trade_no', out_trade_no),
|
||
}
|
||
|
||
# 失败：拼接完整错误信息
|
||
sub_code = biz_resp.get('sub_code', '')
|
||
sub_msg = biz_resp.get('sub_msg', '')
|
||
raise Exception(
|
||
f"支付宝预下单失败: code={code}, msg={biz_resp.get('msg', '')}, "
|
||
f"sub_code={sub_code}, sub_msg={sub_msg}"
|
||
)
|
||
|
||
|
||
def qr_link_to_base64(qr_link):
|
||
"""
|
||
把二维码链接内容转成 PNG base64 data URI（供前端 image 直接展示）。
|
||
依赖 qrcode + Pillow；缺失时抛 ImportError 提示安装。
|
||
"""
|
||
import io
|
||
import base64
|
||
try:
|
||
import qrcode
|
||
except ImportError:
|
||
raise ImportError("缺少 qrcode 库，请执行: pip install qrcode[pil]")
|
||
|
||
img = qrcode.make(qr_link)
|
||
buf = io.BytesIO()
|
||
img.save(buf, format='PNG')
|
||
return 'data:image/png;base64,' + base64.b64encode(buf.getvalue()).decode('ascii')
|
||
|
||
|
||
# ==================== 辅助 ====================
|
||
|
||
def club_id_from_order(order_id):
|
||
"""从订单反查 club_id（与 wechat_pay 一致，暂回落默认）。"""
|
||
from orders.models import Order
|
||
try:
|
||
row = Order.query.filter(OrderID=order_id).first()
|
||
if row and getattr(row, 'ClubID', None):
|
||
return row.ClubID
|
||
except Exception:
|
||
pass
|
||
return CLUB_ID_DEFAULT
|