Skip to content

feat: add application anonymous auth #6671

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hexiaonan-800 merged 1 commit into from
Aug 13, 2026
+350 −242
Merged

feat: add application anonymous auth #6671

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 32 additions & 22 deletions apps/chat/api/chat_authentication_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,10 @@
# coding=utf-8
"""
@project: MaxKB
@Author：虎虎
@file： chat_authentication_api.py
@date：2025/6/6 19:59
@desc:
@project: MaxKB
@Author：虎虎
@file： chat_authentication_api.py
@date：2025/6/6 19:59
@desc:
"""

from django.utils.translation import gettext_lazy as _
Expand All @@ -29,37 +29,47 @@ def get_request():

@staticmethod
def get_parameters():
pass
return [
OpenApiParameter(
name="application_id",
description=_("Application ID"),
type=OpenApiTypes.UUID,
location="query",
required=False,
)
]

@staticmethod
def get_response():
pass


class ChatAuthenticationProfileAPIV2(APIMixin):

@staticmethod
def get_parameters():
return [OpenApiParameter(
name="access_token",
description=_("access_token"),
type=OpenApiTypes.STR,
location='query',
required=True,
)]
return [
OpenApiParameter(
name="access_token",
description=_("access_token"),
type=OpenApiTypes.STR,
location="query",
required=True,
)
]


class ChatAuthenticationProfileAPI(APIMixin):

@staticmethod
def get_parameters():
return [OpenApiParameter(
name="application_id",
description=_("Application ID"),
type=OpenApiTypes.UUID,
location='query',
required=True,
)]
return [
OpenApiParameter(
name="application_id",
description=_("Application ID"),
type=OpenApiTypes.UUID,
location="query",
required=True,
)
]


class ChatOpenAPI(APIMixin):
Expand Down
216 changes: 127 additions & 89 deletions apps/chat/serializers/chat_authentication.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
# coding=utf-8
"""
@project: MaxKB
@Author：虎虎
@file： ChatAuthentication.py
@date：2025/6/6 13:48
@desc:
@project: MaxKB
@Author：虎虎
@file： ChatAuthentication.py
@date：2025/6/6 13:48
@desc:
"""

import uuid_utils.compat as uuid
from django.core import signing
from django.core.cache import cache
Expand All @@ -20,34 +21,52 @@
from common.constants.authentication_type import AuthenticationType
from common.constants.cache_version import Cache_Version
from common.database_model_manage.database_model_manage import DatabaseModelManage
from common.exception.app_exception import NotFound404, AppUnauthorizedFailed
from common.exception.app_exception import NotFound404, AppUnauthorizedFailed, AppApiException
from common.utils.rsa_util import get_key_pair_by_sql


class AnonymousAuthenticationSerializer(serializers.Serializer):
"""v3 匿名认证：application_id 为可选 query 参数。
传入时颁发应用级令牌，未传入时颁发全局令牌。"""

application_id = serializers.UUIDField(required=False, label=_("application_id"))

def auth(self, request):
token = request.META.get('HTTP_AUTHORIZATION')
token = request.META.get("HTTP_AUTHORIZATION")
token_details = {}
try:
# 校验token
if token is not None:
token_details = signing.loads(token[7:])
except Exception as e:
pass
chat_user_id = token_details.get('id') or str(uuid.uuid7())
chat_user_id = token_details.get("id") or str(uuid.uuid7())
_type = AuthenticationType.CHAT_USER
return ChatToken(chat_user_id, _type,
str(Operate.ANNOTATION_AUTH)).to_token(), FileToken(chat_user_id,
_type).to_token()

application_id = self.validated_data.get("application_id")
if application_id:
application_access_token = QuerySet(ApplicationAccessToken).filter(application_id=application_id).first()
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(),
)


class AnonymousAuthenticationV2Serializer(serializers.Serializer):
"""v2 匿名认证：application_id 不在 path，从 access_token 解出并写进 token，
供 ChatUserToken handler 收窄到该应用。"""

access_token = serializers.CharField(required=True, label=_("access_token"))

def auth(self, request, with_valid=True):
token = request.META.get('HTTP_AUTHORIZATION')
token = request.META.get("HTTP_AUTHORIZATION")
token_details = {}
try:
# 校验token
Expand All @@ -61,16 +80,17 @@ def auth(self, request, with_valid=True):
application_access_token = QuerySet(ApplicationAccessToken).filter(access_token=access_token).first()
if application_access_token is None or not application_access_token.is_active:
raise NotFound404(404, _("Invalid access_token"))
chat_user_id = token_details.get('user_id') or token_details.get('id') or str(uuid.uuid7())
chat_user_id = token_details.get("user_id") or token_details.get("id") or str(uuid.uuid7())
_type = AuthenticationType.CHAT_USER
application_id = str(application_access_token.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), application_id=application_id
).to_token(), FileToken(chat_user_id, _type, application_id=application_id).to_token()


class AuthProfileSerializer(serializers.Serializer):
"""v3: 直接通过 application_id 获取认证 profile"""

application_id = serializers.UUIDField(required=True, label=_("application_id"))

def profile(self):
Expand All @@ -81,25 +101,26 @@ def profile(self):
raise NotFound404(404, _("Invalid application_id"))
if not application_access_token.is_active:
raise NotFound404(404, _("Invalid application_id"))
login_value = application_access_token.authentication_value.get('login_value', [])
chat_platform = DatabaseModelManage.get_model('chat_platform')
login_value = application_access_token.authentication_value.get("login_value", [])
chat_platform = DatabaseModelManage.get_model("chat_platform")
if chat_platform is not None:
types = QuerySet(chat_platform).filter(is_active=True, is_valid=True).values_list('auth_type', flat=True)
types = QuerySet(chat_platform).filter(is_active=True, is_valid=True).values_list("auth_type", flat=True)
login_value = list(set(login_value) & set(types))
if 'LOCAL' in application_access_token.authentication_value.get('login_value', []):
login_value.insert(0, 'LOCAL')
if "LOCAL" in application_access_token.authentication_value.get("login_value", []):
login_value.insert(0, "LOCAL")
return {
'application_name': application_access_token.application.name,
'authentication': application_access_token.authentication,
'authentication_type': application_access_token.authentication_value.get('type', 'password'),
'max_attempts': application_access_token.authentication_value.get('max_attempts', 1),
'login_value': login_value,
'rsaKey': get_key_pair_by_sql().get('key')
"application_name": application_access_token.application.name,
"authentication": application_access_token.authentication,
"authentication_type": application_access_token.authentication_value.get("type", "password"),
"max_attempts": application_access_token.authentication_value.get("max_attempts", 1),
"login_value": login_value,
"rsaKey": get_key_pair_by_sql().get("key"),
}


class AuthProfileV2Serializer(serializers.Serializer):
"""v2: 通过 access_token 查表得到 application_id，委托给 AuthProfileSerializer"""

access_token = serializers.CharField(required=True, label=_("access_token"))

def profile(self):
Expand All @@ -110,9 +131,7 @@ def profile(self):
raise NotFound404(404, _("Invalid access_token"))
if not application_access_token.is_active:
raise NotFound404(404, _("Invalid access_token"))
return AuthProfileSerializer(
data={'application_id': application_access_token.application_id}
).profile()
return AuthProfileSerializer(data={"application_id": application_access_token.application_id}).profile()


class ApplicationProfileSerializer(serializers.Serializer):
Expand All @@ -121,18 +140,30 @@ class ApplicationProfileSerializer(serializers.Serializer):
@staticmethod
def reset_application(application, application_version):
update_field_dict = {
'application_name': 'name', 'desc': 'desc', 'prologue': 'prologue', 'dialogue_number': 'dialogue_number',
'user_id': 'user_id', 'model_id': 'model_id', 'knowledge_setting': 'knowledge_setting',
'model_setting': 'model_setting', 'model_params_setting': 'model_params_setting',
'tts_model_params_setting': 'tts_model_params_setting',
'problem_optimization': 'problem_optimization', 'work_flow': 'work_flow',
'problem_optimization_prompt': 'problem_optimization_prompt', 'tts_model_id': 'tts_model_id',
'stt_model_id': 'stt_model_id', 'tts_model_enable': 'tts_model_enable',
'stt_model_enable': 'stt_model_enable', 'tts_type': 'tts_type',
'tts_autoplay': 'tts_autoplay', 'stt_autosend': 'stt_autosend', 'file_upload_enable': 'file_upload_enable',
'file_upload_setting': 'file_upload_setting'
"application_name": "name",
"desc": "desc",
"prologue": "prologue",
"dialogue_number": "dialogue_number",
"user_id": "user_id",
"model_id": "model_id",
"knowledge_setting": "knowledge_setting",
"model_setting": "model_setting",
"model_params_setting": "model_params_setting",
"tts_model_params_setting": "tts_model_params_setting",
"problem_optimization": "problem_optimization",
"work_flow": "work_flow",
"problem_optimization_prompt": "problem_optimization_prompt",
"tts_model_id": "tts_model_id",
"stt_model_id": "stt_model_id",
"tts_model_enable": "tts_model_enable",
"stt_model_enable": "stt_model_enable",
"tts_type": "tts_type",
"tts_autoplay": "tts_autoplay",
"stt_autosend": "stt_autosend",
"file_upload_enable": "file_upload_enable",
"file_upload_setting": "file_upload_setting",
}
for (version_field, app_field) in update_field_dict.items():
for version_field, app_field in update_field_dict.items():
_v = getattr(application_version, version_field)
setattr(application, app_field, _v)

Expand All @@ -144,60 +175,67 @@ def profile(self, with_valid=True):
application_access_token = QuerySet(ApplicationAccessToken).filter(application_id=application.id).first()
if application_access_token is None:
raise AppUnauthorizedFailed(500, _("Illegal User"))
application_setting_model = DatabaseModelManage.get_model('application_setting')
application_version = QuerySet(ApplicationVersion).filter(application_id=application.id).order_by(
'-create_time').first()
application_setting_model = DatabaseModelManage.get_model("application_setting")
application_version = (
QuerySet(ApplicationVersion).filter(application_id=application.id).order_by("-create_time").first()
)
if application_version is not None:
self.reset_application(application, application_version)
license_is_valid = cache.get(Cache_Version.SYSTEM.get_key(key='license_is_valid'),
version=Cache_Version.SYSTEM.get_version())
license_is_valid = cache.get(
Cache_Version.SYSTEM.get_key(key="license_is_valid"), version=Cache_Version.SYSTEM.get_version()
)
application_setting_dict = {}
if application_setting_model is not None and license_is_valid:
application_setting = QuerySet(application_setting_model).filter(
application_id=application_access_token.application_id).first()
application_setting = (
QuerySet(application_setting_model)
.filter(application_id=application_access_token.application_id)
.first()
)
if application_setting is not None:
custom_theme = getattr(application_setting, 'custom_theme', {})
float_location = getattr(application_setting, 'float_location', {})
custom_theme = getattr(application_setting, "custom_theme", {})
float_location = getattr(application_setting, "float_location", {})
if not custom_theme:
application_setting.custom_theme = {
'theme_color': '',
'header_font_color': ''
}
application_setting.custom_theme = {"theme_color": "", "header_font_color": ""}
if not float_location:
application_setting.float_location = {
'x': {'type': '', 'value': ''},
'y': {'type': '', 'value': ''}
"x": {"type": "", "value": ""},
"y": {"type": "", "value": ""},
}
application_setting_dict = {'show_source': application_access_token.show_source,
'show_history': application_setting.show_history,
'draggable': application_setting.draggable,
'show_guide': application_setting.show_guide,
'avatar': application_setting.avatar,
'show_avatar': application_setting.show_avatar,
'float_icon': application_setting.float_icon,
'disclaimer': application_setting.disclaimer,
'disclaimer_value': application_setting.disclaimer_value,
'custom_theme': application_setting.custom_theme,
'user_avatar': application_setting.user_avatar,
'show_user_avatar': application_setting.show_user_avatar,
'show_share': application_setting.show_share,
'float_location': application_setting.float_location,
'chat_background': application_setting.chat_background}
base_node = [node for node in ((application.work_flow or {}).get('nodes', []) or []) if
node.get('id') == 'base-node']
return {**ApplicationSerializerModel(application).data,
'stt_model_id': application.stt_model_id,
'tts_model_id': application.tts_model_id,
'stt_model_enable': application.stt_model_enable,
'tts_model_enable': application.tts_model_enable,
'tts_type': application.tts_type,
'tts_autoplay': application.tts_autoplay,
'stt_autosend': application.stt_autosend,
'file_upload_enable': application.file_upload_enable,
'file_upload_setting': application.file_upload_setting,
'work_flow': {'nodes': base_node} if base_node else None,
'show_source': application_access_token.show_source,
'show_exec': application_access_token.show_exec,
'show_share': True,
'language': application_access_token.language,
**application_setting_dict}
application_setting_dict = {
"show_source": application_access_token.show_source,
"show_history": application_setting.show_history,
"draggable": application_setting.draggable,
"show_guide": application_setting.show_guide,
"avatar": application_setting.avatar,
"show_avatar": application_setting.show_avatar,
"float_icon": application_setting.float_icon,
"disclaimer": application_setting.disclaimer,
"disclaimer_value": application_setting.disclaimer_value,
"custom_theme": application_setting.custom_theme,
"user_avatar": application_setting.user_avatar,
"show_user_avatar": application_setting.show_user_avatar,
"show_share": application_setting.show_share,
"float_location": application_setting.float_location,
"chat_background": application_setting.chat_background,
}
base_node = [
node for node in ((application.work_flow or {}).get("nodes", []) or []) if node.get("id") == "base-node"
]
return {
**ApplicationSerializerModel(application).data,
"stt_model_id": application.stt_model_id,
"tts_model_id": application.tts_model_id,
"stt_model_enable": application.stt_model_enable,
"tts_model_enable": application.tts_model_enable,
"tts_type": application.tts_type,
"tts_autoplay": application.tts_autoplay,
"stt_autosend": application.stt_autosend,
"file_upload_enable": application.file_upload_enable,
"file_upload_setting": application.file_upload_setting,
"work_flow": {"nodes": base_node} if base_node else None,
"show_source": application_access_token.show_source,
"show_exec": application_access_token.show_exec,
"show_share": True,
"language": application_access_token.language,
**application_setting_dict,
}
Loading
Loading