fix: Cross-Workspace Recommended-Problem Read and Overwrite (IDOR)

Author: ziyujiahao

backend/apps/datasource/api/recommended_problem.py

@@ -5,6 +5,7 @@
     save_recommended_problem, get_datasource_recommended_base
 from apps.datasource.models.datasource import RecommendedProblemBase
 from apps.swagger.i18n import PLACEHOLDER_PREFIX
+from apps.system.schemas.permission import SqlbotPermission, require_permissions
 from common.audit.models.log_model import OperationType, OperationModules
 from common.audit.schemas.logger_decorator import LogConfig, system_log
 from common.core.deps import SessionDep, CurrentUser
@@ -13,19 +14,22 @@
 
 
 @router.get("/get_datasource_recommended/{ds_id}", response_model=None, summary=f"{PLACEHOLDER_PREFIX}rp_get")
-async def datasource_recommended(session: SessionDep, ds_id: int):
+@require_permissions(permission=SqlbotPermission(type='ds', keyExpression="ds_id"))
+async def datasource_recommended(session: SessionDep, user: CurrentUser, ds_id: int):
     return get_datasource_recommended(session, ds_id)
 
 
 @router.get("/get_datasource_recommended_base/{ds_id}", response_model=None, summary=f"{PLACEHOLDER_PREFIX}rp_base")
-async def datasource_recommended(session: SessionDep, ds_id: int):
+@require_permissions(permission=SqlbotPermission(type='ds', keyExpression="ds_id"))
+async def datasource_recommended_base(session: SessionDep, user: CurrentUser, ds_id: int):
     return get_datasource_recommended_base(session, ds_id)
 
 
 @router.post("/save_recommended_problem", response_model=None, summary=f"{PLACEHOLDER_PREFIX}rp_save")
+@require_permissions(permission=SqlbotPermission(type='ds', keyExpression="data_info.datasource_id"))
 @system_log(
     LogConfig(operation_type=OperationType.UPDATE, module=OperationModules.DATASOURCE,
               resource_id_expr="data_info.datasource_id"))
-async def datasource_recommended(session: SessionDep, user: CurrentUser, data_info: RecommendedProblemBase):
+async def datasource_recommended_save(session: SessionDep, user: CurrentUser, data_info: RecommendedProblemBase):
     update_ds_recommended_config(session, data_info.datasource_id, data_info.recommended_config)
     return save_recommended_problem(session, user, data_info)

backend/apps/datasource/crud/datasource.py

@@ -119,9 +119,10 @@ def update_ds(session: SessionDep, trans: Trans, user: CurrentUser, ds: CoreData
 
 
 def update_ds_recommended_config(session: SessionDep, datasource_id: int, recommended_config: int):
-    record = session.exec(select(CoreDatasource).where(CoreDatasource.id == datasource_id)).first()
-    record.recommended_config = recommended_config
-    session.add(record)
+    # 使用 update 语句直接更新，避免 ORM 追踪问题
+    from sqlalchemy import update
+    stmt = update(CoreDatasource).where(CoreDatasource.id == datasource_id).values(recommended_config=recommended_config)
+    session.execute(stmt)
     session.commit()
 
 

backend/apps/datasource/crud/recommended_problem.py

@@ -19,15 +19,15 @@ def get_datasource_recommended_chart(session: SessionDep, ds_id: int):
 
 
 def get_datasource_recommended_base(session: SessionDep, ds_id: int):
-    statement = select(CoreDatasource.id,CoreDatasource.recommended_config).where(CoreDatasource.id == ds_id)
+    statement = select(CoreDatasource.id, CoreDatasource.recommended_config).where(CoreDatasource.id == ds_id)
     datasourceBase = session.exec(statement).first()
     if datasourceBase is None:
-        return RecommendedProblemResponse(ds_id,0,None)
+        return RecommendedProblemResponse(ds_id, 0, None)
     elif datasourceBase.recommended_config == 1:
-        return RecommendedProblemResponse(ds_id,1,None)
+        return RecommendedProblemResponse(ds_id, 1, None)
     else:
         dsRecommendedProblems = session.exec(select(DsRecommendedProblem.question).where(DsRecommendedProblem.datasource_id == ds_id)).all()
-        return RecommendedProblemResponse(ds_id,datasourceBase.recommended_config, orjson.dumps(dsRecommendedProblems).decode())
+        return RecommendedProblemResponse(ds_id, datasourceBase.recommended_config, orjson.dumps(dsRecommendedProblems).decode())
 
 def save_recommended_problem(session: SessionDep,user: CurrentUser, data_info: RecommendedProblemBase):
     session.query(DsRecommendedProblem).filter(DsRecommendedProblem.datasource_id == data_info.datasource_id).delete(synchronize_session=False)