From a842af2405d1a46d23c3d28d6bcd0d581e048734 Mon Sep 17 00:00:00 2001 From: Nexus Agent Date: Tue, 9 Jun 2026 08:44:51 +0800 Subject: [PATCH] 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 --- .../2026-06-09-http-api-detail-zh.md | 11 +++-- server/api/auth.py | 40 +++++++++++++------ server/api/auth_jwt.py | 6 +-- server/application/services/auth_service.py | 4 +- server/utils/http_errors_zh.py | 26 ++++++++++++ tests/test_http_errors_zh.py | 18 ++++++++- 6 files changed, 84 insertions(+), 21 deletions(-) diff --git a/docs/changelog/2026-06-09-http-api-detail-zh.md b/docs/changelog/2026-06-09-http-api-detail-zh.md index 0d755da7..39585a69 100644 --- a/docs/changelog/2026-06-09-http-api-detail-zh.md +++ b/docs/changelog/2026-06-09-http-api-detail-zh.md @@ -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` diff --git a/server/api/auth.py b/server/api/auth.py index 72b2c92c..12f70446 100644 --- a/server/api/auth.py +++ b/server/api/auth.py @@ -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} diff --git a/server/api/auth_jwt.py b/server/api/auth_jwt.py index 5dec079e..43506a1c 100644 --- a/server/api/auth_jwt.py +++ b/server/api/auth_jwt.py @@ -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 diff --git a/server/application/services/auth_service.py b/server/application/services/auth_service.py index c9a7e341..0e9666c6 100644 --- a/server/application/services/auth_service.py +++ b/server/application/services/auth_service.py @@ -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未启用"} diff --git a/server/utils/http_errors_zh.py b/server/utils/http_errors_zh.py index bf5e7421..f791814e 100644 --- a/server/utils/http_errors_zh.py +++ b/server/utils/http_errors_zh.py @@ -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): diff --git a/tests/test_http_errors_zh.py b/tests/test_http_errors_zh.py index a4a1cc58..f65d6431 100644 --- a/tests/test_http_errors_zh.py +++ b/tests/test_http_errors_zh.py @@ -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="登录失败") == "登录失败"