kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(security): 防止路径遍历攻击通过使用basename处理task_id

Browse Source
This commit is contained in:
Wenjie Zhang 2026-01-02 22:53:15 +08:00
parent dd5dc53a64
commit 6a94b0626c
1 changed files with 6 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/services/evaluation_service.py
View File

@ -767,7 +767,9 @@ class EvaluationService:
async def get_evaluation_results_by_db(
self, db_id: str, task_id: str, page: int = 1, page_size: int = 20, error_only: bool = False
) -> dict[str, Any]:
result_file_path = os.path.join(self._get_result_dir(db_id), f"{task_id}.json")
# Prevent path traversal by using basename
safe_task_id = os.path.basename(task_id)
result_file_path = os.path.join(self._get_result_dir(db_id), f"{safe_task_id}.json")
if not os.path.exists(result_file_path):
task = await tasker.get_task(task_id)
if task:
@ -833,7 +835,9 @@ class EvaluationService:
return data
async def delete_evaluation_result_by_db(self, db_id: str, task_id: str) -> None:
result_file_path = os.path.join(self._get_result_dir(db_id), f"{task_id}.json")
# Prevent path traversal by using basename
safe_task_id = os.path.basename(task_id)
result_file_path = os.path.join(self._get_result_dir(db_id), f"{safe_task_id}.json")
if os.path.exists(result_file_path):
os.remove(result_file_path)
logger.info(f"成功删除评估结果: {task_id}")