Merge branch 'feature/udf-encrypt' into 'develop'

Feature/udf encrypt

See merge request galaxy/platform/groot-stream!123

7 files changed, 99 insertions, 70 deletions

diff --git a/groot-common/src/main/java/com/geedgenetworks/common/Constants.java b/groot-common/src/main/java/com/geedgenetworks/common/Constants.java
index b523591..27ce8fb 100644
--- a/groot-common/src/main/java/com/geedgenetworks/common/Constants.java
+++ b/groot-common/src/main/java/com/geedgenetworks/common/Constants.java
@@ -2,7 +2,7 @@ package com.geedgenetworks.common;
 
 public final class Constants {
 
-    public static final String DEFAULT_JOB_NAME="groot-stream-job";
+    public static final String DEFAULT_JOB_NAME = "groot-stream-job";
     public static final String SOURCES = "sources";
     public static final String FILTERS = "filters";
     public static final String PREPROCESSING_PIPELINES = "preprocessing_pipelines";
@@ -14,7 +14,7 @@ public final class Constants {
 
     public static final String PROPERTIES = "properties";
     public static final String SPLITS = "splits";
-    public static final String APPLICATION_ENV ="env";
+    public static final String APPLICATION_ENV = "env";
     public static final String APPLICATION_TOPOLOGY = "topology";
     public static final String JOB_NAME = "name";
     public static final String GROOT_LOGO = "\n" +
@@ -49,6 +49,8 @@ public final class Constants {
     public static final String SLIDING_PROCESSING_TIME = "sliding_processing_time";
     public static final String SLIDING_EVENT_TIME = "sliding_event_time";
 
-
+    public static final String SYSPROP_KMS_TYPE_CONFIG = "kms.type";
+    public static final String SYSPROP_ENCRYPT_KMS_KEY_SCHEDULER_INTERVAL_NAME = "scheduler.encrypt.update.kms.key.minutes";
+    public static final String SYSPROP_ENCRYPT_SENSITIVE_FIELDS_SCHEDULER_INTERVAL_NAME = "scheduler.encrypt.update.sensitive.fields.minutes";
 
 }
diff --git a/groot-common/src/main/java/com/geedgenetworks/common/config/CommonConfigDomProcessor.java b/groot-common/src/main/java/com/geedgenetworks/common/config/CommonConfigDomProcessor.java
index eec66fa..b3b17e8 100644
--- a/groot-common/src/main/java/com/geedgenetworks/common/config/CommonConfigDomProcessor.java
+++ b/groot-common/src/main/java/com/geedgenetworks/common/config/CommonConfigDomProcessor.java
@@ -1,8 +1,6 @@
 package com.geedgenetworks.common.config;
 
 import com.hazelcast.internal.config.AbstractDomConfigProcessor;
-import com.hazelcast.logging.ILogger;
-import com.hazelcast.logging.Logger;
 import lombok.extern.slf4j.Slf4j;
 import org.w3c.dom.Node;
 
@@ -16,6 +14,7 @@ import static com.hazelcast.internal.config.DomConfigHelper.*;
 @Slf4j
 public class CommonConfigDomProcessor extends AbstractDomConfigProcessor {
     private final GrootStreamConfig config;
+
     CommonConfigDomProcessor(boolean domLevel3, GrootStreamConfig config) {
         super(domLevel3);
         this.config = config;
@@ -26,16 +25,16 @@ public class CommonConfigDomProcessor extends AbstractDomConfigProcessor {
         final CommonConfig commonConfig = config.getCommonConfig();
         for (Node node : childElements(rootNode)) {
             String name = cleanNodeName(node);
-             if (CommonConfigOptions.KNOWLEDGE_BASE.key().equals(name)) {
-                 commonConfig.setKnowledgeBaseConfig(parseKnowledgeBaseConfig(node));
+            if (CommonConfigOptions.KNOWLEDGE_BASE.key().equals(name)) {
+                commonConfig.setKnowledgeBaseConfig(parseKnowledgeBaseConfig(node));
             } else if (CommonConfigOptions.KMS.key().equals(name)) {
-                 commonConfig.setKmsConfig(parseKmsConfig(node));
-             } else if (CommonConfigOptions.SSL.key().equals(name)) {
-                 commonConfig.setSslConfig(parseSSLConfig(node));
-             } else if (CommonConfigOptions.PROPERTIES.key().equals(name)) {
-                 commonConfig.setPropertiesConfig(parsePropertiesConfig(node));
+                commonConfig.setKmsConfig(parseKmsConfig(node));
+            } else if (CommonConfigOptions.SSL.key().equals(name)) {
+                commonConfig.setSslConfig(parseSSLConfig(node));
+            } else if (CommonConfigOptions.PROPERTIES.key().equals(name)) {
+                commonConfig.setPropertiesConfig(parsePropertiesConfig(node));
             } else {
-                 log.warn("Unrecognized Groot Stream configuration element: {}", name);
+                log.warn("Unrecognized Groot Stream configuration element: {}", name);
             }
 
         }
@@ -43,12 +42,12 @@ public class CommonConfigDomProcessor extends AbstractDomConfigProcessor {
 
     private Map<String, String> parsePropertiesConfig(Node properties) {
 
-            Map<String, String> propertiesMap = new HashMap<>();
-            for (Node node : childElements(properties)) {
-                String name = cleanNodeName(node);
-                propertiesMap.put(name,getTextContent(node));
-            }
-            return propertiesMap;
+        Map<String, String> propertiesMap = new HashMap<>();
+        for (Node node : childElements(properties)) {
+            String name = cleanNodeName(node);
+            propertiesMap.put(name, getTextContent(node));
+        }
+        return propertiesMap;
     }
 
 
@@ -62,7 +61,7 @@ public class CommonConfigDomProcessor extends AbstractDomConfigProcessor {
     }
 
 
-   private KnowledgeBaseConfig parseKnowledgeBaseConfigAsObject(Node kbNode) {
+    private KnowledgeBaseConfig parseKnowledgeBaseConfigAsObject(Node kbNode) {
         KnowledgeBaseConfig knowledgeBaseConfig = new KnowledgeBaseConfig();
         for (Node node : childElements(kbNode)) {
             String name = cleanNodeName(node);
@@ -76,7 +75,7 @@ public class CommonConfigDomProcessor extends AbstractDomConfigProcessor {
                 knowledgeBaseConfig.setFiles(parseKnowledgeBaseFilesConfig(node));
             } else if (CommonConfigOptions.KNOWLEDGE_BASE_PROPERTIES.key().equals(name)) {
                 knowledgeBaseConfig.setProperties(parseKnowledgeBasePropertiesConfig(node));
-            } else{
+            } else {
                 log.warn("Unrecognized KB configuration element: {}", name);
             }
 
@@ -84,18 +83,18 @@ public class CommonConfigDomProcessor extends AbstractDomConfigProcessor {
         return knowledgeBaseConfig;
     }
 
-    private SSLConfig parseSSLConfig (Node sslRootNode) {
+    private SSLConfig parseSSLConfig(Node sslRootNode) {
         SSLConfig sslConfig = new SSLConfig();
         for (Node node : childElements(sslRootNode)) {
             String name = cleanNodeName(node);
-            if (CommonConfigOptions.SSL_ENABLED.key().equals(name)) {
-                sslConfig.setEnabled(getBooleanValue(getTextContent(node)));
-            } else if (CommonConfigOptions.SSL_CERT_FILE.key().equals(name)) {
-                sslConfig.setCertFile(getTextContent(node));
-            } else if (CommonConfigOptions.SSL_KEY_FILE.key().equals(name)) {
-                sslConfig.setKeyFile(getTextContent(node));
-            } else if (CommonConfigOptions.SSL_REQUIRE_CLIENT_AUTH.key().equals(name)) {
-                sslConfig.setRequireClientAuth(getBooleanValue(getTextContent(node)));
+            if (CommonConfigOptions.SKIP_VERIFICATION.key().equals(name)) {
+                sslConfig.setSkipVerification(getBooleanValue(getTextContent(node)));
+            } else if (CommonConfigOptions.CA_CERTIFICATE_PATH.key().equals(name)) {
+                sslConfig.setCaCertificatePath(getTextContent(node));
+            } else if (CommonConfigOptions.CERTIFICATE_PATH.key().equals(name)) {
+                sslConfig.setCertificatePath(getTextContent(node));
+            } else if (CommonConfigOptions.PRIVATE_KEY_PATH.key().equals(name)) {
+                sslConfig.setPrivateKeyPath(getTextContent(node));
             } else {
                 log.warn("Unrecognized SSL configuration element: {}", name);
             }
@@ -120,10 +119,14 @@ public class CommonConfigDomProcessor extends AbstractDomConfigProcessor {
                 kmsConfig.setType(getTextContent(node));
             } else if (CommonConfigOptions.KMS_URL.key().equals(name)) {
                 kmsConfig.setUrl(getTextContent(node));
-            } else if (CommonConfigOptions.KMS_TOKEN.key().equals(name)) {
-                kmsConfig.setToken(getTextContent(node));
-            } else if (CommonConfigOptions.KMS_KEY_PATH.key().equals(name)) {
-                kmsConfig.setKeyPath(getTextContent(node));
+            } else if (CommonConfigOptions.KMS_USERNAME.key().equals(name)) {
+                kmsConfig.setUsername(getTextContent(node));
+            } else if (CommonConfigOptions.KMS_PASSWORD.key().equals(name)) {
+                kmsConfig.setPassword(getTextContent(node));
+            } else if (CommonConfigOptions.KMS_DEFAULT_KEY_PATH.key().equals(name)) {
+                kmsConfig.setDefaultKeyPath(getTextContent(node));
+            } else if (CommonConfigOptions.KMS_PLUGIN_KEY_PATH.key().equals(name)) {
+                kmsConfig.setPluginKeyPath(getTextContent(node));
             } else {
                 log.warn("Unrecognized KMS configuration element: {}", name);
             }
@@ -136,7 +139,7 @@ public class CommonConfigDomProcessor extends AbstractDomConfigProcessor {
         Map<String, String> propertiesMap = new HashMap<>();
         for (Node node : childElements(properties)) {
             String name = cleanNodeName(node);
-            propertiesMap.put(name,getTextContent(node));
+            propertiesMap.put(name, getTextContent(node));
         }
         return propertiesMap;
     }
diff --git a/groot-common/src/main/java/com/geedgenetworks/common/config/CommonConfigOptions.java b/groot-common/src/main/java/com/geedgenetworks/common/config/CommonConfigOptions.java
index d3f1cb9..167fcba 100644
--- a/groot-common/src/main/java/com/geedgenetworks/common/config/CommonConfigOptions.java
+++ b/groot-common/src/main/java/com/geedgenetworks/common/config/CommonConfigOptions.java
@@ -12,7 +12,7 @@ public class CommonConfigOptions {
     public static final Option<Map<String, String>> KNOWLEDGE_BASE_PROPERTIES =
             Options.key("properties")
                     .mapType()
-                    .defaultValue(new HashMap<String,String>())
+                    .defaultValue(new HashMap<String, String>())
                     .withDescription("The properties of knowledge base");
     public static final Option<String> KNOWLEDGE_BASE_NAME =
             Options.key("name")
@@ -47,7 +47,8 @@ public class CommonConfigOptions {
 
     public static final Option<List<KnowledgeBaseConfig>> KNOWLEDGE_BASE =
             Options.key("knowledge_base")
-                    .type(new TypeReference<List<KnowledgeBaseConfig>>() {})
+                    .type(new TypeReference<List<KnowledgeBaseConfig>>() {
+                    })
                     .noDefaultValue()
                     .withDescription("The knowledge base configuration.");
 
@@ -59,7 +60,8 @@ public class CommonConfigOptions {
 
     public static final Option<Map<String, KmsConfig>> KMS =
             Options.key("kms")
-                    .type(new TypeReference<Map<String, KmsConfig>>() {})
+                    .type(new TypeReference<Map<String, KmsConfig>>() {
+                    })
                     .noDefaultValue()
                     .withDescription("The kms configuration.");
 
@@ -73,42 +75,49 @@ public class CommonConfigOptions {
             .defaultValue("")
             .withDescription("The access url of KMS.");
 
-    public static final Option<String> KMS_TOKEN = Options.key("token")
+    public static final Option<String> KMS_USERNAME = Options.key("username")
             .stringType()
             .defaultValue("")
-            .withDescription("The access token of KMS.");
+            .withDescription("The access username of KMS.");
 
-    public static final Option<String> KMS_KEY_PATH = Options.key("key_path")
+    public static final Option<String> KMS_PASSWORD = Options.key("password")
             .stringType()
             .defaultValue("")
-            .withDescription("The key path of KMS.");
+            .withDescription("The access username of KMS.");
+
+    public static final Option<String> KMS_DEFAULT_KEY_PATH = Options.key("default_key_path")
+            .stringType()
+            .defaultValue("")
+            .withDescription("The default key path of KMS.");
+
+    public static final Option<String> KMS_PLUGIN_KEY_PATH = Options.key("plugin_key_path")
+            .stringType()
+            .defaultValue("")
+            .withDescription("The plugin key path of KMS.");
 
     public static final Option<SSLConfig> SSL = Options.key("ssl")
-            .type(new TypeReference<SSLConfig>() {})
+            .type(new TypeReference<SSLConfig>() {
+            })
             .noDefaultValue()
             .withDescription("The ssl configuration.");
 
-    public static final Option<Boolean> SSL_ENABLED = Options.key("enabled")
+    public static final Option<Boolean> SKIP_VERIFICATION = Options.key("skip_verification")
             .booleanType()
             .defaultValue(false)
-            .withDescription("The enabled flag of the configuration.");
+            .withDescription("The skip certificate of the configuration.");
+
+    public static final Option<String> CA_CERTIFICATE_PATH = Options.key("ca_certificate_path")
+            .stringType()
+            .defaultValue("")
+            .withDescription("The ca certificate file path of the configuration.");
 
-    public static final Option<String> SSL_CERT_FILE = Options.key("cert_file")
+    public static final Option<String> CERTIFICATE_PATH = Options.key("certificate_path")
             .stringType()
             .defaultValue("")
             .withDescription("The certificate file path of the configuration.");
 
-    public static final Option<String> SSL_KEY_FILE = Options.key("key_file")
+    public static final Option<String> PRIVATE_KEY_PATH = Options.key("private_key_path")
             .stringType()
             .defaultValue("")
             .withDescription("The private key file path of the configuration.");
-
-    public static final Option<Boolean> SSL_REQUIRE_CLIENT_AUTH = Options.key("require_client_auth")
-            .booleanType()
-            .defaultValue(false)
-            .withDescription("The require client auth flag of the configuration.");
-
-
-
-
 }
diff --git a/groot-common/src/main/java/com/geedgenetworks/common/config/KmsConfig.java b/groot-common/src/main/java/com/geedgenetworks/common/config/KmsConfig.java
index f26062c..f0e213f 100644
--- a/groot-common/src/main/java/com/geedgenetworks/common/config/KmsConfig.java
+++ b/groot-common/src/main/java/com/geedgenetworks/common/config/KmsConfig.java
@@ -6,12 +6,10 @@ import java.io.Serializable;
 
 @Data
 public class KmsConfig implements Serializable {
-
-    private String type =  CommonConfigOptions.KMS_TYPE.defaultValue();
+    private String type = CommonConfigOptions.KMS_TYPE.defaultValue();
     private String url = CommonConfigOptions.KMS_URL.defaultValue();
-    private String token = CommonConfigOptions.KMS_TOKEN.defaultValue();
-    private String keyPath = CommonConfigOptions.KMS_KEY_PATH.defaultValue();
-
-
-
+    private String username = CommonConfigOptions.KMS_USERNAME.defaultValue();
+    private String password = CommonConfigOptions.KMS_PASSWORD.defaultValue();
+    private String defaultKeyPath = CommonConfigOptions.KMS_DEFAULT_KEY_PATH.defaultValue();
+    private String pluginKeyPath = CommonConfigOptions.KMS_PLUGIN_KEY_PATH.defaultValue();
 }
diff --git a/groot-common/src/main/java/com/geedgenetworks/common/config/SSLConfig.java b/groot-common/src/main/java/com/geedgenetworks/common/config/SSLConfig.java
index 7df5c5b..874c163 100644
--- a/groot-common/src/main/java/com/geedgenetworks/common/config/SSLConfig.java
+++ b/groot-common/src/main/java/com/geedgenetworks/common/config/SSLConfig.java
@@ -6,14 +6,11 @@ import java.io.Serializable;
 
 @Data
 public class SSLConfig implements Serializable {
+    private Boolean skipVerification = CommonConfigOptions.SKIP_VERIFICATION.defaultValue();
 
-    private Boolean enabled = CommonConfigOptions.SSL_ENABLED.defaultValue();
-
-    private String certFile = CommonConfigOptions.SSL_CERT_FILE.defaultValue();
-
-    private String keyFile = CommonConfigOptions.SSL_KEY_FILE.defaultValue();
-
-    private Boolean requireClientAuth = CommonConfigOptions.SSL_REQUIRE_CLIENT_AUTH.defaultValue();
+    private String caCertificatePath = CommonConfigOptions.CA_CERTIFICATE_PATH.defaultValue();
 
+    private String certificatePath = CommonConfigOptions.CERTIFICATE_PATH.defaultValue();
 
+    private String privateKeyPath = CommonConfigOptions.PRIVATE_KEY_PATH.defaultValue();
 }
diff --git a/groot-common/src/main/resources/grootstream.yaml b/groot-common/src/main/resources/grootstream.yaml
index 1a9a974..26752e3 100644
--- a/groot-common/src/main/resources/grootstream.yaml
+++ b/groot-common/src/main/resources/grootstream.yaml
@@ -11,6 +11,24 @@ grootstream:
       files:
         - 64af7077-eb9b-4b8f-80cf-2ceebc89bea9
         - 004390bc-3135-4a6f-a492-3662ecb9e289
+
+  kms:
+    local:
+      type: local
+    vault:
+      type: vault
+      url: https://192.168.40.223:8200
+      username: tsg_olap
+      password: tsg_olap
+      default_key_path: tsg_olap/transit
+      plugin_key_path: tsg_olap/plugin/gmsm
+
+  ssl:
+    skip_verification: true
+    ca_certificate_path: ./config/ssl/root.pem
+    certificate_path: ./config/ssl/worker.pem
+    private_key_path: ./config/ssl/worker.key
+
   properties:
     hos.path: http://192.168.44.12:9098/hos
     hos.bucket.name.traffic_file: traffic_file_bucket
diff --git a/groot-common/src/main/resources/udf.plugins b/groot-common/src/main/resources/udf.plugins
index fe7a083..3d6a353 100644
--- a/groot-common/src/main/resources/udf.plugins
+++ b/groot-common/src/main/resources/udf.plugins
@@ -4,11 +4,13 @@ com.geedgenetworks.core.udf.DecodeBase64
 com.geedgenetworks.core.udf.Domain
 com.geedgenetworks.core.udf.Drop
 com.geedgenetworks.core.udf.EncodeBase64
+com.geedgenetworks.core.udf.Encrypt
 com.geedgenetworks.core.udf.Eval
 com.geedgenetworks.core.udf.Flatten
 com.geedgenetworks.core.udf.FromUnixTimestamp
 com.geedgenetworks.core.udf.GenerateStringArray
 com.geedgenetworks.core.udf.GeoIpLookup
+com.geedgenetworks.core.udf.Hmac
 com.geedgenetworks.core.udf.JsonExtract
 com.geedgenetworks.core.udf.PathCombine
 com.geedgenetworks.core.udf.Rename