优化安全问题

app.py

@@ -60,7 +60,7 @@ from router_open_api import router as open_api_router  # 🌐 通用开放API（
 from router_admin_plugins import router as admin_plugins_router  # 🔒 管理后台 - 第三方插件管理API
 
 from database_sql import init_sql_db, get_db, SessionLocal
-from models_sql import Ownership, SystemConfig
+from models_sql import SystemConfig  # 🧹 P2清理：Ownership 在本文件无引用
 
 # 🚀 P2优化：速率限制 (防止暴力攻击)
 from slowapi import Limiter, _rate_limit_exceeded_handler

models.py

@@ -1,6 +1,6 @@
 # models.py
 import re
-from pydantic import BaseModel, Field, validator
+from pydantic import BaseModel, Field, field_validator  # 🧹 P2修复：pydantic v2 已废弃 v1 风格 validator
 from typing import Optional, List
 
 def _validate_email_format(email: str) -> str:
@@ -37,15 +37,26 @@ class UserRegister(BaseModel):
     region: Optional[str] = None
     intro: Optional[str] = None
     
-    _validate_password = validator('password', allow_reuse=True)(_validate_password_length)
-    _validate_email = validator('email', allow_reuse=True)(_validate_email_format)
+    # 🧹 P2修复：v2 中下划线前缀属性会被当作私有属性而非校验器，改用 field_validator 显式挂载（校验逻辑不变）
+    @field_validator('password')
+    @classmethod
+    def _check_password(cls, v):
+        return _validate_password_length(v)
+
+    @field_validator('email')
+    @classmethod
+    def _check_email(cls, v):
+        return _validate_email_format(v)
 
 class UserLogin(BaseModel):
     account: str   
     password: str
     remember: Optional[bool] = True  # 保持登录选项：True=30天, False=24小时
     
-    _validate_password = validator('password', allow_reuse=True)(_validate_password_length)
+    @field_validator('password')
+    @classmethod
+    def _check_password(cls, v):
+        return _validate_password_length(v)
 
 class UserUpdate(BaseModel):
     name: Optional[str] = None
@@ -58,15 +69,7 @@ class UserUpdate(BaseModel):
     avatarDataUrl: Optional[str] = None
     bannerUrl: Optional[str] = None         # 🚀 个人资料卡背景图云端URL
 
-class PasswordReset(BaseModel):
-    old_password: Optional[str] = None
-    new_password: str
-    verifyContact: str
-    verifyType: str
-    code: str
-    account: str
-    
-    _validate_new_password = validator('new_password', allow_reuse=True)(_validate_password_length)
+# 🧹 P2清理：移除零引用的 PasswordReset 模型（重置密码实际走 UserUpdate/专用端点）
 
 class CommentCreate(BaseModel):
     item_id: str
@@ -130,7 +133,8 @@ class InteractionToggle(BaseModel):
 class RatingRequest(BaseModel):
     score: int  # 1-5
 
-    @validator("score")
+    @field_validator("score")
+    @classmethod
     def validate_score(cls, v):
         if not isinstance(v, int):
             raise ValueError("score must be an integer between 1 and 5")
@@ -186,57 +190,24 @@ class TaskUpdate(BaseModel):
     depositRatio: Optional[int] = None
     deadline: Optional[str] = None
 
-class TaskApply(BaseModel):
-    """ 申请接单 """
-    task_id: str
-    applicant: str                          # 申请者账号
-    message: Optional[str] = None           # 申请留言
-
-class TaskAssign(BaseModel):
-    """ 指派接单者 """
-    task_id: str
-    publisher: str                          # 发布者（校验权限）
-    assignee: str                           # 被指派的接单者
-
+# 🔧 P2修复：以下三个 body 模型修复前后端参数通道不匹配
+# （前端以 JSON body 提交，旧版端点用裸标量参数会被 FastAPI 解析为 query 导致 422）
 class TaskSubmit(BaseModel):
     """ 接单者提交成果 """
-    task_id: str
-    assignee: str                           # 接单者（校验权限）
-    deliverables: List[str]                 # 交付物图片URL列表
-    note: Optional[str] = None              # 备注说明
+    deliverables: List[str]                 # 交付成果列表
+    note: Optional[str] = None              # 提交说明
 
-class TaskAccept(BaseModel):
-    """ 发布者验收 """
-    task_id: str
-    publisher: str                          # 发布者（校验权限）
-    is_accepted: bool                       # True=验收通过, False=验收不通过
-    feedback: Optional[str] = None          # 反馈意见
-
-
-# ==========================================
-# ⚖️ P3增强：任务申诉数据模型
-# ==========================================
 class TaskDispute(BaseModel):
     """ 发起申诉 """
-    task_id: str                            # 任务ID
-    initiator: str                          # 申诉发起人（可以是发布者或接单者）
     reason: str                             # 申诉理由
-    evidence: Optional[List[str]] = []      # 证据图片URL列表
+    evidence: Optional[List[str]] = None    # 证据图片URL列表（最多6张）
 
 class TaskDisputeResponse(BaseModel):
-    """ 被申诉方回应 """
-    dispute_id: str                         # 申诉ID
-    respondent: str                         # 回应人
+    """ 被申诉方回应申诉 """
     response: str                           # 回应内容
-    evidence: Optional[List[str]] = []      # 证据图片URL列表
-
-class TaskDisputeResolve(BaseModel):
-    """ 管理员仲裁 """
-    dispute_id: str                         # 申诉ID
-    admin_account: str                      # 管理员账号
-    result: str                             # 仲裁结果: support_initiator / support_respondent / split
-    split_ratio: Optional[int] = 50         # 分成比例（申诉方获得的百分比，仅当result=split时有效）
-    admin_note: Optional[str] = None        # 管理员备注
+    evidence: Optional[List[str]] = None    # 证据图片URL列表（最多6张）
+
+# 🧹 P2清理：移除零引用的 TaskApply/TaskAssign/TaskAccept 模型（对应端点使用 query 参数且前端匹配）
 
 # ==========================================
 # 💬 讨论区数据模型（小红书风格图文社区）

models_sql.py

@@ -55,6 +55,16 @@ class Refund(Base):
     refunded_at = Column(DateTime, default=datetime.datetime.utcnow)
     ban_until = Column(DateTime)                   # 禁止购买截止时间（30天后）
 
+class RechargeOrder(Base):
+    """💰 充值订单预存表（订单号→账号映射持久化，防止实例重启导致回调找不到订单）"""
+    __tablename__ = "recharge_orders"
+    
+    order_id = Column(String, primary_key=True)    # 商户订单号 PAY_xxx
+    account = Column(String, index=True)           # 充值用户
+    amount = Column(Integer)                        # 充值金额
+    status = Column(String, default="pending")     # pending/paid
+    created_at = Column(DateTime, default=datetime.datetime.utcnow)
+
 class Transaction(Base):
     """交易流水表"""
     __tablename__ = "transactions"

requirements.txt

@@ -3,13 +3,14 @@ fastapi==0.104.1
 uvicorn==0.24.0
 pydantic==2.5.2
 huggingface_hub==0.19.4
-datasets==2.15.0
+# 🧹 P2清理：datasets/aiofiles 全库零引用，已移除
 python-multipart==0.0.6
 alibabacloud_dysmsapi20170525==2.0.24
+# 🧹 P2修复：verify_code_engine.py 直接导入 alibabacloud_tea_openapi，显式声明避免隐式依赖漂移
+alibabacloud_tea_openapi==0.3.8
 sqlalchemy==2.0.23
 psycopg2-binary==2.9.9
 httpx==0.25.2
-aiofiles==23.2.1
 # 🚀 P2优化：速率限制
 slowapi==0.1.9
 # 🔒 P0安全增强：bcrypt密码哈希

router_admin_plugins.py

@@ -21,7 +21,7 @@ from typing import Optional, List
 from fastapi import APIRouter, Depends, HTTPException
 from fastapi.responses import JSONResponse
 from sqlalchemy.orm import Session
-from pydantic import BaseModel, Field, validator
+from pydantic import BaseModel, Field, field_validator  # 🧹 P2修复：pydantic v2 已废弃 v1 风格 validator
 
 from database_sql import get_db
 from models_sql import RegisteredPlugin
@@ -120,13 +120,16 @@ class PluginCreateRequest(BaseModel):
     max_deduct_per_day: Optional[int] = Field(1000, ge=1, le=100000)
     permissions: Optional[List[str]] = Field(default=None)
 
-    @validator("plugin_id")
+    @field_validator("plugin_id")
+    @classmethod
     def _validate_plugin_id(cls, v):
         if not _PLUGIN_ID_PATTERN.match(v):
             raise ValueError("plugin_id 仅允许 [a-z0-9_]，长度 3-50")
         return v
 
-    @validator("permissions", always=True)
+    # 🔧 pydantic v2：field_validator 默认对默认值也生效，无需 v1 的 always=True
+    @field_validator("permissions")
+    @classmethod
     def _validate_permissions(cls, v):
         if v is None:
             return _DEFAULT_PERMISSIONS
@@ -144,7 +147,8 @@ class PluginUpdateRequest(BaseModel):
     max_deduct_per_day: Optional[int] = Field(None, ge=1, le=100000)
     permissions: Optional[List[str]] = None
 
-    @validator("permissions")
+    @field_validator("permissions")
+    @classmethod
     def _validate_permissions(cls, v):
         if v is None:
             return v

router_items.py

@@ -95,29 +95,33 @@ async def get_items(type: str = "tool", sort: str = "time", limit: int = 50): #
     filtered_items = sort_cache.get_sorted(cache_key, filtered_items, sort_fn)
         
     price_updated = False
+    result_items = []
     for item in filtered_items:
-        item["commentsData"] = comments_db.get(item["id"], [])
-        item["comments"] = len(item["commentsData"])
-        item["latest_version"] = _get_version_str(versions_db, item["id"])
-        
-        # 💰 检查延迟价格是否已生效
+        # 💰 检查延迟价格是否已生效（在原始字典上执行，确保生效后可持久化）
         old_price = item.get("price", 0)
         _apply_effective_price(item)
         if item.get("price", 0) != old_price:
             price_updated = True
-        
+
+        # 🔒 P0数据修复：浅拷贝后再做下发加工，避免 pop/注入字段污染 items.json 源数据
+        item_data = {**item}
+        item_data["commentsData"] = comments_db.get(item_data["id"], [])
+        item_data["comments"] = len(item_data["commentsData"])
+        item_data["latest_version"] = _get_version_str(versions_db, item_data["id"])
+
         # 🔴 【绝对核心防线】：在下发给前端前，强行在内存中抹除敏感信息！
-        item["has_private_token"] = bool(item.get("github_token"))
-        item.pop("github_token", None)
-        item.pop("netdisk_password", None)  # ☁️ 网盘密码不在列表中显示
-        item.pop("viewed_by", None)  # 👁️ 访问者列表不暴露给前端
+        item_data["has_private_token"] = bool(item_data.get("github_token"))
+        item_data.pop("github_token", None)
+        item_data.pop("netdisk_password", None)  # ☁️ 网盘密码不在列表中显示
+        item_data.pop("viewed_by", None)  # 👁️ 访问者列表不暴露给前端
+        result_items.append(item_data)
     
     # 如果有价格生效了，持久化回 items.json
     if price_updated:
         db.save_data("items.json", items_db)
         invalidate_cache("items.json")
     
-    return {"status": "success", "data": filtered_items[:limit]}
+    return {"status": "success", "data": result_items[:limit]}
 
 def _build_creator_trend_data(account: str, u_items: list, months: list) -> dict:
     """
@@ -559,9 +563,17 @@ async def check_github_updates():
                     latest_sha = commits[0].get("sha", "")[:7]  # 取前7位作为版本标识
                     
                     # 如果版本有变化，更新 versions.json
-                    old_version = versions_db.get(item_id, "")
+                    # 🧹 P2修复（格式统一）：统一以新格式 dict 读写，保留 cached_at/zip_size 等缓存元字段；
+                    # 旧逻辑拿字符串与 dict 比较会把 dict 条目永远判为"有更新"并覆盖成字符串丢失元信息
+                    entry = versions_db.get(item_id)
+                    if isinstance(entry, str):
+                        entry = {"hash": entry}
+                    elif not isinstance(entry, dict):
+                        entry = {}
+                    old_version = entry.get("hash", "")
                     if latest_sha and latest_sha != old_version:
-                        versions_db[item_id] = latest_sha
+                        entry["hash"] = latest_sha
+                        versions_db[item_id] = entry
                         updated_count += 1
                         print(f"[版本更新] {item['title']}: {old_version} -> {latest_sha}")
                         
@@ -588,25 +600,28 @@ async def get_item_by_id(item_id: str):
     
     for item in items_db:
         if item["id"] == item_id:
-            # 添加关联数据
-            item["commentsData"] = comments_db.get(item_id, [])
-            item["comments"] = len(item["commentsData"])
-            item["latest_version"] = _get_version_str(versions_db, item_id)
-            
-            # 💰 检查延迟价格是否已生效
+            # 💰 检查延迟价格是否已生效（在原字典上操作，确保生效结果可持久化）
             old_price = item.get("price", 0)
             _apply_effective_price(item)
             if item.get("price", 0) != old_price:
                 db.save_data("items.json", items_db)
                 invalidate_cache("items.json")
             
-            # 🔴 【安全防线】：抹除敏感信息
-            item["has_private_token"] = bool(item.get("github_token"))
-            item.pop("github_token", None)
-            item.pop("netdisk_password", None)
-            item.pop("viewed_by", None)
+            # 🔒 P2修复（与 get_items 同款）：浅拷贝后再做下发加工，
+            # 避免 pop 敏感字段污染 items.json 源数据/共享缓存（导致 github_token 永久丢失）
+            item_data = {**item}
+            # 添加关联数据
+            item_data["commentsData"] = comments_db.get(item_id, [])
+            item_data["comments"] = len(item_data["commentsData"])
+            item_data["latest_version"] = _get_version_str(versions_db, item_id)
+            
+            # 🔴 【安全防线】：抹除敏感信息（在拷贝上操作）
+            item_data["has_private_token"] = bool(item_data.get("github_token"))
+            item_data.pop("github_token", None)
+            item_data.pop("netdisk_password", None)
+            item_data.pop("viewed_by", None)
             
-            return {"status": "success", "data": item}
+            return {"status": "success", "data": item_data}
     
     raise HTTPException(status_code=404, detail="资源不存在")
 

router_messages.py

@@ -179,8 +179,18 @@ async def get_chat_list(account: str, current_user: str = Depends(require_auth))
     chat_list = []
     for conv_id, msgs in chats_db.items():
         if account in conv_id:
-            other_account = conv_id.replace(account, "").replace("_", "")
             if not msgs: continue
+            # 🔧 P1修复：对方账户从消息的 sender/receiver 字段推导，而非对 conv_id 做 replace
+            # （replace 会替换所有子串，若账户名互为子串或含下划线会解析出错误账户）
+            other_account = None
+            for m in reversed(msgs):
+                if m.get("sender") != account:
+                    other_account = m.get("sender")
+                    break
+                if m.get("receiver") != account:
+                    other_account = m.get("receiver")
+                    break
+            if not other_account: continue
             last_msg = msgs[-1]
             unread_count = sum(1 for m in msgs if m["receiver"] == account and not m.get("is_read"))
             other_user = users_db.get(other_account, {})
@@ -233,6 +243,9 @@ async def get_chat_history(account: str, target_account: str, current_user: str
 # ==========================================
 @router.get("/api/messages/{account}")
 async def get_messages(account: str, count_only: bool = False, current_user: str = Depends(require_auth)):
+    # 🔒 P0安全加固：拦截越权读取他人消息（含 count_only 轮询）
+    if account != current_user:
+        raise HTTPException(status_code=403, detail="无权访问他人消息")
     # 🔥 count_only 模式：轻量级轮询，只返回未读数，不标记已读
     if count_only:
         messages_db = db.load_data("messages.json", default_data={})
@@ -344,10 +357,13 @@ async def get_messages(account: str, count_only: bool = False, current_user: str
     return result_container[0]
 
 @router.post("/api/messages/{account}/read")
-async def mark_messages_read(account: str):
+async def mark_messages_read(account: str, current_user: str = Depends(require_auth)):
     """
     标记消息为已读（原子操作，并发安全）
     """
+    # 🔒 P0安全加固：补全鉴权，防止任意用户篡改他人消息已读状态
+    if account != current_user:
+        raise HTTPException(status_code=403, detail="无权操作他人消息")
     def updater(data):
         user_msgs = data.get(account, [])
         modified = False

router_posts.py

@@ -429,53 +429,21 @@ async def tip_post(post_id: str, amount: int, is_anon: bool = False, current_use
     if amount > MAX_TIP_AMOUNT:
         raise HTTPException(status_code=400, detail=f"打赏金额必须在1-{MAX_TIP_AMOUNT}之间")
     
-    result_container = [None]
-    author_account = [None]  # 用于在原子操作外获取作者账号
-    
-    def updater(data):
-        # 在锁内查找帖子
-        target_post = None
-        for post in data:
-            if post["id"] == post_id:
-                target_post = post
-                break
-        
-        if not target_post:
-            result_container[0] = {"error": "not_found"}
-            return
-        
-        # 不能打赏自己
-        if target_post.get("author") == current_user:
-            result_container[0] = {"error": "self_tip"}
-            return
-        
-        author_account[0] = target_post.get("author")
-        
-        # 更新打赏榜单
-        tip_board = target_post.get("tip_board", [])
-        existing = next((t for t in tip_board if t["account"] == current_user), None)
-        if existing:
-            existing["amount"] += amount
-        else:
-            tip_board.append({"account": current_user, "amount": amount, "is_anon": is_anon})
-        tip_board.sort(key=lambda x: x["amount"], reverse=True)
-        target_post["tip_board"] = tip_board
-        
-        result_container[0] = {"status": "success", "message": f"成功打赏 {amount} 积分"}
-    
-    db.atomic_update("posts.json", updater, default_data=[])
-    
-    result = result_container[0]
-    if result is None or result.get("error") == "not_found":
+    # 🔒 P0数据修复（时序）：先只读校验 → SQL转账扣款 → 提交成功后再写打赏榜
+    # 避免余额不足/重复请求时榜单已被更新的数据不一致
+    posts_db_ro = db.load_data("posts.json", default_data=[])
+    target_post = next((p for p in posts_db_ro if p["id"] == post_id), None)
+    if not target_post:
         raise HTTPException(status_code=404, detail="帖子不存在")
-    if result.get("error") == "self_tip":
+    if target_post.get("author") == current_user:
         raise HTTPException(status_code=400, detail="不能打赏自己的帖子")
+    author = target_post.get("author")
+    if not author:
+        raise HTTPException(status_code=404, detail="作者账户不存在")
+    post_title = target_post.get("title")
     
-    # 💳 使用 SQL Wallet 系统处理余额转账
+    # 💳 使用 SQL Wallet 系统处理余额转账（先扣款）
     try:
-        author = author_account[0]
-        if not author:
-            raise HTTPException(status_code=404, detail="作者账户不存在")
         
         # 🔒 P1幂等性防护：检查最近5秒内是否存在相同交易
         recent_cutoff = datetime.datetime.utcnow() - datetime.timedelta(seconds=5)
@@ -515,20 +483,13 @@ async def tip_post(post_id: str, amount: int, is_anon: bool = False, current_use
         prev_hash_tipper = last_tx_tipper.tx_hash if last_tx_tipper else "GENESIS_HASH"
         prev_hash_author = last_tx_author.tx_hash if last_tx_author else "GENESIS_HASH"
         
-        # 获取用户信息和帖子标题
+        # 获取用户信息（post_title 已在预校验阶段取得）
         users_db = db.load_data("users.json", default_data={})
-        posts_db = db.load_data("posts.json", default_data=[])
         author_info = users_db.get(author, {})
         tipper_info = users_db.get(current_user, {})
         author_name = author_info.get("name", author)
         tipper_name = tipper_info.get("name", current_user)
         
-        post_title = None
-        for post in posts_db:
-            if post["id"] == post_id:
-                post_title = post.get("title")
-                break
-        
         # 打赏方交易记录 (TIP_OUT)
         tx_tipper = Transaction(
             tx_id=tx_id_tipper,
@@ -597,7 +558,26 @@ async def tip_post(post_id: str, amount: int, is_anon: bool = False, current_use
         logger.error(f"POST_TIP_ERROR | from={current_user} | post={post_id} | amount={amount} | error={str(e)}")
         raise HTTPException(status_code=500, detail="打赏处理失败，请稍后重试")
     
-    return result
+    # 🏆 扣款成功后再写打赏榜（写榜失败仅记录日志，不影响资金安全）
+    def board_updater(data):
+        for post in data:
+            if post["id"] == post_id:
+                tip_board = post.get("tip_board", [])
+                existing = next((t for t in tip_board if t["account"] == current_user), None)
+                if existing:
+                    existing["amount"] += amount
+                else:
+                    tip_board.append({"account": current_user, "amount": amount, "is_anon": is_anon})
+                tip_board.sort(key=lambda x: x["amount"], reverse=True)
+                post["tip_board"] = tip_board
+                break
+    
+    try:
+        db.atomic_update("posts.json", board_updater, default_data=[])
+    except Exception as e:
+        logger.error(f"POST_TIP_BOARD_ERROR | post={post_id} | error={str(e)}")
+    
+    return {"status": "success", "message": f"成功打赏 {amount} 积分"}
 
 # ==========================================
 # 💬 评论接口（复用通用评论系统）

router_prompts.py

@@ -726,53 +726,21 @@ async def tip_prompt(prompt_id: str, amount: int, is_anon: bool = False,
     if amount > MAX_TIP_AMOUNT:
         raise HTTPException(status_code=400, detail=f"打赏金额必须在1-{MAX_TIP_AMOUNT}之间")
 
-    result_container = [None]
-    author_account = [None]  # 用于在原子操作外获取作者账号
-
-    def updater(data):
-        # 在锁内查找提示词
-        target_prompt = None
-        for prompt in data:
-            if prompt["id"] == prompt_id:
-                target_prompt = prompt
-                break
-
-        if not target_prompt:
-            result_container[0] = {"error": "not_found"}
-            return
-
-        # 不能打赏自己
-        if target_prompt.get("author") == current_user:
-            result_container[0] = {"error": "self_tip"}
-            return
-
-        author_account[0] = target_prompt.get("author")
-
-        # 更新打赏榜单
-        tip_board = target_prompt.get("tip_board", [])
-        existing = next((t for t in tip_board if t["account"] == current_user), None)
-        if existing:
-            existing["amount"] += amount
-        else:
-            tip_board.append({"account": current_user, "amount": amount, "is_anon": is_anon})
-        tip_board.sort(key=lambda x: x["amount"], reverse=True)
-        target_prompt["tip_board"] = tip_board
-
-        result_container[0] = {"status": "success", "message": f"成功打赏 {amount} 积分"}
-
-    db.atomic_update("prompts.json", updater, default_data=[])
-
-    result = result_container[0]
-    if result is None or result.get("error") == "not_found":
+    # 🔒 P0数据修复（时序）：先只读校验 → SQL转账扣款 → 提交成功后再写打赏榜
+    # 避免余额不足/重复请求时榜单已被更新的数据不一致
+    prompts_db_ro = db.load_data("prompts.json", default_data=[])
+    target_prompt = next((p for p in prompts_db_ro if p["id"] == prompt_id), None)
+    if not target_prompt:
         raise HTTPException(status_code=404, detail="提示词不存在")
-    if result.get("error") == "self_tip":
+    if target_prompt.get("author") == current_user:
         raise HTTPException(status_code=400, detail="不能打赏自己的提示词")
+    author = target_prompt.get("author")
+    if not author:
+        raise HTTPException(status_code=404, detail="作者账户不存在")
+    prompt_title = target_prompt.get("title")
 
-    # 💳 使用 SQL Wallet 系统处理余额转账
+    # 💳 使用 SQL Wallet 系统处理余额转账（先扣款）
     try:
-        author = author_account[0]
-        if not author:
-            raise HTTPException(status_code=404, detail="作者账户不存在")
 
         # 🔒 P1幂等性防护：检查最近5秒内是否存在相同交易
         recent_cutoff = datetime.datetime.utcnow() - datetime.timedelta(seconds=5)
@@ -812,20 +780,13 @@ async def tip_prompt(prompt_id: str, amount: int, is_anon: bool = False,
         prev_hash_tipper = last_tx_tipper.tx_hash if last_tx_tipper else "GENESIS_HASH"
         prev_hash_author = last_tx_author.tx_hash if last_tx_author else "GENESIS_HASH"
 
-        # 获取用户信息和提示词标题
+        # 获取用户信息（prompt_title 已在预校验阶段取得）
         users_db = db.load_data("users.json", default_data={})
         author_info = users_db.get(author, {})
         tipper_info = users_db.get(current_user, {})
         author_name = author_info.get("name", author)
         tipper_name = tipper_info.get("name", current_user)
 
-        prompt_title = None
-        prompts_db = db.load_data("prompts.json", default_data=[])
-        for prompt in prompts_db:
-            if prompt["id"] == prompt_id:
-                prompt_title = prompt.get("title")
-                break
-
         # 打赏方交易记录 (TIP_OUT)
         tx_tipper = Transaction(
             tx_id=tx_id_tipper,
@@ -894,7 +855,26 @@ async def tip_prompt(prompt_id: str, amount: int, is_anon: bool = False,
         logger.error(f"PROMPT_TIP_ERROR | from={current_user} | prompt={prompt_id} | amount={amount} | error={str(e)}")
         raise HTTPException(status_code=500, detail="打赏处理失败，请稍后重试")
 
-    return result
+    # 🏆 扣款成功后再写打赏榜（写榜失败仅记录日志，不影响资金安全）
+    def board_updater(data):
+        for prompt in data:
+            if prompt["id"] == prompt_id:
+                tip_board = prompt.get("tip_board", [])
+                existing = next((t for t in tip_board if t["account"] == current_user), None)
+                if existing:
+                    existing["amount"] += amount
+                else:
+                    tip_board.append({"account": current_user, "amount": amount, "is_anon": is_anon})
+                tip_board.sort(key=lambda x: x["amount"], reverse=True)
+                prompt["tip_board"] = tip_board
+                break
+
+    try:
+        db.atomic_update("prompts.json", board_updater, default_data=[])
+    except Exception as e:
+        logger.error(f"PROMPT_TIP_BOARD_ERROR | prompt={prompt_id} | error={str(e)}")
+
+    return {"status": "success", "message": f"成功打赏 {amount} 积分"}
 
 
 # ==========================================

router_proxy.py

@@ -14,6 +14,7 @@ import urllib.error
 import 数据库连接 as json_db
 from database_sql import get_db
 from models_sql import Ownership
+from 安全认证 import require_auth
 
 router = APIRouter()
 
@@ -22,7 +23,8 @@ class ProxyGithubZipRequest(BaseModel):
     """GitHub ZIP 代理下载请求模型"""
     url: str = Field(..., min_length=1, description="资源URL")
     item_id: str = Field(..., min_length=1, max_length=100, description="资源ID")
-    account: str = Field(..., min_length=1, max_length=50, description="用户账号")
+    # 🔒 P0安全加固：身份以 Authorization Token 为准，account 仅作向后兼容字段，不再作为鉴权依据
+    account: Optional[str] = Field(None, max_length=50, description="用户账号（已废弃，以Token为准）")
     
     @field_validator('item_id')
     @classmethod
@@ -31,31 +33,21 @@ class ProxyGithubZipRequest(BaseModel):
         if not re.match(r'^[a-zA-Z0-9_-]+$', v):
             raise ValueError('item_id只能包含字母、数字、下划线和连字符')
         return v
-    
-    @field_validator('account')
-    @classmethod
-    def validate_account(cls, v: str) -> str:
-        """验证account非空且长度符合要求"""
-        if len(v) < 1:
-            raise ValueError('account不能为空')
-        if len(v) > 50:
-            raise ValueError('account长度不能超过50个字符')
-        return v
 
 @router.post("/api/proxy_github_zip")
-async def proxy_github_zip(req_data: ProxyGithubZipRequest, db: Session = Depends(get_db)):
+async def proxy_github_zip(req_data: ProxyGithubZipRequest, account: str = Depends(require_auth), db: Session = Depends(get_db)):
     """云端代理：校验所有权后，拉取 GitHub 私有库 ZIP 流透传给本地"""
     items_db = json_db.load_data("items.json", default_data=[])
     item = next((i for i in items_db if i["id"] == req_data.item_id), None)
     if not item: return JSONResponse(content={"error": "资源不存在"}, status_code=404)
     
-    # 1. 核心鉴权：验证用户是否购买过该工具
+    # 1. 核心鉴权：验证用户是否购买过该工具（account 来自 Token 验证，不可伪造）
     try:
         price = int(float(item.get("price", 0) or 0))
     except (ValueError, TypeError):
         price = 0
-    if price > 0 and req_data.account != item.get("author"):
-        owned = db.query(Ownership).filter(Ownership.account == req_data.account, Ownership.item_id == req_data.item_id).first()
+    if price > 0 and account != item.get("author"):
+        owned = db.query(Ownership).filter(Ownership.account == account, Ownership.item_id == req_data.item_id).first()
         if not owned:
             return JSONResponse(content={"error": "🚨 拒绝访问：未找到购买记录！"}, status_code=403)
 
@@ -198,7 +190,8 @@ class ProxyDownloadRequest(BaseModel):
     """工作流/应用 JSON 代理下载请求模型"""
     url: str = Field(..., min_length=1, description="下载URL")
     item_id: str = Field(..., min_length=1, max_length=100, description="资源ID")
-    account: str = Field(..., min_length=1, max_length=50, description="用户账号")
+    # 🔒 P0安全加固：身份以 Authorization Token 为准，account 仅作向后兼容字段
+    account: Optional[str] = Field(None, max_length=50, description="用户账号（已废弃，以Token为准）")
     
     @field_validator('item_id')
     @classmethod
@@ -208,16 +201,6 @@ class ProxyDownloadRequest(BaseModel):
             raise ValueError('item_id只能包含字母、数字、下划线和连字符')
         return v
     
-    @field_validator('account')
-    @classmethod
-    def validate_account(cls, v: str) -> str:
-        """验证account非空且长度符合要求"""
-        if len(v) < 1:
-            raise ValueError('account不能为空')
-        if len(v) > 50:
-            raise ValueError('account长度不能超过50个字符')
-        return v
-    
     @field_validator('url')
     @classmethod
     def validate_url(cls, v: str) -> str:
@@ -242,7 +225,7 @@ class ProxyDownloadRequest(BaseModel):
         return v
 
 @router.post("/api/proxy_download")
-async def proxy_download(req_data: ProxyDownloadRequest, db: Session = Depends(get_db)):
+async def proxy_download(req_data: ProxyDownloadRequest, account: str = Depends(require_auth), db: Session = Depends(get_db)):
     """云端代理：校验所有权后，拉取真实的工作流 JSON 文件并透传给本地"""
     items_db = json_db.load_data("items.json", default_data=[])
     item = next((i for i in items_db if i["id"] == req_data.item_id), None)
@@ -250,10 +233,13 @@ async def proxy_download(req_data: ProxyDownloadRequest, db: Session = Depends(g
     if not item: 
         return JSONResponse(content={"error": "云端记录中找不到该资源"}, status_code=404)
     
-    # 1. 核心鉴权：验证用户是否购买/获取过该工作流
-    price = int(item.get("price", 0))
-    if price > 0 and req_data.account != item.get("author"):
-        owned = db.query(Ownership).filter(Ownership.account == req_data.account, Ownership.item_id == req_data.item_id).first()
+    # 1. 核心鉴权：验证用户是否购买/获取过该工作流（account 来自 Token 验证，不可伪造）
+    try:
+        price = int(float(item.get("price", 0) or 0))
+    except (ValueError, TypeError):
+        price = 0
+    if price > 0 and account != item.get("author"):
+        owned = db.query(Ownership).filter(Ownership.account == account, Ownership.item_id == req_data.item_id).first()
         if not owned:
             return JSONResponse(content={"error": "您尚未获取该工作流，请先在社区页面点击获取"}, status_code=403)
 

router_tasks.py

@@ -13,7 +13,7 @@
 
 from fastapi import APIRouter, HTTPException, Depends, Query
 from sqlalchemy.orm import Session
-from models import TaskCreate, TaskUpdate, TaskApply, TaskAssign, TaskSubmit, TaskAccept
+from models import TaskCreate, TaskUpdate, TaskSubmit, TaskDispute, TaskDisputeResponse  # 🔧 P2修复：提交/申诉/回应端点改为 body 模型，匹配前端 JSON 提交
 import 数据库连接 as db
 from 安全认证 import require_auth, is_admin
 from notifications import add_notification
@@ -103,6 +103,11 @@ TASK_STATUS = {
 # 🔄 过期任务自动处理 + 自动退款
 # ==========================================
 import datetime
+import threading
+
+# 🔒 P1修复（CAS）：过期退款串行锁——防止并发请求各持旧 tasks.json 快照，
+# 把同一任务重复判定为过期导致双重退款
+_expired_task_lock = threading.Lock()
 
 def check_and_update_expired_tasks(tasks_db, db_session=None):
     """
@@ -159,60 +164,80 @@ def check_and_update_expired_tasks(tasks_db, db_session=None):
     # ========== 第二阶段：执行SQL事务（钱包退款+交易记录） ==========
     refund_results = []  # 记录退款成功的任务
     sql_committed = False
-    
-    try:
-        for item in expired_task_indices:
-            if item["amount"] > 0:
-                wallet = db_session.query(Wallet).filter(Wallet.account == item["publisher"]).with_for_update().first()
-                if wallet:
-                    wallet.frozen_balance = max(0, wallet.frozen_balance - item["amount"])
-                    wallet.balance += item["amount"]
-                    
-                    # 记录退款交易
-                    create_task_transaction(
-                        db_session, item["publisher"], "TASK_REFUND",
-                        item["amount"], task_id=item["task_id"],
-                        task_title=item["title"]
-                    )
-                    
-                    refund_results.append(item)
-        
-        # 💳 P0修复：先 commit SQL 事务
-        db_session.commit()
-        sql_committed = True
-        logger.info(f"TASK_EXPIRED_SQL_COMMIT | refunded_tasks={len(refund_results)}")
-        
-    except Exception as e:
-        # 💳 P0修复：SQL事务失败，回滚所有操作，不修改JSON
-        db_session.rollback()
-        logger.error(f"TASK_EXPIRED_SQL_ROLLBACK | error={str(e)}")
-        # 不修改 JSON，下次再试
-        return False
-    
-    # ========== 第三阶段：SQL成功后修改JSON任务状态 ==========
-    if sql_committed:
+
+    # 🔒 P1修复（CAS）：整个"复核→退款→写状态"流程串行化，并以磁盘最新数据复核任务状态
+    # （仅退仍为 open 且未退过款的任务），防止并发旧快照导致同一任务重复退款
+    with _expired_task_lock:
+        try:
+            fresh_tasks = {t["id"]: t for t in db.load_data("tasks.json", default_data=[])}
+        except Exception:
+            fresh_tasks = {}  # 复核失败时宁可不退款，等待下次扫描
+
+        def _still_refundable(item):
+            fresh = fresh_tasks.get(item["task_id"])
+            return fresh is not None and fresh.get("status") == "open" and not fresh.get("refunded")
+
+        expired_task_indices = [item for item in expired_task_indices if _still_refundable(item)]
+        if not expired_task_indices:
+            return updated
+
         try:
             for item in expired_task_indices:
-                task = tasks_db[item["index"]]
-                task["status"] = "expired"
-                task["expired_at"] = int(time.time())
                 if item["amount"] > 0:
-                    task["refunded"] = True
-                    task["refund_amount"] = item["amount"]
-            
-            # 💳 P0修复：保存JSON数据
-            db.save_data("tasks.json", tasks_db)
-            logger.info(f"TASK_EXPIRED_JSON_SAVED | tasks={len(expired_task_indices)}")
-            
+                    wallet = db_session.query(Wallet).filter(Wallet.account == item["publisher"]).with_for_update().first()
+                    if wallet:
+                        wallet.frozen_balance = max(0, wallet.frozen_balance - item["amount"])
+                        wallet.balance += item["amount"]
+
+                        # 记录退款交易
+                        create_task_transaction(
+                            db_session, item["publisher"], "TASK_REFUND",
+                            item["amount"], task_id=item["task_id"],
+                            task_title=item["title"]
+                        )
+
+                        refund_results.append(item)
+
+            # 💳 P0修复：先 commit SQL 事务
+            db_session.commit()
+            sql_committed = True
+            logger.info(f"TASK_EXPIRED_SQL_COMMIT | refunded_tasks={len(refund_results)}")
+
         except Exception as e:
-            # 💳 P0修复：JSON保存失败，记录告警但���回滚SQL（资金已转移，可后续手动恢复）
-            import traceback
-            traceback.print_exc()
-            task_ids = ",".join(item["task_id"] for item in expired_task_indices)
-            print(f"[CRITICAL-DATA-INCONSISTENCY] task_id={task_ids}, SQL已退款但JSON状态未更新. Error: {e}")
-            logger.critical(f"TASK_EXPIRED_JSON_SAVE_FAILED | [CRITICAL-DATA-INCONSISTENCY] task_ids={task_ids} | SQL已退款但JSON状态未更新 | error={str(e)}")
-            # 不抛出异常 - 资金安全退回是更高优先级
-            # 定时任务会在下次扫描时修正状态不一致
+            # 💳 P0修复：SQL事务失败，回滚所有操作，不修改JSON
+            db_session.rollback()
+            logger.error(f"TASK_EXPIRED_SQL_ROLLBACK | error={str(e)}")
+            # 不修改 JSON，下次再试
+            return False
+
+        # ========== 第三阶段：SQL成功后修改JSON任务状态 ==========
+        if sql_committed:
+            try:
+                # 🔒 P1修复：按 id 定位任务（而非索引），避免并发修改后索引错位
+                task_map = {t["id"]: t for t in tasks_db}
+                for item in expired_task_indices:
+                    task = task_map.get(item["task_id"])
+                    if not task:
+                        continue
+                    task["status"] = "expired"
+                    task["expired_at"] = int(time.time())
+                    if item["amount"] > 0:
+                        task["refunded"] = True
+                        task["refund_amount"] = item["amount"]
+
+                # 💳 P0修复：保存JSON数据
+                db.save_data("tasks.json", tasks_db)
+                logger.info(f"TASK_EXPIRED_JSON_SAVED | tasks={len(expired_task_indices)}")
+
+            except Exception as e:
+                # 💳 P0修复：JSON保存失败，记录告警但不回滚SQL（资金已转移，可后续手动恢复）
+                import traceback
+                traceback.print_exc()
+                task_ids = ",".join(item["task_id"] for item in expired_task_indices)
+                print(f"[CRITICAL-DATA-INCONSISTENCY] task_id={task_ids}, SQL已退款但JSON状态未更新. Error: {e}")
+                logger.critical(f"TASK_EXPIRED_JSON_SAVE_FAILED | [CRITICAL-DATA-INCONSISTENCY] task_ids={task_ids} | SQL已退款但JSON状态未更新 | error={str(e)}")
+                # 不抛出异常 - 资金安全退回是更高优先级
+                # 定时任务会在下次扫描时修正状态不一致
     
     # 发送退款通知（在事务外执行，失败不影响主流程）
     for item in refund_results:
@@ -748,10 +773,13 @@ async def assign_task(task_id: str, assignee: str, current_user: str = Depends(r
 # ==========================================
 
 @router.post("/api/tasks/{task_id}/submit")
-async def submit_task(task_id: str, deliverables: list, note: str = None, current_user: str = Depends(require_auth)):
+async def submit_task(task_id: str, payload: TaskSubmit, current_user: str = Depends(require_auth)):
     """
     接单者提交成果
     """
+    # 🔧 P2修复：前端以 JSON body 提交，原裸标量参数会被解析为 query 导致 422
+    deliverables = payload.deliverables
+    note = payload.note
     tasks_db = db.load_data("tasks.json", default_data=[])
     
     for task in tasks_db:
@@ -926,11 +954,14 @@ async def accept_task(task_id: str, is_accepted: bool, feedback: str = None, cur
 
 
 @router.post("/api/tasks/{task_id}/dispute")
-async def create_dispute(task_id: str, reason: str, evidence: list = None, current_user: str = Depends(require_auth)):
+async def create_dispute(task_id: str, payload: TaskDispute, current_user: str = Depends(require_auth)):
     """
     发起申诉（发布者或接单者均可）
     - evidence: 证据图片URL列表（可选）
     """
+    # 🔧 P2修复：前端以 JSON body 提交，原裸标量参数会被解析为 query 导致 422
+    reason = payload.reason
+    evidence = payload.evidence
     tasks_db = db.load_data("tasks.json", default_data=[])
     disputes_db = db.load_data("disputes.json", default_data=[])
     
@@ -1030,10 +1061,13 @@ async def get_dispute_detail(dispute_id: str):
     raise HTTPException(status_code=404, detail="申诉不存在")
 
 @router.post("/api/disputes/{dispute_id}/respond")
-async def respond_dispute(dispute_id: str, response: str, evidence: list = None, current_user: str = Depends(require_auth)):
+async def respond_dispute(dispute_id: str, payload: TaskDisputeResponse, current_user: str = Depends(require_auth)):
     """
     被申诉方回应申诉
     """
+    # 🔧 P2修复：前端以 JSON body 提交，原裸标量参数会被解析为 query 导致 422
+    response = payload.response
+    evidence = payload.evidence
     disputes_db = db.load_data("disputes.json", default_data=[])
     
     for dispute in disputes_db:
@@ -1479,54 +1513,25 @@ async def tip_task(task_id: str, amount: int, is_anon: bool = False, current_use
     """
     if amount <= 0:
         raise HTTPException(status_code=400, detail="打赏金额必须大于0")
-    
-    result_container = [None]
-    publisher_account = [None]  # 用于在原子操作外获取发布者账号
-    
-    def updater(data):
-        # 在锁内查找任务
-        target_task = None
-        for task in data:
-            if task["id"] == task_id:
-                target_task = task
-                break
-        
-        if not target_task:
-            result_container[0] = {"error": "not_found"}
-            return
-        
-        # 不能打赏自己
-        if target_task.get("publisher") == current_user:
-            result_container[0] = {"error": "self_tip"}
-            return
-        
-        publisher_account[0] = target_task.get("publisher")
-        
-        # 更新打赏榜单
-        tip_board = target_task.get("tip_board", [])
-        existing = next((t for t in tip_board if t["account"] == current_user), None)
-        if existing:
-            existing["amount"] += amount
-        else:
-            tip_board.append({"account": current_user, "amount": amount, "is_anon": is_anon})
-        tip_board.sort(key=lambda x: x["amount"], reverse=True)
-        target_task["tip_board"] = tip_board
-        
-        result_container[0] = {"status": "success", "message": f"成功打赏 {amount} 积分"}
-    
-    db.atomic_update("tasks.json", updater, default_data=[])
-    
-    result = result_container[0]
-    if result is None or result.get("error") == "not_found":
+    MAX_TIP_AMOUNT = 100000
+    if amount > MAX_TIP_AMOUNT:
+        raise HTTPException(status_code=400, detail=f"打赏金额必须在1-{MAX_TIP_AMOUNT}之间")
+    
+    # 🔒 P0数据修复（时序）：先只读校验 → SQL转账扣款 → 提交成功后再写打赏榜
+    # 避免余额不足/重复请求时榜单已被更新的数据不一致
+    tasks_db_ro = db.load_data("tasks.json", default_data=[])
+    target_task = next((t for t in tasks_db_ro if t["id"] == task_id), None)
+    if not target_task:
         raise HTTPException(status_code=404, detail="任务不存在")
-    if result.get("error") == "self_tip":
+    if target_task.get("publisher") == current_user:
         raise HTTPException(status_code=400, detail="不能打赏自己的任务")
+    publisher = target_task.get("publisher")
+    if not publisher:
+        raise HTTPException(status_code=404, detail="作者账户不存在")
+    task_title = target_task.get("title")
     
-    # 💳 使用 SQL Wallet 系统处理余额转账
+    # 💳 使用 SQL Wallet 系统处理余额转账（先扣款）
     try:
-        publisher = publisher_account[0]
-        if not publisher:
-            raise HTTPException(status_code=404, detail="作者账户不存在")
         
         # 🔒 P1幂等性防护：检查最近5秒内是否存在相同交易
         recent_cutoff = datetime.datetime.utcnow() - datetime.timedelta(seconds=5)
@@ -1566,20 +1571,13 @@ async def tip_task(task_id: str, amount: int, is_anon: bool = False, current_use
         prev_hash_tipper = last_tx_tipper.tx_hash if last_tx_tipper else "GENESIS_HASH"
         prev_hash_author = last_tx_author.tx_hash if last_tx_author else "GENESIS_HASH"
         
-        # 获取用户信息和任务标题
+        # 获取用户信息（task_title 已在预校验阶段取得）
         users_db = db.load_data("users.json", default_data={})
-        tasks_db = db.load_data("tasks.json", default_data=[])
         author_info = users_db.get(publisher, {})
         tipper_info = users_db.get(current_user, {})
         author_name = author_info.get("name", publisher)
         tipper_name = tipper_info.get("name", current_user)
         
-        task_title = None
-        for task in tasks_db:
-            if task["id"] == task_id:
-                task_title = task.get("title")
-                break
-        
         # 打赏方交易记录 (TIP_OUT)
         tx_tipper = Transaction(
             tx_id=tx_id_tipper,
@@ -1648,7 +1646,26 @@ async def tip_task(task_id: str, amount: int, is_anon: bool = False, current_use
         logger.error(f"TASK_TIP_ERROR | task={task_id} | amount={amount} | error={str(e)}")
         raise HTTPException(status_code=500, detail="打赏处理失败，请稍后重试")
     
-    return result
+    # 🏆 扣款成功后再写打赏榜（写榜失败仅记录日志，不影响资金安全）
+    def board_updater(data):
+        for task in data:
+            if task["id"] == task_id:
+                tip_board = task.get("tip_board", [])
+                existing = next((t for t in tip_board if t["account"] == current_user), None)
+                if existing:
+                    existing["amount"] += amount
+                else:
+                    tip_board.append({"account": current_user, "amount": amount, "is_anon": is_anon})
+                tip_board.sort(key=lambda x: x["amount"], reverse=True)
+                task["tip_board"] = tip_board
+                break
+    
+    try:
+        db.atomic_update("tasks.json", board_updater, default_data=[])
+    except Exception as e:
+        logger.error(f"TASK_TIP_BOARD_ERROR | task={task_id} | error={str(e)}")
+    
+    return {"status": "success", "message": f"成功打赏 {amount} 积分"}
 
 # ==========================================
 # 💬 评论接口（复用通用评论系统）

router_wallet.py

@@ -21,7 +21,7 @@ import os
 import datetime
 import logging
 from database_sql import get_db
-from models_sql import Wallet, Transaction, Ownership, Refund
+from models_sql import Wallet, Transaction, Ownership, Refund, RechargeOrder
 from models import RechargeRequest, WithdrawRequest, PurchaseRequest, TipRequest
 import 数据库连接 as json_db
 from notifications import add_notification
@@ -49,8 +49,51 @@ from verify_code_engine import VERIFY_CODES
 router = APIRouter()
 
 # 🔒 P0安全修复：充值订单预存映射，回调时通过订单号查找真实账号，不依赖 subject 字段
+# 🔒 P0数据修复：内存字典仅作热缓存，订单同时落库 RechargeOrder 表，实例重启不丢失
 PENDING_RECHARGE_ORDERS = {}  # {order_id: {"account": str, "amount": int, "created_at": int}}
 
+def _save_pending_order(order_id, account, amount, db):
+    """预存充值订单：内存缓存 + SQL 持久化双写"""
+    PENDING_RECHARGE_ORDERS[order_id] = {
+        "account": account,
+        "amount": amount,
+        "created_at": int(time.time())
+    }
+    try:
+        db.add(RechargeOrder(order_id=order_id, account=account, amount=amount, status="pending"))
+        db.commit()
+    except Exception as e:
+        db.rollback()
+        logger.warning(f"RECHARGE_ORDER_PERSIST_FAILED | order={order_id} | error={str(e)}")
+
+def _get_pending_order(order_id, db):
+    """查找充值订单：优先内存缓存，未命中回查 SQL（重启容灾）"""
+    info = PENDING_RECHARGE_ORDERS.get(order_id)
+    if info:
+        return info
+    try:
+        row = db.query(RechargeOrder).filter(RechargeOrder.order_id == order_id).first()
+        if row:
+            return {
+                "account": row.account,
+                "amount": row.amount,
+                "created_at": int(row.created_at.timestamp()) if row.created_at else 0
+            }
+    except Exception as e:
+        logger.warning(f"RECHARGE_ORDER_QUERY_FAILED | order={order_id} | error={str(e)}")
+    return None
+
+def _mark_order_paid(order_id, db):
+    """订单入账后：清理内存缓存并在 SQL 中标记已支付"""
+    PENDING_RECHARGE_ORDERS.pop(order_id, None)
+    try:
+        row = db.query(RechargeOrder).filter(RechargeOrder.order_id == order_id).first()
+        if row:
+            row.status = "paid"
+            db.commit()
+    except Exception:
+        db.rollback()
+
 # ==========================================
 # 脱敏工具函数
 # ==========================================
@@ -127,7 +170,7 @@ def calculate_tx_hash(tx_id, account, tx_type, amount, prev_hash):
     return hashlib.sha256(data.encode()).hexdigest()
 
 @router.post("/api/wallet/create_recharge_order")
-async def create_recharge_order(req: RechargeRequest, current_user: str = Depends(require_auth)):
+async def create_recharge_order(req: RechargeRequest, current_user: str = Depends(require_auth), db: Session = Depends(get_db)):
     if not alipay:
         # 这里会将真实的错误原因直接弹窗发给前端！
         raise HTTPException(status_code=500, detail="支付网关配置错误，请联系管理员")
@@ -137,12 +180,8 @@ async def create_recharge_order(req: RechargeRequest, current_user: str = Depend
     order_id = f"PAY_{uuid.uuid4().hex}"
     subject = f"ComfyUI Community Points - {authenticated_account}"
     
-    # 🔒 P0安全修复：预存订单号->账号映射，供回调时安全查找
-    PENDING_RECHARGE_ORDERS[order_id] = {
-        "account": authenticated_account,
-        "amount": req.amount,
-        "created_at": int(time.time())
-    }
+    # 🔒 P0安全修复：预存订单号->账号映射，供回调时安全查找（内存+SQL双写）
+    _save_pending_order(order_id, authenticated_account, req.amount, db)
     
     order_string = alipay.api_alipay_trade_precreate(
         out_trade_no=order_id,
@@ -179,8 +218,8 @@ async def alipay_notify(request: Request, db: Session = Depends(get_db)):
         
         amount = int(float(data.get("total_amount", 0)))
         
-        # 🔒 P0安全修复：通过预存订单映射查找真实账号，不依赖 subject 字段提取
-        order_info = PENDING_RECHARGE_ORDERS.get(order_id)
+        # 🔒 P0安全修复：通过预存订单映射查找真实账号，不依赖 subject 字段提取（内存未命中时回查SQL）
+        order_info = _get_pending_order(order_id, db)
         if not order_info:
             logger.warning(f"RECHARGE_ORDER_NOT_FOUND | order={order_id} | subject={data.get('subject', '')}")
             return Response(content="fail", media_type="text/plain")
@@ -192,8 +231,8 @@ async def alipay_notify(request: Request, db: Session = Depends(get_db)):
             logger.warning(f"RECHARGE_ACCOUNT_MISMATCH | order={order_id} | expected={account} | subject={subject_account}")
             return Response(content="fail", media_type="text/plain")
         
-        # 清理已处理的预存订单
-        PENDING_RECHARGE_ORDERS.pop(order_id, None)
+        # 清理已处理的预存订单（内存移除 + SQL 标记 paid）
+        _mark_order_paid(order_id, db)
         
         # 🔒 P0安全修复：充值金额有效性校验
         if amount <= 0:
@@ -243,8 +282,8 @@ async def check_order(order_id: str, current_user: str = Depends(require_auth),
     # 🔒 P0安全修复：使用认证用户账号，不再接受客户端传入的 account 参数
     account = current_user
     
-    # 🔒 纵深防御：优先从预存订单映射查找账号
-    order_info = PENDING_RECHARGE_ORDERS.get(order_id)
+    # 🔒 纵深防御：优先从预存订单映射查找账号（内存未命中时回查SQL）
+    order_info = _get_pending_order(order_id, db)
     if order_info:
         account = order_info["account"]
         # 验证认证用户与订单所属用户一致
@@ -299,7 +338,10 @@ async def check_order(order_id: str, current_user: str = Depends(require_auth),
     return {"status": "PENDING"}
 
 @router.get("/api/wallet/{account}")
-async def get_wallet(account: str, db: Session = Depends(get_db)):
+async def get_wallet(account: str, current_user: str = Depends(require_auth), db: Session = Depends(get_db)):
+    # 🔒 P0安全加固：钱包数据属敏感信息，需登录且仅可查询本人钱包
+    if current_user != account:
+        raise HTTPException(status_code=403, detail="只能查询自己的钱包信息")
     wallet = db.query(Wallet).filter(Wallet.account == account).first()
     
     # 🚀 P1性能优化：使用聚合函数代替 .all() + sum
@@ -813,12 +855,16 @@ async def get_transactions(
     page: int = 1,
     limit: int = 20,
     tx_type: str = None,
+    current_user: str = Depends(require_auth),
     db: Session = Depends(get_db)
 ):
     """
     获取用户交易明细（分页）
     - tx_type: 可选筛选（RECHARGE/PURCHASE/SALE/TIP_OUT/TIP_IN/WITHDRAW/TASK_FREEZE/TASK_DEPOSIT/TASK_PAYMENT/TASK_INCOME/TASK_REFUND）
     """
+    # 🔒 P0安全加固：交易明细属敏感信息，需登录且仅可查询本人记录
+    if current_user != account:
+        raise HTTPException(status_code=403, detail="只能查询自己的交易明细")
     query = db.query(Transaction).filter(Transaction.account == account)
     
     if tx_type:
@@ -865,11 +911,14 @@ async def get_transactions(
     }
 
 @router.get("/api/wallet/{account}/task-stats")
-async def get_task_stats(account: str, db: Session = Depends(get_db)):
+async def get_task_stats(account: str, current_user: str = Depends(require_auth), db: Session = Depends(get_db)):
     """
     📊 获取用户任务收益统计
     🚀 P1性能优化：使用单次查询+分组聚合替代多次查询
     """
+    # 🔒 P0安全加固：收益统计属敏感信息，需登录且仅可查询本人数据
+    if current_user != account:
+        raise HTTPException(status_code=403, detail="只能查询自己的任务统计")
     from sqlalchemy import func as sql_func, case
     
     # 🚀 P1性能优化：使用分组聚合一次查询多种类型的统计
@@ -930,8 +979,11 @@ async def get_task_stats(account: str, db: Session = Depends(get_db)):
     }
 
 @router.get("/api/wallet/{account}/tip-stats")
-async def get_tip_stats(account: str, db: Session = Depends(get_db)):
+async def get_tip_stats(account: str, current_user: str = Depends(require_auth), db: Session = Depends(get_db)):
     """📊 获取用户打赏统计"""
+    # 🔒 P0安全加固：打赏统计属敏感信息，需登录且仅可查询本人数据
+    if current_user != account:
+        raise HTTPException(status_code=403, detail="只能查询自己的打赏统计")
     from sqlalchemy import func as sql_func
     
     stats = db.query(
@@ -963,12 +1015,15 @@ async def get_tip_stats(account: str, db: Session = Depends(get_db)):
     }
 
 @router.get("/api/wallet/{account}/sales-stats")
-async def get_sales_stats(account: str, db: Session = Depends(get_db)):
+async def get_sales_stats(account: str, current_user: str = Depends(require_auth), db: Session = Depends(get_db)):
     """
     📊 获取用户销售统计
     - 销售收入：来自 earn_balance（累计销售收益）
     - 购买支出：来自 PURCHASE 类型的交易记录
     """
+    # 🔒 P0安全加固：销售统计属敏感信息，需登录且仅可查询本人数据
+    if current_user != account:
+        raise HTTPException(status_code=403, detail="只能查询自己的销售统计")
     from sqlalchemy import func as sql_func
     
     # 获取钱包信息（用于读取 earn_balance）
@@ -1057,10 +1112,13 @@ async def get_purchases(
 # ==========================================
 
 @router.get("/api/wallet/{account}/purchase/{item_id}")
-async def get_purchase_status(account: str, item_id: str, db: Session = Depends(get_db)):
+async def get_purchase_status(account: str, item_id: str, current_user: str = Depends(require_auth), db: Session = Depends(get_db)):
     """
     获取购买状态（用���判断是否可退款）
     """
+    # 🔒 P0安全加固：购买状态属敏感信息，需登录且仅可查询本人记录
+    if current_user != account:
+        raise HTTPException(status_code=403, detail="只能查询自己的购买状态")
     ownership = db.query(Ownership).filter(
         Ownership.account == account,
         Ownership.item_id == item_id,

安全认证.py

@@ -27,14 +27,16 @@ import bcrypt
 # ==========================================
 # JWT_SECRET: 用于签名 Token，生产环境应设置环境变量
 # PASSWORD_SALT: 密码哈希加盐，增强安全性
-# 降级策略：环境变量未配置时使用内置默认值（适用于单实例部署）
+# ⚠️ 默认密钥仅供开发环境使用；生产环境必须通过环境变量配置 JWT_SECRET
 _DEFAULT_JWT_SECRET = "ComfyUI-Ranking-JWT-Secret-v2-beta-2026"
 _DEFAULT_PASSWORD_SALT = "ComfyUI-Ranking-Salt-v1"
 
 JWT_SECRET = os.environ.get("JWT_SECRET") or _DEFAULT_JWT_SECRET
 PASSWORD_SALT = os.environ.get("PASSWORD_SALT") or _DEFAULT_PASSWORD_SALT
 
-# 单实例部署时使用内置默认值即可，无需额外配置环境变量
+# 🔒 P0安全加固：未配置 JWT_SECRET 时启动日志告警（默认密钥已公开于源码，必须轮换）
+if not os.environ.get("JWT_SECRET"):
+    print("⚠️ [安全告警] 未设置 JWT_SECRET 环境变量，正在使用内置默认密钥，请尽快配置并轮换！")
 
 # Token 有效期配置（单位：秒）
 TOKEN_EXPIRE_SECONDS = 7 * 24 * 60 * 60  # 默认7天（向后兼容）
@@ -299,7 +301,7 @@ async def require_auth(authorization: str = Header(None, alias="Authorization"))
     
     token = parts[1]
     
-    # 验证 Token（使用 fallback 兼容旧版 mock_token）
+    # 验证 Token（严格 JWT 验证，不再有降级信任）
     is_valid, account, error_msg = verify_token_with_fallback(token)
     
     if not is_valid:
@@ -333,57 +335,33 @@ async def optional_auth(authorization: str = Header(None, alias="Authorization")
 
 
 # ==========================================
-# 🔄 兼容性支持：Mock Token 过渡
+# 🔒 严格 Token 验证（已移除所有降级信任）
 # ==========================================
-# 作用：在过渡期间同时支持旧版 mock_token 和新版 JWT
-# 关联：前端更新后可移除此函数
+# 🔒 P0安全修复历史：旧版实现存在两个高危后门，已全部移除：
+#   1. mock_token_xxx 前缀直接信任后缀为账号（任何人可伪造任意身份）
+#   2. 签名无效/过期的 JWT 仍解析 payload 信任 sub（base64 即可伪造）
 
 def verify_token_with_fallback(token: str) -> Tuple[bool, Optional[str], str]:
     """
-    兼容验证：同时支持 JWT 和旧版 Token（过渡期）
-    
-    验证顺序：
-        1. 标准 JWT 验证（当前密钥）
-        2. 旧版 mock_token 格式兼容
-        3. 签名无效的 JWT：降级提取 payload 中的 sub 字段（过渡期兼容）
-    
-    ⚠️ 过渡期策略：
-        - 当前阶段：宽容模式，允许签名无效的旧 JWT 通过（提取 sub 信任）
-        - 下一版本：移除第3步，强制要求有效签名
+    严格验证：仅信任签名有效且未过期的 JWT
     
     参数：
-        token: Token 字符串（JWT 或 mock_token_xxx）
+        token: Token 字符串
         
     返回：
         (is_valid, account, error_message)
     """
-    # 1. 先尝试标准 JWT 验证
+    # 🔒 拒绝旧版 mock_token：前缀可被任何人构造，等同于任意账号伪造后门
+    if token.startswith("mock_token_"):
+        return False, None, "旧版凭证已失效，请重新登录"
+
     is_valid, payload, error_msg = verify_token(token)
     if is_valid:
-        return True, payload.get("sub"), ""
-    
-    # 2. 降级：检查是否为旧版 mock_token
-    if token.startswith("mock_token_"):
-        account = token.replace("mock_token_", "")
-        print(f"⚠️ [过渡兼容] 旧版 mock_token 用户: {account}")
+        account = payload.get("sub")
+        if not account:
+            return False, None, "Token 中缺少用户信息"
         return True, account, ""
-    
-    # 3. 过渡期兼容：JWT 签名无效时，尝试直接解析 payload 提取账号
-    #    这允许旧版（密钥不同时签发的）JWT 仍能使用
-    #    ⚠️ 下一版本将移除此降级逻辑，届时用户必须重新登录
-    if error_msg in ("Token 签名无效", "Token 已过期，请重新登录"):
-        try:
-            parts = token.split(".")
-            if len(parts) == 3:
-                payload_json = _base64url_decode(parts[1]).decode("utf-8")
-                payload_data = json.loads(payload_json)
-                account = payload_data.get("sub")
-                if account:
-                    print(f"⚠️ [过渡兼容] 旧版 JWT 降级信任，用户: {account}（签名未验证，下版本将失效）")
-                    return True, account, ""
-        except Exception:
-            pass
-    
+
     return False, None, error_msg or "Token 验证失败"
 
 

数据库连接.py

@@ -425,9 +425,13 @@ def save_data(file_name: str, data: Union[Dict, List], force_sync: bool = False)
         except Exception as e:
             # 写入失败，清理临时文件
             if os.path.exists(temp_path):
-                os.remove(temp_path)
+                try:
+                    os.remove(temp_path)
+                except OSError:
+                    pass
             print(f"🚨 保存 {file_name} 失败: {e}")
-            raise
+            # 🔧 回归核查修复：按契约返回 False（旧版 raise 与 -> bool 声明不符，且无调用方依赖异常）
+            return False
     
     # ========== 第五步：标记文件脏，等待批量同步 ==========
     if HF_TOKEN:
@@ -437,6 +441,8 @@ def save_data(file_name: str, data: Union[Dict, List], force_sync: bool = False)
         if force_sync:
             _immediate_sync_file(file_name)
 
+    return True
+
 
 # ==========================================
 # ☁️ HF 批量同步机制

测试脚本.py

@@ -34,12 +34,16 @@ class TestAuthSecurity(unittest.TestCase):
         self.create_token = create_token
         self.verify_token = verify_token
     
-    def test_password_hash_deterministic(self):
-        """测试：相同密码生成相同哈希"""
+    def test_password_hash_unique_and_verifiable(self):
+        """测试：bcrypt 每次加盐不同，哈希唯一且可验证"""
+        # 🧹 P2修正：bcrypt 随机加盐，相同密码每次哈希不同（旧断言基于已废弃的固定盐 SHA256）
         password = "test123456"
         hash1 = self.hash_password(password)
         hash2 = self.hash_password(password)
-        self.assertEqual(hash1, hash2, "相同密码应生成相同哈希")
+        self.assertNotEqual(hash1, hash2, "bcrypt 应每次生成不同哈希（随机盐）")
+        self.assertTrue(hash1.startswith("$2b$"), "哈希应为 bcrypt 格式")
+        self.assertTrue(self.verify_password(password, hash1), "哈希1应验证通过")
+        self.assertTrue(self.verify_password(password, hash2), "哈希2应验证通过")
     
     def test_password_hash_different(self):
         """测试：不同密码生成不同哈希"""
@@ -58,14 +62,19 @@ class TestAuthSecurity(unittest.TestCase):
         hashed = self.hash_password("correctPassword")
         self.assertFalse(self.verify_password("wrongPassword", hashed), "错误密码应验证失败")
     
-    def test_password_verify_plaintext_fallback(self):
-        """测试：明文密码兼容（向后兼容）"""
-        plaintext = "oldPassword123"
-        # 明文密码（旧版格式）
-        self.assertTrue(self.verify_password(plaintext, plaintext), "明文密码应向后兼容")
+    def test_password_verify_legacy_sha256(self):
+        """测试：旧版 SHA256 哈希向后兼容"""
+        # 🧹 P2修正：明文兼容已不存在，实际兼容路径为旧版 SHA256（64位十六进制）
+        import hashlib
+        from 安全认证 import PASSWORD_SALT
+        password = "oldPassword123"
+        legacy_hash = hashlib.sha256(f"{PASSWORD_SALT}{password}".encode("utf-8")).hexdigest()
+        self.assertTrue(self.verify_password(password, legacy_hash), "旧版 SHA256 哈希应向后兼容")
+        self.assertFalse(self.verify_password("wrongPassword", legacy_hash), "错误密码不应通过旧版兼容路径")
     
     def test_token_create_and_verify(self):
         """测试：Token 创建和验证"""
+        # 🧹 P2修正：verify_token 返回 (is_valid, payload, error) 三元组
         account = "test_user"
         token = self.create_token(account)
         
@@ -74,13 +83,15 @@ class TestAuthSecurity(unittest.TestCase):
         self.assertIn(".", token, "Token 应包含点号分隔符")
         
         # 验证 Token
-        verified_account = self.verify_token(token)
-        self.assertEqual(verified_account, account, "验证后的账号应与原账号一致")
+        is_valid, payload, error_msg = self.verify_token(token)
+        self.assertTrue(is_valid, f"有效 Token 应验证通过: {error_msg}")
+        self.assertEqual(payload.get("sub"), account, "验证后的账号应与原账号一致")
     
     def test_token_invalid(self):
         """测试：无效 Token 验证失败"""
-        result = self.verify_token("invalid.token.here")
-        self.assertIsNone(result, "无效 Token 应返回 None")
+        is_valid, payload, error_msg = self.verify_token("invalid.token.here")
+        self.assertFalse(is_valid, "无效 Token 应验证失败")
+        self.assertIsNone(payload, "无效 Token 的 payload 应为 None")
     
     def test_token_with_extra_data(self):
         """测试：Token 携带额外数据"""
@@ -88,18 +99,20 @@ class TestAuthSecurity(unittest.TestCase):
         extra = {"role": "admin", "level": 5}
         token = self.create_token(account, extra_data=extra)
         
-        verified_account = self.verify_token(token)
-        self.assertEqual(verified_account, account, "验证后的账号应正确")
+        is_valid, payload, _ = self.verify_token(token)
+        self.assertTrue(is_valid, "携带额外数据的 Token 应验证通过")
+        self.assertEqual(payload.get("sub"), account, "验证后的账号应正确")
+        self.assertEqual(payload.get("role"), "admin", "额外数据应保留在 payload 中")
     
-    def test_mock_token_support(self):
-        """测试：mock_token 兼容（向后兼容）"""
+    def test_mock_token_rejected(self):
+        """测试：旧版 mock_token 必须被拒绝（P0安全后门已移除）"""
         from 安全认证 import verify_token_with_fallback
         
-        # 旧版 mock_token 格式
+        # 🧹 P2修正：mock_token 前缀可伪造任意身份，P0已移除降级信任，必须拒绝
         mock_token = "mock_token_testuser_1234567890"
         is_valid, account, _ = verify_token_with_fallback(mock_token)
-        self.assertTrue(is_valid, "mock_token 应通过验证")
-        self.assertEqual(account, "testuser_1234567890", "mock_token 应正确解析账号")
+        self.assertFalse(is_valid, "mock_token 应被拒绝（伪造后门已封堵）")
+        self.assertIsNone(account, "mock_token 不应解析出账号")
 
 
 # ==========================================