diff --git a/backend/api/auth_deps.py b/backend/api/auth_deps.py index ce73947..75e96fc 100644 --- a/backend/api/auth_deps.py +++ b/backend/api/auth_deps.py @@ -70,3 +70,23 @@ def get_account_id( aid = int(x_account_id or 1) return require_account_access(aid, user) + +def require_system_admin( + x_admin_token: Optional[str] = Header(default=None, alias="X-Admin-Token"), + user: Dict[str, Any] = Depends(get_admin_user), +) -> Dict[str, Any]: + """ + /api/system/* 管理员保护: + - 启用登录(ATS_AUTH_ENABLED=true):要求 JWT 为 admin + - 未启用登录:兼容旧逻辑,若配置了 SYSTEM_CONTROL_TOKEN,则要求 X-Admin-Token + """ + if _auth_enabled(): + return user + + token = (os.getenv("SYSTEM_CONTROL_TOKEN") or "").strip() + if not token: + return user + if not x_admin_token or x_admin_token != token: + raise HTTPException(status_code=401, detail="Unauthorized") + return user + diff --git a/backend/api/routes/system.py b/backend/api/routes/system.py index fbc5e05..30b2ef0 100644 --- a/backend/api/routes/system.py +++ b/backend/api/routes/system.py @@ -9,20 +9,14 @@ from typing import Any, Dict, Optional, Tuple from fastapi import APIRouter, HTTPException, Header, Depends from pydantic import BaseModel import logging -from api.auth_deps import require_system_admin logger = logging.getLogger(__name__) # 路由统一挂在 /api/system 下,前端直接调用 /api/system/... router = APIRouter(prefix="/api/system") -# JWT 管理员鉴权(启用后替代 X-Admin-Token;未启用登录时仍可使用 X-Admin-Token 做保护) -from api.auth_deps import get_admin_user # noqa: E402 - - -def _auth_enabled() -> bool: - v = (os.getenv("ATS_AUTH_ENABLED") or "true").strip().lower() - return v not in {"0", "false", "no"} +# 管理员鉴权(JWT;未启用登录时兼容 X-Admin-Token) +from api.auth_deps import require_system_admin # noqa: E402 LOG_GROUPS = ("error", "warning", "info") @@ -527,14 +521,8 @@ def _require_admin(token: Optional[str], provided: Optional[str]) -> None: raise HTTPException(status_code=401, detail="Unauthorized") -def require_system_admin( - x_admin_token: Optional[str] = Header(default=None, alias="X-Admin-Token"), - user: Dict[str, Any] = Depends(get_admin_user), -) -> Dict[str, Any]: - # 未启用登录:仍允许使用历史 token 保护 - if not _auth_enabled(): - _require_admin(os.getenv("SYSTEM_CONTROL_TOKEN", "").strip(), x_admin_token) - return user +# +# 注意:require_system_admin 已迁移到 api.auth_deps,避免导入不一致导致 uvicorn 启动失败 def _build_supervisorctl_cmd(args: list[str]) -> list[str]: