fix(auth): use Chinese fallbacks and map reason codes to user messages
Stop exposing raw reason codes like admin_not_found in TOTP setup errors; auth routes and JWT guard now use Chinese literals with auth_failure_detail. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -25,7 +25,12 @@ pytest tests/test_http_errors_zh.py tests/test_validation_errors_zh.py -q
|
||||
# 示例:GET /api/servers/999999 → detail「服务器不存在」
|
||||
```
|
||||
|
||||
## 后续
|
||||
## 后续(同批追加)
|
||||
|
||||
- 新增英文字面量时补 `_DETAIL_EXACT` 或改源码直接写中文
|
||||
- `str(exc)` 透传类错误依赖下层异常已是中文
|
||||
- `auth.py` / `auth_jwt.py`:fallback 与字面量改中文;`auth_failure_detail()` 映射 `reason` 码(避免返回 `admin_not_found`)
|
||||
- `auth_service.py`:`setup_totp` / `enable_totp` 失败路径补 `message`
|
||||
|
||||
## 仍待
|
||||
|
||||
- 其它模块 `str(exc)` 透传
|
||||
- 非 auth 路由字面量可逐步改源码或补 `_DETAIL_EXACT`
|
||||
|
||||
+28
-12
@@ -17,6 +17,7 @@ from server.api.auth_jwt import get_current_admin
|
||||
from server.application.services.auth_service import AuthService, JWT_REFRESH_TOKEN_EXPIRE_DAYS
|
||||
from server.domain.models import Admin, AuditLog
|
||||
from server.utils.client_ip import get_client_ip
|
||||
from server.utils.http_errors_zh import auth_failure_detail
|
||||
|
||||
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
||||
|
||||
@@ -150,7 +151,10 @@ async def login(
|
||||
status_code = 202 # Accepted but needs TOTP
|
||||
elif result.get("reason") == "ip_blocked":
|
||||
return Response(status_code=403)
|
||||
raise HTTPException(status_code=status_code, detail=result.get("message", "Login failed"))
|
||||
raise HTTPException(
|
||||
status_code=status_code,
|
||||
detail=auth_failure_detail(result, fallback="登录失败"),
|
||||
)
|
||||
|
||||
# Set refresh token as HttpOnly cookie (not accessible from JS)
|
||||
_set_refresh_cookie(response, result["refresh_token"], request)
|
||||
@@ -183,14 +187,17 @@ async def refresh_token(
|
||||
refresh_token = payload.refresh_token
|
||||
|
||||
if not refresh_token:
|
||||
raise HTTPException(status_code=401, detail="Missing refresh token")
|
||||
raise HTTPException(status_code=401, detail="缺少刷新令牌")
|
||||
|
||||
ip_address = get_client_ip(request)
|
||||
result = await service.refresh_token(refresh_token, ip_address=ip_address)
|
||||
if not result["success"]:
|
||||
# Clear the cookie on failure (invalid/expired token)
|
||||
_clear_refresh_cookie(response, request)
|
||||
raise HTTPException(status_code=401, detail=result.get("message", "Invalid refresh token"))
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail=auth_failure_detail(result, fallback="刷新令牌无效"),
|
||||
)
|
||||
|
||||
# Rotate: set new refresh token cookie
|
||||
_set_refresh_cookie(response, result["refresh_token"], request)
|
||||
@@ -236,11 +243,14 @@ async def setup_totp(
|
||||
"""
|
||||
# Security: only allow setting up TOTP for the authenticated admin
|
||||
if payload.admin_id != admin.id:
|
||||
raise HTTPException(status_code=403, detail="Can only setup TOTP for yourself")
|
||||
raise HTTPException(status_code=403, detail="只能为自己设置 TOTP")
|
||||
|
||||
result = await service.setup_totp(admin.id)
|
||||
if not result["success"]:
|
||||
raise HTTPException(status_code=400, detail=result.get("reason", "Setup failed"))
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=auth_failure_detail(result, fallback="TOTP 设置失败"),
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
@@ -255,11 +265,14 @@ async def enable_totp(
|
||||
A2: JWT authentication required — admin can only enable TOTP for themselves.
|
||||
"""
|
||||
if payload.admin_id != admin.id:
|
||||
raise HTTPException(status_code=403, detail="Can only enable TOTP for yourself")
|
||||
raise HTTPException(status_code=403, detail="只能为自己启用 TOTP")
|
||||
|
||||
result = await service.enable_totp(admin.id, payload.totp_code)
|
||||
if not result["success"]:
|
||||
raise HTTPException(status_code=400, detail=result.get("message", "Enable failed"))
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=auth_failure_detail(result, fallback="TOTP 启用失败"),
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
@@ -271,7 +284,7 @@ async def disable_totp(
|
||||
):
|
||||
"""Disable TOTP — requires current password + TOTP code when enabled."""
|
||||
if payload.admin_id != admin.id:
|
||||
raise HTTPException(status_code=403, detail="Can only disable TOTP for yourself")
|
||||
raise HTTPException(status_code=403, detail="只能为自己禁用 TOTP")
|
||||
|
||||
result = await service.disable_totp(
|
||||
admin.id,
|
||||
@@ -279,7 +292,10 @@ async def disable_totp(
|
||||
totp_code=payload.totp_code,
|
||||
)
|
||||
if not result.get("success"):
|
||||
raise HTTPException(status_code=400, detail=result.get("message", "Disable failed"))
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=auth_failure_detail(result, fallback="TOTP 禁用失败"),
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
@@ -298,7 +314,7 @@ async def change_password(
|
||||
admin_repo = AdminRepositoryImpl(db)
|
||||
current_admin = await admin_repo.get_by_id(admin.id)
|
||||
if not current_admin:
|
||||
raise HTTPException(status_code=404, detail="Admin not found")
|
||||
raise HTTPException(status_code=404, detail="管理员不存在")
|
||||
|
||||
# Verify current password
|
||||
if not bcrypt.checkpw(payload.current_password.encode(), current_admin.password_hash.encode()):
|
||||
@@ -357,9 +373,9 @@ async def issue_webssh_token(
|
||||
server_repo = ServerRepositoryImpl(db)
|
||||
server = await server_repo.get_by_id(payload.server_id)
|
||||
if not server:
|
||||
raise HTTPException(status_code=404, detail="Server not found")
|
||||
raise HTTPException(status_code=404, detail="服务器不存在")
|
||||
if not server.domain:
|
||||
raise HTTPException(status_code=400, detail="Server has no domain configured")
|
||||
raise HTTPException(status_code=400, detail="服务器未配置域名")
|
||||
|
||||
result = await service.create_webssh_token(admin, payload.server_id)
|
||||
return {**result, "server_name": server.name}
|
||||
|
||||
@@ -189,7 +189,7 @@ async def get_current_admin(
|
||||
if not credentials:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Missing Authorization header",
|
||||
detail="缺少 Authorization 请求头",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
|
||||
@@ -197,7 +197,7 @@ async def get_current_admin(
|
||||
if not admin:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Invalid or expired token",
|
||||
detail="令牌无效或已过期",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
|
||||
@@ -213,7 +213,7 @@ async def require_current_admin(
|
||||
if admin is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Missing Authorization header",
|
||||
detail="缺少 Authorization 请求头",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
return admin
|
||||
|
||||
@@ -317,7 +317,7 @@ class AuthService:
|
||||
secret = base64.b32encode(secrets.token_bytes(20)).decode()
|
||||
admin = await self.admin_repo.get_by_id(admin_id)
|
||||
if not admin:
|
||||
return {"success": False, "reason": "admin_not_found"}
|
||||
return {"success": False, "reason": "admin_not_found", "message": "管理员不存在"}
|
||||
|
||||
admin.totp_secret = secret
|
||||
# Don't enable yet — user must verify once first
|
||||
@@ -341,7 +341,7 @@ class AuthService:
|
||||
"""Enable TOTP after user verifies with a code"""
|
||||
admin = await self.admin_repo.get_by_id(admin_id)
|
||||
if not admin or not admin.totp_secret:
|
||||
return {"success": False, "reason": "no_secret"}
|
||||
return {"success": False, "reason": "no_secret", "message": "请先完成 TOTP 设置"}
|
||||
|
||||
if not self._verify_totp(totp_code, admin.totp_secret):
|
||||
return {"success": False, "reason": "invalid_totp", "message": "验证码错误,TOTP未启用"}
|
||||
|
||||
@@ -43,6 +43,21 @@ _DETAIL_EXACT: dict[str, str] = {
|
||||
"Too many script callback requests": "脚本回调请求过于频繁",
|
||||
}
|
||||
|
||||
# AuthService ``reason`` codes → user-facing Chinese (when ``message`` absent)
|
||||
_AUTH_REASON_ZH: dict[str, str] = {
|
||||
"admin_not_found": "管理员不存在",
|
||||
"no_secret": "请先完成 TOTP 设置",
|
||||
"invalid_totp": "TOTP 验证码错误",
|
||||
"invalid_password": "当前密码错误",
|
||||
"totp_required": "需要 TOTP 验证码",
|
||||
"invalid_token": "令牌无效或已过期",
|
||||
"invalid_credentials": "用户名或密码错误",
|
||||
"account_locked": "登录尝试过多,请 15 分钟后重试",
|
||||
"account_disabled": "账户已被禁用",
|
||||
"ip_blocked": "拒绝访问",
|
||||
"service_unavailable": "认证服务暂不可用,请稍后重试",
|
||||
}
|
||||
|
||||
_DETAIL_PATTERNS: list[tuple[re.Pattern[str], str]] = [
|
||||
(
|
||||
re.compile(r"^Command failed \(exit (\d+)\)$"),
|
||||
@@ -73,6 +88,17 @@ def translate_http_detail_str(msg: str) -> str:
|
||||
return msg
|
||||
|
||||
|
||||
def auth_failure_detail(result: dict[str, Any], *, fallback: str = "操作失败") -> str:
|
||||
"""Map AuthService failure dict to Chinese ``HTTPException.detail``."""
|
||||
msg = result.get("message")
|
||||
if isinstance(msg, str) and msg.strip():
|
||||
return translate_http_detail_str(msg)
|
||||
reason = str(result.get("reason") or "")
|
||||
if reason in _AUTH_REASON_ZH:
|
||||
return _AUTH_REASON_ZH[reason]
|
||||
return fallback if _looks_chinese(fallback) else translate_http_detail_str(fallback)
|
||||
|
||||
|
||||
def translate_http_detail(detail: Any) -> Any:
|
||||
"""Translate ``HTTPException.detail`` (str | list | dict) for JSON responses."""
|
||||
if isinstance(detail, str):
|
||||
|
||||
@@ -1,6 +1,10 @@
|
||||
"""Tests for HTTPException detail Chinese translation."""
|
||||
|
||||
from server.utils.http_errors_zh import translate_http_detail, translate_http_detail_str
|
||||
from server.utils.http_errors_zh import (
|
||||
auth_failure_detail,
|
||||
translate_http_detail,
|
||||
translate_http_detail_str,
|
||||
)
|
||||
|
||||
|
||||
def test_translate_server_not_found():
|
||||
@@ -34,3 +38,15 @@ def test_translate_dict_message_key():
|
||||
def test_translate_list_detail():
|
||||
out = translate_http_detail(["Server not found", "已配置"])
|
||||
assert out == ["服务器不存在", "已配置"]
|
||||
|
||||
|
||||
def test_auth_failure_detail_prefers_message():
|
||||
assert auth_failure_detail({"message": "用户名或密码错误"}) == "用户名或密码错误"
|
||||
|
||||
|
||||
def test_auth_failure_detail_maps_reason_code():
|
||||
assert auth_failure_detail({"reason": "admin_not_found"}) == "管理员不存在"
|
||||
|
||||
|
||||
def test_auth_failure_detail_fallback_zh():
|
||||
assert auth_failure_detail({}, fallback="登录失败") == "登录失败"
|
||||
|
||||
Reference in New Issue
Block a user