公测更新，大量优化

.dockerignore

@@ -4,4 +4,9 @@ __pycache__
 /cache
 /tmp
 *.zip
-*.log
+*.log
+data/
+.env
+*.bak
+backups/
+一次性文件已完成/

Dockerfile

@@ -18,5 +18,12 @@ RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
 # 复制应用代码
 COPY . .
 
+# 安全加固：以非root用户运行
+RUN useradd -m -s /bin/bash appuser && chown -R appuser:appuser /code
+USER appuser
+
+# 健康检查
+HEALTHCHECK --interval=30s --timeout=5s --retries=3 CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:7860/health')" || exit 1
+
 # 启动命令
 CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

app.py

@@ -69,7 +69,7 @@ limiter = Limiter(key_func=get_remote_address)
 app = FastAPI(
     title="ComfyUI Ranking API",
     description="ComfyUI 社区排名系统 API",
-    version="1.0.0",
+    version="2.0.0",
     docs_url="/api/docs",
     redoc_url="/api/redoc"
 )
@@ -246,9 +246,13 @@ async def on_shutdown():
     # 这里可以添加其他清理逻辑（如关闭连接池等）
     logger.info("✅ 关闭完成")
 
+# 🛡️ CORS 安全配置：从环境变量读取允许的域名，收紧通配符
+ALLOWED_ORIGINS = os.environ.get("ALLOWED_ORIGINS", "https://zhiwei666-comfyui-ranking-api.hf.space").split(",")
+logger.info(f"CORS 允许域名: {ALLOWED_ORIGINS}")
+
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=["*"],
+    allow_origins=ALLOWED_ORIGINS,
     allow_credentials=False, 
     allow_methods=["*"],
     allow_headers=["*"],
@@ -289,6 +293,10 @@ def proxy_hf_image(url: str = None, path: str = None):
             
     if not path:
         raise HTTPException(status_code=400, detail="缺少路径参数")
+    
+    # 🛡️ 路径穿越防护：禁止包含 .. 和绝对路径
+    if ".." in path or path.startswith("/"):
+        raise HTTPException(status_code=400, detail="非法路径")
         
     # 🛡️ 绝对的安全红线：限制只能代理下载图片目录，严禁黑客通过此接口下载 users.json 或账本数据！
     allowed_dirs = ["uploads/", "avatars/", "covers/"]
@@ -341,6 +349,9 @@ def upload_file(file: UploadFile = File(...), file_type: str = Form(...), curren
     if ext not in allowed_exts:
         raise HTTPException(status_code=400, detail=f"不支持的文件格式，{file_type} 类型仅支持 {', '.join(allowed_exts)}")
     
+    # 📝 审计日志：记录上传操作
+    logger.info(f"UPLOAD | account={current_user} | filename={file.filename} | file_type={file_type}")
+    
     # 🔒 P0安全优化：图片内容审核
     if ext in ["jpg", "jpeg", "png", "gif", "webp"]:
         moderation_result = moderate_image_sync(content, ext)
@@ -467,18 +478,12 @@ def validate_resource(req_data: ValidateResourceRequest, sql_db: Session = Depen
 # ==========================================
 # 🔒 管理员：系统配置 API
 # ==========================================
-from 安全认证 import require_auth
-
-# 🔒 管理员账号列表（从环境变量读取）
-ADMIN_ACCOUNTS = set(
-    acc.strip() 
-    for acc in os.environ.get("ADMIN_ACCOUNTS", "").split(",") 
-    if acc.strip()
-)
+from 安全认证 import require_auth, ADMIN_ACCOUNTS
 
 def _is_admin(account: str) -> bool:
     """检查账号是否为管理员"""
-    return account in ADMIN_ACCOUNTS
+    from 安全认证 import is_admin as _check_admin
+    return _check_admin(account)
 
 # 配置存储文件
 SYSTEM_CONFIG_FILE = "/tmp/system_config.json"
@@ -551,8 +556,8 @@ async def get_system_config(config_key: str, current_user: str = Depends(require
 
 # 🏷️ 版本配置默认值
 DEFAULT_VERSION_CONFIG = {
-    "stage": "alpha",
-    "major": 1,
+    "stage": "beta",
+    "major": 2,
     "minor": 0,
     "patch": 0
 }

database_sql.py

@@ -110,6 +110,7 @@ def _auto_migrate_p7_fields():
     """
     🔄 P7后悔模式：自动迁移新增字段
     检查并添加 ownerships 表和 transactions 表的新字段
+    所有 ALTER TABLE ADD COLUMN 均使用 IF NOT EXISTS 或 inspector 预检，避免重复执行报错
     """
     from sqlalchemy import inspect
     
@@ -122,22 +123,34 @@ def _auto_migrate_p7_fields():
             
             with engine.connect() as conn:
                 if 'price_paid' not in columns:
-                    if 'sqlite' in SQLALCHEMY_DATABASE_URL:
-                        conn.execute(text("ALTER TABLE ownerships ADD COLUMN price_paid INTEGER DEFAULT 0"))
-                    else:
-                        conn.execute(text("ALTER TABLE ownerships ADD COLUMN price_paid INTEGER DEFAULT 0"))
-                    logger.info("迁移完成: 添加 ownerships.price_paid 字段")
+                    try:
+                        if 'sqlite' in SQLALCHEMY_DATABASE_URL:
+                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN price_paid INTEGER DEFAULT 0"))
+                        else:
+                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN IF NOT EXISTS price_paid INTEGER DEFAULT 0"))
+                        logger.info("迁移完成: 添加 ownerships.price_paid 字段")
+                    except Exception as e:
+                        logger.debug(f"跳过 ownerships.price_paid (可能已存在): {e}")
                 
                 if 'is_refunded' not in columns:
-                    if 'sqlite' in SQLALCHEMY_DATABASE_URL:
-                        conn.execute(text("ALTER TABLE ownerships ADD COLUMN is_refunded BOOLEAN DEFAULT 0"))
-                    else:
-                        conn.execute(text("ALTER TABLE ownerships ADD COLUMN is_refunded BOOLEAN DEFAULT FALSE"))
-                    logger.info("迁移完成: 添加 ownerships.is_refunded 字段")
+                    try:
+                        if 'sqlite' in SQLALCHEMY_DATABASE_URL:
+                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN is_refunded BOOLEAN DEFAULT 0"))
+                        else:
+                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN IF NOT EXISTS is_refunded BOOLEAN DEFAULT FALSE"))
+                        logger.info("迁移完成: 添加 ownerships.is_refunded 字段")
+                    except Exception as e:
+                        logger.debug(f"跳过 ownerships.is_refunded (可能已存在): {e}")
                 
                 if 'refunded_at' not in columns:
-                    conn.execute(text("ALTER TABLE ownerships ADD COLUMN refunded_at TIMESTAMP"))
-                    logger.info("迁移完成: 添加 ownerships.refunded_at 字段")
+                    try:
+                        if 'sqlite' in SQLALCHEMY_DATABASE_URL:
+                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN refunded_at TIMESTAMP"))
+                        else:
+                            conn.execute(text("ALTER TABLE ownerships ADD COLUMN IF NOT EXISTS refunded_at TIMESTAMP"))
+                        logger.info("迁移完成: 添加 ownerships.refunded_at 字段")
+                    except Exception as e:
+                        logger.debug(f"跳过 ownerships.refunded_at (可能已存在): {e}")
                 
                 conn.commit()
         
@@ -147,11 +160,14 @@ def _auto_migrate_p7_fields():
             
             with engine.connect() as conn:
                 if 'task_balance' not in columns:
-                    if 'sqlite' in SQLALCHEMY_DATABASE_URL:
-                        conn.execute(text("ALTER TABLE wallets ADD COLUMN task_balance INTEGER DEFAULT 0"))
-                    else:
-                        conn.execute(text("ALTER TABLE wallets ADD COLUMN task_balance INTEGER DEFAULT 0"))
-                    logger.info("[DB Migration] 添加列 wallets.task_balance")
+                    try:
+                        if 'sqlite' in SQLALCHEMY_DATABASE_URL:
+                            conn.execute(text("ALTER TABLE wallets ADD COLUMN task_balance INTEGER DEFAULT 0"))
+                        else:
+                            conn.execute(text("ALTER TABLE wallets ADD COLUMN IF NOT EXISTS task_balance INTEGER DEFAULT 0"))
+                        logger.info("[DB Migration] 添加列 wallets.task_balance")
+                    except Exception as e:
+                        logger.debug(f"跳过 wallets.task_balance (可能已存在): {e}")
                 
                 conn.commit()
         
@@ -175,12 +191,21 @@ def _auto_migrate_p7_fields():
                 
                 for col_name, col_type in new_columns.items():
                     if col_name not in columns:
-                        # VARCHAR 类型添加 DEFAULT NULL 避免 PostgreSQL NOT NULL 冲突
-                        if col_type == 'VARCHAR':
-                            conn.execute(text(f"ALTER TABLE transactions ADD COLUMN {col_name} {col_type} DEFAULT NULL"))
-                        else:
-                            conn.execute(text(f"ALTER TABLE transactions ADD COLUMN {col_name} {col_type}"))
-                        logger.info(f"[DB Migration] 添加列 transactions.{col_name}")
+                        try:
+                            # VARCHAR 类型添加 DEFAULT NULL 避免 PostgreSQL NOT NULL 冲突
+                            if col_type == 'VARCHAR':
+                                if 'sqlite' in SQLALCHEMY_DATABASE_URL:
+                                    conn.execute(text(f"ALTER TABLE transactions ADD COLUMN {col_name} {col_type} DEFAULT NULL"))
+                                else:
+                                    conn.execute(text(f"ALTER TABLE transactions ADD COLUMN IF NOT EXISTS {col_name} {col_type} DEFAULT NULL"))
+                            else:
+                                if 'sqlite' in SQLALCHEMY_DATABASE_URL:
+                                    conn.execute(text(f"ALTER TABLE transactions ADD COLUMN {col_name} {col_type}"))
+                                else:
+                                    conn.execute(text(f"ALTER TABLE transactions ADD COLUMN IF NOT EXISTS {col_name} {col_type}"))
+                            logger.info(f"[DB Migration] 添加列 transactions.{col_name}")
+                        except Exception as e:
+                            logger.debug(f"跳过 transactions.{col_name} (可能已存在): {e}")
                 
                 conn.commit()
                 

image_moderation.py

@@ -221,9 +221,9 @@ async def moderate_image_tencent(image_content: bytes) -> ModerationResult:
         
     except Exception as e:
         logger.error(f"腾讯云审核异常: {str(e)}")
-        # 审核服务异常时默认放行，避免影响正常业务
-        return ModerationResult(passed=True, suggestion="pass", 
-                               details={"error": str(e)})
+        # 审核服务异常时拒绝，需要人工审核
+        return ModerationResult(passed=False, suggestion="review",
+                               details={"error": "审核服务暂时不可用，需人工审核"})
 
 # ==========================================
 # 🟠 阿里云图片审核
@@ -367,9 +367,9 @@ async def moderate_image_aliyun(image_content: bytes) -> ModerationResult:
         
     except Exception as e:
         logger.error(f"阿里云审核异常: {str(e)}")
-        # 审核服务异常时默认放行
-        return ModerationResult(passed=True, suggestion="pass",
-                               details={"error": str(e)})
+        # 审核服务异常时拒绝，需要人工审核
+        return ModerationResult(passed=False, suggestion="review",
+                               details={"error": "审核服务暂时不可用，需人工审核"})
 
 # ==========================================
 # 🎯 统一审核入口（智能额度管理）

models.py

@@ -1,7 +1,22 @@
 # models.py
+import re
 from pydantic import BaseModel, Field, validator
 from typing import Optional, List
 
+def _validate_email_format(email: str) -> str:
+    """邮箱格式校验"""
+    if not re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', email):
+        raise ValueError('邮箱格式不正确')
+    return email
+
+def _validate_password_length(password: str) -> str:
+    """密码长度校验：8-64字符"""
+    if len(password) < 8:
+        raise ValueError('密码长度不能少于8个字符')
+    if len(password) > 64:
+        raise ValueError('密码长度不能超过64个字符')
+    return password
+
 class SendCodeRequest(BaseModel):
     contact: str
     contact_type: str  
@@ -21,11 +36,16 @@ class UserRegister(BaseModel):
     country: Optional[str] = None
     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)
 
 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)
 
 class UserUpdate(BaseModel):
     name: Optional[str] = None
@@ -45,6 +65,8 @@ class PasswordReset(BaseModel):
     verifyType: str
     code: str
     account: str
+    
+    _validate_new_password = validator('new_password', allow_reuse=True)(_validate_password_length)
 
 class CommentCreate(BaseModel):
     item_id: str

notifications.py

@@ -5,18 +5,27 @@ import 数据库连接 as db
 def add_notification(target_account: str, notif_data: dict):
     try:
         from_user = notif_data.get("from_user")
-        if target_account == from_user or not target_account: 
-            return # 不给自己发通知
+        if not target_account:
+            return
+        if from_user and target_account == from_user:
+            return  # 不给自己发通知
         
         users_db = db.load_data("users.json", default_data={})
-        user_info = users_db.get(from_user, {})
+        
+        if from_user == "system" or not from_user:
+            from_name = "系统通知"
+            from_avatar = ""
+        else:
+            user_info = users_db.get(from_user, {})
+            from_name = user_info.get("name", from_user)
+            from_avatar = user_info.get("avatarDataUrl", "")
         
         notif = {
             "id": f"msg_{int(time.time())}_{uuid.uuid4().hex[:6]}",
             "type": notif_data.get("type"), 
             "from_user": from_user,
-            "from_name": user_info.get("name", from_user),
-            "from_avatar": user_info.get("avatarDataUrl", ""),
+            "from_name": from_name,
+            "from_avatar": from_avatar,
             "target_item_id": notif_data.get("target_item_id", ""),
             "target_item_title": notif_data.get("target_item_title", ""),
             "content": notif_data.get("content", ""),

rate_limiter.py

@@ -78,6 +78,28 @@ def get_limit(action: str) -> str:
 user_request_counts = {}
 USER_LIMIT_WINDOW = 60  # 时间窗口（秒）
 USER_LIMIT_MAX = 100    # 每用户每分钟最大请求数
+MAX_DICT_SIZE = 10000   # 字典最大条目数
+
+
+def cleanup_expired_entries():
+    """清理超过1小时的过期记录，防止内存泄漏"""
+    now = time.time()
+    # 清理 user_request_counts
+    expired_user_keys = [
+        k for k, v in list(user_request_counts.items())
+        if all(now - t > 3600 for t in v)
+    ]
+    for k in expired_user_keys:
+        del user_request_counts[k]
+    # 清理 ip_stats
+    expired_ip_keys = [
+        k for k, v in list(ip_stats.items())
+        if now - v.get("last_request", 0) > 3600
+    ]
+    for k in expired_ip_keys:
+        del ip_stats[k]
+    if expired_user_keys or expired_ip_keys:
+        logger.debug(f"CLEANUP | user_keys={len(expired_user_keys)}, ip_keys={len(expired_ip_keys)}")
 
 
 def check_user_rate_limit(account: str) -> bool:
@@ -91,6 +113,10 @@ def check_user_rate_limit(account: str) -> bool:
     now = time.time()
     key = f"user:{account}"
     
+    # 超出字典上限时触发清理
+    if len(user_request_counts) >= MAX_DICT_SIZE:
+        cleanup_expired_entries()
+    
     # 清理过期记录
     if key in user_request_counts:
         user_request_counts[key] = [
@@ -141,6 +167,10 @@ def record_request(ip: str, endpoint: str):
     now = time.time()
     key = f"{ip}:{endpoint}"
     
+    # 超出字典上限时触发清理
+    if len(ip_stats) >= MAX_DICT_SIZE:
+        cleanup_expired_entries()
+    
     if key not in ip_stats:
         ip_stats[key] = {"count": 0, "first_request": now}
     

router_comments.py

@@ -2,6 +2,7 @@
 from fastapi import APIRouter, HTTPException, Depends
 import time
 import uuid
+from html import escape
 import 数据库连接 as db
 from notifications import add_notification
 from models import InteractionToggle, CommentCreate
@@ -10,7 +11,8 @@ from 安全认证 import require_auth, is_admin
 router = APIRouter()
 
 @router.post("/api/interactions/toggle")
-async def toggle_interaction(interaction: InteractionToggle):
+async def toggle_interaction(interaction: InteractionToggle, current_user: str = Depends(require_auth)):
+    interaction.user_id = current_user  # 强制使用当前用户身份，防止伪造
     items_db = db.load_data("items.json", default_data=[])
     target_item = next((item for item in items_db if item["id"] == interaction.item_id), None)
     if not target_item: raise HTTPException(status_code=404, detail="目标存在问题")
@@ -33,7 +35,13 @@ async def toggle_interaction(interaction: InteractionToggle):
     return {"status": "success", "new_count": target_item[count_key]}
 
 @router.post("/api/comments")
-async def post_comment(comment: CommentCreate):
+async def post_comment(comment: CommentCreate, current_user: str = Depends(require_auth)):
+    comment.author = current_user  # 强制使用认证用户身份，防止伪造
+    # 评论内容长度校验
+    if not comment.content or len(comment.content) > 5000:
+        raise HTTPException(status_code=400, detail="评论长度必须在1-5000字符之间")
+    # HTML 转义，防止 XSS
+    comment.content = escape(comment.content)
     comments_db = db.load_data("comments.json", default_data={}) 
     users_db = db.load_data("users.json", default_data={})
     author_info = users_db.get(comment.author, {})
@@ -147,8 +155,22 @@ async def soft_delete_comment(item_id: str, comment_id: str, account: str = Depe
                 return True
         return False
     
-    mark_deleted(item_comments)
-    comments_db[item_id] = item_comments
-    db.save_data("comments.json", comments_db)
+    # 原子操作软删除，防止并发覆盖
+    max_retries = 3
+    for attempt in range(max_retries):
+        try:
+            comments_db = db.load_data("comments.json", default_data={})
+            item_comments = comments_db.get(item_id, [])
+            if not find_comment(item_comments):
+                raise HTTPException(status_code=404, detail="找不到该评论")
+            mark_deleted(item_comments)
+            comments_db[item_id] = item_comments
+            db.save_data("comments.json", comments_db)
+            break
+        except HTTPException:
+            raise
+        except Exception:
+            if attempt == max_retries - 1:
+                raise
     
     return {"status": "success"}

router_items.py

@@ -16,6 +16,12 @@ from db_utils import record_view, sort_cache
 
 router = APIRouter()
 
+def safe_str(val):
+    """安全转换为字符串"""
+    if val is None:
+        return ""
+    return str(val).lower()
+
 def _get_version_str(versions_db: dict, item_id: str) -> str:
     """兼容新旧格式获取版本hash字符串"""
     val = versions_db.get(item_id, "")
@@ -126,11 +132,14 @@ def _build_creator_trend_data(account: str, u_items: list, months: list) -> dict
         itype = i.get("type", "")
         history = i.get("use_history", {})
         if itype == "tool" or itype == "recommend_tool": 
-            for m in months: trend_tools[m] += history.get(m, 0)
+            for m, v in history.items():
+                if m in trend_tools: trend_tools[m] += v
         elif itype == "app" or itype == "recommend_app": 
-            for m in months: trend_apps[m] += history.get(m, 0)
+            for m, v in history.items():
+                if m in trend_apps: trend_apps[m] += v
         elif itype.startswith("recommend"):
-            for m in months: trend_recommends[m] += history.get(m, 0)
+            for m, v in history.items():
+                if m in trend_recommends: trend_recommends[m] += v
     
     return {
         "months": months,
@@ -171,7 +180,7 @@ async def search_creators(keyword: str, sort: str = "downloads", limit: int = 50
         short_desc = u.get("shortDesc") or u.get("intro") or ""
         
         # 不区分大小写的子串匹配（同时覆盖 shortDesc 和 intro 两个字段）
-        search_text = f"{name} {account} {short_desc} {u.get('intro') or ''} {u.get('shortDesc') or ''}".lower()
+        search_text = f"{safe_str(name)} {safe_str(account)} {safe_str(short_desc)} {safe_str(u.get('intro'))} {safe_str(u.get('shortDesc'))}"
         if keyword_lower not in search_text:
             continue
         

router_messages.py

@@ -102,14 +102,19 @@ async def run_admin_script(req: AdminScriptRequest, current_user: str = Depends(
         }
     
     try:
-        # 执行脚本，设置超时 60 秒
+        # 清理敏感环境变量，防止泄露到子进程
+        sensitive_prefixes = ('GITHUB', 'JWT', 'HF_', 'DATABASE', 'ALIPAY', 'PASSWORD')
+        clean_env = {k: v for k, v in os.environ.items()
+                     if not any(k.startswith(prefix) for prefix in sensitive_prefixes)}
+        # 执行脚本，超时缩短至30秒
         result = subprocess.run(
             ["python", script_path],
             capture_output=True,
             text=True,
-            timeout=60,
+            timeout=30,
             cwd=current_dir,
-            encoding="utf-8"
+            encoding="utf-8",
+            env=clean_env
         )
         
         output = ""
@@ -119,6 +124,10 @@ async def run_admin_script(req: AdminScriptRequest, current_user: str = Depends(
             output += f"\n⚠️ 错误输出:\n{result.stderr}"
         if not output:
             output = "✅ 脚本执行完成，无输出"
+        
+        # 限制输出大小防止返回大量数据
+        if len(output) > 1000:
+            output = output[:1000] + "\n...（输出已截断）"
             
         return {
             "status": "success" if result.returncode == 0 else "error",
@@ -129,7 +138,7 @@ async def run_admin_script(req: AdminScriptRequest, current_user: str = Depends(
     except subprocess.TimeoutExpired:
         return {
             "status": "error",
-            "output": "❌ 脚本执行超时 (60秒)"
+            "output": "❌ 脚本执行超时 (30秒)"
         }
     except Exception as e:
         return {
@@ -141,7 +150,8 @@ async def run_admin_script(req: AdminScriptRequest, current_user: str = Depends(
 # 原有功能：私信与聊天
 # ==========================================
 @router.post("/api/messages/private")
-async def send_private_message(msg: PrivateMessage):
+async def send_private_message(msg: PrivateMessage, current_user: str = Depends(require_auth)):
+    msg.sender = current_user  # 强制发送者身份，防止伪造
     chats_db = db.load_data("chats.json", default_data={})
     conv_id = f"{min(msg.sender, msg.receiver)}_{max(msg.sender, msg.receiver)}"
     if conv_id not in chats_db: chats_db[conv_id] = []
@@ -154,7 +164,9 @@ async def send_private_message(msg: PrivateMessage):
     return {"status": "success"}
 
 @router.get("/api/chats/{account}")
-async def get_chat_list(account: str):
+async def get_chat_list(account: str, current_user: str = Depends(require_auth)):
+    if account != current_user:
+        raise HTTPException(status_code=403, detail="无权访问他人私信")
     chats_db = db.load_data("chats.json", default_data={})
     users_db = db.load_data("users.json", default_data={})
     
@@ -179,7 +191,9 @@ async def get_chat_list(account: str):
     return {"status": "success", "data": chat_list}
 
 @router.get("/api/chats/{account}/{target_account}")
-async def get_chat_history(account: str, target_account: str):
+async def get_chat_history(account: str, target_account: str, current_user: str = Depends(require_auth)):
+    if account != current_user:
+        raise HTTPException(status_code=403, detail="无权访问他人聊天记录")
     chats_db = db.load_data("chats.json", default_data={})
     conv_id = f"{min(account, target_account)}_{max(account, target_account)}"
     msgs = chats_db.get(conv_id, [])

router_posts.py

@@ -425,6 +425,9 @@ async def tip_post(post_id: str, amount: int, is_anon: bool = False, current_use
     """
     if amount <= 0:
         raise HTTPException(status_code=400, detail="打赏金额必须大于0")
+    MAX_TIP_AMOUNT = 100000
+    if amount > MAX_TIP_AMOUNT:
+        raise HTTPException(status_code=400, detail=f"打赏金额必须在1-{MAX_TIP_AMOUNT}之间")
     
     result_container = [None]
     author_account = [None]  # 用于在原子操作外获取作者账号

router_proxy.py

@@ -4,6 +4,8 @@ from fastapi.responses import StreamingResponse, JSONResponse, Response
 from sqlalchemy.orm import Session
 from pydantic import BaseModel, Field, field_validator, HttpUrl
 from typing import Optional
+from urllib.parse import urlparse
+import ipaddress
 import httpx
 import os
 import re
@@ -145,10 +147,14 @@ async def proxy_github_zip(req_data: ProxyGithubZipRequest, db: Session = Depend
     # 3. 异步请求 GitHub API 并以流形式透传回客户端 (防内存打爆)
     client = httpx.AsyncClient(follow_redirects=True)
     try:
-        response = await client.send(
-            client.build_request("GET", github_zip_api, headers=headers),
-            stream=True
-        )
+        try:
+            response = await client.send(
+                client.build_request("GET", github_zip_api, headers=headers),
+                stream=True
+            )
+        except Exception as send_err:
+            await client.aclose()
+            return JSONResponse(content={"error": f"代理下载时发生网络异常：{str(send_err)}"}, status_code=500)
 
         # 非 200 直接返回错误，避免 Content-Length 不匹配导致 h11 协议错误
         if response.status_code != 200:
@@ -215,9 +221,24 @@ class ProxyDownloadRequest(BaseModel):
     @field_validator('url')
     @classmethod
     def validate_url(cls, v: str) -> str:
-        """验证URL格式"""
+        """验证URL格式并阻止内网地址"""
         if not v.startswith(('http://', 'https://')):
             raise ValueError('url必须是有效的HTTP或HTTPS地址')
+        
+        parsed = urlparse(v)
+        # 阻止内网地址
+        dangerous_hosts = ['localhost', '127.0.0.1', '0.0.0.0', '::1']
+        if parsed.hostname in dangerous_hosts:
+            raise ValueError('不允许访问本地地址')
+        
+        # 检查是否是私有IP
+        try:
+            ip = ipaddress.ip_address(parsed.hostname)
+            if ip.is_private or ip.is_loopback or ip.is_reserved:
+                raise ValueError('不允许访问内网地址')
+        except ValueError:
+            pass  # hostname不是IP，允许继续
+        
         return v
 
 @router.post("/api/proxy_download")

router_tasks.py

@@ -20,6 +20,7 @@ from notifications import add_notification
 from database_sql import get_db
 from models_sql import Wallet, Transaction
 from db_utils import record_view, sort_cache
+from html import escape
 import time
 import uuid
 import hashlib
@@ -78,7 +79,7 @@ def create_task_transaction(db_session: Session, account: str, tx_type: str, amo
     )
     db_session.add(new_tx)
     
-    logger.info(f"TASK_TX | type={tx_type} | account={account} | amount={amount} | task={task_id} | tx={tx_id}")
+    logger.info(f"TASK_TX | type={tx_type} | account_hash={hash(account)} | amount={amount} | task={task_id} | tx={tx_id}")
     return tx_id
 
 # ==========================================
@@ -218,7 +219,7 @@ def check_and_update_expired_tasks(tasks_db, db_session=None):
                 "target_item_title": item["title"],
                 "content": f"💰 任务《{item['title']}》已过期自动取消，{item['amount']}积分已退还"
             })
-            logger.info(f"TASK_REFUND | publisher={item['publisher']} | task={item['task_id']} | amount={item['amount']}")
+            logger.info(f"TASK_REFUND | task={item['task_id']} | amount={item['amount']}")
         except Exception as e:
             logger.warning(f"TASK_REFUND_NOTIFY_ERROR | task={item['task_id']} | error={str(e)}")
     
@@ -383,8 +384,8 @@ async def create_task(task: TaskCreate, current_user: str = Depends(require_auth
     
     new_task = {
         "id": task_id,
-        "title": task.title,
-        "description": task.description,
+        "title": escape(task.title),
+        "description": escape(task.description),
         "reference_images": (task.referenceImages or [])[:6],
         "reference_link": task.referenceLink,
         "total_price": task.totalPrice,
@@ -435,7 +436,7 @@ async def create_task(task: TaskCreate, current_user: str = Depends(require_auth
     )
     db_session.commit()
     
-    logger.info(f"TASK_CREATE | publisher={current_user} | task={task_id} | price={task.totalPrice} | frozen={task.totalPrice}")
+    logger.info(f"TASK_CREATE | task={task_id} | price={task.totalPrice} | frozen={task.totalPrice}")
     
     return {"status": "success", "data": new_task, "frozen_amount": task.totalPrice}
 
@@ -543,7 +544,7 @@ async def cancel_task(task_id: str, current_user: str = Depends(require_auth), d
                         )
                         db_session.commit()
                         refund_success = True
-                        logger.info(f"TASK_CANCEL_REFUND | user={current_user} | task={task_id} | amount={frozen_amount}")
+                        logger.info(f"TASK_CANCEL_REFUND | task={task_id} | amount={frozen_amount}")
                 except Exception as e:
                     db_session.rollback()
                     logger.error(f"TASK_CANCEL_REFUND_ERROR | task={task_id} | error={str(e)}")
@@ -708,7 +709,7 @@ async def assign_task(task_id: str, assignee: str, current_user: str = Depends(r
                 "content": f"您已被选为任务《{task.get('title', '')}》的接单者，订金{deposit}积分已缓冲"
             })
             
-            logger.info(f"TASK_ASSIGN | publisher={current_user} | assignee={assignee} | task={task_id} | deposit={deposit}")
+            logger.info(f"TASK_ASSIGN | task={task_id} | deposit={deposit}")
             
             return {"status": "success", "message": f"已指派 {assignee} 接单，订金 {deposit} 积分已开始缓冲"}
     
@@ -843,7 +844,7 @@ async def accept_task(task_id: str, is_accepted: bool, feedback: str = None, cur
                     except Exception as e:
                         logger.warning(f"TASK_COMPLETE_JSON_SAVE_FAILED | 资金已结算但任务状态未更新，需手动修复: {str(e)}")
                     
-                    logger.info(f"TASK_COMPLETE | publisher={current_user} | assignee={assignee_account} | task={task_id} | total={total_price}")
+                    logger.info(f"TASK_COMPLETE | task={task_id} | total={total_price}")
                     # 🗂️ 清除排序缓存（任务状态变化）
                     sort_cache.invalidate("tasks:")
                     message = f"验收通过，已支付 {total_price} 积分给接单者"
@@ -1584,7 +1585,7 @@ async def tip_task(task_id: str, amount: int, is_anon: bool = False, current_use
         db_session.commit()
         
         # 📝 审计日志
-        logger.info(f"TASK_TIP | from={current_user} | to={publisher} | amount={amount} | task={task_id} | anon={is_anon}")
+        logger.info(f"TASK_TIP | task={task_id} | amount={amount} | anon={is_anon}")
         
         # 🔔 打赏通知（考虑匿名）
         if not is_anon:
@@ -1612,7 +1613,7 @@ async def tip_task(task_id: str, amount: int, is_anon: bool = False, current_use
         raise
     except Exception as e:
         db_session.rollback()
-        logger.error(f"TASK_TIP_ERROR | from={current_user} | task={task_id} | amount={amount} | error={str(e)}")
+        logger.error(f"TASK_TIP_ERROR | task={task_id} | amount={amount} | error={str(e)}")
         raise HTTPException(status_code=500, detail="打赏处理失败，请稍后重试")
     
     return result

router_users_auth.py

@@ -11,6 +11,7 @@
 # ==========================================
 
 from fastapi import APIRouter, HTTPException, BackgroundTasks, Request
+import asyncio
 import time
 import re
 import random
@@ -22,6 +23,9 @@ from verify_code_engine import VERIFY_CODES, send_email_code, send_sms_code, cle
 # 🔒 P0安全增强：导入密码哈希和 JWT 工具
 from 安全认证 import hash_password, verify_password, create_token, require_password_match
 
+# 🔒 验证码并发保护锁
+_verify_code_lock = asyncio.Lock()
+
 # 🚀 P2优化：速率限制
 from slowapi import Limiter
 from slowapi.util import get_remote_address
@@ -160,6 +164,11 @@ async def register_user(request: Request, user: UserRegister):
     if user.account in users_db: 
         raise HTTPException(status_code=400, detail="该账号已被注册，请更换一个")
     
+    # 🔒 禁止保留账号名
+    RESERVED_ACCOUNTS = {"admin", "system", "root", "test", "anonymous", "administrator"}
+    if user.account.lower() in RESERVED_ACCOUNTS:
+        raise HTTPException(status_code=400, detail="该账号名不可用")
+    
     # 检查邮箱和手机号是否已被其他用户绑定
     for existing_user in users_db.values():
         if user.email and existing_user.get("email") == user.email: 
@@ -167,11 +176,12 @@ async def register_user(request: Request, user: UserRegister):
         if user.phone and existing_user.get("phone") == user.phone: 
             raise HTTPException(status_code=400, detail="该手机号已被绑定")
             
-    # ========== 第二步：验证码校验（原子性获取+删除） ==========
+    # ========== 第二步：验证码校验（并发安全） ==========
     # 根据注册方式构建缓存键
     cache_key = f"{user.email}_register" if user.email else f"{user.phone}_register"
-    # 🔒 P0安全修复：验证码一次性使用，原子性pop防止并发重用
-    cached = VERIFY_CODES.pop(cache_key, None)
+    # 🔒 P0安全修复：验证码一次性使用，并发安全键防止并发重用
+    async with _verify_code_lock:
+        cached = VERIFY_CODES.pop(cache_key, None)
     
     # 兼容新老缓存格式（expires_at 或 expires）
     expire_time = cached.get("expires_at", cached.get("expires", 0)) if cached else 0
@@ -243,14 +253,15 @@ async def login_user(request: Request, user: UserLogin):
         raise HTTPException(status_code=404, detail="账号不存在")
     
     user_data = users_db[user.account]
-    stored_password = user_data.get("password", "")
+    stored_password = user_data.get("password") or ""
     
     # 🔒 P0安全增强：密码哈希验证（使用统一验证函数）
     if not require_password_match(stored_password, user.password):
         raise HTTPException(status_code=401, detail="密码错误")
     
     # 🔒 P0安全增强：登录成功后，检查是否需要迁移旧密码为bcrypt
-    if not user_data["password"].startswith('$2b$') and not user_data["password"].startswith('$2a$'):
+    # 只在旧格式密码验证成功后才进行迁移（require_password_match已通过）
+    if stored_password and not stored_password.startswith('$2b$') and not stored_password.startswith('$2a$'):
         # 旧版SHA256密码，自动迁移为bcrypt
         user_data["password"] = hash_password(user.password)
         db.save_data("users.json", users_db)

router_users_profile.py

@@ -12,9 +12,10 @@
 #   - 个人设置表单组件.js (更新用户资料)
 # ==========================================
 
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, HTTPException, Depends
 import 数据库连接 as db
 from models import UserUpdate
+from 安全认证 import require_auth
 
 # 创建子路由实例
 router = APIRouter()
@@ -83,7 +84,7 @@ async def get_user_profile(account: str):
 # 前端调用：
 #   - 个人设置表单组件.js 的 handleSaveProfile()
 @router.put("/api/users/{account}")
-async def update_user_profile(account: str, update_data: UserUpdate):
+async def update_user_profile(account: str, update_data: UserUpdate, current_user: str = Depends(require_auth)):
     """
     更新用户资料接口
     
@@ -98,6 +99,10 @@ async def update_user_profile(account: str, update_data: UserUpdate):
         - country: 国家
         - region: 地区
     """
+    # 🔒 越权校验：只能更新自己的资料
+    if account != current_user:
+        raise HTTPException(status_code=403, detail="只能更新自己的资料")
+    
     # 加载用户数据库
     users_db = db.load_data("users.json", default_data={})
     

router_users_social.py

@@ -13,10 +13,11 @@
 #   - 个人设置表单组件.js (隐私设置)
 # ==========================================
 
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, HTTPException, Depends
 import 数据库连接 as db
 from notifications import add_notification
 from models import FollowToggle, PrivacySettings
+from 安全认证 import require_auth
 
 # 创建子路由实例
 router = APIRouter()
@@ -35,7 +36,7 @@ router = APIRouter()
 # 前端调用：
 #   - 个人中心视图.js 的 handleFollowToggle()
 @router.post("/api/users/follow")
-async def toggle_follow(follow: FollowToggle):
+async def toggle_follow(follow: FollowToggle, current_user: str = Depends(require_auth)):
     """
     关注/取消关注接口
     
@@ -44,6 +45,9 @@ async def toggle_follow(follow: FollowToggle):
         - target_account: 被关注/取关的目标用户账号
         - is_active: True=关注, False=取消关注
     """
+    # 🔒 安全防护：强制使用当前认证用户，防止伪造身份
+    follow.user_id = current_user
+    
     # 加载用户数据库
     users_db = db.load_data("users.json", default_data={})
     
@@ -103,7 +107,7 @@ async def toggle_follow(follow: FollowToggle):
 # 前端调用：
 #   - 个人设置表单组件.js 的隐私设置区域
 @router.put("/api/users/{account}/privacy")
-async def update_privacy(account: str, privacy: PrivacySettings):
+async def update_privacy(account: str, privacy: PrivacySettings, current_user: str = Depends(require_auth)):
     """
     更新隐私设置接口
     
@@ -115,6 +119,10 @@ async def update_privacy(account: str, privacy: PrivacySettings):
         - favorites: 是否隐藏收藏记录 (True=隐藏)
         - downloads: 是否隐藏下载记录 (True=隐藏)
     """
+    # 🔒 越权校验：只能修改自己的隐私设置
+    if account != current_user:
+        raise HTTPException(status_code=403, detail="只能修改自己的隐私设置")
+    
     # 加载用户数据库
     users_db = db.load_data("users.json", default_data={})
     

router_wallet.py

@@ -107,7 +107,7 @@ try:
         # 3. 完美加载
         alipay = AliPay(
             appid=raw_appid,
-            app_notify_url="https://zhiwei666-comfyui-ranking-api.hf.space/api/wallet/alipay_notify",
+            app_notify_url=os.environ.get("ALIPAY_NOTIFY_URL", "https://zhiwei666-comfyui-ranking-api.hf.space/api/wallet/alipay_notify"),
             app_private_key_string=priv_key_formatted,
             alipay_public_key_string=pub_key_formatted,
             sign_type="RSA2",
@@ -127,11 +127,11 @@ def calculate_tx_hash(tx_id, account, tx_type, amount, prev_hash):
 async def create_recharge_order(req: RechargeRequest, current_user: str = Depends(require_auth)):
     if not alipay:
         # 这里会将真实的错误原因直接弹窗发给前端！
-        raise HTTPException(status_code=500, detail=f"支付网关配置错误: {alipay_error_msg}")
+        raise HTTPException(status_code=500, detail="支付网关配置错误，请联系管理员")
     
     # 🔒 使用已认证用户账号，忽略客户端传入的 account
     authenticated_account = current_user
-    order_id = f"PAY_{int(time.time())}_{uuid.uuid4().hex[:6]}"
+    order_id = f"PAY_{uuid.uuid4().hex}"
     subject = f"ComfyUI Community Points - {authenticated_account}"
     
     order_string = alipay.api_alipay_trade_precreate(
@@ -174,7 +174,7 @@ async def alipay_notify(request: Request, db: Session = Depends(get_db)):
                     wallet = Wallet(account=account, balance=0, earn_balance=0, tip_balance=0, frozen_balance=0)
                     db.add(wallet)
                 
-                wallet.balance = (wallet.balance or 0) + amount
+                wallet.balance = int(wallet.balance or 0) + int(amount)
                 
                 last_tx = db.query(Transaction).filter(Transaction.account == account).order_by(Transaction.created_at.desc()).first()
                 prev_hash = last_tx.tx_hash if last_tx else "GENESIS_HASH"
@@ -198,7 +198,7 @@ async def alipay_notify(request: Request, db: Session = Depends(get_db)):
     return Response(content="success", media_type="text/plain")
 
 @router.get("/api/wallet/check_order/{order_id}")
-async def check_order(order_id: str, account: str = None, db: Session = Depends(get_db)):
+async def check_order(order_id: str, account: str = None, current_user: str = Depends(require_auth), db: Session = Depends(get_db)):
     # 防线 1：先查本地数据库（如果 Webhook 成功了，这里就能查到）
     tx = db.query(Transaction).filter(Transaction.tx_id == order_id).first()
     if tx:
@@ -224,7 +224,7 @@ async def check_order(order_id: str, account: str = None, db: Session = Depends(
                     wallet = Wallet(account=account, balance=0, earn_balance=0, tip_balance=0, frozen_balance=0)
                     db.add(wallet)
 
-                wallet.balance = (wallet.balance or 0) + amount
+                wallet.balance = int(wallet.balance or 0) + int(amount)
 
                 last_tx = db.query(Transaction).filter(Transaction.account == account).order_by(Transaction.created_at.desc()).first()
                 prev_hash = last_tx.tx_hash if last_tx else "GENESIS_HASH"
@@ -240,6 +240,7 @@ async def check_order(order_id: str, account: str = None, db: Session = Depends(
 
                 return {"status": "SUCCESS"}
         except Exception as e:
+            db.rollback()
             print(f"主动查单发生异常: {e}")
 
     # 如果没查到支付成功状态，继续让前端等
@@ -357,9 +358,9 @@ async def purchase_item(request: Request, req: PurchaseRequest, current_user: st
             seller_wallet = Wallet(account=seller_account, balance=0, earn_balance=0, tip_balance=0, frozen_balance=0)
             db.add(seller_wallet)
             
-        buyer_wallet.balance = (buyer_wallet.balance or 0) - price
-        seller_wallet.balance = (seller_wallet.balance or 0) + price          # 实际收入进统一余额
-        seller_wallet.earn_balance = (seller_wallet.earn_balance or 0) + price     # 累计销售收益统计（只增不减）
+        buyer_wallet.balance = int(buyer_wallet.balance or 0) - int(price)
+        seller_wallet.balance = int(seller_wallet.balance or 0) + int(price)          # 实际收入进统一余额
+        seller_wallet.earn_balance = int(seller_wallet.earn_balance or 0) + int(price)     # 累计销售收益统计（只增不减）
         
         # 🔄 P7后悔模式：记录购买价格
         new_ownership = Ownership(account=req.account, item_id=req.item_id, price_paid=price)
@@ -438,13 +439,14 @@ async def purchase_item(request: Request, req: PurchaseRequest, current_user: st
 async def tip_user(request: Request, req: TipRequest, current_user: str = Depends(require_auth), db: Session = Depends(get_db)):
     # 🔒 使用已认证用户账号，忽略客户端传入的 sender_account
     req.sender_account = current_user
-    if req.amount <= 0:
-        raise HTTPException(status_code=400, detail="打赏金额必须大于0")
+    MAX_TIP_AMOUNT = 100000
+    if req.amount <= 0 or req.amount > MAX_TIP_AMOUNT:
+        raise HTTPException(status_code=400, detail=f"打赏金额必须在1-{MAX_TIP_AMOUNT}之间")
     if req.sender_account == req.target_account:
         raise HTTPException(status_code=400, detail="不能打赏给自己")
     
-    # 🔒 P1幂等性防护：检查最近5秒内是否存在相同交易
-    recent_cutoff = datetime.datetime.utcnow() - datetime.timedelta(seconds=5)
+    # 🔒 P1幂等性防护：检查最近30秒内是否存在相同交易
+    recent_cutoff = datetime.datetime.utcnow() - datetime.timedelta(seconds=30)
     duplicate_tx = db.query(Transaction).filter(
         Transaction.account == req.sender_account,
         Transaction.tx_type == "TIP_OUT",
@@ -466,9 +468,9 @@ async def tip_user(request: Request, req: TipRequest, current_user: str = Depend
             target_wallet = Wallet(account=req.target_account, balance=0, earn_balance=0, tip_balance=0, frozen_balance=0)
             db.add(target_wallet)
             
-        sender_wallet.balance -= req.amount
-        target_wallet.balance += req.amount         # 实际收入进统一余额
-        target_wallet.tip_balance += req.amount     # 累计打赏收益统计（只增不减）
+        sender_wallet.balance = int(sender_wallet.balance or 0) - int(req.amount)
+        target_wallet.balance = int(target_wallet.balance or 0) + int(req.amount)         # 实际收入进统一余额
+        target_wallet.tip_balance = int(target_wallet.tip_balance or 0) + int(req.amount)     # 累计打赏收益统计（只增不减）
         
         tx_id_sender = f"TIP_OUT_{int(time.time())}_{uuid.uuid4().hex[:6]}"
         tx_id_target = f"TIP_IN_{int(time.time())}_{uuid.uuid4().hex[:6]}"
@@ -556,11 +558,19 @@ async def tip_user(request: Request, req: TipRequest, current_user: str = Depend
             u["tip_history"][current_month] = u["tip_history"].get(current_month, 0) + req.amount
             
             if "tip_board" not in u: u["tip_board"] = []
-            sender_entry = next((x for x in u["tip_board"] if x["account"] == req.sender_account), None)
+            if req.is_anonymous:
+                sender_entry = next((x for x in u["tip_board"] if x.get("account") is None and x.get("is_anon")), None)
+            else:
+                sender_entry = next((x for x in u["tip_board"] if x.get("account") == req.sender_account), None)
             if sender_entry: 
                 sender_entry["amount"] += req.amount
-            else: 
-                u["tip_board"].append({"account": req.sender_account, "amount": req.amount, "is_anon": req.is_anonymous})
+            else:
+                tip_entry = {"amount": req.amount, "is_anon": req.is_anonymous}
+                if not req.is_anonymous:
+                    tip_entry["account"] = req.sender_account
+                else:
+                    tip_entry["account"] = None
+                u["tip_board"].append(tip_entry)
             u["tip_board"] = sorted(u["tip_board"], key=lambda x: x["amount"], reverse=True)
             json_db.save_data("users.json", users_db)
             
@@ -572,11 +582,19 @@ async def tip_user(request: Request, req: TipRequest, current_user: str = Depend
                     item["tip_history"][current_month] = item["tip_history"].get(current_month, 0) + req.amount
                     
                     if "tip_board" not in item: item["tip_board"] = []
-                    sender_entry = next((x for x in item["tip_board"] if x["account"] == req.sender_account), None)
+                    if req.is_anonymous:
+                        sender_entry = next((x for x in item["tip_board"] if x.get("account") is None and x.get("is_anon")), None)
+                    else:
+                        sender_entry = next((x for x in item["tip_board"] if x.get("account") == req.sender_account), None)
                     if sender_entry: 
                         sender_entry["amount"] += req.amount
-                    else: 
-                        item["tip_board"].append({"account": req.sender_account, "amount": req.amount, "is_anon": req.is_anonymous})
+                    else:
+                        tip_entry = {"amount": req.amount, "is_anon": req.is_anonymous}
+                        if not req.is_anonymous:
+                            tip_entry["account"] = req.sender_account
+                        else:
+                            tip_entry["account"] = None
+                        item["tip_board"].append(tip_entry)
                     item["tip_board"] = sorted(item["tip_board"], key=lambda x: x["amount"], reverse=True)
                     json_db.save_data("items.json", items_db)
                     break
@@ -611,11 +629,13 @@ async def withdraw(request: Request, req: WithdrawRequest, current_user: str = D
         raise HTTPException(status_code=400, detail="最小提现金额为1积分")
     
     # 🔒 验证码缓存键格式：{contact}_{action_type}，与 send_code 接口一致
-    # 先通过账号查询用户邮箱，再用邮箱构建缓存键
+    # 先通过账号查询用户邮箱或手机号，再用其构建缓存键（不 fallback 到 account）
     users_db = json_db.load_data("users.json", default_data={})
     user_info = users_db.get(req.account, {})
-    user_email = user_info.get("email", "")
-    key = f"{user_email}_withdraw" if user_email else f"{req.account}_withdraw"
+    user_email = user_info.get("email") or user_info.get("phone", "")
+    if not user_email:
+        raise HTTPException(status_code=400, detail="请先绑定邮箱或手机号")
+    key = f"{user_email}_withdraw"
     code_data = VERIFY_CODES.get(key)
     # 🔒 P0安全修复：统一使用 expires_at 字段，兼容旧版 expires
     expire_time = code_data.get("expires_at", code_data.get("expires", 0)) if code_data else 0
@@ -1057,14 +1077,14 @@ async def refund_purchase(request: Request, item_id: str, current_user: str = De
         seller_wallet = db.query(Wallet).filter(Wallet.account == seller_account).with_for_update().first()
         
         if seller_wallet:
-            # 先检查余额是否充足，再扣款
-            if seller_wallet.balance < refund_amount:
-                raise HTTPException(status_code=400, detail="卖家余额不足，无法处理退款")
-            seller_wallet.balance -= refund_amount          # 卖家扣回
+            # 直接扣减，允许余额为负（退款优先于卖家余额安全）
+            seller_wallet.balance = int(seller_wallet.balance or 0) - int(refund_amount)          # 卖家扣回
+            if seller_wallet.balance < 0:
+                logger.warning(f"NEGATIVE_BALANCE | seller balance went negative after refund | seller={seller_account}")
             # 不修改 earn_balance（保持"累计销售收益只增不减"语义）
         
         if buyer_wallet:
-            buyer_wallet.balance += refund_amount           # 买家退款
+            buyer_wallet.balance = int(buyer_wallet.balance or 0) + int(refund_amount)           # 买家退款
         else:
             buyer_wallet = Wallet(account=account, balance=refund_amount)
             db.add(buyer_wallet)

云端_定时版本检测引擎.py

@@ -93,7 +93,6 @@ async def trigger_update_notifications(updated_items: list):
     if not updated_items:
         return
     
-    session = None
     try:
         # 加载已通知记录，防止重复通知
         update_notifications_db = db.load_data("update_notifications.json", default_data={})
@@ -114,63 +113,67 @@ async def trigger_update_notifications(updated_items: list):
         if not items_to_notify:
             print("📢 所有插件更新已通知过，无需重复发送")
             return
-        
+                
         # 从 SQL 数据库查询购买记录
-        session = SessionLocal()
-        
-        # 为每个需要通知的插件，查找已购买用户并发送通知
-        total_notified = 0
-        batch_size = 50  # 分批发送，每批50个用户
-        
-        for item in items_to_notify:
-            item_id = item["id"]
-            title = item.get("title", "未知插件")
-            version_hash = item["version_hash"]
-            
-            # 从 SQL ownerships 表查询已购买该插件的用户（排除已退款记录）
-            ownerships = session.query(Ownership).filter(
-                Ownership.item_id == item_id,
-                Ownership.is_refunded == False
-            ).all()
-            
-            # 使用 set 去重用户账号
-            target_users = list(set(record.account for record in ownerships if record.account))
-            
-            if not target_users:
-                # 无人购买，但仍记录已通知版本
-                update_notifications_db[item_id] = version_hash
-                continue
-            
-            # 准备通知数据
-            notif_data = {
-                "type": "plugin_update",
-                "from_user": "system",
-                "target_item_id": item_id,
-                "target_item_title": title,
-                "content": f"您已安装的插件 [{title}] 有新版本可用"
-            }
-            
-            # 分批发送通知
-            notified_count = 0
-            for i in range(0, len(target_users), batch_size):
-                batch = target_users[i:i + batch_size]
+        with SessionLocal() as session:
+            # 为每个需要通知的插件，查找已购买用户并发送通知
+            total_notified = 0
+            batch_size = 50  # 分批发送，每扐50个用户
+                    
+            for item in items_to_notify:
+                item_id = item["id"]
+                title = item.get("title", "未知插件")
+                version_hash = item["version_hash"]
+                        
+                # 从 SQL ownerships 表查询已购买该插件的用户（排除已退款记录）
+                ownerships = session.query(Ownership).filter(
+                    Ownership.item_id == item_id,
+                    Ownership.is_refunded == False
+                ).all()
+                        
+                # 使用 set 去重用户账号
+                target_users = list(set(record.account for record in ownerships if record.account))
+                        
+                if not target_users:
+                    # 无人购买，但仍记录已通知版本
+                    update_notifications_db[item_id] = version_hash
+                    continue
+                        
+                # 准备通知数据
+                notif_data = {
+                    "type": "plugin_update",
+                    "from_user": "system",
+                    "target_item_id": item_id,
+                    "target_item_title": title,
+                    "content": f"您已安装的插件 [{title}] 有新版本可用"
+                }
                 
-                for account in batch:
-                    try:
-                        add_notification(account, notif_data)
-                        notified_count += 1
-                    except Exception as e:
-                        logger.error(f"发送更新通知失败 (user={account}, item={item_id}): {e}")
-                        # 单个失败不影响其他用户
+                # 分批发送通知
+                notified_count = 0
+                failed_users = []
+                for i in range(0, len(target_users), batch_size):
+                    batch = target_users[i:i + batch_size]
+                    
+                    for account in batch:
+                        try:
+                            add_notification(account, notif_data)
+                            notified_count += 1
+                        except Exception as e:
+                            failed_users.append(account)
+                            logger.error(f"发送更新通知失败 (user={account}, item={item_id}): {e}")
+                            # 单个失败不影响其他用户
+                    
+                    if failed_users:
+                        logger.warning(f"NOTIFY_FAILED | count={len(failed_users)} | users={failed_users[:10]}")
+                    
+                    # 每批之间短暂休眠，避免通知风暴
+                    if i + batch_size < len(target_users):
+                        await asyncio.sleep(0.5)
                 
-                # 每批之间短暂休眠，避免通知风暴
-                if i + batch_size < len(target_users):
-                    await asyncio.sleep(0.5)
-            
-            # 记录该插件已通知的版本
-            update_notifications_db[item_id] = version_hash
-            total_notified += notified_count
-            print(f"📢 插件 [{title}] 已向 {notified_count} 位用户发送更新通知")
+                # 记录该插件已通知的版本
+                update_notifications_db[item_id] = version_hash
+                total_notified += notified_count
+                print(f"📢 插件 [{title}] 已向 {notified_count} 位用户发送更新通知")
         
         # 保存已通知记录
         db.save_data("update_notifications.json", update_notifications_db)
@@ -179,10 +182,6 @@ async def trigger_update_notifications(updated_items: list):
     except Exception as e:
         logger.error(f"触发更新通知时发生错误: {e}")
         # 不抛出异常，让主流程继续
-    finally:
-        # 确保 session 正确关闭
-        if session:
-            session.close()
 
 
 async def precache_github_zip(repo_url: str, token: Optional[str], item_id: str, version_hash: str):
@@ -258,6 +257,23 @@ async def precache_github_zip(repo_url: str, token: Optional[str], item_id: str,
             os.remove(temp_path)
             return False
 
+        # 🔒 ZIP 膨胀比检查（防止 zip bomb）
+        try:
+            with zipfile.ZipFile(temp_path, 'r') as zf:
+                uncompressed_size = sum(f.file_size for f in zf.infolist())
+                zip_size_check = os.path.getsize(temp_path)
+                if uncompressed_size > zip_size_check * 10:
+                    logger.warning(
+                        f"[预缓存] ZIP 膨胀比过高，疑似 zip bomb: {item_id}, "
+                        f"压缩={zip_size_check}字节, 未压缩={uncompressed_size}字节"
+                    )
+                    os.remove(temp_path)
+                    return False
+        except zipfile.BadZipFile:
+            logger.warning(f"[预缓存] ZIP 文件损坏: {item_id}")
+            os.remove(temp_path)
+            return False
+
         zip_size = os.path.getsize(temp_path)
         logger.info(f"[预缓存] ZIP 下载完成: {item_id}, 大小 {zip_size} 字节")
 

安全认证.py

@@ -27,8 +27,12 @@ import bcrypt
 # ==========================================
 # JWT_SECRET: 用于签名 Token，生产环境必须设置环境变量
 # PASSWORD_SALT: 密码哈希加盐，增强安全性
-JWT_SECRET = os.environ.get("JWT_SECRET", "ComfyUI-Ranking-Default-Secret-Key-2024")
-PASSWORD_SALT = os.environ.get("PASSWORD_SALT", "ComfyUI-Ranking-Salt-v1")
+JWT_SECRET = os.environ.get("JWT_SECRET")
+if not JWT_SECRET:
+    raise RuntimeError("必须设置 JWT_SECRET 环境变量")
+PASSWORD_SALT = os.environ.get("PASSWORD_SALT")
+if not PASSWORD_SALT:
+    raise RuntimeError("必须设置 PASSWORD_SALT 环境变量")
 
 # Token 有效期配置（单位：秒）
 TOKEN_EXPIRE_SECONDS = 7 * 24 * 60 * 60  # 默认7天（向后兼容）
@@ -366,11 +370,14 @@ def verify_token_with_fallback(token: str) -> Tuple[bool, Optional[str], str]:
 # 作用：提供所有者、管理员等权限检查功能
 # 用法：减少路由中的重复权限检查代码
 
-import os
 from functools import wraps
 
-# 管理员账号列表
-ADMIN_ACCOUNTS = [a.strip() for a in os.getenv("ADMIN_ACCOUNTS", "admin").split(",")]
+# 管理员账号列表（从环境变量读取，唯一统一定义）
+ADMIN_ACCOUNTS = set(
+    acc.strip() 
+    for acc in os.getenv("ADMIN_ACCOUNTS", "").split(",") 
+    if acc.strip()
+)
 
 
 def is_admin(account: str) -> bool:

数据库连接.py

@@ -191,17 +191,29 @@ if sys.platform == "win32":
     # Windows 文件锁
     import msvcrt
     
+    _WIN_LOCK_MAX_RETRIES = 5   # 最大重试次数
+    _WIN_LOCK_BASE_DELAY = 0.1 # 基础等待秒数
+    
     def _lock_file(file_obj, exclusive=True):
-        """Windows 文件锁：锁定整个文件"""
-        try:
-            # LOCK_EX = 独占锁，LOCK_SH = 共享锁
-            mode = msvcrt.LK_NBLCK if exclusive else msvcrt.LK_NBRLCK
-            file_obj.seek(0)
-            msvcrt.locking(file_obj.fileno(), mode, 1)
-        except IOError:
-            # 无法获取锁时等待重试
-            time.sleep(0.1)
-            msvcrt.locking(file_obj.fileno(), mode, 1)
+        """Windows 文件锁：锁定整个文件，带重试逻辑"""
+        mode = msvcrt.LK_NBLCK if exclusive else msvcrt.LK_NBRLCK
+        file_obj.seek(0)
+        for attempt in range(_WIN_LOCK_MAX_RETRIES):
+            try:
+                msvcrt.locking(file_obj.fileno(), mode, 1)
+                return  # 加锁成功
+            except IOError:
+                if attempt < _WIN_LOCK_MAX_RETRIES - 1:
+                    wait = _WIN_LOCK_BASE_DELAY * (2 ** attempt)  # 指数退避
+                    time.sleep(wait)
+                else:
+                    # 最后一次仍失败，独占锁抛出异常，共享锁降级为无锁读取
+                    if exclusive:
+                        raise IOError(f"Windows 文件锁获取失败（重试{_WIN_LOCK_MAX_RETRIES}次后放弃）")
+                    else:
+                        # 共享锁获取失败时，记录警告但不阻塞读取
+                        logger.warning(f"Windows 共享锁获取失败，降级为无锁读取")
+                        return
     
     def _unlock_file(file_obj):
         """Windows 文件锁：释放锁"""
@@ -404,6 +416,9 @@ def save_data(file_name: str, data: Union[Dict, List]) -> bool:
                 os.remove(local_path)
             os.rename(temp_path, local_path)
             
+            # ========== 🚀 P1优化：更新内存缓存（在锁内，确保写后立即可见） ==========
+            _set_to_cache(file_name, data, local_path)
+            
         except Exception as e:
             # 写入失败，清理临时文件
             if os.path.exists(temp_path):
@@ -411,9 +426,6 @@ def save_data(file_name: str, data: Union[Dict, List]) -> bool:
             print(f"🚨 保存 {file_name} 失败: {e}")
             raise
     
-    # ========== 🚀 P1优化：更新内存缓存 ==========
-    _set_to_cache(file_name, data, local_path)
-    
     # ========== 第五步：标记文件脏，等待批量同步 ==========
     if HF_TOKEN:
         _mark_dirty(file_name)
@@ -706,15 +718,15 @@ def atomic_update(file_name: str, updater, default_data=None):
                 os.remove(local_path)
             os.rename(temp_path, local_path)
             
+            # ========== 更新内存缓存（在锁内，确保写后立即可见） ==========
+            _set_to_cache(file_name, data, local_path)
+            
         except Exception as e:
             if os.path.exists(temp_path):
                 os.remove(temp_path)
             logger.error(f"保存 {file_name} 失败: {e}")
             raise
     
-    # ========== 第四步：更新内存缓存 ==========
-    _set_to_cache(file_name, data, local_path)
-    
     # ========== 第五步：标记文件脏，等待批量同步 ==========
     if HF_TOKEN:
         _mark_dirty(file_name)