Merge branch 'dev' into dev-datavines

Author: zixi0825

docs/docs/en/contribute/join/security-model.md

@@ -87,4 +87,8 @@ The following are some erroneous vulnerabilities raised by users and developers
    There are multiple input boxes in Apache DolphinScheduler, allowing users to customize configurations as needed. As an open source task scheduling system, Apache DolphinScheduler requires administrators to fully trust all authorized operations of the target user in the process of deployment, authorization, and other security-related processes. If the user's behavior of adding and modifying configurations through pages or calling interfaces is within the scope of permissions, then the behavior of attacking or other operations in this way does not belong to security vulnerabilities.
 5. Attack or other operations by modifying the image or providing an unsafe image to run
    Apache DolphinScheduler itself and task operations both support k8s clusters. Before the service or task runs, the user needs to ensure the image's functions and configured parameters, and trust all operations during the service and task running process. Therefore, modifying tasks or parameters by any means before the image runs to attack or complete other operations does not constitute a security vulnerability.
+6. Attacks by obtaining certain sensitive information printed in service logs
+   Apache DolphinScheduler prints some sensitive information in its service logs, which can be used by service deployers to view detailed information about the program's operation. Service deployers are considered trusted users, and we do not believe that service deployers will attack the program, so this type of issue is not a vulnerability.
+7. Security problems caused by system administrators accessing untrusted third-party websites
+   System administrators using Apache DolphinScheduler may access untrusted third-party websites, resulting in system attacks; such issues are not considered security vulnerabilities. System administrators are considered to be trusted users, and we believe that system administrators have a basic awareness of security precautions. Problems caused by weak security precautions on the part of system administrators are not considered vulnerabilities.
 

docs/docs/zh/contribute/join/security-model.md

@@ -85,3 +85,7 @@ Apache DolphinScheduler的部署和使用认为用户网络是安全和值得信
 在Apache DolphinScheduler有多个输入框，允许用户按需进行自定义配置，Apache DolphinScheduler作为开源任务调度系统，管理员用户在部署、授权等涉及安全的过程中需要完全信任目标用户的所有授权范围内的操作，用户通过页面或调用接口增加、修改配置的行为如果属于权限范围内的操作，则通过该方式进行攻击或其他操作的行为不属于安全漏洞。
 5.通过修改镜像或提供不安全的镜像运行进行攻击或其他操作
 Apache DolphinScheduler的本身和任务运行均支持k8s集群，在服务或任务运行之前，用户需要确保镜像的功能和所配置参数，信任服务、任务运行过程中的所有操作。所以在镜像运行之前通过任何途径将任务或参数进行修改进行攻击或其他完成其他操作的行为不属于安全漏洞。
+6.通过获取服务日志中打印的某些敏感信息进行攻击
+Apache DolphinScheduler的服务日志中会打印部分敏感信息，服务部署人员可以通过日志查看程序运行的详细信息。服务部署人员被认为是可信任的用户，我们认为服务部署人员不会攻击程序，所以该类型问题不属于漏洞。
+7.系统管理员通过访问不受信任的三方网站导致的安全问题
+系统管理员在使用Apache DolphinScheduler的过程中，可能会访问不受信任的三方网站，导致系统被攻击，这种问题不属于安全漏洞。系统管理员被认为是可信任的用户，我们认为系统管理员具备基本的安全防范意识，由于系统管理员的安全防范意识薄弱所引发的问题不属于漏洞。

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/RecoverFailureTaskCommandHandler.java

@@ -25,6 +25,7 @@
 import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
 import org.apache.dolphinscheduler.dao.repository.WorkflowInstanceDao;
 import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
+import org.apache.dolphinscheduler.server.master.config.MasterConfig;
 import org.apache.dolphinscheduler.server.master.engine.graph.IWorkflowGraph;
 import org.apache.dolphinscheduler.server.master.engine.graph.WorkflowExecutionGraph;
 import org.apache.dolphinscheduler.server.master.engine.graph.WorkflowGraphTopologyLogicalVisitor;
@@ -67,6 +68,9 @@ public class RecoverFailureTaskCommandHandler extends AbstractCommandHandler {
     @Autowired
     private TaskInstanceFactories taskInstanceFactories;
 
+    @Autowired
+    private MasterConfig masterConfig;
+
     /**
      * Generate the recover workflow instance.
      * <p> Will use the origin workflow instance, but will update the following fields. Need to note we cannot not
@@ -90,6 +94,7 @@ protected void assembleWorkflowInstance(
         workflowInstance.setVarPool(null);
         workflowInstance.setStateWithDesc(WorkflowExecutionStatus.RUNNING_EXECUTION, command.getCommandType().name());
         workflowInstance.setCommandType(command.getCommandType());
+        workflowInstance.setHost(masterConfig.getMasterAddress());
         workflowInstanceDao.updateById(workflowInstance);
 
         workflowExecuteContextBuilder.setWorkflowInstance(workflowInstance);

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/handler/WorkflowFailoverCommandHandler.java

@@ -24,6 +24,7 @@
 import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
 import org.apache.dolphinscheduler.dao.repository.WorkflowInstanceDao;
 import org.apache.dolphinscheduler.extract.master.command.WorkflowFailoverCommandParam;
+import org.apache.dolphinscheduler.server.master.config.MasterConfig;
 import org.apache.dolphinscheduler.server.master.engine.ITaskGroupCoordinator;
 import org.apache.dolphinscheduler.server.master.engine.graph.IWorkflowGraph;
 import org.apache.dolphinscheduler.server.master.engine.graph.WorkflowExecutionGraph;
@@ -58,6 +59,9 @@ public class WorkflowFailoverCommandHandler extends AbstractCommandHandler {
     @Autowired
     private ApplicationContext applicationContext;
 
+    @Autowired
+    private MasterConfig masterConfig;
+
     /**
      * Generate the recover workflow instance.
      * <p> Will use the origin workflow instance, but will update the following fields. Need to note we cannot not
@@ -86,6 +90,7 @@ protected void assembleWorkflowInstance(
                     "The WorkflowFailoverCommandParam: " + command.getCommandParam() + " is invalid");
         }
         workflowInstance.setState(workflowFailoverCommandParam.getWorkflowExecutionStatus());
+        workflowInstance.setHost(masterConfig.getMasterAddress());
         workflowInstanceDao.updateById(workflowInstance);
 
         workflowExecuteContextBuilder.setWorkflowInstance(workflowInstance);

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/AbstractMasterIntegrationTestCase.java

@@ -18,6 +18,8 @@
 package org.apache.dolphinscheduler.server.master;
 
 import org.apache.dolphinscheduler.dao.DaoConfiguration;
+import org.apache.dolphinscheduler.registry.api.RegistryClient;
+import org.apache.dolphinscheduler.server.master.config.MasterConfig;
 import org.apache.dolphinscheduler.server.master.integration.MasterContainer;
 import org.apache.dolphinscheduler.server.master.integration.Repository;
 import org.apache.dolphinscheduler.server.master.integration.WorkflowOperator;
@@ -52,4 +54,10 @@ public abstract class AbstractMasterIntegrationTestCase {
 
     @Autowired
     protected MasterContainer masterContainer;
+
+    @Autowired
+    protected RegistryClient registryClient;
+
+    @Autowired
+    protected MasterConfig masterConfig;
 }

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowInstanceFailoverTestCase.java

@@ -24,7 +24,12 @@
 import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
+import org.apache.dolphinscheduler.extract.master.IWorkflowControlClient;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopResponse;
 import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
+import org.apache.dolphinscheduler.registry.api.utils.RegistryUtils;
 import org.apache.dolphinscheduler.server.master.AbstractMasterIntegrationTestCase;
 import org.apache.dolphinscheduler.server.master.engine.system.SystemEventBus;
 import org.apache.dolphinscheduler.server.master.engine.system.event.GlobalMasterFailoverEvent;
@@ -555,4 +560,59 @@ public void testGlobalFailover_readyStopWorkflow_withKilledTasks() {
         masterContainer.assertAllResourceReleased();
     }
 
+    @Test
+    public void testGlobalFailover_runningWorkflow_fromAnotherMaster() {
+        final String yaml = "/it/failover/running_workflowInstance_from_another_master.yaml";
+        final WorkflowTestCaseContext context = workflowTestCaseContextFactory.initializeContextFromYaml(yaml);
+        final WorkflowDefinition workflow = context.getOneWorkflow();
+
+        systemEventBus.publish(GlobalMasterFailoverEvent.of(new Date()));
+
+        final String masterFailoverNodePath = RegistryUtils.getFailoveredNodePathWhichStartupTimeIsUnknown(
+                "127.0.0.1:15678");
+        // wait failover process
+        await()
+                .atMost(Duration.ofMinutes(3))
+                .untilAsserted(() -> {
+                    assertThat(registryClient.exists(masterFailoverNodePath)).isTrue();
+                });
+
+        // check workflow's status and can stop it
+        await()
+                .atMost(Duration.ofMinutes(1))
+                .untilAsserted(() -> {
+                    assertThat(repository.queryWorkflowInstance(workflow))
+                            .hasSize(1)
+                            .anySatisfy(workflowInstance -> {
+                                assertThat(workflowInstance.getState())
+                                        .isEqualTo(WorkflowExecutionStatus.RUNNING_EXECUTION);
+                                assertThat(workflowInstance.getName())
+                                        .isEqualTo("workflow_with_one_fake_task_running-20250322201900000");
+
+                                final WorkflowInstanceStopResponse stopResponse = Clients
+                                        .withService(IWorkflowControlClient.class)
+                                        .withHost(workflowInstance.getHost())
+                                        .stopWorkflowInstance(
+                                                new WorkflowInstanceStopRequest(workflowInstance.getId()));
+
+                                assertThat((stopResponse != null && stopResponse.isSuccess())).isTrue();
+                            });
+                });
+
+        await()
+                .atMost(Duration.ofMinutes(1))
+                .untilAsserted(() -> {
+                    assertThat(repository.queryWorkflowInstance(workflow))
+                            .hasSize(1)
+                            .anySatisfy(workflowInstance -> {
+                                assertThat(workflowInstance.getState())
+                                        .isEqualTo(WorkflowExecutionStatus.STOP);
+                                assertThat(workflowInstance.getName())
+                                        .isEqualTo("workflow_with_one_fake_task_running-20250322201900000");
+                            });
+                });
+
+        masterContainer.assertAllResourceReleased();
+
+    }
 }

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowInstanceRecoverFailureTaskTestCase.java

@@ -23,7 +23,12 @@
 import org.apache.dolphinscheduler.common.enums.Flag;
 import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
 import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
+import org.apache.dolphinscheduler.extract.master.IWorkflowControlClient;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopResponse;
 import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
 import org.apache.dolphinscheduler.server.master.AbstractMasterIntegrationTestCase;
 import org.apache.dolphinscheduler.server.master.integration.WorkflowTestCaseContext;
@@ -89,4 +94,50 @@ public void testRepeatRunningWorkflow_with_taskOnly() {
         masterContainer.assertAllResourceReleased();
     }
 
+    @Test
+    @DisplayName("Test recover a failure workflow from another master")
+    public void testRecoverFailureWorkflow_from_another_master() {
+        final String yaml = "/it/recover_failure_tasks/failure_workflow_from_another_master.yaml";
+        final WorkflowTestCaseContext context = workflowTestCaseContextFactory.initializeContextFromYaml(yaml);
+        final WorkflowDefinition workflow = context.getOneWorkflow();
+
+        final Integer workflowInstanceId = context.getWorkflowInstance().getId();
+        workflowOperator.recoverFailureTasks(workflowInstanceId);
+
+        await()
+                .atMost(Duration.ofMinutes(1))
+                .untilAsserted(() -> {
+                    assertThat(repository.queryWorkflowInstance(workflow))
+                            .hasSize(1)
+                            .anySatisfy(workflowInstance -> {
+                                assertThat(workflowInstance.getState())
+                                        .isEqualTo(WorkflowExecutionStatus.RUNNING_EXECUTION);
+                                assertThat(workflowInstance.getName())
+                                        .isEqualTo("workflow_with_one_fake_task_killed-20250322201900000");
+
+                                final WorkflowInstanceStopResponse stopResponse = Clients
+                                        .withService(IWorkflowControlClient.class)
+                                        .withHost(workflowInstance.getHost())
+                                        .stopWorkflowInstance(
+                                                new WorkflowInstanceStopRequest(workflowInstance.getId()));
+
+                                assertThat(stopResponse != null && stopResponse.isSuccess()).isTrue();
+                            });
+                });
+
+        await()
+                .atMost(Duration.ofMinutes(1))
+                .untilAsserted(() -> {
+                    assertThat(repository.queryWorkflowInstance(workflow))
+                            .hasSize(1)
+                            .anySatisfy(workflowInstance -> {
+                                assertThat(workflowInstance.getState())
+                                        .isEqualTo(WorkflowExecutionStatus.STOP);
+                                assertThat(workflowInstance.getName())
+                                        .isEqualTo("workflow_with_one_fake_task_killed-20250322201900000");
+                            });
+                });
+
+        masterContainer.assertAllResourceReleased();
+    }
 }

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowInstanceRecoverStopTestCase.java

@@ -24,7 +24,11 @@
 import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
 import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
 import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
+import org.apache.dolphinscheduler.extract.master.IWorkflowControlClient;
 import org.apache.dolphinscheduler.extract.master.command.RunWorkflowCommandParam;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopRequest;
+import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopResponse;
 import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
 import org.apache.dolphinscheduler.server.master.AbstractMasterIntegrationTestCase;
 import org.apache.dolphinscheduler.server.master.integration.WorkflowOperator;
@@ -121,4 +125,50 @@ public void testRecoverStoppedWorkflow_with_subWorkflowTask_success() {
         masterContainer.assertAllResourceReleased();
     }
 
+    @Test
+    @DisplayName("Test recover a stopped workflow from another master")
+    public void testRecoverStoppedWorkflow_from_another_master() {
+        final String yaml = "/it/recover_stopped/stopped_workflow_from_another_master.yaml";
+        final WorkflowTestCaseContext context = workflowTestCaseContextFactory.initializeContextFromYaml(yaml);
+        final WorkflowDefinition workflow = context.getOneWorkflow();
+
+        final Integer workflowInstanceId = context.getWorkflowInstance().getId();
+        assertThat(workflowOperator.recoverSuspendWorkflowInstance(workflowInstanceId).isSuccess()).isTrue();
+
+        await()
+                .atMost(Duration.ofMinutes(1))
+                .untilAsserted(() -> {
+                    assertThat(repository.queryWorkflowInstance(workflow))
+                            .hasSize(1)
+                            .anySatisfy(workflowInstance -> {
+                                assertThat(workflowInstance.getState())
+                                        .isEqualTo(WorkflowExecutionStatus.RUNNING_EXECUTION);
+                                assertThat(workflowInstance.getName())
+                                        .isEqualTo("workflow_with_one_fake_task_killed-20250322201900000");
+
+                                final WorkflowInstanceStopResponse stopResponse = Clients
+                                        .withService(IWorkflowControlClient.class)
+                                        .withHost(workflowInstance.getHost())
+                                        .stopWorkflowInstance(
+                                                new WorkflowInstanceStopRequest(workflowInstance.getId()));
+
+                                assertThat(stopResponse != null && stopResponse.isSuccess()).isTrue();
+                            });
+                });
+
+        await()
+                .atMost(Duration.ofMinutes(1))
+                .untilAsserted(() -> {
+                    assertThat(repository.queryWorkflowInstance(workflow))
+                            .hasSize(1)
+                            .anySatisfy(workflowInstance -> {
+                                assertThat(workflowInstance.getState())
+                                        .isEqualTo(WorkflowExecutionStatus.STOP);
+                                assertThat(workflowInstance.getName())
+                                        .isEqualTo("workflow_with_one_fake_task_killed-20250322201900000");
+                            });
+                });
+
+        masterContainer.assertAllResourceReleased();
+    }
 }