Add token related interfaces

Author: wu-clan

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

@@ -12,6 +12,7 @@
 from backend.app.admin.api.v1.sys.menu import router as menu_router
 from backend.app.admin.api.v1.sys.notice import router as notice_router
 from backend.app.admin.api.v1.sys.role import router as role_router
+from backend.app.admin.api.v1.sys.token import router as token_router
 from backend.app.admin.api.v1.sys.user import router as user_router
 
 router = APIRouter(prefix='/sys')
@@ -27,3 +28,4 @@
 router.include_router(user_router, prefix='/users', tags=['系统用户'])
 router.include_router(data_rule_router, prefix='/data-rules', tags=['系统数据权限规则'])
 router.include_router(notice_router, prefix='/notices', tags=['系统通知公告'])
+router.include_router(token_router, prefix='/tokens', tags=['系统令牌'])

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

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+import json
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Query, Request
+
+from backend.app.admin.schema.token import GetTokenDetail, KickOutToken
+from backend.common.enums import StatusType
+from backend.common.response.response_schema import ResponseModel, ResponseSchemaModel, response_base
+from backend.common.security.jwt import DependsJwtAuth, jwt_decode, superuser_verify
+from backend.common.security.permission import RequestPermission
+from backend.common.security.rbac import DependsRBAC
+from backend.core.conf import settings
+from backend.database.redis import redis_client
+
+router = APIRouter()
+
+
+@router.get('', summary='获取令牌列表', dependencies=[DependsJwtAuth])
+async def get_tokens(username: Annotated[str | None, Query()] = None) -> ResponseSchemaModel[list[GetTokenDetail]]:
+    token_keys = await redis_client.keys(f'{settings.TOKEN_REDIS_PREFIX}:*')
+    token_online = await redis_client.smembers(settings.TOKEN_ONLINE_REDIS_PREFIX)
+    data = []
+    for key in token_keys:
+        token = await redis_client.get(key)
+        token_payload = jwt_decode(token)
+        session_uuid = token_payload.session_uuid
+        token_detail = GetTokenDetail(
+            session_uuid=session_uuid,
+            username='未知',
+            nickname='未知',
+            ip='未知',
+            os='未知',
+            browser='未知',
+            device='未知',
+            status=StatusType.disable if session_uuid not in token_online else StatusType.enable,
+            last_login_time='未知',
+            expire_time=token_payload.expire_time,
+        )
+        extra_info = await redis_client.get(f'{settings.TOKEN_EXTRA_INFO_REDIS_PREFIX}:{session_uuid}')
+        if extra_info:
+
+            def append_token_detail():
+                data.append(
+                    token_detail.model_copy(
+                        update={
+                            'username': extra_info.get('username'),
+                            'nickname': extra_info.get('nickname'),
+                            'ip': extra_info.get('ip'),
+                            'os': extra_info.get('os'),
+                            'browser': extra_info.get('browser'),
+                            'device': extra_info.get('device'),
+                            'last_login_time': extra_info.get('last_login_time'),
+                        }
+                    )
+                )
+
+            extra_info = json.loads(extra_info)
+            if extra_info.get('login_type') != 'swagger':
+                if username:
+                    if username == extra_info.get('username'):
+                        append_token_detail()
+                else:
+                    append_token_detail()
+        else:
+            data.append(token_detail)
+    return response_base.success(data=data)
+
+
+@router.post(
+    '/kick',
+    summary='踢下线',
+    dependencies=[
+        Depends(RequestPermission('sys:token:kick')),
+        DependsRBAC,
+    ],
+)
+async def kick_out(request: Request, obj: KickOutToken) -> ResponseModel:
+    superuser_verify(request)
+    await redis_client.delete(f'{settings.TOKEN_REDIS_PREFIX}:{obj.user_id}:{obj.session_uuid}')
+    return response_base.success()

backend/app/admin/schema/token.py

@@ -3,6 +3,7 @@
 from datetime import datetime
 
 from backend.app.admin.schema.user import GetUserInfoNoRelationDetail
+from backend.common.enums import StatusType
 from backend.common.schema import SchemaBase
 
 
@@ -14,8 +15,8 @@ class GetSwaggerToken(SchemaBase):
 
 class AccessTokenBase(SchemaBase):
     access_token: str
-    access_token_type: str = 'Bearer'
     access_token_expire_time: datetime
+    session_uuid: str
 
 
 class GetNewToken(AccessTokenBase):
@@ -24,3 +25,21 @@ class GetNewToken(AccessTokenBase):
 
 class GetLoginToken(AccessTokenBase):
     user: GetUserInfoNoRelationDetail
+
+
+class KickOutToken(SchemaBase):
+    user_id: int
+    session_uuid: str
+
+
+class GetTokenDetail(SchemaBase):
+    session_uuid: str
+    username: str
+    nickname: str
+    ip: str
+    os: str
+    browser: str
+    device: str
+    status: StatusType
+    last_login_time: str
+    expire_time: datetime

backend/app/admin/service/auth_service.py

@@ -44,9 +44,13 @@ async def user_verify(db: AsyncSession, username: str, password: str) -> User:
     async def swagger_login(self, *, obj: HTTPBasicCredentials) -> tuple[str, User]:
         async with async_db_session.begin() as db:
             user = await self.user_verify(db, obj.username, obj.password)
-            user_id = user.id
-            a_token = await create_access_token(str(user_id), user.is_multi_login)
             await user_dao.update_login_time(db, obj.username)
+            a_token = await create_access_token(
+                str(user.id),
+                user.is_multi_login,
+                # extra info
+                login_type='swagger',
+            )
             return a_token.access_token, user
 
     async def login(
@@ -61,9 +65,29 @@ async def login(
                     raise errors.AuthorizationError(msg='验证码失效，请重新获取')
                 if captcha_code.lower() != obj.captcha.lower():
                     raise errors.CustomError(error=CustomErrorCode.CAPTCHA_ERROR)
-                user_id = user.id
-                a_token = await create_access_token(str(user_id), user.is_multi_login)
-                r_token = await create_refresh_token(str(user_id), user.is_multi_login)
+                await redis_client.delete(f'{admin_settings.CAPTCHA_LOGIN_REDIS_PREFIX}:{request.state.ip}')
+                await user_dao.update_login_time(db, obj.username)
+                await db.refresh(user)
+                a_token = await create_access_token(
+                    str(user.id),
+                    user.is_multi_login,
+                    # extra info
+                    username=user.username,
+                    nickname=user.nickname,
+                    last_login_time=timezone.t_str(user.last_login_time),
+                    ip=request.state.ip,
+                    os=request.state.os,
+                    browser=request.state.browser,
+                    device=request.state.device,
+                )
+                r_token = await create_refresh_token(str(user.id), user.is_multi_login)
+                response.set_cookie(
+                    key=settings.COOKIE_REFRESH_TOKEN_KEY,
+                    value=r_token.refresh_token,
+                    max_age=settings.COOKIE_REFRESH_TOKEN_EXPIRE_SECONDS,
+                    expires=timezone.f_utc(r_token.refresh_token_expire_time),
+                    httponly=True,
+                )
             except errors.NotFoundError as e:
                 log.error('登陆错误: 用户名不存在')
                 raise errors.NotFoundError(msg=e.msg)
@@ -99,19 +123,10 @@ async def login(
                         msg='登录成功',
                     ),
                 )
-                await redis_client.delete(f'{admin_settings.CAPTCHA_LOGIN_REDIS_PREFIX}:{request.state.ip}')
-                await user_dao.update_login_time(db, obj.username)
-                response.set_cookie(
-                    key=settings.COOKIE_REFRESH_TOKEN_KEY,
-                    value=r_token.refresh_token,
-                    max_age=settings.COOKIE_REFRESH_TOKEN_EXPIRE_SECONDS,
-                    expires=timezone.f_utc(r_token.refresh_token_expire_time),
-                    httponly=True,
-                )
-                await db.refresh(user)
                 data = GetLoginToken(
                     access_token=a_token.access_token,
                     access_token_expire_time=a_token.access_token_expire_time,
+                    session_uuid=a_token.session_uuid,
                     user=user,  # type: ignore
                 )
                 return data
@@ -122,23 +137,31 @@ async def new_token(*, request: Request, response: Response) -> GetNewToken:
         if not refresh_token:
             raise errors.TokenError(msg='Refresh Token 丢失，请重新登录')
         try:
-            user_id = jwt_decode(refresh_token)
+            user_id = jwt_decode(refresh_token).user_id
         except Exception:
             raise errors.TokenError(msg='Refresh Token 无效')
         if request.user.id != user_id:
             raise errors.TokenError(msg='Refresh Token 无效')
         async with async_db_session() as db:
+            token = get_token(request)
             user = await user_dao.get(db, user_id)
             if not user:
                 raise errors.NotFoundError(msg='用户名或密码有误')
             elif not user.status:
                 raise errors.AuthorizationError(msg='用户已被锁定, 请联系统管理员')
-            current_token = get_token(request)
             new_token = await create_new_token(
-                sub=str(user.id),
-                token=current_token,
+                user_id=str(user.id),
+                token=token,
                 refresh_token=refresh_token,
                 multi_login=user.is_multi_login,
+                # extra info
+                username=user.username,
+                nickname=user.nickname,
+                last_login_time=timezone.t_str(user.last_login_time),
+                ip=request.state.ip,
+                os=request.state.os,
+                browser=request.state.browser,
+                device_type=request.state.device,
             )
             response.set_cookie(
                 key=settings.COOKIE_REFRESH_TOKEN_KEY,
@@ -150,25 +173,27 @@ async def new_token(*, request: Request, response: Response) -> GetNewToken:
             data = GetNewToken(
                 access_token=new_token.new_access_token,
                 access_token_expire_time=new_token.new_access_token_expire_time,
+                session_uuid=new_token.session_uuid,
             )
             return data
 
     @staticmethod
     async def logout(*, request: Request, response: Response) -> None:
         token = get_token(request)
+        token_payload = jwt_decode(token)
         refresh_token = request.cookies.get(settings.COOKIE_REFRESH_TOKEN_KEY)
         response.delete_cookie(settings.COOKIE_REFRESH_TOKEN_KEY)
         if request.user.is_multi_login:
-            key = f'{settings.TOKEN_REDIS_PREFIX}:{request.user.id}:{token}'
-            await redis_client.delete(key)
+            await redis_client.delete(f'{settings.TOKEN_REDIS_PREFIX}:{request.user.id}:{token_payload.session_uuid}')
             if refresh_token:
-                key = f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{request.user.id}:{refresh_token}'
-                await redis_client.delete(key)
+                await redis_client.delete(f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{request.user.id}:{refresh_token}')
         else:
-            key_prefix = f'{settings.TOKEN_REDIS_PREFIX}:{request.user.id}:'
-            await redis_client.delete_prefix(key_prefix)
-            key_prefix = f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{request.user.id}:'
-            await redis_client.delete_prefix(key_prefix)
+            key_prefix = [
+                f'{settings.TOKEN_REDIS_PREFIX}:{request.user.id}:',
+                f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{request.user.id}:',
+            ]
+            for prefix in key_prefix:
+                await redis_client.delete_prefix(prefix)
 
 
 auth_service: AuthService = AuthService()

backend/app/admin/service/oauth2_service.py

@@ -31,37 +31,50 @@ async def create_with_login(
     ) -> GetLoginToken | None:
         async with async_db_session.begin() as db:
             # 获取 OAuth2 平台用户信息
-            _id = user.get('id')
-            _username = user.get('username')
+            social_id = user.get('id')
+            social_username = user.get('username')
             if social == UserSocialType.github:
-                _username = user.get('login')
-            _nickname = user.get('name')
-            _email = user.get('email')
-            if social == UserSocialType.linuxdo:
-                _email = f'{_username}@linux.do'
-            if not _email:
+                social_username = user.get('login')
+            social_nickname = user.get('name')
+            social_email = user.get('email')
+            if social == UserSocialType.linuxdo:  # 不提供明文邮箱的平台
+                social_email = f'{social_username}@linux.do'
+            if not social_email:
                 raise AuthorizationError(msg=f'授权失败，{social.value} 账户未绑定邮箱')
             # 创建系统用户
-            sys_user = await user_dao.check_email(db, _email)
+            sys_user = await user_dao.check_email(db, social_email)
             if not sys_user:
-                sys_user = await user_dao.get_by_username(db, _username)
+                sys_user = await user_dao.get_by_username(db, social_username)
                 if sys_user:
-                    _username = f'{_username}#{text_captcha(5)}'
-                sys_user = await user_dao.get_by_nickname(db, _nickname)
+                    username = f'{social_username}#{text_captcha(5)}'
+                sys_user = await user_dao.get_by_nickname(db, social_nickname)
                 if sys_user:
-                    _nickname = f'{_nickname}#{text_captcha(5)}'
-                new_sys_user = RegisterUserParam(username=_username, password=None, nickname=_nickname, email=_email)
+                    nickname = f'{social_nickname}#{text_captcha(5)}'
+                new_sys_user = RegisterUserParam(
+                    username=username, password=None, nickname=nickname, email=social_email
+                )
                 await user_dao.create(db, new_sys_user, social=True)
                 await db.flush()
-                sys_user = await user_dao.check_email(db, _email)
+                sys_user = await user_dao.check_email(db, social_email)
             # 绑定社交用户
             sys_user_id = sys_user.id
             user_social = await user_social_dao.get(db, sys_user_id, social.value)
             if not user_social:
-                new_user_social = CreateUserSocialParam(source=social.value, uid=str(_id), user_id=sys_user_id)
+                new_user_social = CreateUserSocialParam(source=social.value, uid=str(social_id), user_id=sys_user_id)
                 await user_social_dao.create(db, new_user_social)
             # 创建 token
-            access_token = await jwt.create_access_token(str(sys_user_id), sys_user.is_multi_login)
+            access_token = await jwt.create_access_token(
+                str(sys_user_id),
+                sys_user.is_multi_login,
+                # extra info
+                username=sys_user.username,
+                nickname=sys_user.nickname,
+                last_login_time=timezone.t_str(user.last_login_time),
+                ip=request.state.ip,
+                os=request.state.os,
+                browser=request.state.browser,
+                device=request.state.device,
+            )
             refresh_token = await jwt.create_refresh_token(str(sys_user_id), multi_login=sys_user.is_multi_login)
             await user_dao.update_login_time(db, sys_user.username)
             await db.refresh(sys_user)
@@ -87,6 +100,7 @@ async def create_with_login(
                 access_token=access_token.access_token,
                 access_token_expire_time=access_token.access_token_expire_time,
                 user=sys_user,  # type: ignore
+                session_uuid=access_token.session_uuid,
             )
             return data