feat: Optimize interface authentication logic

apps/chat/serializers/chat_authentication.py

@@ -15,7 +15,7 @@
 from rest_framework import serializers
 
 from application.models import ApplicationAccessToken, Application, ApplicationVersion
-from common.auth.common import FileToken, ChatToken
+from common.auth.common import ChatToken
 from common.auth.constants.operate_constants import Operate
 from common.constants.authentication_type import AuthenticationType
 from common.constants.cache_version import Cache_Version
@@ -48,14 +48,10 @@ def auth(self, request):
             if application_access_token is None or not application_access_token.is_active:
                 raise AppApiException(500, _("Invalid application_id"))
             application_id = str(application_id)
-            return (
-                ChatToken(chat_user_id, _type, str(Operate.ANNOTATION_AUTH), application_id=application_id).to_token(),
-                FileToken(chat_user_id, _type, application_id=application_id).to_token(),
-            )
-        return (
-            ChatToken(chat_user_id, _type, str(Operate.ANNOTATION_AUTH)).to_token(),
-            FileToken(chat_user_id, _type).to_token(),
-        )
+            return ChatToken(
+                chat_user_id, _type, str(Operate.ANNOTATION_AUTH), application_id=application_id
+            ).to_token()
+        return (ChatToken(chat_user_id, _type, str(Operate.ANNOTATION_AUTH)).to_token(),)
 
 
 class AnonymousAuthenticationV2Serializer(serializers.Serializer):

apps/chat/views/v2/chat.py

@@ -43,11 +43,9 @@
 )
 from common.auth import ChatTokenAuth
 from common.auth.authentication import has_permissions
-from common.auth.common import FileToken
 from common.auth.constants.chat_permission_constants import ChatPermissionConstants
 from common.constants.authentication_type import AuthenticationType
 from common.constants.cache_version import Cache_Version
-from common.auth.common import ChatAuthentication
 from common.exception.app_exception import AppAuthenticationFailed, AppApiException
 from common.log.log import _get_ip_address, log
 from common.result import result
@@ -443,7 +441,7 @@ def create_token_and_cache(access_token, user, request):
         token = ChatUserAccessTokenSerializer.create_token_and_cache(access_token, user, request)
         version, get_key = Cache_Version.CHAT_USER_TOKEN.value
         cache.set(get_key(token), user, timeout=60 * 60 * 2, version=version)
-        return token, FileToken(str(user.id), AuthenticationType.CHAT_USER.value).to_token()
+        return token
 
     @classmethod
     def generate(self, request, f_token: str, response: HttpResponse, path: str = "/chat"):
@@ -474,9 +472,9 @@ class LocalLoginView(BaseAuthView):
     def post(self, request: Request, access_token: str = None):
         user = ChatUserAccessTokenSerializer.local_login(request.data, access_token)
         user.source = "LOCAL"
-        token, f_token = self.create_token_and_cache(access_token, user, request)
+        token = self.create_token_and_cache(access_token, user, request)
         response = result.success({"token": token})
-        return self.generate(request, f_token, response, path=f"/chat/{access_token}/")
+        return self.generate(request, token, response, path=f"/chat/{access_token}/")
 
 
 class Logout(APIView):

apps/chat/views/v3/chat.py

@@ -38,7 +38,7 @@
 )
 from common.auth import ChatTokenAuth
 from common.auth.authentication import has_permissions
-from common.auth.common import FileToken, ChatToken
+from common.auth.common import ChatToken
 from common.auth.constants.chat_permission_constants import ChatPermissionConstants
 from common.auth.constants.operate_constants import Operate
 from common.constants.authentication_type import AuthenticationType
@@ -140,7 +140,7 @@ def options(self, request, *args, **kwargs):
     def post(self, request: Request):
         serializer = AnonymousAuthenticationSerializer(data=request.query_params)
         serializer.is_valid(raise_exception=True)
-        token, f_token = serializer.auth(request)
+        token = serializer.auth(request)
         response = result.success(
             token,
             headers={
@@ -156,7 +156,7 @@ def post(self, request: Request):
         cookie_path = f"{CONFIG.get_chat_path()}/{application_id}" if application_id else CONFIG.get_chat_path()
         response.set_cookie(
             key="mk_file_auth",
-            value=f_token,
+            value=token,
             max_age=7 * 24 * 3600,
             path=cookie_path,
             secure=is_https,
@@ -450,7 +450,7 @@ def create_token_and_cache(user, access_token, operate):
         ).to_token()
         version, get_key = Cache_Version.CHAT_USER_TOKEN.value
         cache.set(get_key(token), user, timeout=60 * 60 * 2, version=version)
-        return token, FileToken(str(user.id), AuthenticationType.CHAT_USER.value).to_token()
+        return token
 
     @classmethod
     def generate(self, request, f_token: str, response: HttpResponse, path: str = "/chat"):
@@ -482,9 +482,9 @@ def post(self, request: Request):
         user = ChatUserAccessTokenV3Serializer.local_login(request.data)
         user.source = "LOCAL"
         access_token = request.query_params.get("accessToken")
-        token, f_token = self.create_token_and_cache(user, access_token, Operate.LOCAL)
+        token = self.create_token_and_cache(user, access_token, Operate.LOCAL)
         response = result.success({"token": token})
-        return self.generate(request, f_token, response, path=f"/chat/{access_token + '/' if access_token else ''}")
+        return self.generate(request, token, response, path=f"/chat/{access_token + '/' if access_token else ''}")
 
 
 class Logout(APIView):

apps/common/auth/common.py

@@ -7,138 +7,57 @@
 @desc:
 """
 
-import hashlib
-import json
-import threading
+from django.core import signing
 
-from django.core import signing, cache
-
-from application.models import ChatUserType
 from common.constants.authentication_type import AuthenticationType
-from common.constants.cache_version import Cache_Version
-from common.utils.rsa_util import encrypt, decrypt
-
-authentication_cache = cache.cache
-lock = threading.Lock()
-
-
-def _decrypt(authentication: str):
-    cache_key = hashlib.sha256(authentication.encode()).hexdigest()
-    result = authentication_cache.get(key=cache_key, version=Cache_Version.CHAT.value)
-    if result is None:
-        with lock:
-            result = authentication_cache.get(cache_key, version=Cache_Version.CHAT.value)
-            if result is None:
-                result = decrypt(authentication)
-                authentication_cache.set(cache_key, result, version=Cache_Version.CHAT.value, timeout=60 * 60 * 2)
-
-    return result
-
-
-class ChatAuthentication:
-    def __init__(self, auth_type: str | None, **kwargs):
-        self.auth_type = auth_type
-        for k, v in kwargs.items():
-            self.__setattr__(k, v)
-
-    def to_dict(self):
-        return self.__dict__
-
-    def to_string(self):
-        value = json.dumps(self.to_dict())
-        authentication = encrypt(value)
-        cache_key = hashlib.sha256(authentication.encode()).hexdigest()
-        authentication_cache.set(cache_key, value, version=Cache_Version.CHAT.get_version(), timeout=60 * 60 * 2)
-        return authentication
-
-    @staticmethod
-    def new_instance(authentication: str):
-        auth = json.loads(_decrypt(authentication))
-        return ChatAuthentication(**auth)
+from common.exception.app_exception import AppAuthenticationFailed
 
 
-class FileToken:
-    def __init__(self, user_id, _type, application_id: str = None):
-        self.user_id = user_id
+class SystemToken:
+    def __init__(self, user_id, _type: AuthenticationType, **kwargs):
+        self.id = user_id
         self.type = _type
-        self.application_id = application_id
-
-    def to_dict(self):
-        return (
-            {"user_id": self.user_id, "type": str(self.type), "application_id": self.application_id}
-            if self.application_id
-            else {"user_id": self.user_id, "type": str(self.type)}
-        )
-
-    def to_token(self):
-        return signing.dumps(self.to_dict())
-
-    @staticmethod
-    def new_instance(token):
-        token_dict = signing.loads(token)
-        return FileToken(token_dict.get("user_id"), token_dict.get("type"), token_dict.get("application_id"))
-
-
-class ChatUserToken:
-    def __init__(
-        self,
-        application_id,
-        user_id,
-        access_token,
-        _type,
-        chat_user_type,
-        chat_user_id,
-        authentication: ChatAuthentication,
-    ):
-        self.application_id = application_id
-        self.user_id = user_id
-        self.access_token = access_token
-        self.type = _type
-        self.chat_user_type = chat_user_type
-        self.chat_user_id = chat_user_id
-        self.authentication = authentication
+        self.kwargs = kwargs
 
     def to_dict(self):
-        return {
-            "application_id": str(self.application_id),
-            "user_id": str(self.user_id),
-            "access_token": self.access_token,
-            "type": str(self.type.value),
-            "chat_user_type": str(self.chat_user_type),
-            "chat_user_id": str(self.chat_user_id),
-            "authentication": self.authentication.to_string(),
-        }
+        if self.kwargs:
+            return {"user_id": self.id, "type": str(self.type.value), "kwargs": self.kwargs}
+        return {"id": str(self.id), "type": str(self.type.value)}
 
     def to_token(self):
         return signing.dumps(self.to_dict())
 
-    @staticmethod
-    def new_instance(token_dict):
-        return ChatUserToken(
-            token_dict.get("application_id"),
-            token_dict.get("user_id"),
-            token_dict.get("access_token"),
-            token_dict.get("type"),
-            token_dict.get("chat_user_type"),
-            token_dict.get("chat_user_id"),
-            ChatAuthentication.new_instance(token_dict.get("authentication")),
-        )
-
 
 class ChatToken:
     def __init__(self, user_id, _type: AuthenticationType, login_type: str, **kwargs):
-        self.user_id = user_id
+        self.id = user_id
         self.type = _type
         self.login_type = login_type
         self.kwargs = kwargs
 
     def to_dict(self):
+        if self.kwargs:
+            return {
+                "id": str(self.id),
+                "type": str(self.type.value),
+                "login_type": str(self.login_type),
+                "kwargs": self.kwargs,
+            }
         return {
-            "user_id": str(self.user_id),
+            "id": str(self.id),
             "type": str(self.type.value),
             "login_type": str(self.login_type),
-            "kwargs": self.kwargs,
         }
 
     def to_token(self):
         return signing.dumps(self.to_dict())
+
+
+def parse_token(token):
+    details = signing.loads(token)
+    _type = details.get("type")
+    if _type:
+        if _type == AuthenticationType.SYSTEM_USER.value:
+            return SystemToken(details.get("id"), details.get("type"), **details.get("kwargs", {}))
+        return ChatToken(details.get("id"), details.get("type"), details.get("login_type"), **details.get("kwargs", {}))
+    raise AppAuthenticationFailed(1001, "")

apps/common/auth/constants/chat_permission_constants.py

@@ -1,11 +1,12 @@
 # coding=utf-8
 """
-    @project: MaxKB
-    @Author：虎虎虎
-    @file： chat_permission_constants.py
-    @date：2026/8/6 16:38
-    @desc:
+@project: MaxKB
+@Author：虎虎虎
+@file： chat_permission_constants.py
+@date：2026/8/6 16:38
+@desc:
 """
+
 from enum import Enum
 
 from common.auth.constants.group_constants import Group
@@ -16,34 +17,36 @@
 
 class ChatPermissionConstants(Enum):
     CHAT_USER_ANONYMOUS = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.ANNOTATION_AUTH, 0)
-    CHAT_USER_PASSWORD = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.PASSWORD, 1)
-    CHAT_USER_LOCAL = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.LOCAL, 2)
-    CHAT_USER_CAS = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.CAS, 3)
-    CHAT_USER_DINGTALK = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.DINGTALK, 4)
-    CHAT_USER_WECOM = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.WECOM, 5)
-    CHAT_USER_LARK = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.LARK, 6)
-    CHAT_USER_OIDC = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.OIDC, 7)
-    CHAT_USER_LDAP = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.LDAP, 8)
-    CHAT_USER_OAUTH2 = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.OAUTH2, 9)
+    CHAT_USER_LOCAL = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.LOCAL, 1)
+    CHAT_USER_CAS = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.CAS, 2)
+    CHAT_USER_DINGTALK = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.DINGTALK, 3)
+    CHAT_USER_WECOM = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.WECOM, 4)
+    CHAT_USER_LARK = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.LARK, 5)
+    CHAT_USER_OIDC = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.OIDC, 6)
+    CHAT_USER_LDAP = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.LDAP, 7)
+    CHAT_USER_OAUTH2 = Permission(Group.CHAT_USER, Group.CHAT_USER, Operate.OAUTH2, 8)
 
     def get_permission(self):
-        return self._build_workspace_permission('application_id')
+        return self._build_workspace_permission("application_id")
 
     def _build_workspace_permission(self, resource_id_key=None):
         def permission_factory(_, **kwargs):
-            return Permission(group=self.value.group,
-                              sub_group=self.value.sub_group,
-                              operate=self.value.operate,
-                              bit_index=self.value.bit_index,
-                              workspace_id=kwargs.get('workspace_id'),
-                              resource_id=kwargs.get(resource_id_key) if resource_id_key else None)
+            return Permission(
+                group=self.value.group,
+                sub_group=self.value.sub_group,
+                operate=self.value.operate,
+                bit_index=self.value.bit_index,
+                workspace_id=kwargs.get("workspace_id"),
+                resource_id=kwargs.get(resource_id_key) if resource_id_key else None,
+            )
 
         return permission_factory
 
     @staticmethod
     def get_aggregate_permissions():
         return AggregatePermission(
-            permissions=[_permission.get_permission() for _permission in ChatPermissionConstants])
+            permissions=[_permission.get_permission() for _permission in ChatPermissionConstants]
+        )
 
 
 # 权限字符串与权限对象的Map