Explorar o código

fix(backend): unify auth 401/403 JSON envelope

Co-authored-by: Cursor <cursoragent@cursor.com>
zlwu hai 2 meses
pai
achega
c2e77c6d10
Modificáronse 4 ficheiros con 42 adicións e 8 borrados
  1. 7 6
      backend/app/core/deps.py
  2. 10 0
      backend/app/core/response.py
  3. 11 2
      backend/app/main.py
  4. 14 0
      backend/tests/test_auth.py

+ 7 - 6
backend/app/core/deps.py

@@ -1,11 +1,12 @@
 """FastAPI 依赖:当前用户与菜单权限校验"""
 """FastAPI 依赖:当前用户与菜单权限校验"""
 from typing import Annotated, Callable
 from typing import Annotated, Callable
 
 
-from fastapi import Depends, HTTPException
+from fastapi import Depends
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from sqlalchemy.orm import Session
 from sqlalchemy.orm import Session
 
 
 from app.core.database import get_db
 from app.core.database import get_db
+from app.core.response import AppHTTPError
 from app.core.security import TokenError, safe_decode_token
 from app.core.security import TokenError, safe_decode_token
 from app.crud import menu as menu_crud
 from app.crud import menu as menu_crud
 from app.crud import user as user_crud
 from app.crud import user as user_crud
@@ -19,17 +20,17 @@ def get_current_user(
     creds: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer_scheme)],
     creds: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer_scheme)],
 ) -> SysUser:
 ) -> SysUser:
     if creds is None or not creds.credentials:
     if creds is None or not creds.credentials:
-        raise HTTPException(status_code=401, detail="未登录或令牌无效")
+        raise AppHTTPError(401, "未登录或令牌无效")
     try:
     try:
         payload = safe_decode_token(creds.credentials)
         payload = safe_decode_token(creds.credentials)
     except TokenError:
     except TokenError:
-        raise HTTPException(status_code=401, detail="未登录或令牌无效") from None
+        raise AppHTTPError(401, "未登录或令牌无效") from None
     user_id = payload.get("sub")
     user_id = payload.get("sub")
     if not user_id:
     if not user_id:
-        raise HTTPException(status_code=401, detail="未登录或令牌无效")
+        raise AppHTTPError(401, "未登录或令牌无效")
     user = user_crud.get_user_by_id(db, user_id)
     user = user_crud.get_user_by_id(db, user_id)
     if not user or user.status != "0":
     if not user or user.status != "0":
-        raise HTTPException(status_code=401, detail="未登录或令牌无效")
+        raise AppHTTPError(401, "未登录或令牌无效")
     return user
     return user
 
 
 
 
@@ -42,7 +43,7 @@ def require_menu(*codes: str) -> Callable:
         db: Annotated[Session, Depends(get_db)],
         db: Annotated[Session, Depends(get_db)],
     ) -> SysUser:
     ) -> SysUser:
         if not menu_crud.role_has_any_menu_code(db, user.role_id, needed):
         if not menu_crud.role_has_any_menu_code(db, user.role_id, needed):
-            raise HTTPException(status_code=403, detail="无权限")
+            raise AppHTTPError(403, "无权限")
         return user
         return user
 
 
     return _dep
     return _dep

+ 10 - 0
backend/app/core/response.py

@@ -22,3 +22,13 @@ def error_response(code: int = 400, message: str = "error", data: Any = None) ->
         "data": data,
         "data": data,
         "timestamp": int(datetime.now().timestamp() * 1000)
         "timestamp": int(datetime.now().timestamp() * 1000)
     }
     }
+
+
+class AppHTTPError(Exception):
+    """业务/鉴权错误,由全局 handler 转为统一信封 JSON。"""
+
+    def __init__(self, status_code: int, message: str, data: Any = None):
+        self.status_code = status_code
+        self.message = message
+        self.data = data
+        super().__init__(message)

+ 11 - 2
backend/app/main.py

@@ -1,12 +1,13 @@
 from contextlib import asynccontextmanager
 from contextlib import asynccontextmanager
 
 
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
 
 
 from app.api.auth import router as auth_router
 from app.api.auth import router as auth_router
 from app.core.config import get_settings
 from app.core.config import get_settings
 from app.core.database import SessionLocal, init_db
 from app.core.database import SessionLocal, init_db
-from app.core.response import success_response
+from app.core.response import AppHTTPError, error_response, success_response
 from app.services.seed import seed_if_empty
 from app.services.seed import seed_if_empty
 
 
 settings = get_settings()
 settings = get_settings()
@@ -40,6 +41,14 @@ app.add_middleware(
 app.include_router(auth_router, prefix=settings.API_V1_PREFIX)
 app.include_router(auth_router, prefix=settings.API_V1_PREFIX)
 
 
 
 
+@app.exception_handler(AppHTTPError)
+async def app_http_error_handler(_request: Request, exc: AppHTTPError):
+    return JSONResponse(
+        status_code=exc.status_code,
+        content=error_response(exc.status_code, exc.message, exc.data),
+    )
+
+
 @app.get(f"{settings.API_V1_PREFIX}/health")
 @app.get(f"{settings.API_V1_PREFIX}/health")
 def health_check():
 def health_check():
     return success_response({"status": "ok"})
     return success_response({"status": "ok"})

+ 14 - 0
backend/tests/test_auth.py

@@ -45,12 +45,26 @@ def test_login_wrong_password(client):
     assert r.json()["code"] == 401
     assert r.json()["code"] == 401
 
 
 
 
+def _assert_error_envelope(body: dict, code: int):
+    assert body["code"] == code
+    assert "message" in body
+    assert "timestamp" in body
+    assert "detail" not in body
+
+
 def test_current_user_invalid_token(client):
 def test_current_user_invalid_token(client):
     r = client.get(
     r = client.get(
         "/api/web/auth/current-user",
         "/api/web/auth/current-user",
         headers={"Authorization": "Bearer not-a-valid-token"},
         headers={"Authorization": "Bearer not-a-valid-token"},
     )
     )
     assert r.status_code == 401
     assert r.status_code == 401
+    _assert_error_envelope(r.json(), 401)
+
+
+def test_current_user_missing_token(client):
+    r = client.get("/api/web/auth/current-user")
+    assert r.status_code == 401
+    _assert_error_envelope(r.json(), 401)
 
 
 
 
 def test_logout(client):
 def test_logout(client):