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

secure(48): 验证 task_id 格式以防止路径遍历漏洞

Browse Source
This commit is contained in:
Wenjie Zhang 2026-01-15 12:56:47 +08:00
parent c7a417c68b
commit beb62436fd
1 changed files with 9 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
src/services/evaluation_service.py
View File

@ -2,6 +2,7 @@ import asyncio
import glob import glob
import json import json
import os import os
import re
import uuid import uuid
from datetime import datetime from datetime import datetime
from typing import Any from typing import Any
@ -767,9 +768,10 @@ class EvaluationService:
async def get_evaluation_results_by_db( 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 self, db_id: str, task_id: str, page: int = 1, page_size: int = 20, error_only: bool = False
) -> dict[str, Any]: ) -> dict[str, Any]:
# Prevent path traversal by using basename # Validate task_id format to prevent path traversal
safe_task_id = os.path.basename(task_id) if not re.match(r'^eval_[a-f0-9]{8}$', task_id):
result_file_path = os.path.join(self._get_result_dir(db_id), f"{safe_task_id}.json") 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): if not os.path.exists(result_file_path):
task = await tasker.get_task(task_id) task = await tasker.get_task(task_id)
if task: if task:
@ -835,9 +837,10 @@ class EvaluationService:
return data return data
async def delete_evaluation_result_by_db(self, db_id: str, task_id: str) -> None: async def delete_evaluation_result_by_db(self, db_id: str, task_id: str) -> None:
# Prevent path traversal by using basename # Validate task_id format to prevent path traversal
safe_task_id = os.path.basename(task_id) if not re.match(r'^eval_[a-f0-9]{8}$', task_id):
result_file_path = os.path.join(self._get_result_dir(db_id), f"{safe_task_id}.json") 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): if os.path.exists(result_file_path):
os.remove(result_file_path) os.remove(result_file_path)
logger.info(f"成功删除评估结果: {task_id}") logger.info(f"成功删除评估结果: {task_id}")