[Improvement-17994][Seatunnel] harden startupScript and -i args (#17996)

* [Improvement-17994][Seatunnel] harden startupScript and -i args

* [Improvement-17994][Seatunnel] update docs

* [Improvement-17994][Seatunnel] update docs to 2.3.3

* [Improvement-17994][Seatunnel] add UT

---------

Co-authored-by: zengyi <zengyi@chinatelecom.cn>
Co-authored-by: xiangzihao <460888207@qq.com>

Author: 3 people

docs/docs/en/guide/task/seatunnel.md

@@ -2,7 +2,7 @@
 
 ## Overview
 
-`SeaTunnel` task type for creating and executing `SeaTunnel` tasks. When the worker executes this task, it will parse the config file through the `start-seatunnel-spark.sh` , `start-seatunnel-flink.sh` or `seatunnel.sh` command.
+`SeaTunnel` task type for creating and executing `SeaTunnel` tasks. When the worker executes this task, it will parse and run the config file through the startup scripts under `${SEATUNNEL_HOME}/bin/` (such as `seatunnel.sh` / `start-seatunnel-*-connector-v2.sh`).
 Click [here](https://seatunnel.apache.org/) for more information about `Apache SeaTunnel`.
 
 ## Create Task
@@ -16,7 +16,7 @@ Click [here](https://seatunnel.apache.org/) for more information about `Apache S
 [//]: # (- Please refer to [DolphinScheduler Task Parameters Appendix](appendix.md#default-task-parameters) `Default Task Parameters` section for default parameters.)
 
 - Please refer to [DolphinScheduler Task Parameters Appendix](appendix.md) `Default Task Parameters` section for default parameters.
-- Startup script: Select script name to start the task, including `seatunnel.sh`, `start-seatunnel-flink-13-connector-v2.sh`, `start-seatunnel-flink-15-connector-v2.sh`, `start-seatunnel-flink-connector-v2.sh`, `start-seatunnel-flink.sh`, `start-seatunnel-spark-2-connector-v2.sh`, `start-seatunnel-spark-3-connector-v2.sh`, `start-seatunnel-spark-connector-v2.sh`, `start-seatunnel-spark.sh`
+- Startup script: Select script name to start the task (it may vary across SeaTunnel distributions, please check `${SEATUNNEL_HOME}/bin/`), including `seatunnel.sh`, `start-seatunnel-flink-13-connector-v2.sh`, `start-seatunnel-flink-15-connector-v2.sh`, `start-seatunnel-flink-connector-v2.sh`, `start-seatunnel-flink.sh`, `start-seatunnel-spark-2-connector-v2.sh`, `start-seatunnel-spark-3-connector-v2.sh`, `start-seatunnel-spark-connector-v2.sh`, `start-seatunnel-spark.sh`
 - FLINK
 - Run model: supports `run` and `run-application` modes
 - Option parameters: used to add the parameters of the Flink engine, such as `-m yarn-cluster -ynm seatunnel`
@@ -82,7 +82,7 @@ sink {
 
 ### Support SeaTunnel Version
 
-- v2.3.1
-- v2.3.2
-- v2.3.3
+- The examples in this doc are based on the `2.3.x` CLI options and startup scripts
+- Verified: v2.3.1, v2.3.2, v2.3.3
+- Other versions: this task type is essentially a wrapper of SeaTunnel CLI. Newer versions usually work as long as the startup scripts and CLI options are compatible (please run regression tests after upgrading).
 

docs/docs/zh/guide/task/seatunnel.md

@@ -2,7 +2,7 @@
 
 ## 综述
 
-`SeaTunnel` 任务类型，用于创建并执行 `SeaTunnel` 类型任务。worker 执行该任务的时候，会通过 `start-seatunnel-spark.sh` 、 `start-seatunnel-flink.sh` 和 `seatunnel.sh` 命令解析 config 文件。
+`SeaTunnel` 任务类型，用于创建并执行 `SeaTunnel` 类型任务。worker 执行该任务的时候，会通过 `${SEATUNNEL_HOME}/bin/` 下的启动脚本（如 `seatunnel.sh` / `start-seatunnel-*-connector-v2.sh`）解析并执行 config 文件。
 点击 [这里](https://seatunnel.apache.org/) 获取更多关于 `Apache SeaTunnel` 的信息。
 
 ## 创建任务
@@ -16,7 +16,7 @@
 [//]: # (- 默认参数说明请参考[DolphinScheduler任务参数附录](appendix.md#默认任务参数)`默认任务参数`一栏。)
 
 - 默认参数说明请参考[DolphinScheduler任务参数附录](appendix.md)`默认任务参数`一栏。
-- 启动脚本：选择你想要运行任务的启动脚本，包括 `seatunnel.sh`, `start-seatunnel-flink-13-connector-v2.sh`, `start-seatunnel-flink-15-connector-v2.sh`, `start-seatunnel-flink-connector-v2.sh`, `start-seatunnel-flink.sh`, `start-seatunnel-spark-2-connector-v2.sh`, `start-seatunnel-spark-3-connector-v2.sh`, `start-seatunnel-spark-connector-v2.sh`, `start-seatunnel-spark.sh`
+- 启动脚本：选择你想要运行任务的启动脚本（不同 SeaTunnel 发行包可能存在差异，以实际 `${SEATUNNEL_HOME}/bin/` 为准），包括 `seatunnel.sh`, `start-seatunnel-flink-13-connector-v2.sh`, `start-seatunnel-flink-15-connector-v2.sh`, `start-seatunnel-flink-connector-v2.sh`, `start-seatunnel-flink.sh`, `start-seatunnel-spark-2-connector-v2.sh`, `start-seatunnel-spark-3-connector-v2.sh`, `start-seatunnel-spark-connector-v2.sh`, `start-seatunnel-spark.sh`
 - FLINK
 - 运行模型：支持 `run` 和 `run-application` 两种模式
 - 选项参数：用于添加 Flink 引擎本身参数，例如 `-m yarn-cluster -ynm seatunnel`
@@ -82,7 +82,7 @@ sink {
 
 ### 支持 SeaTunnel 版本
 
-- 2.3.1
-- 2.3.2
-- 2.3.3
+- 文档示例基于 `2.3.x` 版本的命令行参数与启动脚本
+- 已验证：2.3.1、2.3.2、2.3.3
+- 其他版本：该任务类型本质是对 SeaTunnel CLI 的封装，如 SeaTunnel 启动脚本与命令行参数保持兼容，通常可直接使用更高版本（建议升级后先做回归验证）
 

dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/SeatunnelParameters.java

@@ -25,7 +25,7 @@
 import org.apache.commons.lang3.StringUtils;
 
 import java.util.List;
-import java.util.Objects;
+import java.util.regex.Pattern;
 
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@@ -36,6 +36,8 @@
 @NoArgsConstructor
 public class SeatunnelParameters extends AbstractParameters {
 
+    private static final Pattern STARTUP_SCRIPT_PATTERN = Pattern.compile("^[A-Za-z0-9][A-Za-z0-9._-]*\\.sh$");
+
     private String startupScript;
 
     private Boolean useCustom;
@@ -49,12 +51,16 @@ public class SeatunnelParameters extends AbstractParameters {
 
     @Override
     public boolean checkParameters() {
-        return Objects.nonNull(startupScript)
+        return isValidStartupScript(startupScript)
                 && ((BooleanUtils.isTrue(useCustom) && StringUtils.isNotBlank(rawScript))
                         || (BooleanUtils.isFalse(useCustom) && CollectionUtils.isNotEmpty(resourceList)
                                 && resourceList.size() == 1));
     }
 
+    private static boolean isValidStartupScript(String startupScript) {
+        return StringUtils.isNotBlank(startupScript) && STARTUP_SCRIPT_PATTERN.matcher(startupScript).matches();
+    }
+
     @Override
     public List<ResourceInfo> getResourceFilesList() {
         return resourceList;

dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/SeatunnelTask.java

@@ -178,11 +178,19 @@ private List<String> generateTaskParameters() {
         List<String> parameters = new ArrayList<>();
         variables.forEach((k, v) -> {
             parameters.add("-i");
-            parameters.add(String.format("%s='%s'", k, v));
+            parameters.add(String.format("%s=%s", k, quoteForBash(v)));
         });
         return parameters;
     }
 
+    private static String quoteForBash(String value) {
+        if (value == null) {
+            return "''";
+        }
+        // Escape single quotes in a bash-safe way: abc'def -> 'abc'"'"'def'
+        return "'" + value.replace("'", "'\"'\"'") + "'";
+    }
+
     private String buildCustomConfigContent() {
         log.info("raw custom config content : {}", seatunnelParameters.getRawScript());
         String script = seatunnelParameters.getRawScript().replaceAll("\\r\\n", System.lineSeparator());

dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/test/java/org/apache/dolphinscheduler/plugin/task/seatunnel/SeatunnelParametersTest.java

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.plugin.task.seatunnel;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class SeatunnelParametersTest {
+
+    @Test
+    public void testInvalidStartupScript() {
+        SeatunnelParameters seatunnelParameters = new SeatunnelParameters();
+        seatunnelParameters.setUseCustom(true);
+        seatunnelParameters.setRawScript("env { execution.parallelism = 1 }");
+
+        seatunnelParameters.setStartupScript("../../../etc/passwd");
+        Assertions.assertFalse(seatunnelParameters.checkParameters());
+
+        seatunnelParameters.setStartupScript("script.sh; rm -rf /");
+        Assertions.assertFalse(seatunnelParameters.checkParameters());
+
+        seatunnelParameters.setStartupScript("script");
+        Assertions.assertFalse(seatunnelParameters.checkParameters());
+
+        seatunnelParameters.setStartupScript(".sh");
+        Assertions.assertFalse(seatunnelParameters.checkParameters());
+    }
+}

dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/test/java/org/apache/dolphinscheduler/plugin/task/seatunnel/SeatunnelTaskTest.java

@@ -27,6 +27,7 @@
 
 import org.apache.commons.io.FileUtils;
 
+import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -133,6 +134,20 @@ public void testParameterPass() throws Exception {
         Assertions.assertEquals(expectedCommand, command);
     }
 
+    @Test
+    public void testQuoteForBash() throws Exception {
+        Assertions.assertEquals("'value'", invokeQuoteForBash("value"));
+        Assertions.assertEquals("'abc'\"'\"'def'", invokeQuoteForBash("abc'def"));
+        Assertions.assertEquals("''", invokeQuoteForBash(null));
+        Assertions.assertEquals("'$(rm -rf /)'", invokeQuoteForBash("$(rm -rf /)"));
+    }
+
+    private String invokeQuoteForBash(String value) throws Exception {
+        Method quoteForBash = SeatunnelTask.class.getDeclaredMethod("quoteForBash", String.class);
+        quoteForBash.setAccessible(true);
+        return (String) quoteForBash.invoke(null, value);
+    }
+
     private static final String RAW_SCRIPT = "env {\n" +
             "  execution.parallelism = 2\n" +
             "  job.mode = \"BATCH\"\n" +