From 453a6374806c6ff0575015d88ce09ac2f4cd65c0 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Tue, 23 Jun 2026 06:51:01 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E4=BF=AE=E5=A4=8D9=E9=A1=B9?= =?UTF-8?q?=E4=B8=AD=E7=AD=89=E6=9E=B6=E6=9E=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 安全修复: - WebSocket 端点添加认证（cookie/token），防止未授权窃听日志 - SPA serve_spa 添加路径遍历防护（resolve + relative_to 检查） - Token 改用 httpOnly Cookie 存储，移除前端 localStorage token（防 XSS 窃取） - 添加安全响应头中间件（X-Content-Type-Options/X-Frame-Options/Referrer-Policy） - HTTP 请求日志脱敏请求体中的 password/secret/token 等敏感字段 - 权限检查统一使用 user_has_permission（考虑自定义权限，修复 has_permission 忽略 custom_permissions 的缺陷） 性能与稳定性: - cookies.py 列表接口修复 N+1 查询（改为批量查询 Account） - login_service.py run() 结束时关闭 DB Session（防止连接泄漏） - _active_batches/_active_tests 全局字典添加 threading.Lock（防止并发竞态） 配置优化: - CORS 源支持环境变量 CORS_ORIGINS 配置 - Uvicorn reload 支持环境变量 UVICORN_RELOAD 控制（生产环境默认关闭） - Cookie 安全标志支持环境变量 COOKIE_SECURE 配置（HTTPS 部署时启用） - logs.py 权限不足返回 HTTP 403（原来返回 200 + message） --- .env.example | 9 ++ docker-compose.yml | 6 + start_web.sh | 6 +- utils/http_logger.py | 43 ++++++- web/backend/deps.py | 38 ++++++- web/backend/main.py | 36 +++++- web/backend/permissions.py | 13 ++- web/backend/routers/accounts.py | 10 +- web/backend/routers/auth.py | 24 +++- web/backend/routers/cookies.py | 16 ++- web/backend/routers/login.py | 39 ++++--- web/backend/routers/logs.py | 16 +-- web/backend/routers/proxy.py | 23 +++- web/backend/services/login_service.py | 157 +++++++++++++------------- web/frontend/src/App.tsx | 4 +- web/frontend/src/api/index.ts | 11 +- web/frontend/src/pages/LoginPage.tsx | 2 +- web/frontend/src/store/auth.ts | 17 ++- 18 files changed, 317 insertions(+), 153 deletions(-) diff --git a/.env.example b/.env.example index c1616b0..880786f 100644 --- a/.env.example +++ b/.env.example @@ -9,3 +9,12 @@ JWT_SECRET_KEY= # 默认管理员账号（仅首次启动建库时生效） ADMIN_USERNAME=admin ADMIN_PASSWORD=admin123 + +# Cookie 安全标志（生产环境 HTTPS 部署时设为 true） +COOKIE_SECURE=false + +# CORS 允许的源（逗号分隔，不设则默认开发环境） +# CORS_ORIGINS=https://example.com,https://www.example.com + +# Uvicorn reload（开发模式设为 true，生产环境保持 false） +UVICORN_RELOAD=false diff --git a/docker-compose.yml b/docker-compose.yml index 4ecdbbe..99c0067 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -16,6 +16,12 @@ services: # 默认管理员账号密码（仅首次启动建库时生效） - ADMIN_USERNAME=${ADMIN_USERNAME:-admin} - ADMIN_PASSWORD=${ADMIN_PASSWORD:-admin123} + # Cookie 安全标志（HTTPS 部署时设为 true） + - COOKIE_SECURE=${COOKIE_SECURE:-false} + # CORS 允许的源（逗号分隔） + - CORS_ORIGINS=${CORS_ORIGINS:-} + # Uvicorn reload（生产环境保持 false） + - UVICORN_RELOAD=${UVICORN_RELOAD:-false} healthcheck: test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/api/health')"] interval: 30s diff --git a/start_web.sh b/start_web.sh index 3a66f6c..2c2f537 100755 --- a/start_web.sh +++ b/start_web.sh @@ -45,7 +45,11 @@ trap cleanup EXIT INT TERM # 启动后端 echo "启动后端 (FastAPI :$BACKEND_PORT)..." -"$PYTHON" -m uvicorn web.backend.main:app --host 0.0.0.0 --port $BACKEND_PORT --reload & +RELOAD_FLAG="" +if [ "${UVICORN_RELOAD:-true}" = "true" ]; then + RELOAD_FLAG="--reload" +fi +"$PYTHON" -m uvicorn web.backend.main:app --host 0.0.0.0 --port $BACKEND_PORT $RELOAD_FLAG & BACKEND_PID=$! # 等后端就绪 diff --git a/utils/http_logger.py b/utils/http_logger.py index dbc35c5..f39b57a 100644 --- a/utils/http_logger.py +++ b/utils/http_logger.py @@ -51,6 +51,47 @@ def _safe_headers(headers: Any) -> dict: return safe +# 请求体中需要脱敏的字段名（小写匹配，包含即遮罩） +_SENSITIVE_BODY_KEYS = { + 'password', 'pwd', 'passwd', 'secret', 'token', 'apikey', 'api_key', + 'email_password', 'mm', 'authorization', 'credential', +} + + +def _safe_body(body: Any) -> str: + """清理请求体中的敏感字段值。 + + 支持 dict、JSON 字符串、其他类型。 + 敏感字段的值会被替换为 ***，其余内容保留（仍受 _truncate 限制）。 + """ + if body is None: + return "" + + # 尝试解析 JSON 字符串 + parsed = body + if isinstance(body, str): + try: + parsed = json.loads(body) + except (json.JSONDecodeError, ValueError): + # 非 JSON 字符串，直接截断 + return _truncate(body) + + # dict 类型：遮罩敏感字段 + if isinstance(parsed, dict): + safe = {} + for k, v in parsed.items(): + if any(s in k.lower() for s in _SENSITIVE_BODY_KEYS): + safe[k] = '***' + elif isinstance(v, (dict, list)): + safe[k] = _safe_body(v) if isinstance(v, dict) else _truncate(str(v)) + else: + safe[k] = v + return _truncate(json.dumps(safe, ensure_ascii=False)) + + # 其他类型：直接截断 + return _truncate(str(body)) + + def log_http( category: str, method: str, @@ -93,7 +134,7 @@ def log_http( "proxy": proxy, "request": { "headers": _safe_headers(request_headers), - "body": _truncate(request_body), + "body": _safe_body(request_body), }, "response": { "status_code": status_code, diff --git a/web/backend/deps.py b/web/backend/deps.py index 6498b9d..3d3d58c 100644 --- a/web/backend/deps.py +++ b/web/backend/deps.py @@ -1,20 +1,23 @@ """FastAPI 依赖注入""" -from fastapi import Depends, HTTPException, status +from typing import Optional +from fastapi import Depends, HTTPException, Request, status, WebSocket from fastapi.security import OAuth2PasswordBearer from sqlalchemy.orm import Session from jose import JWTError -from .database import get_db +from .database import get_db, SessionLocal from .security import decode_access_token from .models import User from .permissions import get_user_permissions -oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login") +# auto_error=False: 允许 token 为空（后续从 cookie 读取） +oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login", auto_error=False) def get_current_user( - token: str = Depends(oauth2_scheme), + request: Request, + token: Optional[str] = Depends(oauth2_scheme), db: Session = Depends(get_db), ) -> User: credentials_exc = HTTPException( @@ -22,6 +25,11 @@ def get_current_user( detail="无效的认证凭据", headers={"WWW-Authenticate": "Bearer"}, ) + # 优先从 Authorization Bearer header 读取，回退到 httpOnly cookie + if not token: + token = request.cookies.get("access_token") + if not token: + raise credentials_exc try: payload = decode_access_token(token) if payload is None: @@ -48,3 +56,25 @@ def require_permission(permission: str): raise HTTPException(status_code=403, detail=f"无权限: {permission}") return current_user return checker + + +def authenticate_websocket(websocket: WebSocket) -> Optional[User]: + """WebSocket 认证：从 cookie 或 query param token 中验证用户身份。 + + Returns: + User 如果认证成功，None 如果失败。 + """ + token = websocket.cookies.get("access_token") or websocket.query_params.get("token") + if not token: + return None + payload = decode_access_token(token) + if not payload or not payload.get("sub"): + return None + db = SessionLocal() + try: + user = db.query(User).filter(User.id == int(payload["sub"])).first() + if user and user.is_active: + return user + return None + finally: + db.close() diff --git a/web/backend/main.py b/web/backend/main.py index 4122930..513706c 100644 --- a/web/backend/main.py +++ b/web/backend/main.py @@ -8,6 +8,7 @@ from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse +from starlette.middleware.base import BaseHTTPMiddleware from .database import init_db from .routers import auth, users, accounts, login, proxy, cookies, logs @@ -25,15 +26,34 @@ app = FastAPI( lifespan=lifespan, ) -# CORS（开发期允许前端 localhost:5173） +# CORS（支持通过环境变量配置，逗号分隔；默认开发环境） +_cors_env = os.getenv("CORS_ORIGINS", "") +if _cors_env: + _cors_origins = [o.strip() for o in _cors_env.split(",") if o.strip()] +else: + _cors_origins = ["http://localhost:5173", "http://localhost:3000"] + app.add_middleware( CORSMiddleware, - allow_origins=["http://localhost:5173", "http://localhost:3000"], + allow_origins=_cors_origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) + +# 安全响应头中间件 +class SecurityHeadersMiddleware(BaseHTTPMiddleware): + async def dispatch(self, request: Request, call_next): + response = await call_next(request) + response.headers["X-Content-Type-Options"] = "nosniff" + response.headers["X-Frame-Options"] = "DENY" + response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin" + return response + + +app.add_middleware(SecurityHeadersMiddleware) + # 注册路由 app.include_router(auth.router) app.include_router(users.router) @@ -66,8 +86,13 @@ if _INDEX_HTML.exists(): # 排除 API 路径 if full_path.startswith("api"): return {"detail": "Not Found"} - # 尝试返回静态文件 - file_path = _FRONTEND_DIST / full_path + # 尝试返回静态文件（防护路径遍历） + file_path = (_FRONTEND_DIST / full_path).resolve() + try: + file_path.relative_to(_FRONTEND_DIST.resolve()) + except ValueError: + # 路径逃逸出 dist 目录，拒绝访问 + return {"detail": "Not Found"} if file_path.is_file(): return FileResponse(str(file_path)) # SPA fallback 到 index.html @@ -75,7 +100,8 @@ if _INDEX_HTML.exists(): def run(): - uvicorn.run("web.backend.main:app", host="0.0.0.0", port=8000, reload=True) + _reload = os.getenv("UVICORN_RELOAD", "false").lower() == "true" + uvicorn.run("web.backend.main:app", host="0.0.0.0", port=8000, reload=_reload) if __name__ == "__main__": diff --git a/web/backend/permissions.py b/web/backend/permissions.py index 0c44b91..a6181df 100644 --- a/web/backend/permissions.py +++ b/web/backend/permissions.py @@ -75,5 +75,16 @@ def get_user_permissions(user) -> list[str]: def has_permission(role: str, permission: str) -> bool: - """检查角色是否拥有某权限。""" + """检查角色是否拥有某权限（仅检查角色默认权限，不考虑自定义权限）。 + + 注意：对于需要考虑用户自定义权限的场景，请使用 user_has_permission()。 + """ return permission in get_role_permissions(role) + + +def user_has_permission(user, permission: str) -> bool: + """检查用户是否拥有某权限（考虑自定义权限）。 + + 优先使用用户的 custom_permissions，若为 None 则回退到角色默认权限。 + """ + return permission in get_user_permissions(user) diff --git a/web/backend/routers/accounts.py b/web/backend/routers/accounts.py index dff849c..b85ae57 100644 --- a/web/backend/routers/accounts.py +++ b/web/backend/routers/accounts.py @@ -9,7 +9,7 @@ from ..database import get_db from ..models import User, Account, AuditLog, LoginTask from ..schemas import AccountImport, AccountAssign, AccountTag, AccountOut, BatchAssign from ..deps import get_current_user, require_permission -from ..permissions import has_permission +from ..permissions import user_has_permission from sqlalchemy import func from sqlalchemy.orm import joinedload @@ -53,13 +53,13 @@ def list_accounts( query = query.filter(Account.id.in_(_cookie_account_ids_query(db))) # 权限控制：客服只能看分配给自己的 - if not has_permission(current.role, "account:view_all"): - if has_permission(current.role, "account:view_assigned"): + if not user_has_permission(current, "account:view_all"): + if user_has_permission(current, "account:view_assigned"): query = query.filter(Account.assigned_to == current.id) else: raise HTTPException(status_code=403, detail="无权查看账号") - if assigned_only and has_permission(current.role, "account:view_all"): + if assigned_only and user_has_permission(current, "account:view_all"): query = query.filter(Account.assigned_to.isnot(None)) if tag: @@ -76,7 +76,7 @@ def list_accounts( created_at=acc.created_at, ) # 运营+超管可看完整字段 - if has_permission(current.role, "account:view_all"): + if user_has_permission(current, "account:view_all"): item.password = acc.password item.email = acc.email item.email_password = acc.email_password diff --git a/web/backend/routers/auth.py b/web/backend/routers/auth.py index c912cd9..31429c0 100644 --- a/web/backend/routers/auth.py +++ b/web/backend/routers/auth.py @@ -1,21 +1,25 @@ """认证路由""" +import os from datetime import datetime -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Response from sqlalchemy.orm import Session from ..database import get_db from ..models import User, AuditLog -from ..security import verify_password, create_access_token +from ..security import verify_password, create_access_token, ACCESS_TOKEN_EXPIRE_HOURS from ..permissions import get_user_permissions, ROLE_LABELS from ..schemas import LoginRequest, TokenResponse from ..deps import get_current_user router = APIRouter(prefix="/api/auth", tags=["认证"]) +# Cookie 安全配置（生产环境 HTTPS 部署时设置 COOKIE_SECURE=true） +_COOKIE_SECURE = os.getenv("COOKIE_SECURE", "false").lower() == "true" + @router.post("/login", response_model=TokenResponse) -def login(req: LoginRequest, db: Session = Depends(get_db)): +def login(req: LoginRequest, response: Response, db: Session = Depends(get_db)): user = db.query(User).filter(User.username == req.username).first() if not user or not verify_password(req.password, user.password_hash): raise HTTPException(status_code=401, detail="用户名或密码错误") @@ -25,6 +29,17 @@ def login(req: LoginRequest, db: Session = Depends(get_db)): token = create_access_token({"sub": str(user.id), "role": user.role}) perms = get_user_permissions(user) + # 设置 httpOnly cookie（前端无需 JS 读取 token，防 XSS 窃取） + response.set_cookie( + key="access_token", + value=token, + httponly=True, + secure=_COOKIE_SECURE, + samesite="lax", + max_age=ACCESS_TOKEN_EXPIRE_HOURS * 3600, + path="/", + ) + # 审计 db.add(AuditLog(user_id=user.id, username=user.username, action="login", target="auth")) db.commit() @@ -51,7 +66,8 @@ def me(current_user: User = Depends(get_current_user)): @router.post("/logout") -def logout(current_user: User = Depends(get_current_user), db: Session = Depends(get_db)): +def logout(response: Response, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)): + response.delete_cookie(key="access_token", path="/") db.add(AuditLog(user_id=current_user.id, username=current_user.username, action="logout", target="auth")) db.commit() return {"message": "已登出"} diff --git a/web/backend/routers/cookies.py b/web/backend/routers/cookies.py index aa275de..6f92915 100644 --- a/web/backend/routers/cookies.py +++ b/web/backend/routers/cookies.py @@ -10,7 +10,7 @@ import csv from ..database import get_db from ..models import User, LoginTask, Account from ..deps import get_current_user, require_permission -from ..permissions import has_permission +from ..permissions import user_has_permission def _fmt_dt(dt) -> str | None: @@ -33,15 +33,23 @@ def list_cookies( query = db.query(LoginTask).filter(LoginTask.status == "success") # 客服只能看自己账号的 - if not has_permission(current.role, "login:view_all"): + if not user_has_permission(current, "login:view_all"): query = query.join(Account, LoginTask.account_id == Account.id).filter( Account.assigned_to == current.id ) tasks = query.order_by(LoginTask.finished_at.desc()).all() + + # 批量查账号，避免 N+1 查询 + account_ids = [t.account_id for t in tasks] + accounts_map = {} + if account_ids: + accs = db.query(Account).filter(Account.id.in_(account_ids)).all() + accounts_map = {a.id: a for a in accs} + result = [] for t in tasks: - acc = db.query(Account).filter(Account.id == t.account_id).first() + acc = accounts_map.get(t.account_id) item = { "id": t.id, "batch_id": t.batch_id, @@ -52,7 +60,7 @@ def list_cookies( "created_at": _fmt_dt(t.finished_at), } # 只有有 cookie:view 权限才返回 cookie 内容 - if has_permission(current.role, "cookie:view"): + if user_has_permission(current, "cookie:view"): cookie = t.cookie or "" item["cookie"] = cookie item["cookie_preview"] = cookie[:50] + "..." if len(cookie) > 50 else cookie diff --git a/web/backend/routers/login.py b/web/backend/routers/login.py index 703f905..c89186e 100644 --- a/web/backend/routers/login.py +++ b/web/backend/routers/login.py @@ -9,14 +9,15 @@ from sqlalchemy.orm import Session from ..database import get_db, SessionLocal from ..models import User, Account, LoginTask, ProxyConfig as ProxyConfigModel from ..schemas import LoginBatchRequest, LoginTaskOut -from ..deps import get_current_user, require_permission -from ..permissions import has_permission +from ..deps import get_current_user, require_permission, authenticate_websocket +from ..permissions import user_has_permission, get_user_permissions from ..services.login_service import LoginBatchRunner router = APIRouter(prefix="/api/login", tags=["登录任务"]) # 运行中的批次: batch_id -> {log_queue, loop, runner} _active_batches: dict[str, dict] = {} +_active_batches_lock = threading.Lock() @router.post("/batch") @@ -38,7 +39,7 @@ async def create_batch( acc = db.query(Account).filter(Account.id == aid).first() if not acc: continue - if not has_permission(current.role, "login:view_all"): + if not user_has_permission(current, "login:view_all"): if acc.assigned_to != current.id: continue valid_ids.append(aid) @@ -56,7 +57,7 @@ async def create_batch( db=thread_db, account_ids=valid_ids, created_by=current.id, - creator_role=current.role, + creator_permissions=get_user_permissions(current), max_geetest_retries=req.max_geetest_retries, max_proxy_retries=req.max_proxy_retries, proxy_config=proxy, @@ -68,11 +69,12 @@ async def create_batch( batch_id = runner.batch_id # 先注册到全局，再启动线程，确保 WebSocket 连接时能找到 - _active_batches[batch_id] = { - "log_queue": log_queue, - "loop": loop, - "runner": runner, - } + with _active_batches_lock: + _active_batches[batch_id] = { + "log_queue": log_queue, + "loop": loop, + "runner": runner, + } # 启动线程 thread = threading.Thread(target=runner.run, daemon=True) @@ -91,7 +93,7 @@ def list_tasks( query = db.query(LoginTask).join(Account, LoginTask.account_id == Account.id) # 客服只能看自己账号的任务 - if not has_permission(current.role, "login:view_all"): + if not user_has_permission(current, "login:view_all"): query = query.filter(Account.assigned_to == current.id) if batch_id: @@ -153,7 +155,8 @@ def stop_batch( batch_id: str, current: User = Depends(require_permission("login:batch")), ): - batch = _active_batches.get(batch_id) + with _active_batches_lock: + batch = _active_batches.get(batch_id) if batch: batch["runner"].stop() return {"message": "已发送停止信号", "success": True} @@ -162,11 +165,18 @@ def stop_batch( @router.websocket("/ws/login/{batch_id}") async def ws_login_logs(websocket: WebSocket, batch_id: str): - """WebSocket 推送登录实时日志。""" + """WebSocket 推送登录实时日志（需认证）。""" + # 认证：从 cookie 或 token query param 验证用户身份 + user = authenticate_websocket(websocket) + if not user: + await websocket.close(code=1008, reason="未授权") + return + await websocket.accept() # 从已注册的批次中获取 log_queue（由 create_batch 创建） - batch = _active_batches.get(batch_id) + with _active_batches_lock: + batch = _active_batches.get(batch_id) if not batch: await websocket.send_json({"level": "error", "message": "批次不存在或已结束"}) await websocket.close() @@ -188,4 +198,5 @@ async def ws_login_logs(websocket: WebSocket, batch_id: str): except WebSocketDisconnect: pass finally: - _active_batches.pop(batch_id, None) + with _active_batches_lock: + _active_batches.pop(batch_id, None) diff --git a/web/backend/routers/logs.py b/web/backend/routers/logs.py index c4b1251..39dffaf 100644 --- a/web/backend/routers/logs.py +++ b/web/backend/routers/logs.py @@ -1,10 +1,10 @@ """请求日志路由 - 查看 HTTP 请求/响应详情日志""" -from fastapi import APIRouter, Depends, Query +from fastapi import APIRouter, Depends, HTTPException, Query from typing import Optional from ..deps import get_current_user -from ..permissions import has_permission +from ..permissions import user_has_permission from utils.http_logger import read_http_logs, clear_http_logs router = APIRouter(prefix="/api/logs", tags=["日志"]) @@ -20,10 +20,10 @@ def list_http_logs( current=Depends(get_current_user), ): """查看 HTTP 请求/响应详情日志（需要审计日志查看权限）""" - if not has_permission(current.role, "audit:view"): + if not user_has_permission(current, "audit:view"): # 运营也可以查看请求日志（用于排查登录问题） - if not has_permission(current.role, "login:batch"): - return {"items": [], "total": 0, "message": "无权限"} + if not user_has_permission(current, "login:batch"): + raise HTTPException(status_code=403, detail="无权限") items, total = read_http_logs( limit=limit, @@ -38,9 +38,9 @@ def list_http_logs( @router.delete("/http") def clear_http_logs_api(current=Depends(get_current_user)): """清空 HTTP 请求/响应详情日志""" - if not has_permission(current.role, "audit:view"): - if not has_permission(current.role, "login:batch"): - return {"success": False, "message": "无权限"} + if not user_has_permission(current, "audit:view"): + if not user_has_permission(current, "login:batch"): + raise HTTPException(status_code=403, detail="无权限") count = clear_http_logs() return {"success": True, "cleared": count} diff --git a/web/backend/routers/proxy.py b/web/backend/routers/proxy.py index d306629..4cfc233 100644 --- a/web/backend/routers/proxy.py +++ b/web/backend/routers/proxy.py @@ -9,13 +9,14 @@ from sqlalchemy.orm import Session from ..database import get_db from ..models import User, ProxyConfig as ProxyConfigModel, AuditLog from ..schemas import ProxyConfigOut, ProxyConfigUpdate -from ..deps import require_permission +from ..deps import require_permission, authenticate_websocket from core.douyu.proxy import resolve_working_proxy, verify_proxy_url, parse_proxy_response router = APIRouter(prefix="/api/proxy", tags=["代理与白名单"]) # 运行中的测试: test_id -> {log_queue, loop, result} _active_tests: dict[str, dict] = {} +_active_tests_lock = threading.Lock() def _get_or_create(db: Session) -> ProxyConfigModel: @@ -63,10 +64,17 @@ def update_proxy_config( @router.websocket("/ws/test/{test_id}") async def ws_test_logs(websocket: WebSocket, test_id: str): - """WebSocket 推送代理/白名单测试实时日志。""" + """WebSocket 推送代理/白名单测试实时日志（需认证）。""" + # 认证：从 cookie 或 token query param 验证用户身份 + user = authenticate_websocket(websocket) + if not user: + await websocket.close(code=1008, reason="未授权") + return + await websocket.accept() - test = _active_tests.get(test_id) + with _active_tests_lock: + test = _active_tests.get(test_id) if not test: await websocket.send_json({"level": "error", "message": "测试任务不存在"}) await websocket.close() @@ -88,7 +96,8 @@ async def ws_test_logs(websocket: WebSocket, test_id: str): except WebSocketDisconnect: pass finally: - _active_tests.pop(test_id, None) + with _active_tests_lock: + _active_tests.pop(test_id, None) # ---- 异步测试执行 ---- @@ -263,7 +272,8 @@ async def test_proxy( log_queue = asyncio.Queue() loop = asyncio.get_running_loop() - _active_tests[test_id] = {"log_queue": log_queue, "loop": loop} + with _active_tests_lock: + _active_tests[test_id] = {"log_queue": log_queue, "loop": loop} thread = threading.Thread(target=_run_proxy_test, args=(cfg, log_queue, loop), daemon=True) thread.start() @@ -283,7 +293,8 @@ async def test_whitelist( log_queue = asyncio.Queue() loop = asyncio.get_running_loop() - _active_tests[test_id] = {"log_queue": log_queue, "loop": loop} + with _active_tests_lock: + _active_tests[test_id] = {"log_queue": log_queue, "loop": loop} thread = threading.Thread(target=_run_whitelist_test, args=(cfg, log_queue, loop), daemon=True) thread.start() diff --git a/web/backend/services/login_service.py b/web/backend/services/login_service.py index 8047460..637765a 100644 --- a/web/backend/services/login_service.py +++ b/web/backend/services/login_service.py @@ -13,7 +13,6 @@ from core.douyu import DouyuLogin from core.models import Account, ProxyConfig as DouyuProxyConfig from core.douyu.proxy import resolve_working_proxy, get_proxy_manager from ..models import Account as AccountModel, LoginTask, ProxyConfig as ProxyConfigModel -from ..permissions import has_permission class LoginBatchRunner: @@ -24,7 +23,7 @@ class LoginBatchRunner: db: Session, account_ids: list[int], created_by: int, - creator_role: str, + creator_permissions: list[str], max_geetest_retries: int = 5, max_proxy_retries: int = 10, proxy_config: Optional[ProxyConfigModel] = None, @@ -35,7 +34,7 @@ class LoginBatchRunner: self.db = db self.account_ids = account_ids self.created_by = created_by - self.creator_role = creator_role + self.creator_permissions = creator_permissions self.max_geetest_retries = max_geetest_retries self.max_proxy_retries = max_proxy_retries self.proxy_config = proxy_config @@ -175,87 +174,91 @@ class LoginBatchRunner: concurrency = self.concurrency self._push_log("info", f"批量登录任务 {batch_id} 开始，共 {len(self.account_ids)} 个账号，并发数: {concurrency}") - # 创建或复用任务记录（顺序执行，线程安全） - task_infos: list[dict] = [] # {task_id, acc_info} - for aid in self.account_ids: - acc = self.db.query(AccountModel).filter(AccountModel.id == aid).first() - if not acc: - continue - # 权限检查：客服只能跑分配给自己的 - if not has_permission(self.creator_role, "login:view_all"): - if acc.assigned_to != self.created_by: - self._push_log("warning", f"跳过无权账号: {acc.username}") + try: + # 创建或复用任务记录（顺序执行，线程安全） + task_infos: list[dict] = [] # {task_id, acc_info} + for aid in self.account_ids: + acc = self.db.query(AccountModel).filter(AccountModel.id == aid).first() + if not acc: continue + # 权限检查：客服只能跑分配给自己的 + if "login:view_all" not in self.creator_permissions: + if acc.assigned_to != self.created_by: + self._push_log("warning", f"跳过无权账号: {acc.username}") + continue - # 复用该账号最近一条失败任务记录，避免重复产生多条 - existing_task = ( - self.db.query(LoginTask) - .filter(LoginTask.account_id == aid, LoginTask.status.in_(["failed", "error"])) - .order_by(LoginTask.id.desc()) - .first() - ) - if existing_task: - existing_task.batch_id = batch_id - existing_task.status = "pending" - existing_task.cookie = "" - existing_task.message = "" - existing_task.finished_at = None - task = existing_task - else: - task = LoginTask( - batch_id=batch_id, - account_id=aid, - status="pending", - created_by=self.created_by, + # 复用该账号最近一条失败任务记录，避免重复产生多条 + existing_task = ( + self.db.query(LoginTask) + .filter(LoginTask.account_id == aid, LoginTask.status.in_(["failed", "error"])) + .order_by(LoginTask.id.desc()) + .first() ) - self.db.add(task) + if existing_task: + existing_task.batch_id = batch_id + existing_task.status = "pending" + existing_task.cookie = "" + existing_task.message = "" + existing_task.finished_at = None + task = existing_task + else: + task = LoginTask( + batch_id=batch_id, + account_id=aid, + status="pending", + created_by=self.created_by, + ) + self.db.add(task) - self.db.flush() # 获取 task.id + self.db.flush() # 获取 task.id - task_infos.append({ - "task_id": task.id, - "acc_info": { - "username": acc.username, - "password": acc.password, - "email": acc.email, - "email_password": acc.email_password, - "email_imap_server": acc.email_imap_server or "", - "email_imap_port": acc.email_imap_port or 993, - "email_imap_ssl": acc.email_imap_ssl if acc.email_imap_ssl is not None else True, - }, - }) + task_infos.append({ + "task_id": task.id, + "acc_info": { + "username": acc.username, + "password": acc.password, + "email": acc.email, + "email_password": acc.email_password, + "email_imap_server": acc.email_imap_server or "", + "email_imap_port": acc.email_imap_port or 993, + "email_imap_ssl": acc.email_imap_ssl if acc.email_imap_ssl is not None else True, + }, + }) - self.db.commit() - total = len(task_infos) - if total == 0: - self._push_log("warning", "没有可执行的账号") + self.db.commit() + total = len(task_infos) + if total == 0: + self._push_log("warning", "没有可执行的账号") + self._push_log("result", "") + return + + # 并发执行登录，每个账号独立获取代理 + with ThreadPoolExecutor(max_workers=concurrency) as executor: + futures = [] + for item in task_infos: + if self._stop.is_set(): + self._push_log("warning", "任务已停止，跳过剩余账号") + break + future = executor.submit( + self._execute_one, + item["task_id"], + item["acc_info"], + total, + ) + futures.append(future) + + # 等待所有任务完成 + for future in as_completed(futures): + try: + future.result() + except Exception as e: + self._push_log("error", f"Worker 异常: {e}") + + self._push_log("info", f"批量登录任务 {batch_id} 完成") self._push_log("result", "") - return - - # 并发执行登录，每个账号独立获取代理 - with ThreadPoolExecutor(max_workers=concurrency) as executor: - futures = [] - for item in task_infos: - if self._stop.is_set(): - self._push_log("warning", "任务已停止，跳过剩余账号") - break - future = executor.submit( - self._execute_one, - item["task_id"], - item["acc_info"], - total, - ) - futures.append(future) - - # 等待所有任务完成 - for future in as_completed(futures): - try: - future.result() - except Exception as e: - self._push_log("error", f"Worker 异常: {e}") - - self._push_log("info", f"批量登录任务 {batch_id} 完成") - self._push_log("result", "") + finally: + # 确保 DB Session 被关闭，避免连接泄漏 + self.db.close() # 在模块末尾导入 SessionLocal（避免循环导入） diff --git a/web/frontend/src/App.tsx b/web/frontend/src/App.tsx index 8102b2b..1e15c44 100644 --- a/web/frontend/src/App.tsx +++ b/web/frontend/src/App.tsx @@ -12,14 +12,14 @@ import ProxyPage from './pages/ProxyPage'; import UsersPage from './pages/UsersPage'; import CookiePage from './pages/CookiePage'; import HttpLogsPage from './pages/HttpLogsPage'; -import { getToken } from './store/auth'; +import { getUser } from './store/auth'; import { ThemeProvider, useTheme } from './store/theme'; function AppContent() { // 用 state 驱动重渲染，登录/登出时调 refreshAuth() const [authVersion, setAuthVersion] = useState(0); const refreshAuth = useCallback(() => setAuthVersion((v) => v + 1), []); - const isLoggedIn = !!getToken(); + const isLoggedIn = !!getUser(); const { isDark } = useTheme(); return ( diff --git a/web/frontend/src/api/index.ts b/web/frontend/src/api/index.ts index 99b3d7a..076b0c2 100644 --- a/web/frontend/src/api/index.ts +++ b/web/frontend/src/api/index.ts @@ -3,15 +3,7 @@ import axios from 'axios'; const api = axios.create({ baseURL: '/api', timeout: 30000, -}); - -// 请求拦截：携带 token -api.interceptors.request.use((config) => { - const token = localStorage.getItem('token'); - if (token) { - config.headers.Authorization = `Bearer ${token}`; - } - return config; + withCredentials: true, // 携带 httpOnly cookie }); // 响应拦截：统一错误处理 @@ -19,7 +11,6 @@ api.interceptors.response.use( (response) => response.data, (error) => { if (error.response?.status === 401) { - localStorage.removeItem('token'); localStorage.removeItem('user'); window.location.href = '/login'; } diff --git a/web/frontend/src/pages/LoginPage.tsx b/web/frontend/src/pages/LoginPage.tsx index efa7c99..377d25f 100644 --- a/web/frontend/src/pages/LoginPage.tsx +++ b/web/frontend/src/pages/LoginPage.tsx @@ -23,7 +23,7 @@ export default function LoginPage({ onLogin }: { onLogin?: () => void }) { role: result.role, permissions: result.permissions, }; - setAuth(result.access_token, user); + setAuth(user); message.success('登录成功'); onLogin?.(); // 触发 App 重渲染 navigate('/', { replace: true }); diff --git a/web/frontend/src/store/auth.ts b/web/frontend/src/store/auth.ts index 90220b0..a452d9e 100644 --- a/web/frontend/src/store/auth.ts +++ b/web/frontend/src/store/auth.ts @@ -8,12 +8,10 @@ export interface AuthUser { permissions: string[]; } -const TOKEN_KEY = 'token'; const USER_KEY = 'user'; -export function getToken(): string | null { - return localStorage.getItem(TOKEN_KEY); -} +// Token 存储在 httpOnly cookie 中，JS 无法读取，防 XSS 窃取 +// 前端仅用 localStorage 存储用户信息（非敏感），用于判断登录状态和显示 export function getUser(): AuthUser | null { const raw = localStorage.getItem(USER_KEY); @@ -25,28 +23,27 @@ export function getUser(): AuthUser | null { } } -export function setAuth(token: string, user: AuthUser) { - localStorage.setItem(TOKEN_KEY, token); +export function setAuth(user: AuthUser) { localStorage.setItem(USER_KEY, JSON.stringify(user)); } export function clearAuth() { - localStorage.removeItem(TOKEN_KEY); localStorage.removeItem(USER_KEY); } export async function fetchCurrentUser(): Promise { - const token = getToken(); - if (!token) return null; + // 通过 /me 接口验证 cookie 是否有效 try { const data = await authApi.me(); - return { + const user: AuthUser = { id: data.id, username: data.username, role: data.role, role_label: data.role_label, permissions: data.permissions, }; + setAuth(user); + return user; } catch { clearAuth(); return null;