This commit is contained in:
薇薇安 2026-01-20 19:03:19 +08:00
parent 8832b83ced
commit 4c20a7a488
2 changed files with 24 additions and 16 deletions

View File

@ -70,3 +70,23 @@ def get_account_id(
aid = int(x_account_id or 1) aid = int(x_account_id or 1)
return require_account_access(aid, user) 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

View File

@ -9,20 +9,14 @@ from typing import Any, Dict, Optional, Tuple
from fastapi import APIRouter, HTTPException, Header, Depends from fastapi import APIRouter, HTTPException, Header, Depends
from pydantic import BaseModel from pydantic import BaseModel
import logging import logging
from api.auth_deps import require_system_admin
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# 路由统一挂在 /api/system 下,前端直接调用 /api/system/... # 路由统一挂在 /api/system 下,前端直接调用 /api/system/...
router = APIRouter(prefix="/api/system") router = APIRouter(prefix="/api/system")
# JWT 管理员鉴权(启用后替代 X-Admin-Token未启用登录时仍可使用 X-Admin-Token 做保护) # 管理员鉴权JWT未启用登录时兼容 X-Admin-Token
from api.auth_deps import get_admin_user # noqa: E402 from api.auth_deps import require_system_admin # noqa: E402
def _auth_enabled() -> bool:
v = (os.getenv("ATS_AUTH_ENABLED") or "true").strip().lower()
return v not in {"0", "false", "no"}
LOG_GROUPS = ("error", "warning", "info") 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") raise HTTPException(status_code=401, detail="Unauthorized")
def require_system_admin( #
x_admin_token: Optional[str] = Header(default=None, alias="X-Admin-Token"), # 注意require_system_admin 已迁移到 api.auth_deps避免导入不一致导致 uvicorn 启动失败
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
def _build_supervisorctl_cmd(args: list[str]) -> list[str]: def _build_supervisorctl_cmd(args: list[str]) -> list[str]: