[DSIP-105][API] Encrypt sensitive definition params with PasswordUtils

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowDefinitionServiceImpl.java

@@ -265,7 +265,8 @@ public WorkflowDefinition createWorkflowDefinition(User loginUser,
 
         List<TaskDefinitionLog> taskDefinitionLogs = generateTaskDefinitionList(taskDefinitionJson);
         List<WorkflowTaskRelationLog> taskRelationList = generateTaskRelationList(taskRelationJson, taskDefinitionLogs);
-        SensitivePropertyUtils.requireNoPlaceholder(GlobalParameterUtils.deserializeGlobalParameter(globalParams));
+        List<Property> submittedGlobalParams = GlobalParameterUtils.deserializeGlobalParameter(globalParams);
+        SensitivePropertyUtils.requireNoPlaceholder(submittedGlobalParams);
         for (TaskDefinitionLog taskDefinitionLog : CollectionUtils.emptyIfNull(taskDefinitionLogs)) {
             taskDefinitionLog.setTaskParams(
                     SensitivePropertyUtils.mergeLocalParams(taskDefinitionLog.getTaskParams(), null));
@@ -277,15 +278,44 @@ public WorkflowDefinition createWorkflowDefinition(User loginUser,
                         globalParams, locations, timeout, loginUser.getId());
         workflowDefinition.setExecutionType(executionType);
 
-        return createDagDefine(loginUser, taskRelationList, workflowDefinition, taskDefinitionLogs);
+        // encodeSensitiveParams=true: plaintext through checks, encode inside createDagDefine before save*.
+        return createDagDefine(loginUser, taskRelationList, workflowDefinition, taskDefinitionLogs, true);
+    }
+
+    /** Persist-boundary: encode task localParams immediately before {@code saveTaskDefine}. */
+    private void encodeTaskParamsForPersist(List<TaskDefinitionLog> taskDefinitionLogs) {
+        for (TaskDefinitionLog taskDefinitionLog : CollectionUtils.emptyIfNull(taskDefinitionLogs)) {
+            taskDefinitionLog.setTaskParams(
+                    SensitivePropertyUtils.encodeLocalParamsInTaskParams(taskDefinitionLog.getTaskParams()));
+        }
+    }
+
+    /** Persist-boundary: encode workflow globalParams immediately before {@code saveWorkflowDefine}. */
+    private void encodeGlobalParamsForPersist(WorkflowDefinition workflowDefinition) {
+        if (workflowDefinition != null && StringUtils.isNotEmpty(workflowDefinition.getGlobalParams())) {
+            workflowDefinition.setGlobalParams(
+                    SensitivePropertyUtils.encodeGlobalParams(workflowDefinition.getGlobalParams()));
+        }
     }
 
     protected WorkflowDefinition createDagDefine(User loginUser,
                                                  List<WorkflowTaskRelationLog> taskRelationList,
                                                  WorkflowDefinition workflowDefinition,
                                                  List<TaskDefinitionLog> taskDefinitionLogs) {
+        // Copy / import paths already persist ciphertext — do not encode again.
+        return createDagDefine(loginUser, taskRelationList, workflowDefinition, taskDefinitionLogs, false);
+    }
+
+    protected WorkflowDefinition createDagDefine(User loginUser,
+                                                 List<WorkflowTaskRelationLog> taskRelationList,
+                                                 WorkflowDefinition workflowDefinition,
+                                                 List<TaskDefinitionLog> taskDefinitionLogs,
+                                                 boolean encodeSensitiveParams) {
         taskDatasourcePermissionChecker.checkPermission(loginUser, taskDefinitionLogs);
         taskSubWorkflowPermissionChecker.checkPermission(loginUser, taskDefinitionLogs);
+        if (encodeSensitiveParams) {
+            encodeTaskParamsForPersist(taskDefinitionLogs);
+        }
         int saveTaskResult = processService.saveTaskDefine(loginUser, workflowDefinition.getProjectCode(),
                 taskDefinitionLogs, Boolean.TRUE);
         if (saveTaskResult == Constants.EXIT_CODE_SUCCESS) {
@@ -295,6 +325,9 @@ protected WorkflowDefinition createDagDefine(User loginUser,
             log.error("Save task definition error.");
             throw new ServiceException(Status.CREATE_TASK_DEFINITION_ERROR);
         }
+        if (encodeSensitiveParams) {
+            encodeGlobalParamsForPersist(workflowDefinition);
+        }
         int insertVersion =
                 processService.saveWorkflowDefine(loginUser, workflowDefinition, Boolean.TRUE, Boolean.TRUE);
         if (insertVersion == 0) {
@@ -690,9 +723,12 @@ public WorkflowDefinition updateWorkflowDefinition(User loginUser,
         List<Property> submittedGlobalParams =
                 GlobalParameterUtils.deserializeGlobalParameter(globalParams);
         if (CollectionUtils.isNotEmpty(submittedGlobalParams)) {
+            List<Property> existingRaw =
+                    GlobalParameterUtils.deserializeGlobalParameter(workflowDefinition.getGlobalParams());
+            List<Property> merged = SensitivePropertyUtils.merge(submittedGlobalParams,
+                    SensitivePropertyUtils.decodeAfterLoad(existingRaw));
             globalParams = GlobalParameterUtils.serializeGlobalParameter(
-                    SensitivePropertyUtils.merge(submittedGlobalParams,
-                            GlobalParameterUtils.deserializeGlobalParameter(workflowDefinition.getGlobalParams())));
+                    SensitivePropertyUtils.normalizeResubmittedCiphertext(merged, existingRaw));
         }
         List<TaskDefinition> versionKeys = new ArrayList<>();
         for (TaskDefinitionLog taskDefinitionLog : CollectionUtils.emptyIfNull(taskDefinitionLogs)) {
@@ -712,6 +748,7 @@ public WorkflowDefinition updateWorkflowDefinition(User loginUser,
             String existingTaskParams = submitted.getCode() <= 0 || submitted.getVersion() <= 0
                     ? null
                     : existingTaskParamsMap.get(submitted.getCode() + "_" + submitted.getVersion());
+            // Keep plaintext; encode inside updateDagDefine immediately before save*.
             submitted.setTaskParams(SensitivePropertyUtils.mergeLocalParams(
                     submitted.getTaskParams(), existingTaskParams));
         }
@@ -720,7 +757,7 @@ public WorkflowDefinition updateWorkflowDefinition(User loginUser,
         workflowDefinition.set(projectCode, name, description, globalParams, locations, timeout);
         workflowDefinition.setExecutionType(executionType);
         return updateDagDefine(loginUser, taskRelationList, workflowDefinition, workflowDefinitionDeepCopy,
-                taskDefinitionLogs);
+                taskDefinitionLogs, true);
     }
 
     /**
@@ -760,8 +797,22 @@ protected WorkflowDefinition updateDagDefine(User loginUser,
                                                  WorkflowDefinition workflowDefinition,
                                                  WorkflowDefinition workflowDefinitionDeepCopy,
                                                  List<TaskDefinitionLog> taskDefinitionLogs) {
+        // Move path already has ciphertext — do not encode again.
+        return updateDagDefine(loginUser, taskRelationList, workflowDefinition, workflowDefinitionDeepCopy,
+                taskDefinitionLogs, false);
+    }
+
+    protected WorkflowDefinition updateDagDefine(User loginUser,
+                                                 List<WorkflowTaskRelationLog> taskRelationList,
+                                                 WorkflowDefinition workflowDefinition,
+                                                 WorkflowDefinition workflowDefinitionDeepCopy,
+                                                 List<TaskDefinitionLog> taskDefinitionLogs,
+                                                 boolean encodeSensitiveParams) {
         taskDatasourcePermissionChecker.checkPermission(loginUser, taskDefinitionLogs);
         taskSubWorkflowPermissionChecker.checkPermission(loginUser, taskDefinitionLogs);
+        if (encodeSensitiveParams) {
+            encodeTaskParamsForPersist(taskDefinitionLogs);
+        }
         int saveTaskResult = processService.saveTaskDefine(loginUser, workflowDefinition.getProjectCode(),
                 taskDefinitionLogs, Boolean.TRUE);
         if (saveTaskResult == Constants.EXIT_CODE_SUCCESS) {
@@ -772,6 +823,10 @@ protected WorkflowDefinition updateDagDefine(User loginUser,
                     workflowDefinition.getProjectCode(), workflowDefinition.getCode());
             throw new ServiceException(Status.UPDATE_TASK_DEFINITION_ERROR);
         }
+        // Encode globals before equals/save so comparison matches DB ciphertext (deepCopy).
+        if (encodeSensitiveParams) {
+            encodeGlobalParamsForPersist(workflowDefinition);
+        }
         boolean isChange = false;
         if (workflowDefinition.equals(workflowDefinitionDeepCopy) && saveTaskResult == Constants.EXIT_CODE_SUCCESS) {
             List<WorkflowTaskRelationLog> workflowTaskRelationLogList = workflowTaskRelationLogMapper

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowInstanceServiceImpl.java

@@ -413,12 +413,13 @@ public WorkflowDefinition updateWorkflowInstance(User loginUser, long projectCod
         }
 
         List<Property> submittedGlobalParams = GlobalParameterUtils.deserializeGlobalParameter(globalParams);
+        String instanceGlobalParams = globalParams;
         if (CollectionUtils.isNotEmpty(submittedGlobalParams)) {
-            globalParams = GlobalParameterUtils.serializeGlobalParameter(
+            instanceGlobalParams = GlobalParameterUtils.serializeGlobalParameter(
                     SensitivePropertyUtils.merge(submittedGlobalParams,
                             GlobalParameterUtils.deserializeGlobalParameter(workflowInstance.getGlobalParams())));
         }
-        setWorkflowInstance(workflowInstance, scheduleTime, globalParams, timeout, timezoneId);
+        setWorkflowInstance(workflowInstance, scheduleTime, instanceGlobalParams, timeout, timezoneId);
         List<TaskDefinitionLog> taskDefinitionLogs = JSONUtils.toList(taskDefinitionJson, TaskDefinitionLog.class);
         if (taskDefinitionLogs.isEmpty()) {
             log.warn("Parameter taskDefinitionJson is empty");
@@ -448,27 +449,44 @@ public WorkflowDefinition updateWorkflowInstance(User loginUser, long projectCod
             String existingTaskParams = submitted.getCode() <= 0 || submitted.getVersion() <= 0
                     ? null
                     : existingTaskParamsMap.get(submitted.getCode() + "_" + submitted.getVersion());
+            // Keep plaintext through permission / DAG checks; encode only at each save* below.
             submitted.setTaskParams(SensitivePropertyUtils.mergeLocalParams(
                     submitted.getTaskParams(), existingTaskParams));
         }
         taskDatasourcePermissionChecker.checkPermission(loginUser, taskDefinitionLogs);
         taskSubWorkflowPermissionChecker.checkPermission(loginUser, taskDefinitionLogs);
-        int saveTaskResult = processService.saveTaskDefine(loginUser, projectCode, taskDefinitionLogs, syncDefine);
-        if (saveTaskResult == Constants.DEFINITION_FAILURE) {
-            log.error("Update task definition error, projectCode:{}, workflowInstanceId:{}", projectCode,
-                    workflowInstanceId);
-            throw new ServiceException(Status.UPDATE_TASK_DEFINITION_ERROR);
-        }
-        WorkflowDefinition workflowDefinition =
-                workflowDefinitionDao.queryByCode(workflowInstance.getWorkflowDefinitionCode()).orElse(null);
         List<WorkflowTaskRelationLog> taskRelationList =
                 JSONUtils.toList(taskRelationJson, WorkflowTaskRelationLog.class);
-        // check workflow json is valid (throws ServiceException on validation failures)
+        // Validate on plaintext before any definition persist.
         workflowDefinitionService.checkWorkflowNodeList(taskRelationJson, taskDefinitionLogs);
 
+        WorkflowDefinition workflowDefinition =
+                workflowDefinitionDao.queryByCode(workflowInstance.getWorkflowDefinitionCode()).orElse(null);
+        // Definition snapshot uses instance-resolved globals (****** already merged); do not merge again
+        // against the latest definition, or syncDefine would diverge when instance overrode a secret.
+        String definitionGlobalParams = CollectionUtils.isNotEmpty(submittedGlobalParams)
+                ? instanceGlobalParams
+                : globalParams;
         workflowDefinition.set(projectCode, workflowDefinition.getName(), workflowDefinition.getDescription(),
-                globalParams, locations, timeout);
+                definitionGlobalParams, locations, timeout);
         workflowDefinition.setUpdateTime(new Date());
+
+        // Persist boundary: encode definition snapshots immediately before each DB write.
+        // Instance global_params stay plaintext; only the definition copy is encoded here.
+        for (TaskDefinitionLog submitted : taskDefinitionLogs) {
+            submitted.setTaskParams(
+                    SensitivePropertyUtils.encodeLocalParamsInTaskParams(submitted.getTaskParams()));
+        }
+        int saveTaskResult = processService.saveTaskDefine(loginUser, projectCode, taskDefinitionLogs, syncDefine);
+        if (saveTaskResult == Constants.DEFINITION_FAILURE) {
+            log.error("Update task definition error, projectCode:{}, workflowInstanceId:{}", projectCode,
+                    workflowInstanceId);
+            throw new ServiceException(Status.UPDATE_TASK_DEFINITION_ERROR);
+        }
+        if (CollectionUtils.isNotEmpty(submittedGlobalParams)) {
+            workflowDefinition.setGlobalParams(
+                    SensitivePropertyUtils.encodeGlobalParams(workflowDefinition.getGlobalParams()));
+        }
         int insertVersion = processService.saveWorkflowDefine(loginUser, workflowDefinition, syncDefine, Boolean.FALSE);
         if (insertVersion == 0) {
             log.error("Update workflow definition error, projectCode:{}, workflowDefinitionName:{}.", projectCode,