[Feature][Config] Support custom config keys for encrypt/decrypt

Author: wanqifei

docs/en/connector-v2/Config-Encryption-Decryption.md

@@ -8,17 +8,19 @@ In most production environments, sensitive configuration items such as passwords
 
 SeaTunnel comes with the function of base64 encryption and decryption, but it is not recommended for production use, it is recommended that users implement custom encryption and decryption logic. You can refer to this chapter [How to implement user-defined encryption and decryption](#How to implement user-defined encryption and decryption) get more details about it.
 
-Base64 encryption support encrypt the following parameters:
+Base64 encryption support encrypt the following parameters by default:
 - username
 - password
 - auth
 - token
 - access_key
 - secret_key
 
+And users can add custom parameters to `shade.options` for encryption and decryption.
+
 Next, I'll show how to quickly use SeaTunnel's own `base64` encryption:
 
-1. And a new option `shade.identifier` in env block of config file, this option indicate what the encryption method that you want to use, in this example, we should add `shade.identifier = base64` in config as the following shown:
+1. And new option `shade.identifier` and `shade.options` in env block of config file, `shade.identifier` indicate what the encryption method that you want to use, while `shade.options` specifies which parameters should be encrypted/decrypted. In this example, we should add `shade.identifier = base64` in config as the following shown:
 
    ```hocon
    #
@@ -41,6 +43,7 @@ Next, I'll show how to quickly use SeaTunnel's own `base64` encryption:
    env {
      parallelism = 1
      shade.identifier = "base64"
+     shade.options = ["username", "password", "f1", "config1.f1",  "config2.list"]
    }
 
    source {
@@ -55,6 +58,10 @@ Next, I'll show how to quickly use SeaTunnel's own `base64` encryption:
        database-name = "inventory_vwyw0n"
        table-name = "products"
        base-url = "jdbc:mysql://localhost:56725"
+       f1 = "seatunnel"
+       # custom shade options
+       config1.f1 = "seatunnel"
+       config2.list = ["seatunnel", "seatunnel", "seatunnel"]
      }
    }
 
@@ -103,7 +110,9 @@ Next, I'll show how to quickly use SeaTunnel's own `base64` encryption:
                "table-name" : "products",
                "plugin_name" : "MySQL-CDC",
                "server-id" : 5656,
-               "username" : "c2VhdHVubmVs"
+               "username" : "c2VhdHVubmVs",
+               "config1.f1": "c2VhdHVubmVs",
+               "config2.list": ["c2VhdHVubmVs","c2VhdHVubmVs","c2VhdHVubmVs"]
            }
        ],
        "transform" : [],

seatunnel-common/src/main/java/org/apache/seatunnel/common/config/TypesafeConfigUtils.java

@@ -77,7 +77,7 @@ public static <T> T getConfig(
                     ? (T) Boolean.valueOf(config.getString(configKey))
                     : defaultValue;
         }
-        if (defaultValue instanceof Map) {
+        if (defaultValue instanceof Map || defaultValue instanceof List) {
             return config.hasPath(configKey) ? (T) config.getAnyRef(configKey) : defaultValue;
         }
         throw new RuntimeException("Unsupported config type, configKey: " + configKey);

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

@@ -53,10 +53,10 @@ public void execute() throws CommandExecuteException, ConfigCheckException {
                         .resolveWith(
                                 ConfigFactory.systemProperties(),
                                 ConfigResolveOptions.defaults().setAllowUnresolved(true));
-        Config encryptConfig = ConfigShadeUtils.decryptConfig(config);
+        Config decryptConfig = ConfigShadeUtils.decryptConfig(config);
         log.info(
-                "Encrypt config: \n{}",
-                encryptConfig
+                "Decrypt config: \n{}",
+                decryptConfig
                         .root()
                         .render(ConfigRenderOptions.defaults().setOriginComments(false)));
     }

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

@@ -37,18 +37,17 @@
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
 import static org.apache.seatunnel.common.utils.PlaceholderUtils.replacePlaceholders;
-import static org.apache.seatunnel.core.starter.utils.ConfigShadeUtils.DEFAULT_SENSITIVE_KEYWORDS;
 
 /** Used to build the {@link Config} from config file. */
 @Slf4j
@@ -91,10 +90,12 @@ public static Config of(@NonNull Path filePath, List<String> variables) {
                 adapterSupplier
                         .map(adapter -> of(adapter, filePath, variables))
                         .orElseGet(() -> ofInner(filePath, variables));
-        boolean isJson = filePath.getFileName().toString().endsWith(".json");
         log.info(
                 "Parsed config file: \n{}",
-                mapToString(configDesensitization(config.root().unwrapped())));
+                mapToString(
+                        configDesensitization(
+                                config.root().unwrapped(),
+                                ConfigShadeUtils.getSensitiveOptions(config))));
         return config;
     }
 
@@ -116,23 +117,39 @@ public static Config of(
         }
         log.info(
                 "Parsed config file: \n{}",
-                mapToString(configDesensitization(config.root().unwrapped())));
+                mapToString(
+                        configDesensitization(
+                                config.root().unwrapped(),
+                                ConfigShadeUtils.getSensitiveOptions(config))));
         return config;
     }
 
-    public static Map<String, Object> configDesensitization(Map<String, Object> configMap) {
+    public static Map<String, Object> configDesensitization(
+            Map<String, Object> configMap, Set<String> sensitiveKeywords) {
         return configMap.entrySet().stream()
                 .collect(
                         LinkedHashMap::new,
                         (m, p) -> {
                             String key = p.getKey();
                             Object value = p.getValue();
-                            if (Arrays.asList(DEFAULT_SENSITIVE_KEYWORDS)
-                                    .contains(key.toLowerCase())) {
-                                m.put(key, "******");
+                            if (sensitiveKeywords.contains(key.toLowerCase())) {
+                                if (value instanceof List<?>) {
+                                    List<Object> maskedList =
+                                            ((List<?>) value)
+                                                    .stream()
+                                                            .map(v -> "******")
+                                                            .collect(Collectors.toList());
+                                    m.put(key, maskedList);
+                                } else {
+                                    m.put(key, "******");
+                                }
                             } else {
                                 if (value instanceof Map<?, ?>) {
-                                    m.put(key, configDesensitization((Map<String, Object>) value));
+                                    m.put(
+                                            key,
+                                            configDesensitization(
+                                                    (Map<String, Object>) value,
+                                                    sensitiveKeywords));
                                 } else if (value instanceof List<?>) {
                                     List<?> listValue = (List<?>) value;
                                     List<Object> newList =
@@ -141,8 +158,8 @@ public static Map<String, Object> configDesensitization(Map<String, Object> conf
                                                             v -> {
                                                                 if (v instanceof Map<?, ?>) {
                                                                     return configDesensitization(
-                                                                            (Map<String, Object>)
-                                                                                    v);
+                                                                            (Map<String, Object>) v,
+                                                                            sensitiveKeywords);
                                                                 } else {
                                                                     return v;
                                                                 }

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

@@ -35,10 +35,12 @@
 import java.util.Arrays;
 import java.util.Base64;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.ServiceLoader;
+import java.util.Set;
 import java.util.function.BiFunction;
 
 /** Config shade utilities */
@@ -47,6 +49,7 @@ public final class ConfigShadeUtils {
 
     private static final String SHADE_IDENTIFIER_OPTION = "shade.identifier";
     private static final String SHADE_PROPS_OPTION = "shade.properties";
+    private static final String SHADE_OPTIONS_OPTION = "shade.options";
 
     public static final String[] DEFAULT_SENSITIVE_KEYWORDS =
             new String[] {"password", "username", "auth", "token", "access_key", "secret_key"};
@@ -147,14 +150,24 @@ private static Config processConfig(
         // call open method before the encrypt/decrypt
         configShade.open(props);
 
-        List<String> sensitiveOptions = new ArrayList<>(Arrays.asList(DEFAULT_SENSITIVE_KEYWORDS));
+        Set<String> sensitiveOptions = new HashSet<>(getSensitiveOptions(config));
         sensitiveOptions.addAll(Arrays.asList(configShade.sensitiveOptions()));
-        BiFunction<String, Object, String> processFunction =
+        BiFunction<String, Object, Object> processFunction =
                 (key, value) -> {
-                    if (isDecrypted) {
-                        return configShade.decrypt(value.toString());
+                    if (value instanceof List) {
+                        List<String> list = (List<String>) value;
+                        List<String> processedList = new ArrayList<>();
+                        for (String element : list) {
+                            processedList.add(
+                                    isDecrypted
+                                            ? configShade.decrypt(element)
+                                            : configShade.encrypt(element));
+                        }
+                        return processedList;
                     } else {
-                        return configShade.encrypt(value.toString());
+                        return isDecrypted
+                                ? configShade.decrypt((String) value)
+                                : configShade.encrypt((String) value);
                     }
                 };
         String jsonString = config.root().render(ConfigRenderOptions.concise());
@@ -185,6 +198,19 @@ private static Config processConfig(
         return ConfigFactory.parseMap(configMap);
     }
 
+    public static Set<String> getSensitiveOptions(Config config) {
+        Set<String> sensitiveOptions =
+                new HashSet<>(
+                        TypesafeConfigUtils.getConfig(
+                                config != null && config.hasPath(Constants.ENV)
+                                        ? config.getConfig(Constants.ENV)
+                                        : ConfigFactory.empty(),
+                                SHADE_OPTIONS_OPTION,
+                                new ArrayList<>()));
+        sensitiveOptions.addAll(new ArrayList<>(Arrays.asList(DEFAULT_SENSITIVE_KEYWORDS)));
+        return sensitiveOptions;
+    }
+
     public static class Base64ConfigShade implements ConfigShade {
 
         private static final Base64.Encoder ENCODER = Base64.getEncoder();

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

@@ -38,7 +38,9 @@ public void testConfigDesensitizationSort() {
         config.put("e", "1");
         config.put("f", "1");
 
-        Map<String, Object> desensitizationConfig = ConfigBuilder.configDesensitization(config);
+        Map<String, Object> desensitizationConfig =
+                ConfigBuilder.configDesensitization(
+                        config, ConfigShadeUtils.getSensitiveOptions(null));
         List<String> keys = new ArrayList<>(desensitizationConfig.keySet());
         Assertions.assertIterableEquals(Arrays.asList("a", "b", "c", "d", "e", "f"), keys);
     }

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

@@ -39,6 +39,7 @@
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Paths;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Base64;
 import java.util.List;
 import java.util.Map;
@@ -88,7 +89,9 @@ public void testUsePrivacyHandlerHocon() throws URISyntaxException {
         Config config = ConfigBuilder.of(Paths.get(resource.toURI()), Lists.newArrayList());
         config =
                 ConfigFactory.parseMap(
-                                ConfigBuilder.configDesensitization(config.root().unwrapped()))
+                                ConfigBuilder.configDesensitization(
+                                        config.root().unwrapped(),
+                                        ConfigShadeUtils.getSensitiveOptions(config)))
                         .resolve(ConfigResolveOptions.defaults().setAllowUnresolved(true))
                         .resolveWith(
                                 ConfigFactory.systemProperties(),
@@ -101,6 +104,12 @@ public void testUsePrivacyHandlerHocon() throws URISyntaxException {
                 config.getConfigList("source").get(0).getString("access_key"), "******");
         Assertions.assertEquals(
                 config.getConfigList("source").get(0).getString("secret_key"), "******");
+        Assertions.assertEquals(config.getConfigList("source").get(0).getString("f1"), "******");
+        Assertions.assertEquals(
+                config.getConfigList("source").get(0).getString("config1.f1"), "******");
+        Assertions.assertEquals(
+                config.getConfigList("source").get(0).getStringList("config2.list"),
+                Arrays.asList("******", "******", "******"));
         String conf = ConfigBuilder.mapToString(config.root().unwrapped());
         Assertions.assertTrue(conf.contains("\"password\" : \"******\""));
     }
@@ -112,7 +121,9 @@ public void testUsePrivacyHandlerJson() throws URISyntaxException {
         Config config = ConfigBuilder.of(Paths.get(resource.toURI()), Lists.newArrayList());
         config =
                 ConfigFactory.parseMap(
-                                ConfigBuilder.configDesensitization(config.root().unwrapped()))
+                                ConfigBuilder.configDesensitization(
+                                        config.root().unwrapped(),
+                                        ConfigShadeUtils.getSensitiveOptions(config)))
                         .resolve(ConfigResolveOptions.defaults().setAllowUnresolved(true))
                         .resolveWith(
                                 ConfigFactory.systemProperties(),
@@ -121,6 +132,13 @@ public void testUsePrivacyHandlerJson() throws URISyntaxException {
                 config.getConfigList("source").get(0).getString("username"), "******");
         Assertions.assertEquals(
                 config.getConfigList("source").get(0).getString("password"), "******");
+        Assertions.assertEquals(config.getConfigList("source").get(0).getString("f1"), "******");
+        Assertions.assertEquals(
+                config.getConfigList("source").get(0).getString("config1.f1"), "******");
+        Assertions.assertEquals(
+                config.getConfigList("source").get(0).getStringList("config2.list"),
+                Arrays.asList("******", "******", "******"));
+        String conf = ConfigBuilder.mapToString(config.root().unwrapped());
         String json = ConfigBuilder.mapToString(config.root().unwrapped());
         Assertions.assertTrue(json.contains("\"password\" : \"******\""));
     }
@@ -132,7 +150,9 @@ public void testConfNull() throws URISyntaxException {
         Config config = ConfigBuilder.of(Paths.get(resource.toURI()), Lists.newArrayList());
         config =
                 ConfigFactory.parseMap(
-                                ConfigBuilder.configDesensitization(config.root().unwrapped()))
+                                ConfigBuilder.configDesensitization(
+                                        config.root().unwrapped(),
+                                        ConfigShadeUtils.getSensitiveOptions(config)))
                         .resolve(ConfigResolveOptions.defaults().setAllowUnresolved(true))
                         .resolveWith(
                                 ConfigFactory.systemProperties(),

seatunnel-core/seatunnel-core-starter/src/test/resources/config.shade.conf

@@ -18,6 +18,7 @@
 env {
   parallelism = 1
   shade.identifier = "base64"
+  shade.options = ["username", "password", "f1", "config1.f1",  "config2.list", "f2"]
 }
 
 source {
@@ -43,6 +44,11 @@ source {
     # test properties
     access_key = "YWNjZXNzX2tleQ=="
     secret_key = "c2VjcmV0X2tleQ=="
+
+    # test shade options
+    f1 = "c2VhdHVubmVs"
+    config1.f1 = "c2VhdHVubmVs"
+    config2.list = ["c2VhdHVubmVsX3Bhc3N3b3Jk", "c2VhdHVubmVsX3Bhc3N3b3Jk", "c2VhdHVubmVsX3Bhc3N3b3Jk"]
   }
 }
 

seatunnel-core/seatunnel-core-starter/src/test/resources/config.shade.json

@@ -1,7 +1,8 @@
 {
   "env" : {
     "shade.identifier" : "base64",
-    "parallelism" : 1
+    "parallelism" : 1,
+    "shade.options": ["username", "password", "f1", "config1.f1",  "config2.list", "f2"]
   },
   "source" : [
     {
@@ -22,7 +23,13 @@
           "sex" : "boolean"
         }
       },
-      "plugin_output" : "fake"
+      "plugin_output" : "fake",
+      "f1": "c2VhdHVubmVs",
+      "config1.f1": "c2VhdHVubmVs",
+      "config2.list": ["c2VhdHVubmVsX3Bhc3N3b3Jk", "c2VhdHVubmVsX3Bhc3N3b3Jk", "c2VhdHVubmVsX3Bhc3N3b3Jk"],
+      "config3": {
+        "f2": "c2VhdHVubmVs"
+      }
     }
   ],
   "transform" : [],