fix: prevent path traversal via task_id in cleaning task log/delete e…

backend/services/data-management-service/src/main/java/com/datamate/datamanagement/application/DatasetFileApplicationService.java

@@ -493,6 +493,16 @@ public String preUpload(UploadFilesPreRequest chunkUploadRequest, String dataset
             if (!prefix.isEmpty()) {
                 uploadPath = uploadPath + File.separator + prefix.replace("/", File.separator);
             }
+
+            // 防止路径穿越：规范化后校验 uploadPath 仍在 datasetBasePath 下
+            Path normalizedUploadPath = Paths.get(uploadPath).normalize().toAbsolutePath();
+            Path normalizedBase = Paths.get(datasetBasePath).normalize().toAbsolutePath();
+            if (!normalizedUploadPath.startsWith(normalizedBase)) {
+                log.warn("Path traversal attempt in preUpload: datasetId={}, prefix={}, uploadPath={}",
+                    datasetId, prefix, normalizedUploadPath);
+                throw BusinessException.of(CommonErrorCode.PARAM_ERROR,
+                    "Upload path outside allowed directory");
+            }
         }
 
         ChunkUploadPreRequest request = ChunkUploadPreRequest.builder().build();

backend/services/data-management-service/src/main/java/com/datamate/datamanagement/interfaces/validation/ValidFilePathValidator.java

@@ -31,6 +31,15 @@ public boolean isValid(String value, ConstraintValidatorContext context) {
             return true; // 空值由 @NotBlank 等其他注解处理
         }
 
+        // 检查是否包含路径遍历序列 ..
+        if (value.contains("..")) {
+            context.disableDefaultConstraintViolation();
+            context.buildConstraintViolationWithTemplate(
+                "文件路径不允许包含 '..'"
+            ).addConstraintViolation();
+            return false;
+        }
+
         boolean isValid = FILE_PATH_PATTERN.matcher(value).matches();
 
         if (!isValid) {

backend/services/data-management-service/src/main/java/com/datamate/datamanagement/interfaces/validation/ValidPathValidator.java

@@ -54,6 +54,15 @@ public boolean isValid(String value, ConstraintValidatorContext context) {
             return false;
         }
 
+        // 检查是否包含路径遍历序列 ..
+        if (value.contains("..")) {
+            context.disableDefaultConstraintViolation();
+            context.buildConstraintViolationWithTemplate(
+                "路径不允许包含 '..'"
+            ).addConstraintViolation();
+            return false;
+        }
+
         // 检查是否包含非法字符
         if (!PATH_PATTERN.matcher(value).matches()) {
             context.disableDefaultConstraintViolation();

backend/shared/domain-common/src/main/java/com/datamate/common/domain/utils/ChunksSaver.java

@@ -49,6 +49,14 @@ public static Optional<File> save(ChunkUploadRequest fileUploadRequest, ChunkUpl
         }
 
         File finalFile = new File(preUploadReq.getUploadPath(), fileUploadRequest.getFileName());
+        // 防止路径穿越：规范化后校验仍在 uploadPath 下
+        Path uploadBasePath = Paths.get(preUploadReq.getUploadPath()).normalize().toAbsolutePath();
+        Path finalFilePath = finalFile.toPath().normalize().toAbsolutePath();
+        if (!finalFilePath.startsWith(uploadBasePath)) {
+            log.error("Path traversal attempt detected in chunk save: fileName={}, finalPath={}",
+                fileUploadRequest.getFileName(), finalFilePath);
+            throw BusinessException.of(SystemErrorCode.FILE_SYSTEM_ERROR);
+        }
         // 确保父目录存在（处理嵌套文件夹上传的情况）
         File parentDir = finalFile.getParentFile();
         if (parentDir != null && !parentDir.exists()) {
@@ -91,6 +99,14 @@ private static InputStream getFileInputStream(MultipartFile file) {
     public static File saveFile(ChunkUploadRequest fileUploadRequest, ChunkUploadPreRequest preUploadReq) {
         // 保存文件
         File targetFile = new File(preUploadReq.getUploadPath(), fileUploadRequest.getFileName());
+        // 防止路径穿越：规范化后校验仍在 uploadPath 下
+        Path uploadBasePath = Paths.get(preUploadReq.getUploadPath()).normalize().toAbsolutePath();
+        Path targetFilePath = targetFile.toPath().normalize().toAbsolutePath();
+        if (!targetFilePath.startsWith(uploadBasePath)) {
+            log.error("Path traversal attempt detected in saveFile: fileName={}, targetPath={}",
+                fileUploadRequest.getFileName(), targetFilePath);
+            throw BusinessException.of(SystemErrorCode.FILE_SYSTEM_ERROR);
+        }
         // 确保父目录存在（处理嵌套文件夹上传的情况）
         File parentDir = targetFile.getParentFile();
         if (parentDir != null && !parentDir.exists()) {

runtime/datamate-python/app/module/cleaning/interface/cleaning_task_routes.py

@@ -284,7 +284,16 @@
     import re
     from pathlib import Path
 
-    FLOW_PATH = "/flow"
+    flow_base = Path("/flow").resolve()
+
+    # 校验 task_id 格式，防止不受控数据直接用于路径构造
+    if not re.fullmatch(r"[A-Za-z0-9_-]+", task_id):
+        logger.warning(f"Invalid task_id in stream_log: task_id={task_id}")
+
+        async def invalid_error_generator():
+            yield f"data: {json.dumps({'level': 'ERROR', 'message': 'Invalid task_id'}, ensure_ascii=False)}\n\n"
+
+        return StreamingResponse(invalid_error_generator(), media_type="text/event-stream")
 
     task_service = _get_task_service(db)
     task = await task_service.task_repo.find_task_by_id(db, task_id)
@@ -296,9 +305,19 @@
 
         return StreamingResponse(error_generator(), media_type="text/event-stream")
 
-    log_path = Path(f"{FLOW_PATH}/{task_id}/output.log")
-    if retry_count > 0:
-        log_path = Path(f"{FLOW_PATH}/{task_id}/output.log.{retry_count}")
+    log_filename = "output.log" if retry_count == 0 else f"output.log.{retry_count}"
+    log_path = (flow_base / task_id / log_filename).resolve()
+
+    # 防止路径穿越：relative_to 校验仍在 flow_base 下
+    try:
+        log_path.relative_to(flow_base)
+    except ValueError:
+        logger.warning(f"Path traversal attempt in stream_log: task_id={task_id}")
+
+        async def traversal_error_generator():
+            yield f"data: {json.dumps({'level': 'ERROR', 'message': 'Invalid task_id'}, ensure_ascii=False)}\n\n"
+
+        return StreamingResponse(traversal_error_generator(), media_type="text/event-stream")
 
     standard_level_pattern = re.compile(
         r"\b(DEBUG|Debug|INFO|Info|WARN|Warn|WARNING|Warning|ERROR|Error|FATAL|Fatal)\b"
@@ -414,7 +433,16 @@
     from pathlib import Path
     from fastapi.responses import FileResponse
 
-    FLOW_PATH = "/flow"
+    flow_base = Path("/flow").resolve()
+
+    # 校验 task_id 格式，防止不受控数据直接用于路径构造
+    if not re.fullmatch(r"[A-Za-z0-9_-]+", task_id):
+        logger.warning(f"Invalid task_id in download_log: task_id={task_id}")
+        from app.core.exception import BusinessError, ErrorCodes
+        raise BusinessError(
+            ErrorCodes.CLEANING_TASK_LOG_NOT_FOUND,
+            f"Invalid task_id: {task_id}",
+        )
 
     task_service = _get_task_service(db)
     task = await task_service.task_repo.find_task_by_id(db, task_id)
@@ -424,9 +452,19 @@
 
         raise BusinessError(ErrorCodes.CLEANING_TASK_NOT_FOUND, task_id)
 
-    log_path = Path(f"{FLOW_PATH}/{task_id}/output.log")
-    if retry_count > 0:
-        log_path = Path(f"{FLOW_PATH}/{task_id}/output.log.{retry_count}")
+    log_filename = "output.log" if retry_count == 0 else f"output.log.{retry_count}"
+    log_path = (flow_base / task_id / log_filename).resolve()
+
+    # 防止路径穿越：relative_to 校验仍在 flow_base 下
+    try:
+        log_path.relative_to(flow_base)
+    except ValueError:
+        logger.warning(f"Path traversal attempt in download_log: task_id={task_id}")
+        from app.core.exception import BusinessError, ErrorCodes
+        raise BusinessError(
+            ErrorCodes.CLEANING_TASK_LOG_NOT_FOUND,
+            f"Invalid task_id: {task_id}",
+        )
 
     if not log_path.exists():
         from app.core.exception import BusinessError, ErrorCodes

runtime/datamate-python/app/module/cleaning/service/cleaning_task_service.py

@@ -1,4 +1,5 @@
 import json
+import os
 import re
 import shutil
 import uuid
@@ -382,11 +383,20 @@
         """Get task log"""
         self.validator.check_task_id(task_id)
 
-        log_path = Path(f"{FLOW_PATH}/{task_id}/output.log")
+        flow_root = Path(FLOW_PATH).resolve()
+        log_path = flow_root / task_id / "output.log"
         if retry_count > 0:
-            log_path = Path(f"{FLOW_PATH}/{task_id}/output.log.{retry_count}")
+            log_path = flow_root / task_id / f"output.log.{retry_count}"
 
-        if not log_path.exists():
+        # 防止路径穿越：规范化后校验仍在 FLOW_PATH 下
+        resolved_log_path = log_path.resolve()
+        try:
+            resolved_log_path.relative_to(flow_root)
+        except ValueError:
+            logger.warning(f"Path traversal attempt detected: task_id={task_id}")
+            return []
+
+        if not resolved_log_path.exists():
             return []
 
         logs = []
@@ -397,7 +407,7 @@
         )
         exception_suffix_pattern = re.compile(r"\b\w+(Warning|Error|Exception)\b")
 
-        with open(log_path, "r", encoding="utf-8") as f:
+        with open(resolved_log_path, "r", encoding="utf-8") as f:
             for line in f:
                 last_level = self._get_log_level(
                     line, last_level, standard_level_pattern, exception_suffix_pattern
@@ -447,7 +457,18 @@
         await self.result_repo.delete_by_instance_id(db, task_id)
 
         # 删除任务相关文件
-        task_path = Path(f"{FLOW_PATH}/{task_id}")
+        flow_root = Path(FLOW_PATH).resolve()
+        task_path = (flow_root / task_id).resolve()
+        # 防止路径穿越：relative_to 校验目标路径仍在 flow_root 下
+        try:
+            task_path.relative_to(flow_root)
+        except ValueError:
+            logger.warning(f"Path traversal attempt in delete_task: task_id={task_id}")
+            raise BusinessError(
+                ErrorCodes.CLEANING_TASK_NOT_FOUND,
+                f"Invalid task_id: {task_id}",
+            )
+
         if task_path.exists():
             try:
                 shutil.rmtree(task_path)

runtime/datamate-python/app/module/cleaning/service/cleaning_task_validator.py

@@ -1,9 +1,16 @@
+import re
+
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.exception import BusinessError, ErrorCodes
 from app.module.cleaning.schema import OperatorInstanceDto
 from app.module.operator.constants import CATEGORY_DATA_JUICER_ID, CATEGORY_DATAMATE_ID
 
+# UUID pattern for task_id validation (prevents path traversal)
+_TASK_ID_PATTERN = re.compile(
+    r'^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$'
+)
+
 
 class CleaningTaskValidator:
     """Validator for cleaning tasks and templates"""
@@ -95,6 +102,11 @@ def check_and_get_executor_type(instances: list[OperatorInstanceDto]) -> str:
 
     @staticmethod
     def check_task_id(task_id: str) -> None:
-        """Validate task ID"""
+        """Validate task ID — rejects non-UUID and path traversal patterns"""
         if not task_id:
             raise BusinessError(ErrorCodes.CLEANING_TASK_ID_REQUIRED)
+        if not _TASK_ID_PATTERN.match(task_id):
+            raise BusinessError(
+                ErrorCodes.CLEANING_TASK_ID_REQUIRED,
+                f"Invalid task_id format: {task_id}",
+            )

runtime/datamate-python/app/module/shared/chunks_saver.py

@@ -57,6 +57,16 @@ def save(
 
         final_file = Path(upload_path) / file_upload_request.file_name
 
+        # 防止路径穿越：规范化后校验仍在 upload_path 下
+        resolved_final = final_file.resolve()
+        resolved_base = Path(upload_path).resolve()
+        if not str(resolved_final).startswith(str(resolved_base) + os.sep):
+            logger.error(
+                f"Path traversal attempt in chunk save: file_name={file_upload_request.file_name}, "
+                f"resolved_path={resolved_final}"
+            )
+            raise ValueError("Path traversal detected in file name")
+
         try:
             temp_file.rename(final_file)
         except OSError as e:
@@ -89,6 +99,16 @@ def save_file(
         """
         target_file = Path(upload_path) / file_upload_request.file_name
 
+        # 防止路径穿越：规范化后校验仍在 upload_path 下
+        resolved_target = target_file.resolve()
+        resolved_base = Path(upload_path).resolve()
+        if not str(resolved_target).startswith(str(resolved_base) + os.sep):
+            logger.error(
+                f"Path traversal attempt in save_file: file_name={file_upload_request.file_name}, "
+                f"resolved_path={resolved_target}"
+            )
+            raise ValueError("Path traversal detected in file name")
+
         logger.info(f"file path {target_file}, file size {len(file_content)}")
 
         try:

runtime/datax/glusterfsreader/src/main/java/com/datamate/plugin/reader/glusterfsreader/GlusterfsReader.java

@@ -98,7 +98,16 @@ public void startRead(RecordSender recordSender) {
                 readPath = this.mountPoint + "/" + this.subPath.replaceFirst("^/+", "");
             }
 
-            try (Stream<Path> stream = Files.list(Paths.get(readPath))) {
+            // 防止路径穿越：规范化后校验读取路径仍在 mount 点下
+            Path normalizedReadPath = Paths.get(readPath).normalize().toAbsolutePath();
+            Path normalizedMount = Paths.get(this.mountPoint).normalize().toAbsolutePath();
+            if (!normalizedReadPath.startsWith(normalizedMount)) {
+                LOG.error("Path traversal detected in reader: readPath={} outside mountPoint={}",
+                    normalizedReadPath, normalizedMount);
+                throw new RuntimeException("Read path outside mount point: " + readPath);
+            }
+
+            try (Stream<Path> stream = Files.list(normalizedReadPath)) {
                 List<String> fileList = stream.filter(Files::isRegularFile)
                         .filter(file -> fileType.isEmpty() || fileType.contains(getFileSuffix(file)))
                         .map(path -> path.getFileName().toString())

runtime/datax/glusterfswriter/src/main/java/com/datamate/plugin/writer/glusterfswriter/GlusterfsWriter.java

@@ -13,6 +13,8 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
@@ -26,6 +28,9 @@ public class GlusterfsWriter extends Writer {
 
     private static final Logger LOG = LoggerFactory.getLogger(GlusterfsWriter.class);
 
+    /** 允许的目标路径基础目录 */
+    private static final String ALLOWED_DEST_BASE = "/dataset";
+
     public static class Job extends Writer.Job {
         private Configuration jobConfig;
         private String mountPoint;
@@ -49,7 +54,13 @@ public void prepare() {
             GlusterfsMountUtil.mount(remote, mountPoint);
 
             String destPath = this.jobConfig.getString("destPath");
-            new File(destPath).mkdirs();
+            // 防止路径穿越：规范化后校验目标路径在允许目录内
+            Path normalizedDest = Paths.get(destPath).normalize().toAbsolutePath();
+            if (!normalizedDest.startsWith(ALLOWED_DEST_BASE)) {
+                throw new IllegalArgumentException(
+                    "destPath is outside allowed directory: " + destPath);
+            }
+            new File(normalizedDest.toString()).mkdirs();
         }
 
         @Override
@@ -86,6 +97,12 @@ public void init() {
             this.mountPoint = this.jobConfig.getString("mountPoint");
             this.subPath = this.jobConfig.getString("path", "");
             this.files = this.jobConfig.getList("files", Collections.emptyList(), String.class);
+
+            // 防止路径穿越：subPath 不能包含 ..
+            if (StringUtils.isNotBlank(this.subPath) && this.subPath.contains("..")) {
+                throw new IllegalArgumentException(
+                    "Invalid subPath (path traversal): " + this.subPath);
+            }
         }
 
         @Override
@@ -95,6 +112,23 @@ public void startWrite(RecordReceiver lineReceiver) {
                 sourcePath = this.mountPoint + "/" + this.subPath.replaceFirst("^/+", "");
             }
 
+            // 防止路径穿越：规范化后校验源路径仍在 mount 点下
+            Path normalizedSourcePath = Paths.get(sourcePath).normalize().toAbsolutePath();
+            Path normalizedMount = Paths.get(this.mountPoint).normalize().toAbsolutePath();
+            if (!normalizedSourcePath.startsWith(normalizedMount)) {
+                LOG.error("Path traversal detected: sourcePath={} outside mountPoint={}",
+                    normalizedSourcePath, normalizedMount);
+                throw DataXException.asDataXException(CommonErrorCode.RUNTIME_ERROR,
+                    "Source path outside mount point: " + sourcePath);
+            }
+
+            // 防止路径穿越：校验目标路径在允许目录内
+            Path normalizedDest = Paths.get(this.destPath).normalize().toAbsolutePath();
+            if (!normalizedDest.startsWith(ALLOWED_DEST_BASE)) {
+                throw DataXException.asDataXException(CommonErrorCode.RUNTIME_ERROR,
+                    "destPath outside allowed directory: " + this.destPath);
+            }
+
             try {
                 Record record;
                 while ((record = lineReceiver.getFromReader()) != null) {
@@ -106,9 +140,15 @@ public void startWrite(RecordReceiver lineReceiver) {
                         continue;
                     }
 
-                    String filePath = sourcePath + "/" + fileName;
+                    // 防止路径穿越：fileName 不能包含路径分隔符或 ..
+                    if (fileName.contains("..") || fileName.contains("/") || fileName.contains("\\")) {
+                        LOG.warn("Skipping file with suspicious name: {}", fileName);
+                        continue;
+                    }
+
+                    String filePath = normalizedSourcePath + "/" + fileName;
                     ShellUtil.runCommand("rsync", Arrays.asList("--no-links", "--chmod=754", "--", filePath,
-                            this.destPath + "/" + fileName));
+                            normalizedDest + "/" + fileName));
                 }
             } catch (Exception e) {
                 LOG.error("Error writing files from GlusterFS: {}", e.getMessage(), e);