支持开发环境局域网登录测试

backend/app/api/v1/auth.py

@@ -22,6 +22,11 @@ class LoginRequest(BaseModel):
     ciphertext: str = Field(min_length=1)
 
 
+class DevLoginRequest(BaseModel):
+    email: str = Field(min_length=1)
+    password: str = Field(min_length=1)
+
+
 class LoginKeyResponse(BaseModel):
     key_id: str
     public_key: str
@@ -84,6 +89,29 @@ async def authenticate_encrypted_password(payload: LoginRequest, db: AsyncSessio
     return db_user
 
 
+async def authenticate_plain_password(payload: DevLoginRequest, db: AsyncSession):
+    db_user = await user_crud.get_by_email(db, payload.email)
+    if not db_user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="账号不存在",
+        )
+    if not verify_password(payload.password, db_user.password_hash):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="密码错误",
+        )
+    return db_user
+
+
+def ensure_user_active(db_user) -> None:
+    if db_user.status != UserStatus.ACTIVE:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="账号未审核或不可用",
+        )
+
+
 @router.post("/register", status_code=status.HTTP_201_CREATED)
 async def register(
     payload: UserRegisterRequest,
@@ -112,15 +140,22 @@ async def login_for_access_token(
     payload: LoginRequest, db: AsyncSession = Depends(get_db_session)
 ) -> Token:
     db_user = await authenticate_encrypted_password(payload, db)
-    if db_user.status != UserStatus.ACTIVE:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="账号未审核或不可用",
-        )
+    ensure_user_active(db_user)
 
     return issue_user_token(db_user)
 
 
+@router.post("/dev-login", response_model=Token)
+async def dev_login_for_access_token(
+    payload: DevLoginRequest, db: AsyncSession = Depends(get_db_session)
+) -> Token:
+    if settings.ENV != "development":
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
+    db_user = await authenticate_plain_password(payload, db)
+    ensure_user_active(db_user)
+    return issue_user_token(db_user)
+
+
 @router.get("/me", response_model=UserRead)
 async def read_me(current_user=Depends(get_current_user)) -> UserRead:
     return current_user