From beb62436fd62bd00410842be102a1c4007a721d6 Mon Sep 17 00:00:00 2001 From: Wenjie Zhang Date: Thu, 15 Jan 2026 12:56:47 +0800 Subject: [PATCH] =?UTF-8?q?secure(48):=20=E9=AA=8C=E8=AF=81=20task=5Fid=20?= =?UTF-8?q?=E6=A0=BC=E5=BC=8F=E4=BB=A5=E9=98=B2=E6=AD=A2=E8=B7=AF=E5=BE=84?= =?UTF-8?q?=E9=81=8D=E5=8E=86=E6=BC=8F=E6=B4=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/services/evaluation_service.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/src/services/evaluation_service.py b/src/services/evaluation_service.py index 89b0a241..b74cfddb 100644 --- a/src/services/evaluation_service.py +++ b/src/services/evaluation_service.py @@ -2,6 +2,7 @@ import asyncio import glob import json import os +import re import uuid from datetime import datetime from typing import Any @@ -767,9 +768,10 @@ 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]: - # 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") + # Validate task_id format to prevent path traversal + if not re.match(r'^eval_[a-f0-9]{8}$', task_id): + raise ValueError("Invalid task_id format") + result_file_path = os.path.join(self._get_result_dir(db_id), f"{task_id}.json") if not os.path.exists(result_file_path): task = await tasker.get_task(task_id) if task: @@ -835,9 +837,10 @@ class EvaluationService: return data async def delete_evaluation_result_by_db(self, db_id: str, task_id: str) -> None: - # 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") + # Validate task_id format to prevent path traversal + if not re.match(r'^eval_[a-f0-9]{8}$', task_id): + raise ValueError("Invalid task_id format") + result_file_path = os.path.join(self._get_result_dir(db_id), f"{task_id}.json") if os.path.exists(result_file_path): os.remove(result_file_path) logger.info(f"成功删除评估结果: {task_id}")