fix: refactor file retrieval and upload logic to enhance permission checks and streamline code

apps/oss/views/file.py

@@ -1,94 +1,123 @@
 # coding=utf-8
+from application.models import Chat
+from maxkb.const import CONFIG
 from common.auth import AllTokenAuth, TokenAuth
 from common.auth.authentication import has_permissions
 from common.constants.permission_constants import ChatAuth, RoleConstants
+from common.exception.app_exception import AppUnauthorizedFailed
 from common.log.log import log
 from common.result import result
+from django.db.models import QuerySet
 from django.utils.translation import gettext_lazy as _
 from drf_spectacular.utils import extend_schema
 from knowledge.api.file import FileGetAPI, FileUploadAPI, GetUrlContentAPI
+from knowledge.models import FileSourceType
 from oss.serializers.file import FileSerializer, get_url_content
 from rest_framework.parsers import MultiPartParser
 from rest_framework.views import APIView, Request
 
 
 class FileRetrievalView(APIView):
     @extend_schema(
-        methods=['GET'],
-        summary=_('Get file'),
-        description=_('Get file'),
-        operation_id=_('Get file'),  # type: ignore
+        methods=["GET"],
+        summary=_("Get file"),
+        description=_("Get file"),
+        operation_id=_("Get file"),  # type: ignore
         parameters=FileGetAPI.get_parameters(),
         responses=FileGetAPI.get_response(),
-        tags=[_('File')]  # type: ignore
+        tags=[_("File")],  # type: ignore
     )
     def get(self, request: Request, file_id: str):
-        return FileSerializer.Operate(data={
-            'id': file_id,
-            'http_range': request.headers.get('Range', ''),
-        }).get(mk_file_auth=request.COOKIES.get('mk_file_auth'))
+        return FileSerializer.Operate(
+            data={
+                "id": file_id,
+                "http_range": request.headers.get("Range", ""),
+            }
+        ).get(mk_file_auth=request.COOKIES.get("mk_file_auth"))
 
 
 class FileView(APIView):
     authentication_classes = [AllTokenAuth]
     parser_classes = [MultiPartParser]
 
     @extend_schema(
-        methods=['POST'],
-        summary=_('Upload file'),
-        description=_('Upload file'),
-        operation_id=_('Upload file'),  # type: ignore
+        methods=["POST"],
+        summary=_("Upload file"),
+        description=_("Upload file"),
+        operation_id=_("Upload file"),  # type: ignore
         parameters=FileUploadAPI.get_parameters(),
         request=FileUploadAPI.get_request(),
         responses=FileUploadAPI.get_response(),
-        tags=[_('File')]  # type: ignore
+        tags=[_("File")],  # type: ignore
     )
-    @log(menu='file', operate='Upload file')
+    @log(menu="file", operate="Upload file")
     def post(self, request: Request):
-        return result.success(FileSerializer(data={
-            'file': request.FILES.get('file'),
-            'source_id': request.data.get('source_id'),
-            'source_type': request.data.get('source_type'),
-        }).upload(user_id=(str(request.user.id) if request.user else request.auth.chat_user_id)))
+        source_id = request.data.get("source_id")
+        source_type = request.data.get("source_type") or FileSourceType.TEMPORARY_120_MINUTE.value
+        # 聊天路径（/chat/...）或匿名会话下只能上传聊天文件，禁止将文件归属到
+        # Application/Knowledge 等其他受保护资源，无论调用者是否登录。
+        is_chat_path = request.path.startswith(CONFIG.get_chat_path())
+        if request.user is None or is_chat_path:
+            if source_type != FileSourceType.CHAT.value:
+                raise AppUnauthorizedFailed(403, _("No permission"))
+            # 聊天文件必须归属于调用者所在应用下的会话
+            if source_type == FileSourceType.CHAT.value:
+                chat = QuerySet(Chat).filter(id=source_id, application_id=request.auth.application_id).first()
+                if chat is None:
+                    raise AppUnauthorizedFailed(403, _("No permission"))
+        return result.success(
+            FileSerializer(
+                data={
+                    "file": request.FILES.get("file"),
+                    "source_id": source_id,
+                    "source_type": source_type,
+                }
+            ).upload(user_id=(str(request.user.id) if request.user else request.auth.chat_user_id))
+        )
 
     class Operate(APIView):
         authentication_classes = [TokenAuth]
 
         @extend_schema(
-            methods=['DELETE'],
-            summary=_('Delete file'),
-            description=_('Delete file'),
-            operation_id=_('Delete file'),  # type: ignore
+            methods=["DELETE"],
+            summary=_("Delete file"),
+            description=_("Delete file"),
+            operation_id=_("Delete file"),  # type: ignore
             parameters=FileGetAPI.get_parameters(),
             responses=FileGetAPI.get_response(),
-            tags=[_('File')]  # type: ignore
+            tags=[_("File")],  # type: ignore
         )
-        @log(menu='file', operate='Delete file')
+        @log(menu="file", operate="Delete file")
         @has_permissions(RoleConstants.ADMIN, RoleConstants.WORKSPACE_MANAGE, RoleConstants.USER)
         def delete(self, request: Request, file_id: str):
-            return result.success(FileSerializer.Operate(
-                data={
-                    "id": file_id,
-                    "http_range": request.headers.get("Range", ""),
-                }
-            ).delete(mk_file_auth=request.COOKIES.get("mk_file_auth")))
+            return result.success(
+                FileSerializer.Operate(
+                    data={
+                        "id": file_id,
+                        "http_range": request.headers.get("Range", ""),
+                    }
+                ).delete(mk_file_auth=request.COOKIES.get("mk_file_auth"))
+            )
 
 
 class GetUrlView(APIView):
     authentication_classes = [AllTokenAuth]
 
     @extend_schema(
-        methods=['GET'],
-        summary=_('Get url'),
+        methods=["GET"],
+        summary=_("Get url"),
         parameters=GetUrlContentAPI.get_parameters(),
-        description=_('Get url'),
-        operation_id=_('Get url'),  # type: ignore
-        tags=[_('Chat')]  # type: ignore
+        description=_("Get url"),
+        operation_id=_("Get url"),  # type: ignore
+        tags=[_("Chat")],  # type: ignore
     )
     def get(self, request: Request, application_id: str):
-        if isinstance(request.auth, ChatAuth) and request.auth.application_id and str(
-                request.auth.application_id) != application_id:
-            return result.error(_('No permission'))
-        url = request.query_params.get('url')
+        if (
+            isinstance(request.auth, ChatAuth)
+            and request.auth.application_id
+            and str(request.auth.application_id) != application_id
+        ):
+            return result.error(_("No permission"))
+        url = request.query_params.get("url")
         result_data = get_url_content(url, application_id)
         return result.success(result_data)