更新备份

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": "创建供应商",

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:
-        # 获取请求体
+        # 解析 JSON
-        body = await request.json()
+        import json
+        body = json.loads(body_str)
         
-        # 获取 PayPal 签名头（用于验证）
+        # 获取 PayPal 签名头
-        headers = {
+        transmission_id = request.headers.get("PayPal-Transmission-Id")
-            "PayPal-Transmission-Id": request.headers.get("PayPal-Transmission-Id"),
+        transmission_time = request.headers.get("PayPal-Transmission-Time")
-            "PayPal-Transmission-Time": request.headers.get("PayPal-Transmission-Time"),
+        transmission_sig = request.headers.get("PayPal-Transmission-Sig")
-            "PayPal-Transmission-Sig": request.headers.get("PayPal-Transmission-Sig"),
+        cert_url = request.headers.get("PayPal-Cert-Url")
-            "PayPal-Cert-Url": request.headers.get("PayPal-Cert-Url"),
+        auth_algo = request.headers.get("PayPal-Auth-Algo")
-            "PayPal-Auth-Algo": request.headers.get("PayPal-Auth-Algo"),
-        }
-        
-        # TODO: 验证 Webhook 签名
-        # 生产环境必须验证签名，防止伪造请求
-        # 使用 PayPal Webhook Signature Verification API
         
         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")
+            success = await _process_capture_completed(resource, db, request)
-            # 根据 capture_id 查找并更新订单
+            logger.info(
-            # 这里作为备份机制，主要逻辑在 capture-order 接口
+                "PayPal 支付完成 Webhook 处理完毕",
-            logger.info("PayPal 支付完成 Webhook", capture_id=capture_id)
+                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)}