Fix: unexpected operation of document management

api/apps/document_app.py

@@ -24,6 +24,7 @@
 from flask_login import current_user, login_required
 
 from api import settings
+from api.common.check_team_permission import check_kb_team_permission
 from api.constants import FILE_NAME_LEN_LIMIT, IMG_BASE64_PREFIX
 from api.db import VALID_FILE_TYPES, VALID_TASK_STATUS, FileSource, FileType, ParserType, TaskStatus
 from api.db.db_models import File, Task
@@ -68,8 +69,10 @@ def upload():
     e, kb = KnowledgebaseService.get_by_id(kb_id)
     if not e:
         raise LookupError("Can't find this knowledgebase!")
-    err, files = FileService.upload_document(kb, file_objs, current_user.id)
+    if not check_kb_team_permission(kb, current_user.id):
+        return get_json_result(data=False, message="No authorization.", code=settings.RetCode.AUTHENTICATION_ERROR)
 
+    err, files = FileService.upload_document(kb, file_objs, current_user.id)
     if err:
         return get_json_result(data=files, message="\n".join(err), code=settings.RetCode.SERVER_ERROR)
 
@@ -94,6 +97,8 @@ def web_crawl():
     e, kb = KnowledgebaseService.get_by_id(kb_id)
     if not e:
         raise LookupError("Can't find this knowledgebase!")
+    if check_kb_team_permission(kb, current_user.id):
+        return get_json_result(data=False, message="No authorization.", code=settings.RetCode.AUTHENTICATION_ERROR)
 
     blob = html2pdf(url)
     if not blob:

api/apps/file_app.py

@@ -21,6 +21,7 @@
 from flask import request
 from flask_login import login_required, current_user
 
+from api.common.check_team_permission import check_file_team_permission
 from api.db.services.document_service import DocumentService
 from api.db.services.file2document_service import File2DocumentService
 from api.utils.api_utils import server_error_response, get_data_error_result, validate_request
@@ -178,6 +179,9 @@ def list_files():
         if not e:
             return get_data_error_result(message="Folder not found!")
 
+        if not check_file_team_permission(file, current_user.id):
+            return get_json_result(data=False, message='No authorization.', code=settings.RetCode.AUTHENTICATION_ERROR)
+
         files, total = FileService.get_by_pf_id(
             current_user.id, pf_id, page_number, items_per_page, orderby, desc, keywords)
 
@@ -209,6 +213,9 @@ def get_parent_folder():
         if not e:
             return get_data_error_result(message="Folder not found!")
 
+        if not check_file_team_permission(file, current_user.id):
+            return get_json_result(data=False, message='No authorization.', code=settings.RetCode.AUTHENTICATION_ERROR)
+
         parent_folder = FileService.get_parent_folder(file_id)
         return get_json_result(data={"parent_folder": parent_folder.to_json()})
     except Exception as e:
@@ -224,6 +231,9 @@ def get_all_parent_folders():
         if not e:
             return get_data_error_result(message="Folder not found!")
 
+        if not check_file_team_permission(file, current_user.id):
+            return get_json_result(data=False, message='No authorization.', code=settings.RetCode.AUTHENTICATION_ERROR)
+
         parent_folders = FileService.get_all_parent_folders(file_id)
         parent_folders_res = []
         for parent_folder in parent_folders:
@@ -246,7 +256,7 @@ def rm():
                 return get_data_error_result(message="File or Folder not found!")
             if not file.tenant_id:
                 return get_data_error_result(message="Tenant not found!")
-            if file.tenant_id != current_user.id:
+            if not check_file_team_permission(file, current_user.id):
                 return get_json_result(data=False, message='No authorization.', code=settings.RetCode.AUTHENTICATION_ERROR)
             if file.source_type == FileSource.KNOWLEDGEBASE:
                 continue
@@ -294,7 +304,7 @@ def rename():
         e, file = FileService.get_by_id(req["file_id"])
         if not e:
             return get_data_error_result(message="File not found!")
-        if file.tenant_id != current_user.id:
+        if not check_file_team_permission(file, current_user.id):
             return get_json_result(data=False, message='No authorization.', code=settings.RetCode.AUTHENTICATION_ERROR)
         if file.type != FileType.FOLDER.value \
             and pathlib.Path(req["name"].lower()).suffix != pathlib.Path(
@@ -332,7 +342,7 @@ def get(file_id):
         e, file = FileService.get_by_id(file_id)
         if not e:
             return get_data_error_result(message="Document not found!")
-        if file.tenant_id != current_user.id:
+        if not check_file_team_permission(file, current_user.id):
             return get_json_result(data=False, message='No authorization.', code=settings.RetCode.AUTHENTICATION_ERROR)
 
         blob = STORAGE_IMPL.get(file.parent_id, file.location)
@@ -373,7 +383,7 @@ def move():
                 return get_data_error_result(message="File or Folder not found!")
             if not file.tenant_id:
                 return get_data_error_result(message="Tenant not found!")
-            if file.tenant_id != current_user.id:
+            if not check_file_team_permission(file, current_user.id):
                 return get_json_result(data=False, message='No authorization.', code=settings.RetCode.AUTHENTICATION_ERROR)
         fe, _ = FileService.get_by_id(parent_id)
         if not fe:

api/common/check_team_permission.py

@@ -0,0 +1,59 @@
+#
+#  Copyright 2025 The InfiniFlow Authors. All Rights Reserved.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+#
+
+
+from api.db import TenantPermission
+from api.db.db_models import File, Knowledgebase
+from api.db.services.file_service import FileService
+from api.db.services.knowledgebase_service import KnowledgebaseService
+from api.db.services.user_service import TenantService
+
+
+def check_kb_team_permission(kb: dict | Knowledgebase, other: str) -> bool:
+    kb = kb.to_dict() if isinstance(kb, Knowledgebase) else kb
+
+    kb_tenant_id = kb["tenant_id"]
+
+    if kb_tenant_id == other:
+        return True
+
+    if kb["permission"] != TenantPermission.TEAM:
+        return False
+
+    joined_tenants = TenantService.get_joined_tenants_by_user_id(other)
+    return any(tenant["tenant_id"] == kb_tenant_id for tenant in joined_tenants)
+
+
+def check_file_team_permission(file: dict | File, other: str) -> bool:
+    file = file.to_dict() if isinstance(file, File) else file
+
+    file_tenant_id = file["tenant_id"]
+    if file_tenant_id == other:
+        return True
+
+    file_id = file["id"]
+
+    kb_ids = [kb_info["kb_id"] for kb_info in FileService.get_kb_id_by_file_id(file_id)]
+
+    for kb_id in kb_ids:
+        ok, kb = KnowledgebaseService.get_by_id(kb_id)
+        if not ok:
+            continue
+
+        if check_kb_team_permission(kb, other):
+            return True
+
+    return False

api/db/services/user_service.py

@@ -315,4 +315,4 @@ def filter_by_tenant_and_user_id(cls, tenant_id, user_id):
             ).first()
             return user_tenant
         except peewee.DoesNotExist:
-            return None
+            return None