205 lines
11 KiB
Python
205 lines
11 KiB
Python
# coding=utf-8
|
||
"""
|
||
@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
|
||
from django.db.models import QuerySet
|
||
from django.utils.translation import gettext_lazy as _
|
||
from rest_framework import serializers
|
||
|
||
from application.models import ApplicationAccessToken, ChatUserType, Application, ApplicationVersion
|
||
from application.serializers.application import ApplicationSerializerModel
|
||
from common.auth.common import ChatUserToken, ChatAuthentication
|
||
from common.auth.handle.impl.user_token import UserToken
|
||
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.utils.rsa_util import get_key_pair_by_sql
|
||
|
||
|
||
class AnonymousAuthenticationSerializer(serializers.Serializer):
|
||
access_token = serializers.CharField(required=True, label=_("access_token"))
|
||
|
||
@staticmethod
|
||
def get_platform_user(request):
|
||
platform_auth = request.META.get('HTTP_X_PLATFORM_AUTHORIZATION')
|
||
if platform_auth is None or not platform_auth.startswith('Bearer '):
|
||
return None
|
||
token = platform_auth[7:]
|
||
token_details = None
|
||
|
||
def get_token_details():
|
||
nonlocal token_details
|
||
if token_details is None:
|
||
token_details = signing.loads(token)
|
||
return token_details
|
||
|
||
try:
|
||
user, _ = UserToken().handle(request, token, get_token_details)
|
||
return user
|
||
except Exception:
|
||
return None
|
||
|
||
def auth(self, request, with_valid=True):
|
||
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
|
||
if with_valid:
|
||
self.is_valid(raise_exception=True)
|
||
access_token = self.data.get("access_token")
|
||
application_access_token = QuerySet(ApplicationAccessToken).filter(access_token=access_token).first()
|
||
if application_access_token is not None and application_access_token.is_active:
|
||
platform_user = self.get_platform_user(request)
|
||
if platform_user is not None:
|
||
chat_user_id = str(platform_user.id)
|
||
chat_user_type = ChatUserType.PLATFORM_USER.value
|
||
user_id = platform_user.id
|
||
else:
|
||
chat_user_id = token_details.get('chat_user_id') or str(uuid.uuid7())
|
||
chat_user_type = ChatUserType.ANONYMOUS_USER.value
|
||
user_id = None
|
||
_type = AuthenticationType.CHAT_ANONYMOUS_USER
|
||
token = ChatUserToken(application_access_token.application_id, user_id, access_token, _type,
|
||
chat_user_type, chat_user_id, ChatAuthentication(None)).to_token()
|
||
return {'token': token, 'chat_user_type': chat_user_type}
|
||
else:
|
||
raise NotFound404(404, _("Invalid access_token"))
|
||
|
||
|
||
class AuthProfileSerializer(serializers.Serializer):
|
||
access_token = serializers.CharField(required=True, label=_("access_token"))
|
||
|
||
def profile(self):
|
||
self.is_valid(raise_exception=True)
|
||
access_token = self.data.get("access_token")
|
||
application_access_token = QuerySet(ApplicationAccessToken).filter(access_token=access_token).first()
|
||
if application_access_token is None:
|
||
raise NotFound404(404, _("Invalid access_token"))
|
||
if not application_access_token.is_active:
|
||
raise NotFound404(404, _("Invalid access_token"))
|
||
application_id = application_access_token.application_id
|
||
profile = {
|
||
'authentication': False
|
||
}
|
||
application_setting_model = DatabaseModelManage.get_model('application_setting')
|
||
chat_platform = DatabaseModelManage.get_model('chat_platform')
|
||
if application_setting_model and chat_platform:
|
||
application_setting = QuerySet(application_setting_model).filter(application_id=application_id).first()
|
||
types = QuerySet(chat_platform).filter(is_active=True, is_valid=True).values_list('auth_type', flat=True)
|
||
login_value = application_access_token.authentication_value.get('login_value', [])
|
||
max_attempts = application_access_token.authentication_value.get('max_attempts', 1)
|
||
final_login_value = list(set(login_value) & set(types))
|
||
if 'LOCAL' in login_value:
|
||
final_login_value.insert(0, 'LOCAL')
|
||
if application_setting is not None:
|
||
profile = {
|
||
'icon': application_setting.application.icon,
|
||
'application_name': application_setting.application.name,
|
||
'bg_icon': application_setting.chat_background,
|
||
'authentication': application_access_token.authentication,
|
||
'authentication_type': application_access_token.authentication_value.get(
|
||
'type', 'password'),
|
||
'max_attempts': max_attempts,
|
||
'login_value': final_login_value,
|
||
'rasKey' : get_key_pair_by_sql().get('key')
|
||
}
|
||
return profile
|
||
|
||
|
||
class ApplicationProfileSerializer(serializers.Serializer):
|
||
application_id = serializers.UUIDField(required=True, label=_("Application ID"))
|
||
|
||
@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'
|
||
}
|
||
for (version_field, app_field) in update_field_dict.items():
|
||
_v = getattr(application_version, version_field)
|
||
setattr(application, app_field, _v)
|
||
|
||
def profile(self, with_valid=True):
|
||
if with_valid:
|
||
self.is_valid()
|
||
application_id = self.data.get("application_id")
|
||
application = QuerySet(Application).get(id=application_id)
|
||
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()
|
||
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())
|
||
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()
|
||
if application_setting is not None:
|
||
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': ''
|
||
}
|
||
if not float_location:
|
||
application_setting.float_location = {
|
||
'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,
|
||
'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,
|
||
'language': application_access_token.language,
|
||
**application_setting_dict}
|