fix: Cross-Workspace Assistant Modification and Deletion (IDOR)

Author: fit2cloud-chenyw

backend/apps/system/api/assistant.py

@@ -13,7 +13,7 @@
 from apps.db.constant import DB
 from apps.swagger.i18n import PLACEHOLDER_PREFIX
 from apps.system.crud.assistant import AssistantOutDs, AssistantOutDsFactory, get_assistant_info
-from apps.system.crud.assistant_manage import dynamic_upgrade_cors, save
+from apps.system.crud.assistant_manage import dynamic_upgrade_cors, get_ws_assistant, save
 from apps.system.models.system_model import AssistantModel
 from apps.system.schemas.auth import CacheName, CacheNamespace
 from apps.system.schemas.permission import SqlbotPermission, require_permissions
@@ -177,13 +177,11 @@ def iterfile():
 @router.patch('/ui', summary=f"{PLACEHOLDER_PREFIX}assistant_ui_api", description=f"{PLACEHOLDER_PREFIX}assistant_ui_api")
 @require_permissions(permission=SqlbotPermission(role=['ws_admin']))
 @system_log(LogConfig(operation_type=OperationType.UPDATE, module=OperationModules.APPLICATION, result_id_expr="id"))
-async def ui(session: SessionDep, data: str = Form(), files: List[UploadFile] = []):
+async def ui(session: SessionDep, current_user: CurrentUser, trans: Trans, data: str = Form(), files: List[UploadFile] = []):
     json_data = json.loads(data)
     uiSchema = AssistantUiSchema(**json_data)
     id = uiSchema.id
-    db_model = session.get(AssistantModel, id)
-    if not db_model:
-        raise ValueError(f"AssistantModel with id {id} not found")
+    db_model = get_ws_assistant(session, id, current_user, trans)
     configuration = db_model.configuration
     config_obj = json.loads(configuration) if configuration else {}
 
@@ -332,12 +330,15 @@ async def add(request: Request, session: SessionDep, current_user: CurrentUser,
 @require_permissions(permission=SqlbotPermission(role=['ws_admin']))
 @clear_cache(namespace=CacheNamespace.EMBEDDED_INFO, cacheName=CacheName.ASSISTANT_INFO, keyExpression="editor.id")
 @system_log(LogConfig(operation_type=OperationType.UPDATE, module=OperationModules.APPLICATION, resource_id_expr="editor.id"))
-async def update(request: Request, session: SessionDep, editor: AssistantDTO):
+async def update(request: Request, session: SessionDep, current_user: CurrentUser, trans: Trans, editor: AssistantDTO):
     id = editor.id
-    db_model = session.get(AssistantModel, id)
-    if not db_model:
-        raise ValueError(f"AssistantModel with id {id} not found")
+    db_model = get_ws_assistant(session, id, current_user, trans)
     update_data = AssistantModel.model_validate(editor)
+    if not current_user.isAdmin:
+        # 忽略请求体 oid（含默认值 1）：非 admin 不允许跨工作空间移动/接管
+        update_data.oid = db_model.oid
+    # create_time 不在请求体中，model_validate 会填默认值 0，更新时保留原值
+    update_data.create_time = db_model.create_time
     db_model.sqlmodel_update(update_data)
     session.add(db_model)
     session.commit()
@@ -357,10 +358,8 @@ async def get_one(session: SessionDep, id: int = Path(description="ID")):
 @require_permissions(permission=SqlbotPermission(role=['ws_admin']))
 @clear_cache(namespace=CacheNamespace.EMBEDDED_INFO, cacheName=CacheName.ASSISTANT_INFO, keyExpression="id")
 @system_log(LogConfig(operation_type=OperationType.DELETE, module=OperationModules.APPLICATION, resource_id_expr="id"))
-async def delete(request: Request, session: SessionDep, id: int = Path(description="ID")):
-    db_model = session.get(AssistantModel, id)
-    if not db_model:
-        raise ValueError(f"AssistantModel with id {id} not found")
+async def delete(request: Request, session: SessionDep, current_user: CurrentUser, trans: Trans, id: int = Path(description="ID")):
+    db_model = get_ws_assistant(session, id, current_user, trans)
     session.delete(db_model)
     session.commit()
     dynamic_upgrade_cors(request=request, session=session)

backend/apps/system/crud/assistant_manage.py

@@ -1,10 +1,10 @@
 
 
 from typing import Optional
-from fastapi import FastAPI, Request
+from fastapi import FastAPI, HTTPException, Request
 from sqlmodel import Session, select
 from starlette.middleware.cors import CORSMiddleware
-from apps.system.schemas.system_schema import AssistantBase
+from apps.system.schemas.system_schema import AssistantBase, UserInfoDTO
 from common.core.config import settings
 from apps.system.models.system_model import AssistantModel
 from common.utils.time import get_timestamp
@@ -60,4 +60,16 @@ async def save(request: Request, session: Session, creator: AssistantBase, oid:
     session.add(db_model)
     session.commit()
     dynamic_upgrade_cors(request=request, session=session)
-    return db_model
+    return db_model
+
+
+def get_ws_assistant(session: Session, id: int, current_user: UserInfoDTO, trans=None) -> AssistantModel:
+    """按 id 加载小助手并校验工作空间归属。资源不存在与越权访问统一返回 404，不泄露资源存在性。
+
+    系统管理员（isAdmin=True）保留跨工作空间管理能力。
+    """
+    db_model = session.get(AssistantModel, id)
+    if not db_model or (not current_user.isAdmin and db_model.oid != current_user.oid):
+        raise HTTPException(status_code=404,
+                            detail=trans('i18n_permission.permission_resource_limit') if trans else 'Assistant not found')
+    return db_model