fix: 添加用户登录失败锁定机制及相关测试用例 #604

backend/package/yuxi/storage/postgres/models_business.py

@@ -1,5 +1,6 @@
 """PostgreSQL 业务数据模型 - 用户、部门、对话等相关表"""
 
+from datetime import timedelta
 from typing import Any
 
 from sqlalchemy import (
@@ -21,6 +22,9 @@ from yuxi.utils.datetime_utils import format_utc_datetime, utc_now_naive
 
 Base = declarative_base()
 
+MAX_LOGIN_FAILED_ATTEMPTS = 5
+LOGIN_LOCK_DURATION_SECONDS = 300
+
 
 class Department(Base):
     """部门模型"""
@@ -112,6 +116,13 @@ class User(Base):
         remaining = int((self.login_locked_until - utc_now_naive()).total_seconds())
         return max(0, remaining)
 
+    def increment_failed_login(self):
+        """增加登录失败计数，并在达到阈值后锁定登录"""
+        self.login_failed_count += 1
+        self.last_failed_login = utc_now_naive()
+        if self.login_failed_count >= MAX_LOGIN_FAILED_ATTEMPTS:
+            self.login_locked_until = self.last_failed_login + timedelta(seconds=LOGIN_LOCK_DURATION_SECONDS)
+
     def reset_failed_login(self):
         """重置登录失败相关字段"""
         self.login_failed_count = 0

backend/test/integration/api/test_auth_router.py

@@ -17,6 +17,28 @@ async def test_login_with_invalid_credentials(test_client):
     assert "detail" in response.json()
 
 
+async def test_user_is_locked_after_repeated_failed_logins(test_client, standard_user):
+    user_id = standard_user["user"]["user_id"]
+
+    for attempt in range(1, 5):
+        response = await test_client.post("/api/auth/token", data={"username": user_id, "password": "wrong-password"})
+        assert response.status_code == 401, response.text
+        assert response.json()["detail"] == "用户名或密码错误"
+
+    locked_response = await test_client.post("/api/auth/token", data={"username": user_id, "password": "wrong-password"})
+    assert locked_response.status_code == 423, locked_response.text
+    assert "X-Lock-Remaining" in locked_response.headers
+    assert "账户已被锁定" in locked_response.json()["detail"]
+
+    still_locked_response = await test_client.post(
+        "/api/auth/token",
+        data={"username": user_id, "password": standard_user["password"]},
+    )
+    assert still_locked_response.status_code == 423, still_locked_response.text
+    assert "X-Lock-Remaining" in still_locked_response.headers
+    assert "登录被锁定" in still_locked_response.json()["detail"]
+
+
 async def test_admin_can_login_and_fetch_profile(test_client, admin_headers):
     profile_response = await test_client.get("/api/auth/me", headers=admin_headers)
     assert profile_response.status_code == 200