"""模块：支付宝适配器 实现支付宝 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': '生产环境需调用支付宝转账接口'}, )