Update user and login security configs

backend/app/admin/api/v1/auth/auth.py

@@ -20,7 +20,7 @@ async def login_swagger(
     db: CurrentSessionTransaction, obj: Annotated[HTTPBasicCredentials, Depends()]
 ) -> GetSwaggerToken:
     token, user = await auth_service.swagger_login(db=db, obj=obj)
-    return GetSwaggerToken(access_token=token, user=user)
+    return GetSwaggerToken(access_token=token, user=user)  # type: ignore
 
 
 @router.post(

backend/app/admin/api/v1/auth/captcha.py

@@ -1,4 +1,4 @@
-from uuid import uuid4
+import uuid
 
 from fast_captcha import img_captcha
 from fastapi import APIRouter, Depends
@@ -8,7 +8,9 @@
 from backend.app.admin.schema.captcha import GetCaptchaDetail
 from backend.common.response.response_schema import ResponseSchemaModel, response_base
 from backend.core.conf import settings
+from backend.database.db import CurrentSession
 from backend.database.redis import redis_client
+from backend.utils.dynamic_config import load_login_config
 
 router = APIRouter()
 
@@ -18,17 +20,19 @@
     summary='获取登录验证码',
     dependencies=[Depends(RateLimiter(times=5, seconds=10))],
 )
-async def get_captcha() -> ResponseSchemaModel[GetCaptchaDetail]:
-    """
-    此接口可能存在性能损耗，尽管是异步接口，但是验证码生成是IO密集型任务，使用线程池尽量减少性能损耗
-    """
-    img_type: str = 'base64'
-    img, code = await run_in_threadpool(img_captcha, img_byte=img_type)
-    uuid = str(uuid4())
+async def get_captcha(db: CurrentSession) -> ResponseSchemaModel[GetCaptchaDetail]:
+    await load_login_config(db)
+    img, code = await run_in_threadpool(img_captcha, img_byte='base64')
+    captcha_uuid = str(uuid.uuid4())
     await redis_client.set(
-        f'{settings.CAPTCHA_LOGIN_REDIS_PREFIX}:{uuid}',
+        f'{settings.LOGIN_CAPTCHA_REDIS_PREFIX}:{captcha_uuid}',
         code,
-        ex=settings.CAPTCHA_LOGIN_EXPIRE_SECONDS,
+        ex=settings.LOGIN_CAPTCHA_EXPIRE_SECONDS,
+    )
+    data = GetCaptchaDetail(
+        is_enabled=settings.LOGIN_CAPTCHA_ENABLED,
+        expire_seconds=settings.LOGIN_CAPTCHA_EXPIRE_SECONDS,
+        uuid=captcha_uuid,
+        image=img,
     )
-    data = GetCaptchaDetail(uuid=uuid, img_type=img_type, image=img)
     return response_base.success(data=data)

backend/app/admin/api/v1/sys/user.py

@@ -102,9 +102,7 @@ async def update_user_permission(
 async def update_user_password(
     db: CurrentSessionTransaction, request: Request, obj: ResetPasswordParam
 ) -> ResponseModel:
-    count = await user_service.update_password(
-        db=db, user_id=request.user.id, hash_password=request.user.password, obj=obj
-    )
+    count = await user_service.update_password(db=db, user_id=request.user.id, obj=obj)
     if count > 0:
         return response_base.success()
     return response_base.fail()

backend/app/admin/crud/crud_user.py

@@ -24,7 +24,7 @@
     AddUserRoleParam,
     UpdateUserParam,
 )
-from backend.common.security.jwt import get_hash_password
+from backend.app.admin.utils.password_security import get_hash_password
 from backend.plugin.oauth2.crud.crud_user_social import user_social_dao
 from backend.utils.serializers import select_join_serialize
 from backend.utils.timezone import timezone
@@ -63,15 +63,47 @@ async def get_by_nickname(self, db: AsyncSession, nickname: str) -> User | None:
         """
         return await self.select_model_by_column(db, nickname=nickname)
 
-    async def update_login_time(self, db: AsyncSession, username: str) -> int:
+    async def check_email(self, db: AsyncSession, email: str) -> User | None:
         """
-        更新用户最后登录时间
+        检查邮箱是否已被绑定
 
         :param db: 数据库会话
+        :param email: 电子邮箱
+        :return:
+        """
+        return await self.select_model_by_column(db, email=email)
+
+    async def get_select(self, dept: int | None, username: str | None, phone: str | None, status: int | None) -> Select:
+        """
+        获取用户列表查询表达式
+
+        :param dept: 部门 ID
         :param username: 用户名
+        :param phone: 电话号码
+        :param status: 用户状态
         :return:
         """
-        return await self.update_model_by_column(db, {'last_login_time': timezone.now()}, username=username)
+        filters = {}
+
+        if dept:
+            filters['dept_id'] = dept
+        if username:
+            filters['username__like'] = f'%{username}%'
+        if phone:
+            filters['phone__like'] = f'%{phone}%'
+        if status is not None:
+            filters['status'] = status
+
+        return await self.select_order(
+            'id',
+            'desc',
+            join_conditions=[
+                JoinConfig(model=Dept, join_on=Dept.id == self.model.dept_id, fill_result=True),
+                JoinConfig(model=user_role, join_on=user_role.c.user_id == self.model.id),
+                JoinConfig(model=Role, join_on=Role.id == user_role.c.role_id, fill_result=True),
+            ],
+            **filters,
+        )
 
     async def add(self, db: AsyncSession, obj: AddUserParam) -> None:
         """
@@ -119,90 +151,85 @@ async def add_by_oauth2(self, db: AsyncSession, obj: AddOAuth2UserParam) -> None
         user_role_stmt = insert(user_role).values(AddUserRoleParam(user_id=new_user.id, role_id=role.id).model_dump())
         await db.execute(user_role_stmt)
 
-    async def update(self, db: AsyncSession, input_user: User, obj: UpdateUserParam) -> int:
+    async def update(self, db: AsyncSession, user_id: int, obj: UpdateUserParam) -> int:
         """
         更新用户信息
 
         :param db: 数据库会话
-        :param input_user: 用户 ID
+        :param user_id: 用户 ID
         :param obj: 更新用户参数
         :return:
         """
         role_ids = obj.roles
         del obj.roles
 
-        count = await self.update_model(db, input_user.id, obj)
+        count = await self.update_model(db, user_id, obj)
 
         role_stmt = select(Role).where(Role.id.in_(role_ids))
         result = await db.execute(role_stmt)
         roles = result.scalars().all()
 
-        user_role_stmt = delete(user_role).where(user_role.c.user_id == input_user.id)
+        user_role_stmt = delete(user_role).where(user_role.c.user_id == user_id)
         await db.execute(user_role_stmt)
 
-        user_role_data = [AddUserRoleParam(user_id=input_user.id, role_id=role.id).model_dump() for role in roles]
+        user_role_data = [AddUserRoleParam(user_id=user_id, role_id=role.id).model_dump() for role in roles]
         user_role_stmt = insert(user_role)
         await db.execute(user_role_stmt, user_role_data)
 
         return count
 
-    async def update_nickname(self, db: AsyncSession, user_id: int, nickname: str) -> int:
+    async def update_login_time(self, db: AsyncSession, username: str) -> int:
         """
-        更新用户昵称
+        更新用户上次登录时间
 
         :param db: 数据库会话
-        :param user_id: 用户 ID
-        :param nickname: 用户昵称
+        :param username: 用户名
         :return:
         """
-        return await self.update_model(db, user_id, {'nickname': nickname})
+        return await self.update_model_by_column(db, {'last_login_time': timezone.now()}, username=username)
 
-    async def update_avatar(self, db: AsyncSession, user_id: int, avatar: str) -> int:
+    async def update_password_changed_time(self, db: AsyncSession, user_id: int) -> int:
         """
-        更新用户头像
+        更新用户上次密码变更时间
 
         :param db: 数据库会话
         :param user_id: 用户 ID
-        :param avatar: 头像地址
         :return:
         """
-        return await self.update_model(db, user_id, {'avatar': avatar})
+        return await self.update_model(db, user_id, {'last_password_changed_time': timezone.now()})
 
-    async def update_email(self, db: AsyncSession, user_id: int, email: str) -> int:
+    async def update_nickname(self, db: AsyncSession, user_id: int, nickname: str) -> int:
         """
-        更新用户邮箱
+        更新用户昵称
 
         :param db: 数据库会话
         :param user_id: 用户 ID
-        :param email: 邮箱
+        :param nickname: 用户昵称
         :return:
         """
-        return await self.update_model(db, user_id, {'email': email})
+        return await self.update_model(db, user_id, {'nickname': nickname})
 
-    async def delete(self, db: AsyncSession, user_id: int) -> int:
+    async def update_avatar(self, db: AsyncSession, user_id: int, avatar: str) -> int:
         """
-        删除用户
+        更新用户头像
 
         :param db: 数据库会话
         :param user_id: 用户 ID
+        :param avatar: 头像地址
         :return:
         """
-        user_role_stmt = delete(user_role).where(user_role.c.user_id == user_id)
-        await db.execute(user_role_stmt)
-
-        await user_social_dao.delete_by_user_id(db, user_id)
-
-        return await self.delete_model(db, user_id)
+        return await self.update_model(db, user_id, {'avatar': avatar})
 
-    async def check_email(self, db: AsyncSession, email: str) -> User | None:
+    async def update_email(self, db: AsyncSession, user_id: int, email: str) -> int:
         """
-        检查邮箱是否已被绑定
+        更新用户邮箱
 
         :param db: 数据库会话
-        :param email: 电子邮箱
+        :param user_id: 用户 ID
+        :param email: 邮箱
         :return:
         """
-        return await self.select_model_by_column(db, email=email)
+        return await self.update_model(db, user_id, {'email': email})
 
     async def reset_password(self, db: AsyncSession, pk: int, password: str) -> int:
         """
@@ -215,39 +242,7 @@ async def reset_password(self, db: AsyncSession, pk: int, password: str) -> int:
         """
         salt = bcrypt.gensalt()
         new_pwd = get_hash_password(password, salt)
-        return await self.update_model(db, pk, {'password': new_pwd, 'salt': salt})
-
-    async def get_select(self, dept: int | None, username: str | None, phone: str | None, status: int | None) -> Select:
-        """
-        获取用户列表查询表达式
-
-        :param dept: 部门 ID
-        :param username: 用户名
-        :param phone: 电话号码
-        :param status: 用户状态
-        :return:
-        """
-        filters = {}
-
-        if dept:
-            filters['dept_id'] = dept
-        if username:
-            filters['username__like'] = f'%{username}%'
-        if phone:
-            filters['phone__like'] = f'%{phone}%'
-        if status is not None:
-            filters['status'] = status
-
-        return await self.select_order(
-            'id',
-            'desc',
-            join_conditions=[
-                JoinConfig(model=Dept, join_on=Dept.id == self.model.dept_id, fill_result=True),
-                JoinConfig(model=user_role, join_on=user_role.c.user_id == self.model.id),
-                JoinConfig(model=Role, join_on=Role.id == user_role.c.role_id, fill_result=True),
-            ],
-            **filters,
-        )
+        return await self.update_model(db, pk, {'password': new_pwd, 'salt': salt}, flush=True)
 
     async def set_super(self, db: AsyncSession, user_id: int, *, is_super: bool) -> int:
         """
@@ -293,6 +288,21 @@ async def set_multi_login(self, db: AsyncSession, user_id: int, *, multi_login:
         """
         return await self.update_model(db, user_id, {'is_multi_login': multi_login})
 
+    async def delete(self, db: AsyncSession, user_id: int) -> int:
+        """
+        删除用户
+
+        :param db: 数据库会话
+        :param user_id: 用户 ID
+        :return:
+        """
+        user_role_stmt = delete(user_role).where(user_role.c.user_id == user_id)
+        await db.execute(user_role_stmt)
+
+        await user_social_dao.delete_by_user_id(db, user_id)
+
+        return await self.delete_model(db, user_id)
+
     async def get_join(
         self,
         db: AsyncSession,

backend/app/admin/crud/crud_user_password_history.py

@@ -0,0 +1,34 @@
+from collections.abc import Sequence
+
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy_crud_plus import CRUDPlus
+
+from backend.app.admin.model.user_password_history import UserPasswordHistory
+from backend.app.admin.schema.user_password_history import CreateUserPasswordHistoryParam
+
+
+class CRUDUserPasswordHistory(CRUDPlus[UserPasswordHistory]):
+    """用户密码历史记录数据库操作类"""
+
+    async def create(self, db: AsyncSession, obj: CreateUserPasswordHistoryParam) -> None:
+        """
+        创建密码历史记录
+
+        :param db: 数据库会话
+        :param obj: 创建密码历史记录参数
+        :return:
+        """
+        await self.create_model(db, obj)
+
+    async def get_by_user_id(self, db: AsyncSession, user_id: int) -> Sequence[UserPasswordHistory]:
+        """
+        获取用户的密码历史记录
+
+        :param db: 数据库会话
+        :param user_id: 用户 ID
+        :return:
+        """
+        return await self.select_models_order(db, 'id', 'desc', self.model.user_id == user_id)
+
+
+user_password_history_dao: CRUDUserPasswordHistory = CRUDUserPasswordHistory(UserPasswordHistory)

backend/app/admin/model/__init__.py

@@ -10,3 +10,4 @@
 from backend.app.admin.model.opera_log import OperaLog as OperaLog
 from backend.app.admin.model.role import Role as Role
 from backend.app.admin.model.user import User as User
+from backend.app.admin.model.user_password_history import UserPasswordHistory as UserPasswordHistory

backend/app/admin/model/user.py

@@ -29,7 +29,10 @@ class User(Base):
     is_multi_login: Mapped[bool] = mapped_column(default=False, comment='是否重复登陆(0否 1是)')
     join_time: Mapped[datetime] = mapped_column(TimeZone, init=False, default_factory=timezone.now, comment='注册时间')
     last_login_time: Mapped[datetime | None] = mapped_column(
-        TimeZone, init=False, onupdate=timezone.now, comment='上次登录'
+        TimeZone, init=False, onupdate=timezone.now, comment='上次登录时间'
+    )
+    last_password_changed_time: Mapped[datetime | None] = mapped_column(
+        TimeZone, init=False, default_factory=timezone.now, comment='上次密码变更时间'
     )
 
     # 逻辑外键

backend/app/admin/model/user_password_history.py

@@ -0,0 +1,24 @@
+from datetime import datetime
+
+import sqlalchemy as sa
+
+from sqlalchemy.orm import Mapped, mapped_column
+
+from backend.common.model import DataClassBase, TimeZone, id_key
+from backend.utils.timezone import timezone
+
+
+class UserPasswordHistory(DataClassBase):
+    """用户密码历史记录表"""
+
+    __tablename__ = 'sys_user_password_history'
+
+    id: Mapped[id_key] = mapped_column(init=False)
+    user_id: Mapped[int] = mapped_column(sa.BigInteger, index=True, comment='用户 ID')
+    password: Mapped[str] = mapped_column(sa.String(256), comment='历史密码')
+    created_time: Mapped[datetime] = mapped_column(
+        TimeZone,
+        init=False,
+        default_factory=timezone.now,
+        comment='创建时间',
+    )

backend/app/admin/schema/captcha.py

@@ -6,6 +6,7 @@
 class GetCaptchaDetail(SchemaBase):
     """验证码详情"""
 
+    is_enabled: bool = Field(description='是否启用')
+    expire_seconds: int = Field(description='过期秒数')
     uuid: str = Field(description='图片唯一标识')
-    img_type: str = Field(description='图片类型')
     image: str = Field(description='图片内容')

backend/app/admin/schema/token.py

@@ -30,6 +30,7 @@ class GetNewToken(AccessTokenBase):
 class GetLoginToken(AccessTokenBase):
     """获取登录令牌"""
 
+    password_expire_days_remaining: int | None = Field(None, description='密码过期剩余天数')
     user: GetUserInfoDetail = Field(description='用户信息')