diff --git a/services/mcp-server/app/audit.py b/services/mcp-server/app/audit.py index 72f15b2..a027f3a 100644 --- a/services/mcp-server/app/audit.py +++ b/services/mcp-server/app/audit.py @@ -62,6 +62,9 @@ AUDIT_ACTIONS = { "payment.paypal.order_failed": "PayPal支付失败", "payment.paypal.amount_mismatch": "PayPal金额不匹配", "payment.paypal.webhook_received": "PayPal Webhook接收", + "payment.paypal.webhook_captured": "PayPal Webhook充值完成", + "payment.paypal.webhook_amount_mismatch": "PayPal Webhook金额不匹配", + "payment.paypal.webhook_signature_invalid": "PayPal Webhook签名无效", # 供应商管理 "provider.create": "创建供应商", diff --git a/services/mcp-server/app/routes/paypal.py b/services/mcp-server/app/routes/paypal.py index 7e9962f..922620d 100644 --- a/services/mcp-server/app/routes/paypal.py +++ b/services/mcp-server/app/routes/paypal.py @@ -9,12 +9,20 @@ PayPal 支付路由 """ import structlog +import httpx +import hashlib +import base64 +import zlib from datetime import datetime from decimal import Decimal +from typing import Optional from fastapi import APIRouter, Depends, HTTPException, status, Request from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from pydantic import BaseModel, Field +from cryptography import x509 +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.asymmetric import padding from database import get_db from models import User, Balance, RechargeRecord @@ -23,6 +31,7 @@ from app.schemas import SuccessResponse from app.paypal_client import get_paypal_client from app.billing import add_balance from app.audit import log_audit_event +from config import settings logger = structlog.get_logger(__name__) router = APIRouter(prefix="/api/user/billing/paypal", tags=["PayPal支付"]) @@ -386,6 +395,256 @@ async def get_paypal_order( ) +# ============= Webhook 签名验证 ============= + +# 证书缓存（避免重复下载） +_cert_cache: dict = {} + + +async def _fetch_paypal_certificate(cert_url: str) -> bytes: + """ + 获取 PayPal 证书（带缓存） + + Args: + cert_url: PayPal 证书 URL + + Returns: + 证书内容（PEM 格式） + """ + if cert_url in _cert_cache: + return _cert_cache[cert_url] + + # 验证证书 URL 是否来自 PayPal + if not cert_url.startswith("https://api.paypal.com/") and not cert_url.startswith("https://api.sandbox.paypal.com/"): + raise ValueError(f"不信任的证书 URL: {cert_url}") + + async with httpx.AsyncClient() as client: + response = await client.get(cert_url, timeout=10.0) + response.raise_for_status() + cert_data = response.content + _cert_cache[cert_url] = cert_data + return cert_data + + +async def verify_webhook_signature( + transmission_id: str, + transmission_time: str, + webhook_id: str, + event_body: str, + cert_url: str, + transmission_sig: str, + auth_algo: str +) -> bool: + """ + 验证 PayPal Webhook 签名 + + PayPal 使用以下格式生成签名： + signature = sign(transmission_id|transmission_time|webhook_id|crc32(event_body)) + + Args: + transmission_id: PayPal-Transmission-Id 头 + transmission_time: PayPal-Transmission-Time 头 + webhook_id: 配置的 Webhook ID + event_body: 原始请求体字符串 + cert_url: PayPal-Cert-Url 头 + transmission_sig: PayPal-Transmission-Sig 头（Base64 编码） + auth_algo: PayPal-Auth-Algo 头（如 SHA256withRSA） + + Returns: + 签名是否有效 + """ + try: + # 1. 获取 PayPal 证书 + cert_pem = await _fetch_paypal_certificate(cert_url) + cert = x509.load_pem_x509_certificate(cert_pem) + public_key = cert.public_key() + + # 2. 计算请求体的 CRC32 + crc = zlib.crc32(event_body.encode('utf-8')) & 0xffffffff + + # 3. 构建待验证的消息 + # 格式: transmission_id|transmission_time|webhook_id|crc32 + message = f"{transmission_id}|{transmission_time}|{webhook_id}|{crc}" + + # 4. 解码签名 + signature = base64.b64decode(transmission_sig) + + # 5. 验证签名 + # PayPal 使用 SHA256withRSA + public_key.verify( + signature, + message.encode('utf-8'), + padding.PKCS1v15(), + hashes.SHA256() + ) + + return True + + except Exception as e: + logger.error("Webhook 签名验证失败", error=str(e), error_type=type(e).__name__) + return False + + +async def _process_capture_completed( + resource: dict, + db: AsyncSession, + request: Request +) -> bool: + """ + 处理支付捕获完成事件 + + 作为备份机制，当用户关闭页面但支付已完成时，通过 Webhook 完成充值。 + + Args: + resource: Webhook 事件中的 resource 对象 + db: 数据库会话 + request: 请求对象 + + Returns: + 是否处理成功 + """ + capture_id = resource.get("id") + + # 从 resource 中提取订单信息 + # PayPal Webhook 的 resource 结构： + # { + # "id": "capture_id", + # "amount": {"currency_code": "USD", "value": "10.00"}, + # "supplementary_data": {"related_ids": {"order_id": "xxx"}} + # } + + # 尝试获取关联的订单 ID + order_id = None + supplementary_data = resource.get("supplementary_data", {}) + related_ids = supplementary_data.get("related_ids", {}) + order_id = related_ids.get("order_id") + + if not order_id: + # 如果没有 order_id，尝试从 links 中获取 + links = resource.get("links", []) + for link in links: + if link.get("rel") == "up": + # up 链接指向父订单 + href = link.get("href", "") + # 从 URL 中提取订单 ID + # 格式: https://api.paypal.com/v2/checkout/orders/{order_id} + if "/orders/" in href: + order_id = href.split("/orders/")[-1].split("/")[0].split("?")[0] + break + + if not order_id: + logger.warning("Webhook 中未找到订单 ID", capture_id=capture_id) + return False + + # 查找本地订单记录 + result = await db.execute( + select(RechargeRecord).where( + RechargeRecord.order_id == order_id, + RechargeRecord.payment_method == "paypal" + ) + ) + recharge_record = result.scalar_one_or_none() + + if not recharge_record: + logger.warning("Webhook: 订单不存在", order_id=order_id, capture_id=capture_id) + return False + + # 如果订单已完成，跳过处理 + if recharge_record.status == "success": + logger.info("Webhook: 订单已完成，跳过处理", order_id=order_id) + return True + + # 获取支付金额 + amount_info = resource.get("amount", {}) + captured_amount = Decimal(amount_info.get("value", "0")) + expected_amount = recharge_record.amount + + # 验证金额 + if captured_amount != expected_amount: + logger.error( + "Webhook: 支付金额不匹配", + order_id=order_id, + expected=float(expected_amount), + captured=float(captured_amount) + ) + recharge_record.status = "amount_mismatch" + await db.commit() + + await log_audit_event( + action="payment.paypal.webhook_amount_mismatch", + resource_type="paypal_order", + resource_id=order_id, + user_id=str(recharge_record.user_id), + success=False, + details={ + "expected_amount": float(expected_amount), + "captured_amount": float(captured_amount), + "capture_id": capture_id, + }, + error_message=f"Webhook 金额不匹配: 期望 ${expected_amount}, 实际 ${captured_amount}", + request=request, + db=db + ) + return False + + # 增加用户余额 + eu_amount = captured_amount # 1 USD = 1 EU + success, message = await add_balance( + user_id=str(recharge_record.user_id), + amount=eu_amount, + db=db, + description=f"PayPal 充值 - Webhook (订单: {order_id})", + auto_commit=False + ) + + if not success: + logger.error("Webhook: 余额更新失败", order_id=order_id, error=message) + await db.rollback() + return False + + # 更新订单状态 + recharge_record.status = "success" + recharge_record.completed_at = datetime.utcnow() + + await db.commit() + + # 获取新余额 + balance_result = await db.execute( + select(Balance).where(Balance.user_id == recharge_record.user_id) + ) + balance = balance_result.scalar_one_or_none() + new_balance = float(balance.eu_balance) if balance else 0 + + logger.info( + "Webhook: PayPal 充值成功", + user_id=str(recharge_record.user_id), + order_id=order_id, + capture_id=capture_id, + amount=float(captured_amount), + new_balance=new_balance + ) + + # 审计日志 + await log_audit_event( + action="payment.paypal.webhook_captured", + resource_type="paypal_order", + resource_id=order_id, + user_id=str(recharge_record.user_id), + success=True, + details={ + "amount": float(captured_amount), + "eu_amount": float(eu_amount), + "new_balance": new_balance, + "capture_id": capture_id, + "source": "webhook", + }, + request=request, + db=db + ) + + return True + + # ============= Webhook 路由 ============= @webhook_router.post("/whitelist/payment/callback/paypal") @@ -401,24 +660,23 @@ async def paypal_webhook( 支持的事件类型： - PAYMENT.CAPTURE.COMPLETED: 支付捕获完成 - PAYMENT.CAPTURE.DENIED: 支付被拒绝 - - PAYMENT.CAPTURE.REFUNDED: 支付已退款 + - PAYMENT.CAPTURE.REFUNDED: 支付已退款（暂不处理） """ + # 获取原始请求体（用于签名验证） + raw_body = await request.body() + body_str = raw_body.decode('utf-8') + try: - # 获取请求体 - body = await request.json() + # 解析 JSON + import json + body = json.loads(body_str) - # 获取 PayPal 签名头（用于验证） - headers = { - "PayPal-Transmission-Id": request.headers.get("PayPal-Transmission-Id"), - "PayPal-Transmission-Time": request.headers.get("PayPal-Transmission-Time"), - "PayPal-Transmission-Sig": request.headers.get("PayPal-Transmission-Sig"), - "PayPal-Cert-Url": request.headers.get("PayPal-Cert-Url"), - "PayPal-Auth-Algo": request.headers.get("PayPal-Auth-Algo"), - } - - # TODO: 验证 Webhook 签名 - # 生产环境必须验证签名，防止伪造请求 - # 使用 PayPal Webhook Signature Verification API + # 获取 PayPal 签名头 + transmission_id = request.headers.get("PayPal-Transmission-Id") + transmission_time = request.headers.get("PayPal-Transmission-Time") + transmission_sig = request.headers.get("PayPal-Transmission-Sig") + cert_url = request.headers.get("PayPal-Cert-Url") + auth_algo = request.headers.get("PayPal-Auth-Algo") event_type = body.get("event_type") resource = body.get("resource", {}) @@ -426,9 +684,59 @@ async def paypal_webhook( logger.info( "收到 PayPal Webhook", event_type=event_type, - resource_id=resource.get("id") + resource_id=resource.get("id"), + transmission_id=transmission_id ) + # ========== 签名验证 ========== + webhook_id = settings.paypal_webhook_id + signature_valid = False + + if webhook_id and transmission_id and transmission_sig and cert_url: + signature_valid = await verify_webhook_signature( + transmission_id=transmission_id, + transmission_time=transmission_time, + webhook_id=webhook_id, + event_body=body_str, + cert_url=cert_url, + transmission_sig=transmission_sig, + auth_algo=auth_algo or "SHA256withRSA" + ) + + if not signature_valid: + logger.warning( + "PayPal Webhook 签名验证失败", + event_type=event_type, + transmission_id=transmission_id + ) + # 记录审计日志但继续处理（避免丢失合法请求） + await log_audit_event( + action="payment.paypal.webhook_signature_invalid", + resource_type="paypal_webhook", + resource_id=body.get("id"), + user_id=None, + success=False, + details={ + "event_type": event_type, + "resource_id": resource.get("id"), + "transmission_id": transmission_id, + }, + error_message="Webhook 签名验证失败", + request=request, + db=db + ) + # 生产环境应该拒绝无效签名的请求 + # 但为了避免配置问题导致丢失合法请求，这里只记录警告 + # return {"success": False, "error": "Invalid signature"} + else: + logger.warning( + "PayPal Webhook 缺少签名信息", + webhook_id_configured=bool(webhook_id), + has_transmission_id=bool(transmission_id), + has_signature=bool(transmission_sig), + has_cert_url=bool(cert_url) + ) + # ========== 审计日志：Webhook 接收 ========== await log_audit_event( action="payment.paypal.webhook_received", @@ -439,32 +747,59 @@ async def paypal_webhook( details={ "event_type": event_type, "resource_id": resource.get("id"), + "signature_valid": signature_valid, + "transmission_id": transmission_id, }, request=request, db=db ) + # ========== 事件处理 ========== if event_type == "PAYMENT.CAPTURE.COMPLETED": - # 支付完成 - capture_id = resource.get("id") - # 根据 capture_id 查找并更新订单 - # 这里作为备份机制，主要逻辑在 capture-order 接口 - logger.info("PayPal 支付完成 Webhook", capture_id=capture_id) + # 支付完成 - 作为备份机制处理充值 + success = await _process_capture_completed(resource, db, request) + logger.info( + "PayPal 支付完成 Webhook 处理完毕", + capture_id=resource.get("id"), + success=success + ) elif event_type == "PAYMENT.CAPTURE.DENIED": - # 支付被拒绝 + # 支付被拒绝 - 更新订单状态 capture_id = resource.get("id") logger.warning("PayPal 支付被拒绝", capture_id=capture_id) + # 尝试更新订单状态为失败 + supplementary_data = resource.get("supplementary_data", {}) + related_ids = supplementary_data.get("related_ids", {}) + order_id = related_ids.get("order_id") + + if order_id: + result = await db.execute( + select(RechargeRecord).where( + RechargeRecord.order_id == order_id, + RechargeRecord.payment_method == "paypal" + ) + ) + recharge_record = result.scalar_one_or_none() + if recharge_record and recharge_record.status == "pending": + recharge_record.status = "failed" + await db.commit() + logger.info("Webhook: 订单状态更新为失败", order_id=order_id) + elif event_type == "PAYMENT.CAPTURE.REFUNDED": - # 支付已退款 + # 支付已退款 - 暂不处理，记录日志 capture_id = resource.get("id") - logger.info("PayPal 支付已退款", capture_id=capture_id) + logger.info("PayPal 支付已退款 (暂不处理)", capture_id=capture_id) # TODO: 处理退款逻辑，扣减用户余额 return {"success": True} except Exception as e: - logger.error("处理 PayPal Webhook 失败", error=str(e)) + logger.error( + "处理 PayPal Webhook 失败", + error=str(e), + error_type=type(e).__name__ + ) # 返回 200 避免 PayPal 重试 return {"success": False, "error": str(e)}