[Fix][Core] Compatible with key encryption and decryption repeated operations

seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/command/AbstractCommandArgs.java

@@ -19,7 +19,9 @@
 
 import org.apache.seatunnel.common.Constants;
 import org.apache.seatunnel.common.config.DeployMode;
+import org.apache.seatunnel.core.starter.enums.CryptoMode;
 
+import com.beust.jcommander.IStringConverter;
 import com.beust.jcommander.Parameter;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
@@ -64,14 +66,36 @@ public abstract class AbstractCommandArgs extends CommandArgs {
     @Parameter(
             names = {"--encrypt"},
             description =
-                    "Encrypt config file, when both --decrypt and --encrypt are specified, only --encrypt will take effect")
+                    "Enable encryption. If set, encryption is applied with default rule unless --crypto-mode is specified.",
+            arity = 0)
     protected boolean encrypt = false;
 
     @Parameter(
             names = {"--decrypt"},
             description =
-                    "Decrypt config file, When both --decrypt and --encrypt are specified, only --encrypt will take effect")
+                    "Enable decryption. If set, decryption is applied with default rule unless --crypto-mode is specified.",
+            arity = 0)
     protected boolean decrypt = false;
 
+    @Parameter(
+            names = {"--crypto-mode"},
+            description =
+                    "Encryption/Decryption mode: 'default' or 'legacy'. When provided, the specified mode is applied.",
+            converter = EncryptionModeConverter.class)
+    protected CryptoMode cryptoMode = CryptoMode.DEFAULT;
+
     public abstract DeployMode getDeployMode();
+
+    /** Custom converter for --encrypt and --decrypt parameters. */
+    public static class EncryptionModeConverter implements IStringConverter<CryptoMode> {
+        @Override
+        public CryptoMode convert(String value) {
+            // If no value is provided, treat it as DEFAULT
+            if (value == null || value.isEmpty()) {
+                return CryptoMode.DEFAULT;
+            }
+            // Otherwise, parse the value as a string and convert to EncryptionMode
+            return CryptoMode.fromValue(value);
+        }
+    }
 }

seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/command/ConfDecryptCommand.java

@@ -22,6 +22,7 @@
 import org.apache.seatunnel.shade.com.typesafe.config.ConfigRenderOptions;
 import org.apache.seatunnel.shade.com.typesafe.config.ConfigResolveOptions;
 
+import org.apache.seatunnel.core.starter.enums.CryptoMode;
 import org.apache.seatunnel.core.starter.exception.CommandExecuteException;
 import org.apache.seatunnel.core.starter.exception.ConfigCheckException;
 import org.apache.seatunnel.core.starter.utils.ConfigShadeUtils;
@@ -45,6 +46,7 @@ public ConfDecryptCommand(AbstractCommandArgs abstractCommandArgs) {
     @Override
     public void execute() throws CommandExecuteException, ConfigCheckException {
         String decryptConfigFile = abstractCommandArgs.getConfigFile();
+        CryptoMode encryptMode = abstractCommandArgs.getCryptoMode();
         Path configPath = Paths.get(decryptConfigFile);
         checkConfigExist(configPath);
         Config config =
@@ -53,7 +55,7 @@ public void execute() throws CommandExecuteException, ConfigCheckException {
                         .resolveWith(
                                 ConfigFactory.systemProperties(),
                                 ConfigResolveOptions.defaults().setAllowUnresolved(true));
-        Config decryptConfig = ConfigShadeUtils.decryptConfig(config);
+        Config decryptConfig = ConfigShadeUtils.decryptConfig(config, encryptMode);
         log.info(
                 "Decrypt config: \n{}",
                 decryptConfig

seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/command/ConfEncryptCommand.java

@@ -22,6 +22,7 @@
 import org.apache.seatunnel.shade.com.typesafe.config.ConfigRenderOptions;
 import org.apache.seatunnel.shade.com.typesafe.config.ConfigResolveOptions;
 
+import org.apache.seatunnel.core.starter.enums.CryptoMode;
 import org.apache.seatunnel.core.starter.exception.CommandExecuteException;
 import org.apache.seatunnel.core.starter.exception.ConfigCheckException;
 import org.apache.seatunnel.core.starter.utils.ConfigShadeUtils;
@@ -50,6 +51,7 @@ public void execute() throws CommandExecuteException, ConfigCheckException {
                     "When both --decrypt and --encrypt are specified, only --encrypt will take effect");
         }
         String encryptConfigFile = abstractCommandArgs.getConfigFile();
+        CryptoMode encryptMode = abstractCommandArgs.getCryptoMode();
         Path configPath = Paths.get(encryptConfigFile);
         checkConfigExist(configPath);
         Config config =
@@ -66,7 +68,7 @@ public void execute() throws CommandExecuteException, ConfigCheckException {
                             ConfigFactory.systemProperties(),
                             ConfigResolveOptions.defaults().setAllowUnresolved(true));
         }
-        Config encryptConfig = ConfigShadeUtils.encryptConfig(config);
+        Config encryptConfig = ConfigShadeUtils.encryptConfig(config, encryptMode);
         log.info(
                 "Encrypt config: \n{}",
                 encryptConfig

seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/enums/CryptoMode.java

@@ -0,0 +1,38 @@
+/*
+ * 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.seatunnel.core.starter.enums;
+
+public enum CryptoMode {
+    DEFAULT("default"),
+    LEGACY("legacy");
+
+    private final String value;
+
+    CryptoMode(String value) {
+        this.value = value;
+    }
+
+    public static CryptoMode fromValue(String value) {
+        for (CryptoMode mode : values()) {
+            if (mode.value.equalsIgnoreCase(value)) {
+                return mode;
+            }
+        }
+        throw new IllegalArgumentException("Invalid encryption mode value: " + value);
+    }
+}

seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigBuilder.java

@@ -29,6 +29,7 @@
 import org.apache.seatunnel.api.sink.TablePlaceholder;
 import org.apache.seatunnel.common.utils.JsonUtils;
 import org.apache.seatunnel.common.utils.ParserException;
+import org.apache.seatunnel.core.starter.enums.CryptoMode;
 import org.apache.seatunnel.core.starter.exception.ConfigCheckException;
 
 import lombok.NonNull;
@@ -62,34 +63,40 @@ private ConfigBuilder() {
         // utility class and cannot be instantiated
     }
 
-    private static Config ofInner(@NonNull Path filePath, List<String> variables) {
+    private static Config ofInner(
+            @NonNull Path filePath, List<String> variables, CryptoMode cryptoMode) {
         Config config =
                 ConfigFactory.parseFile(filePath.toFile())
                         .resolve(ConfigResolveOptions.defaults().setAllowUnresolved(true));
-        return ConfigShadeUtils.decryptConfig(backfillUserVariables(config, variables));
+        return ConfigShadeUtils.decryptConfig(backfillUserVariables(config, variables), cryptoMode);
     }
 
     public static Config of(@NonNull String filePath) {
         Path path = Paths.get(filePath);
         return of(path);
     }
 
-    public static Config of(@NonNull String filePath, List<String> variables) {
-        Path path = Paths.get(filePath);
-        return of(path, variables);
-    }
-
     public static Config of(@NonNull Path filePath) {
         return of(filePath, null);
     }
 
     public static Config of(@NonNull Path filePath, List<String> variables) {
+        return of(filePath, variables, CryptoMode.DEFAULT);
+    }
+
+    public static Config of(
+            @NonNull String filePath, List<String> variables, CryptoMode cryptoMode) {
+        Path path = Paths.get(filePath);
+        return of(path, variables);
+    }
+
+    public static Config of(@NonNull Path filePath, List<String> variables, CryptoMode cryptoMode) {
         log.info("Loading config file from path: {}", filePath);
         Optional<ConfigAdapter> adapterSupplier = ConfigAdapterUtils.selectAdapter(filePath);
         Config config =
                 adapterSupplier
-                        .map(adapter -> of(adapter, filePath, variables))
-                        .orElseGet(() -> ofInner(filePath, variables));
+                        .map(adapter -> of(adapter, filePath, variables, cryptoMode))
+                        .orElseGet(() -> ofInner(filePath, variables, cryptoMode));
         log.info(
                 "Parsed config file: \n{}",
                 mapToString(
@@ -99,12 +106,15 @@ public static Config of(@NonNull Path filePath, List<String> variables) {
         return config;
     }
 
-    public static Config of(@NonNull Map<String, Object> objectMap) {
-        return of(objectMap, false, false);
+    public static Config of(@NonNull Map<String, Object> objectMap, CryptoMode cryptoMode) {
+        return of(objectMap, false, false, cryptoMode);
     }
 
     public static Config of(
-            @NonNull Map<String, Object> objectMap, boolean isEncrypt, boolean isJson) {
+            @NonNull Map<String, Object> objectMap,
+            boolean isEncrypt,
+            boolean isJson,
+            CryptoMode cryptoMode) {
         log.info("Loading config file from objectMap");
         Config config =
                 ConfigFactory.parseMap(objectMap)
@@ -113,7 +123,7 @@ public static Config of(
                                 ConfigFactory.systemProperties(),
                                 ConfigResolveOptions.defaults().setAllowUnresolved(true));
         if (!isEncrypt) {
-            config = ConfigShadeUtils.decryptConfig(config);
+            config = ConfigShadeUtils.decryptConfig(config, cryptoMode);
         }
         log.info(
                 "Parsed config file: \n{}",
@@ -175,19 +185,23 @@ public static Map<String, Object> configDesensitization(
     }
 
     public static Config of(
-            @NonNull ConfigAdapter configAdapter, @NonNull Path filePath, List<String> variables) {
+            @NonNull ConfigAdapter configAdapter,
+            @NonNull Path filePath,
+            List<String> variables,
+            CryptoMode cryptoMode) {
         log.info("With config adapter spi {}", configAdapter.getClass().getName());
         try {
             Map<String, Object> flattenedMap = configAdapter.loadConfig(filePath);
             Config config = ConfigFactory.parseMap(flattenedMap);
-            return ConfigShadeUtils.decryptConfig(backfillUserVariables(config, variables));
+            return ConfigShadeUtils.decryptConfig(
+                    backfillUserVariables(config, variables), cryptoMode);
         } catch (ParserException | IllegalArgumentException e) {
             throw e;
         } catch (Exception warn) {
             log.warn(
                     "Loading config failed with spi {}, fallback to HOCON loader.",
                     configAdapter.getClass().getName());
-            return ofInner(filePath, variables);
+            return ofInner(filePath, variables, cryptoMode);
         }
     }
 

seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigShadeUtils.java

@@ -27,6 +27,7 @@
 import org.apache.seatunnel.common.Constants;
 import org.apache.seatunnel.common.config.TypesafeConfigUtils;
 import org.apache.seatunnel.common.utils.JsonUtils;
+import org.apache.seatunnel.core.starter.enums.CryptoMode;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -97,7 +98,7 @@ public static String decryptOption(String identifier, String content) {
         return configShade.decrypt(content);
     }
 
-    public static Config decryptConfig(Config config) {
+    public static Config decryptConfig(Config config, CryptoMode cryptoMode) {
         String identifier =
                 TypesafeConfigUtils.getConfig(
                         config.hasPath(Constants.ENV)
@@ -112,10 +113,10 @@ public static Config decryptConfig(Config config) {
                                 : ConfigFactory.empty(),
                         SHADE_PROPS_OPTION,
                         new HashMap<>());
-        return decryptConfig(identifier, config, props);
+        return decryptConfig(identifier, config, props, cryptoMode);
     }
 
-    public static Config encryptConfig(Config config) {
+    public static Config encryptConfig(Config config, CryptoMode cryptoMode) {
         String identifier =
                 TypesafeConfigUtils.getConfig(
                         config.hasPath(Constants.ENV)
@@ -130,28 +131,37 @@ public static Config encryptConfig(Config config) {
                                 : ConfigFactory.empty(),
                         SHADE_PROPS_OPTION,
                         new HashMap<>());
-        return encryptConfig(identifier, config, props);
+        return encryptConfig(identifier, config, props, cryptoMode);
     }
 
     private static Config decryptConfig(
-            String identifier, Config config, Map<String, Object> props) {
-        return processConfig(identifier, config, true, props);
+            String identifier, Config config, Map<String, Object> props, CryptoMode cryptoMode) {
+        return processConfig(identifier, config, true, props, cryptoMode);
     }
 
     private static Config encryptConfig(
-            String identifier, Config config, Map<String, Object> props) {
-        return processConfig(identifier, config, false, props);
+            String identifier, Config config, Map<String, Object> props, CryptoMode cryptoMode) {
+        return processConfig(identifier, config, false, props, cryptoMode);
     }
 
     @SuppressWarnings("unchecked")
     private static Config processConfig(
-            String identifier, Config config, boolean isDecrypted, Map<String, Object> props) {
+            String identifier,
+            Config config,
+            boolean isDecrypted,
+            Map<String, Object> props,
+            CryptoMode cryptoMode) {
         ConfigShade configShade = CONFIG_SHADES.getOrDefault(identifier, DEFAULT_SHADE);
         // call open method before the encrypt/decrypt
         configShade.open(props);
 
         Set<String> sensitiveOptions = new HashSet<>(getSensitiveOptions(config));
         sensitiveOptions.addAll(Arrays.asList(configShade.sensitiveOptions()));
+        if (cryptoMode == CryptoMode.DEFAULT) {
+            Set<String> uniqueKeys = new HashSet<>(sensitiveOptions);
+            sensitiveOptions.clear();
+            sensitiveOptions.addAll(uniqueKeys);
+        }
         BiFunction<String, Object, Object> processFunction =
                 (key, value) -> {
                     if (value instanceof List) {

seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigShadeTest.java

@@ -25,6 +25,7 @@
 
 import org.apache.seatunnel.api.configuration.ConfigShade;
 import org.apache.seatunnel.common.utils.JsonUtils;
+import org.apache.seatunnel.core.starter.enums.CryptoMode;
 import org.apache.seatunnel.core.starter.exception.ConfigCheckException;
 
 import org.junit.jupiter.api.Assertions;
@@ -309,7 +310,7 @@ public void testDecryptWithProps() throws URISyntaxException {
         Assertions.assertEquals(
                 rawPassword, decryptedProps.getConfigList("source").get(0).getString("password"));
 
-        Config encryptedConfig = ConfigShadeUtils.encryptConfig(decryptedProps);
+        Config encryptedConfig = ConfigShadeUtils.encryptConfig(decryptedProps, CryptoMode.DEFAULT);
         Assertions.assertEquals(
                 rawUsername + suffix,
                 encryptedConfig.getConfigList("source").get(0).getString("username"));

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestConstant.java

@@ -47,6 +47,8 @@ public class RestConstant {
 
     public static final String METRICS = "metrics";
 
+    public static final String CRYPTO_MODE = "cryptoMode";
+
     public static final String HOCON = "hocon";
 
     public static final String TABLE_SOURCE_RECEIVED_COUNT = "TableSourceReceivedCount";

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestHttpPostCommandProcessor.java

@@ -85,7 +85,7 @@ public void handle(HttpPostCommand httpPostCommand) {
             } else if (uri.startsWith(CONTEXT_PATH + REST_URL_STOP_JOB)) {
                 handleStopJob(httpPostCommand);
             } else if (uri.startsWith(CONTEXT_PATH + REST_URL_ENCRYPT_CONFIG)) {
-                handleEncrypt(httpPostCommand);
+                handleEncrypt(httpPostCommand, uri);
             } else if (uri.startsWith(CONTEXT_PATH + REST_URL_UPDATE_TAGS)) {
                 handleUpdateTags(httpPostCommand);
             } else {
@@ -123,9 +123,12 @@ private void handleStopJob(HttpPostCommand httpPostCommand) {
         this.prepareResponse(httpPostCommand, jobInfoService.stopJob(httpPostCommand.getData()));
     }
 
-    private void handleEncrypt(HttpPostCommand httpPostCommand) {
+    private void handleEncrypt(HttpPostCommand httpPostCommand, String uri) {
+        Map<String, String> requestParams = new HashMap<>();
+        RestUtil.buildRequestParams(requestParams, uri);
         this.prepareResponse(
-                httpPostCommand, encryptConfigService.encryptConfig(httpPostCommand.getData()));
+                httpPostCommand,
+                encryptConfigService.encryptConfig(httpPostCommand.getData(), requestParams));
     }
 
     private void handleUpdateTags(HttpPostCommand httpPostCommand) {