From cb3eaa2c8c3825fc02dc92c8d09098da885fc53d Mon Sep 17 00:00:00 2001 From: yumoqing Date: Wed, 17 Dec 2025 11:16:17 +0800 Subject: [PATCH] bugfix --- unipay/providers/alipay.py | 300 ++++++++++++++++++------------------- unipay/providers/stripe.py | 66 ++++---- 2 files changed, 183 insertions(+), 183 deletions(-) diff --git a/unipay/providers/alipay.py b/unipay/providers/alipay.py index b43da92..fd8b92f 100644 --- a/unipay/providers/alipay.py +++ b/unipay/providers/alipay.py @@ -19,184 +19,184 @@ ALIPAY_GATEWAY = "https://openapi.alipay.com/gateway.do" class AlipayGateway(Gateway): - """ - 生产级支付宝接口（H5 支付 / 退款 / 查询 / 回调验签） - """ - def __init__( - self, - app_id: str, - app_private_key_pem: str, - alipay_public_key_pem: str, - ): - self.app_id = app_id + """ + 生产级支付宝接口（H5 支付 / 退款 / 查询 / 回调验签） + """ + def __init__( + self, + app_id: str, + app_private_key_pem: str, + alipay_public_key_pem: str, + ): + self.app_id = app_id - # -------- load keys ------- - self._private_key = serialization.load_pem_private_key( - app_private_key_pem, password=None - ) - self._alipay_public_key = serialization.load_pem_public_key( - alipay_public_key_pem - ) + # -------- load keys ------- + self._private_key = serialization.load_pem_private_key( + app_private_key_pem, password=None + ) + self._alipay_public_key = serialization.load_pem_public_key( + alipay_public_key_pem + ) - self.session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=20)) + self.session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=20)) - # ============================================================================== - # 工具函数（签名 / 验签） - # ============================================================================== + # ============================================================================== + # 工具函数（签名 / 验签） + # ============================================================================== - def _sign(self, unsigned_str: str) -> str: - """ - RSA2（SHA256）签名 - """ - signature = self._private_key.sign( - unsigned_str.encode(), - padding.PKCS1v15(), - hashes.SHA256(), - ) - return base64.b64encode(signature).decode() + def _sign(self, unsigned_str: str) -> str: + """ + RSA2（SHA256）签名 + """ + signature = self._private_key.sign( + unsigned_str.encode(), + padding.PKCS1v15(), + hashes.SHA256(), + ) + return base64.b64encode(signature).decode() - def _verify(self, data: str, sign: str) -> bool: - """ - 验证支付宝回调签名 - """ - try: - self._alipay_public_key.verify( - base64.b64decode(sign), - data.encode(), - padding.PKCS1v15(), - hashes.SHA256() - ) - return True - except Exception: - return False + def _verify(self, data: str, sign: str) -> bool: + """ + 验证支付宝回调签名 + """ + try: + self._alipay_public_key.verify( + base64.b64decode(sign), + data.encode(), + padding.PKCS1v15(), + hashes.SHA256() + ) + return True + except Exception: + return False - def _build_sign_content(self, params: Dict[str, str]) -> str: - """ - 生成 "k=v&k2=v2" 的签名串 - """ - sorted_items = sorted((k, v) for k, v in params.items() if v is not None) - return "&".join(f"{k}={v}" for k, v in sorted_items) + def _build_sign_content(self, params: Dict[str, str]) -> str: + """ + 生成 "k=v&k2=v2" 的签名串 + """ + sorted_items = sorted((k, v) for k, v in params.items() if v is not None) + return "&".join(f"{k}={v}" for k, v in sorted_items) - # ============================================================================== - # 支付 - H5 页面跳转 - # ============================================================================== + # ============================================================================== + # 支付 - H5 页面跳转 + # ============================================================================== - async def create_payment(self, payload: Dict[str, Any]) -> str: - """ - 返回一个可以在 H5 里直接重定向的支付宝支付 URL - """ + async def create_payment(self, payload: Dict[str, Any]) -> str: + """ + 返回一个可以在 H5 里直接重定向的支付宝支付 URL + """ debug(f'{payload=}') - biz_content = { - "out_trade_no": payload["out_trade_no"], - "total_amount": payload["amount"], - "subject": payload["payment_name"], - "product_code": "QUICK_WAP_WAY" - } + biz_content = { + "out_trade_no": payload["out_trade_no"], + "total_amount": payload["amount"], + "subject": payload["payment_name"], + "product_code": "QUICK_WAP_WAY" + } - params = { - "app_id": self.app_id, - "method": "alipay.trade.wap.pay", - "format": "JSON", - "charset": "utf-8", - "sign_type": "RSA2", - "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), - "version": "1.0", - "notify_url": payload["notify_url"], - "biz_content": json.dumps(biz_content, separators=(",", ":")) - } + params = { + "app_id": self.app_id, + "method": "alipay.trade.wap.pay", + "format": "JSON", + "charset": "utf-8", + "sign_type": "RSA2", + "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), + "version": "1.0", + "notify_url": payload["notify_url"], + "biz_content": json.dumps(biz_content, separators=(",", ":")) + } - # 生成签名 - unsigned_str = self._build_sign_content(params) - sign = self._sign(unsigned_str) - params["sign"] = sign + # 生成签名 + unsigned_str = self._build_sign_content(params) + sign = self._sign(unsigned_str) + params["sign"] = sign debug(f'{params=}') - query_str = urllib.parse.urlencode(params) - return f"{ALIPAY_GATEWAY}?{query_str}" + query_str = urllib.parse.urlencode(params) + return f"{ALIPAY_GATEWAY}?{query_str}" - # ============================================================================== - # 查询订单 - # ============================================================================== + # ============================================================================== + # 查询订单 + # ============================================================================== - async def query(self, out_trade_no: str) -> Dict[str, Any]: - biz_content = { - "out_trade_no": out_trade_no, - } + async def query(self, out_trade_no: str) -> Dict[str, Any]: + biz_content = { + "out_trade_no": out_trade_no, + } - params = { - "app_id": self.app_id, - "method": "alipay.trade.query", - "format": "JSON", - "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=(",", ":")) - } + params = { + "app_id": self.app_id, + "method": "alipay.trade.query", + "format": "JSON", + "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=(",", ":")) + } - # 签名 - unsigned_str = self._build_sign_content(params) - params["sign"] = self._sign(unsigned_str) + # 签名 + unsigned_str = self._build_sign_content(params) + params["sign"] = self._sign(unsigned_str) - async with self.session.post(ALIPAY_GATEWAY, data=params) as resp: - data = await resp.json() - return data + async with self.session.post(ALIPAY_GATEWAY, data=params) as resp: + data = await resp.json() + return data - # ============================================================================== - # 退款 - # ============================================================================== + # ============================================================================== + # 退款 + # ============================================================================== - async def refund(self, *, out_trade_no: str, refund_amount: str, out_request_no: str) -> Dict[str, Any]: - """ - out_request_no 必须全局唯一（一个退款请求一个编号） - """ + async def refund(self, *, out_trade_no: str, refund_amount: str, out_request_no: str) -> Dict[str, Any]: + """ + out_request_no 必须全局唯一（一个退款请求一个编号） + """ - biz_content = { - "out_trade_no": out_trade_no, - "refund_amount": refund_amount, - "out_request_no": out_request_no - } + biz_content = { + "out_trade_no": out_trade_no, + "refund_amount": refund_amount, + "out_request_no": out_request_no + } - params = { - "app_id": self.app_id, - "method": "alipay.trade.refund", - "format": "JSON", - "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=(",", ":")) - } + params = { + "app_id": self.app_id, + "method": "alipay.trade.refund", + "format": "JSON", + "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=(",", ":")) + } - unsigned_str = self._build_sign_content(params) - params["sign"] = self._sign(unsigned_str) + unsigned_str = self._build_sign_content(params) + params["sign"] = self._sign(unsigned_str) - async with self.session.post(ALIPAY_GATEWAY, data=params) as resp: - return await resp.json() + async with self.session.post(ALIPAY_GATEWAY, data=params) as resp: + return await resp.json() - # ============================================================================== - # 回调 / 异步通知（验签） - # ============================================================================== + # ============================================================================== + # 回调 / 异步通知（验签） + # ============================================================================== - async def handle_notify(self, request) -> Dict[str, Any]: - """ - 支付宝异步通知验签 - """ - form = await request.post() - params = dict(form) + async def handle_notify(self, request) -> Dict[str, Any]: + """ + 支付宝异步通知验签 + """ + form = await request.post() + params = dict(form) - sign = params.pop("sign", None) - sign_type = params.pop("sign_type", None) + sign = params.pop("sign", None) + sign_type = params.pop("sign_type", None) - unsigned_str = self._build_sign_content(params) + unsigned_str = self._build_sign_content(params) - if not sign: - return {"verified": False, "msg": "no sign"} + if not sign: + return {"verified": False, "msg": "no sign"} - ok = self._verify(unsigned_str, sign) + ok = self._verify(unsigned_str, sign) - return { - "verified": ok, - "provider": "alipay", - "data": params, - } + return { + "verified": ok, + "provider": "alipay", + "data": params, + } diff --git a/unipay/providers/stripe.py b/unipay/providers/stripe.py index 206246e..a1acb6b 100644 --- a/unipay/providers/stripe.py +++ b/unipay/providers/stripe.py @@ -3,37 +3,37 @@ import aiohttp, json from ..core import Gateway, GatewayError class StripeGateway(Gateway): - def __init__(self, api_key: str): - self.api_key = api_key - self.base = "https://api.stripe.com/v1" - async def create_payment(self, payload): - # 使用 PaymentIntent -> 前端使用 stripe.js 完成卡片采集 - url = self.base + "/payment_intents" - body = { - "amount": str(payload["amount_total"]), # in cents - "currency": payload.get("currency","usd"), - "payment_method_types[]": "card", - "description": payload.get("description",""), - "metadata[out_trade_no]": payload.get("out_trade_no","") - } - async with aiohttp.ClientSession() as s: - async with s.post(url, data=body, auth=aiohttp.BasicAuth(self.api_key, "")) as r: - return {"provider":"stripe","data": await r.json()} - async def refund(self, payload): - url = self.base + "/refunds" - body = {"charge": payload["charge_id"], "amount": str(payload.get("refund_amount"))} - async with aiohttp.ClientSession() as s: - async with s.post(url, data=body, auth=aiohttp.BasicAuth(self.api_key, "")) as r: - return {"provider":"stripe","data": await r.json()} - async def query(self, payload): - # query payment intent or charge - pid = payload.get("payment_intent_id") or payload.get("charge_id") - if not pid: - raise GatewayError("need payment_intent_id or charge_id") - async with aiohttp.ClientSession() as s: - async with s.get(self.base + f"/payment_intents/{pid}", auth=aiohttp.BasicAuth(self.api_key, "")) as r: - return {"provider":"stripe","data": await r.json()} - async def handle_notify(self, headers, body): - # stripe webhook: verify signature header (Stripe-Signature) — production use official lib or implement verification - return {"provider":"stripe","data": json.loads(body)} + def __init__(self, api_key: str): + self.api_key = api_key + self.base = "https://api.stripe.com/v1" + async def create_payment(self, payload): + # 使用 PaymentIntent -> 前端使用 stripe.js 完成卡片采集 + url = self.base + "/payment_intents" + body = { + "amount": str(payload["amount_total"]), # in cents + "currency": payload.get("currency","usd"), + "payment_method_types[]": "card", + "description": payload.get("description",""), + "metadata[out_trade_no]": payload.get("out_trade_no","") + } + async with aiohttp.ClientSession() as s: + async with s.post(url, data=body, auth=aiohttp.BasicAuth(self.api_key, "")) as r: + return {"provider":"stripe","data": await r.json()} + async def refund(self, payload): + url = self.base + "/refunds" + body = {"charge": payload["charge_id"], "amount": str(payload.get("refund_amount"))} + async with aiohttp.ClientSession() as s: + async with s.post(url, data=body, auth=aiohttp.BasicAuth(self.api_key, "")) as r: + return {"provider":"stripe","data": await r.json()} + async def query(self, payload): + # query payment intent or charge + pid = payload.get("payment_intent_id") or payload.get("charge_id") + if not pid: + raise GatewayError("need payment_intent_id or charge_id") + async with aiohttp.ClientSession() as s: + async with s.get(self.base + f"/payment_intents/{pid}", auth=aiohttp.BasicAuth(self.api_key, "")) as r: + return {"provider":"stripe","data": await r.json()} + async def handle_notify(self, headers, body): + # stripe webhook: verify signature header (Stripe-Signature) — production use official lib or implement verification + return {"provider":"stripe","data": json.loads(body)}