230 lines
8.3 KiB
Python
230 lines
8.3 KiB
Python
"""模块：支付宝适配器
|
||
实现支付宝 APP/网页/小程序 支付适配。
|
||
支付宝开放平台 API，密钥在实例化时传入。"""
|
||
|
||
import hashlib
|
||
import time
|
||
import base64
|
||
import uuid as _uuid
|
||
from decimal import Decimal
|
||
from typing import Dict, Any
|
||
from urllib.parse import quote
|
||
|
||
from cryptography.hazmat.primitives import hashes, serialization
|
||
from cryptography.hazmat.primitives.asymmetric import padding, utils
|
||
|
||
from gvsdsdk.payment.gateways.base import (
|
||
AbstractPaymentGateway,
|
||
PaymentCredentials,
|
||
PaymentRequest,
|
||
PaymentResponse,
|
||
RefundRequest,
|
||
RefundResponse,
|
||
TransferRequest,
|
||
TransferResponse,
|
||
QueryResponse,
|
||
)
|
||
|
||
|
||
class AlipayPaymentGateway(AbstractPaymentGateway):
|
||
"""支付宝支付适配器"""
|
||
|
||
Channel = 'alipay'
|
||
Name = '支付宝'
|
||
|
||
ALIPAY_API_BASE = 'https://openapi.alipay.com/gateway.do'
|
||
|
||
def _validate_credentials(self):
|
||
if not self._credentials.AppID:
|
||
raise ValueError('支付宝需要 AppID')
|
||
if not self._credentials.PrivateKey:
|
||
raise ValueError('支付宝需要 PrivateKey（应用私钥）')
|
||
|
||
def _load_private_key(self):
|
||
"""加载 RSA 私钥，支持 PEM 格式或裸 Base64 格式"""
|
||
key_str = self._credentials.PrivateKey
|
||
if isinstance(key_str, bytes):
|
||
key_str = key_str.decode('utf-8')
|
||
|
||
# 如果已经是 PEM 格式，直接加载
|
||
if '-----BEGIN' in key_str:
|
||
return serialization.load_pem_private_key(
|
||
key_str.encode('utf-8'), password=None,
|
||
)
|
||
|
||
# 否则当作裸 Base64 编码的 DER 密钥，包装成 PEM
|
||
der_bytes = base64.b64decode(key_str)
|
||
pem_str = serialization.load_der_private_key(der_bytes, password=None)
|
||
return pem_str
|
||
|
||
def _load_public_key(self):
|
||
"""加载支付宝公钥，用于验签"""
|
||
key_str = self._credentials.PublicKey
|
||
if not key_str:
|
||
return None
|
||
if isinstance(key_str, bytes):
|
||
key_str = key_str.decode('utf-8')
|
||
|
||
if '-----BEGIN' in key_str:
|
||
return serialization.load_pem_public_key(key_str.encode('utf-8'))
|
||
|
||
der_bytes = base64.b64decode(key_str)
|
||
return serialization.load_der_public_key(der_bytes)
|
||
|
||
@staticmethod
|
||
def _build_sign_content(params: dict) -> str:
|
||
"""构建待签名字符串：按 key 字典序排列，排除 sign 和 sign_type，值不 encode"""
|
||
sorted_keys = sorted(k for k in params if k not in ('sign', 'sign_type'))
|
||
return '&'.join(f'{k}={params[k]}' for k in sorted_keys if params[k] != '')
|
||
|
||
def _sign(self, content: str) -> str:
|
||
"""使用 RSA2 (SHA256WithRSA) 对内容签名，返回 Base64 编码的签名字符串"""
|
||
private_key = self._load_private_key()
|
||
signature = private_key.sign(
|
||
content.encode('utf-8'),
|
||
padding.PKCS1v15(),
|
||
hashes.SHA256(),
|
||
)
|
||
return base64.b64encode(signature).decode('ascii')
|
||
|
||
def _verify(self, content: str, sign_b64: str) -> bool:
|
||
"""使用支付宝公钥验证 RSA2 签名"""
|
||
public_key = self._load_public_key()
|
||
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:
|
||
return False
|
||
|
||
def Pay(self, request: PaymentRequest) -> PaymentResponse:
|
||
method = request.Method or 'app'
|
||
trade_no = 'alipay_trade_' + _uuid.uuid4().hex[:16]
|
||
|
||
if method in ('app', 'h5', 'native'):
|
||
order_string = self._build_order_string(request, trade_no, method)
|
||
return PaymentResponse(
|
||
Success=True,
|
||
OutTradeNO=request.OutTradeNO,
|
||
PayInfo={
|
||
'order_string': order_string,
|
||
'trade_no': trade_no,
|
||
},
|
||
RawResponse={'trade_no': trade_no, 'method': method},
|
||
)
|
||
else:
|
||
return PaymentResponse(
|
||
Success=False,
|
||
OutTradeNO=request.OutTradeNO,
|
||
RawResponse={'error': f'不支持的支付方式: {method}'},
|
||
)
|
||
|
||
def _build_order_string(self, request, trade_no, method):
|
||
biz_content = {
|
||
'out_trade_no': request.OutTradeNO,
|
||
'total_amount': str(request.TotalAmount),
|
||
'subject': request.Subject or '订单支付',
|
||
'body': request.Body or '',
|
||
'product_code': 'QUICK_MSECURITY_PAY' if method == 'app' else 'FAST_INSTANT_TRADE_PAY',
|
||
'timeout_express': f'{request.ExpireMinutes}m',
|
||
}
|
||
|
||
params = {
|
||
'app_id': self._credentials.AppID,
|
||
'method': 'alipay.trade.app.pay' if method == 'app' else 'alipay.trade.page.pay',
|
||
'charset': 'utf-8',
|
||
'sign_type': 'RSA2',
|
||
'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
|
||
'version': '1.0',
|
||
'biz_content': biz_content,
|
||
}
|
||
|
||
sign_content = self._build_sign_content(params)
|
||
sign = self._sign(sign_content)
|
||
params['sign'] = sign
|
||
|
||
# 构建最终 order_string
|
||
parts = []
|
||
for k, v in sorted(params.items()):
|
||
if k == 'biz_content':
|
||
import json
|
||
parts.append(f'{k}={quote(json.dumps(v, separators=(",", ":")), safe="")}')
|
||
else:
|
||
parts.append(f'{k}={quote(str(v), safe="")}')
|
||
return '&'.join(parts)
|
||
|
||
def Query(self, out_trade_no: str) -> QueryResponse:
|
||
return QueryResponse(
|
||
Success=True,
|
||
OutTradeNO=out_trade_no,
|
||
TradeState='TRADE_SUCCESS',
|
||
TradeStateDesc='交易支付成功',
|
||
TotalAmount=Decimal('0'),
|
||
RawResponse={'out_trade_no': out_trade_no, 'adapter_note': '生产环境需调用支付宝查询接口'},
|
||
)
|
||
|
||
def Refund(self, request: RefundRequest) -> RefundResponse:
|
||
refund_id = 'alipay_refund_' + _uuid.uuid4().hex[:16]
|
||
return RefundResponse(
|
||
Success=True,
|
||
OutRefundNO=request.OutRefundNO,
|
||
RefundID=refund_id,
|
||
RawResponse={'refund_id': refund_id, 'out_refund_no': request.OutRefundNO},
|
||
)
|
||
|
||
def Close(self, out_trade_no: str) -> bool:
|
||
return True
|
||
|
||
def VerifyNotify(self, raw_data: bytes, headers: Dict[str, str]) -> bool:
|
||
"""验证支付宝异步通知签名"""
|
||
if not raw_data or not self._credentials.PublicKey:
|
||
return False
|
||
try:
|
||
import urllib.parse
|
||
body = urllib.parse.unquote(raw_data.decode('utf-8'))
|
||
params = dict(urllib.parse.parse_qsl(body))
|
||
|
||
sign = params.get('sign')
|
||
sign_type = params.get('sign_type')
|
||
if not sign:
|
||
return False
|
||
if sign_type and sign_type.upper() != 'RSA2':
|
||
return False
|
||
|
||
sign_content = self._build_sign_content(params)
|
||
return self._verify(sign_content, sign)
|
||
except Exception:
|
||
return False
|
||
|
||
def ParseNotify(self, raw_data: bytes) -> Dict[str, Any]:
|
||
try:
|
||
import urllib.parse
|
||
body = urllib.parse.unquote(raw_data.decode('utf-8'))
|
||
params = dict(urllib.parse.parse_qsl(body))
|
||
return {
|
||
'out_trade_no': params.get('out_trade_no', ''),
|
||
'transaction_id': params.get('trade_no', ''),
|
||
'trade_state': params.get('trade_status', ''),
|
||
'total_amount': params.get('total_amount', '0'),
|
||
'buyer_id': params.get('buyer_id', ''),
|
||
'gmt_payment': params.get('gmt_payment', ''),
|
||
}
|
||
except Exception:
|
||
return {'out_trade_no': '', 'error': '回调解析失败'}
|
||
|
||
def Transfer(self, request: TransferRequest) -> TransferResponse:
|
||
transfer_id = 'alipay_transfer_' + _uuid.uuid4().hex[:16]
|
||
return TransferResponse(
|
||
Success=True,
|
||
OutTransferNO=request.OutTransferNO,
|
||
TransferID=transfer_id,
|
||
RawResponse={'transfer_id': transfer_id, 'note': '生产环境需调用支付宝转账接口'},
|
||
)
|