diff options
25 files changed, 1131 insertions, 92 deletions
diff --git a/config/grootstream.yaml b/config/grootstream.yaml index fdefe44..ec661f0 100644 --- a/config/grootstream.yaml +++ b/config/grootstream.yaml @@ -11,21 +11,24 @@ grootstream: files: - 64af7077-eb9b-4b8f-80cf-2ceebc89bea9 - 004390bc-3135-4a6f-a492-3662ecb9e289 + kms: - local: - type: local + # local: + # type: local + # secret_key: .geedgenetworks. vault: type: vault - url: <vault-url> - token: <vault-token> - key_path: <vault-key-path> + 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: - enabled: false - cert_file: ./config/ssl/cert.pem - key_file: ./config/ssl/key.pem - require_client_auth: false - + 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 diff --git a/config/udf.plugins b/config/udf.plugins index 7cd3b0e..bca60bc 100644 --- a/config/udf.plugins +++ b/config/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 diff --git a/groot-bootstrap/src/main/java/com/geedgenetworks/bootstrap/command/AES128GCM96Shade.java b/groot-bootstrap/src/main/java/com/geedgenetworks/bootstrap/command/AES128GCM96Shade.java new file mode 100644 index 0000000..03ed1af --- /dev/null +++ b/groot-bootstrap/src/main/java/com/geedgenetworks/bootstrap/command/AES128GCM96Shade.java @@ -0,0 +1,72 @@ +package com.geedgenetworks.bootstrap.command; + +import cn.hutool.core.util.RandomUtil; +import com.geedgenetworks.common.crypto.CryptoShade; + +import javax.crypto.Cipher; +import javax.crypto.spec.GCMParameterSpec; +import javax.crypto.spec.SecretKeySpec; +import java.nio.charset.StandardCharsets; +import java.security.Key; +import java.util.Base64; + +public class AES128GCM96Shade implements CryptoShade { + private static final String IDENTIFIER = "aes-128-gcm96"; + private static final String ALGORITHM = "AES"; + private static final String TRANSFORMATION = "AES/GCM/NoPadding"; + private static final int GCM_TAG_LENGTH = 128; + private static final int GCM_NONCE_LENGTH = 12; + private static final byte[] NONCE = RandomUtil.randomBytes(GCM_NONCE_LENGTH); + + private static final String[] SENSITIVE_OPTIONS = + new String[]{"secret_key", "connection.user", "connection.password", "kafka.sasl.jaas.config", "kafka.ssl.keystore.password", "kafka.ssl.truststore.password", "kafka.ssl.key.password"}; + + private static final Key SECURITY_KEY = new SecretKeySpec(".geedgenetworks.".getBytes(StandardCharsets.UTF_8), ALGORITHM); + + @Override + public String[] sensitiveOptions() { + return SENSITIVE_OPTIONS; + } + + @Override + public String getIdentifier() { + return IDENTIFIER; + } + + @Override + public String encrypt(String content) { + String encryptedString = ""; + try { + Cipher cipher = Cipher.getInstance(TRANSFORMATION); + GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, NONCE); + cipher.init(Cipher.ENCRYPT_MODE, SECURITY_KEY, gcmSpec); + byte[] encryptedBytes = cipher.doFinal(content.getBytes()); + byte[] combinedBytes = new byte[GCM_NONCE_LENGTH + encryptedBytes.length]; + System.arraycopy(NONCE, 0, combinedBytes, 0, GCM_NONCE_LENGTH); + System.arraycopy(encryptedBytes, 0, combinedBytes, GCM_NONCE_LENGTH, encryptedBytes.length); + encryptedString = Base64.getEncoder().encodeToString(combinedBytes); + } catch (Exception e) { + throw new RuntimeException(e); + } + return encryptedString; + } + + @Override + public String decrypt(String content) { + String decryptedString = ""; + try { + byte[] combined = Base64.getDecoder().decode(content); + byte[] encryptedBytes = new byte[combined.length - GCM_NONCE_LENGTH]; + System.arraycopy(combined, 0, NONCE, 0, GCM_NONCE_LENGTH); + System.arraycopy(combined, GCM_NONCE_LENGTH, encryptedBytes, 0, encryptedBytes.length); + Cipher cipher = Cipher.getInstance(TRANSFORMATION); + GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, NONCE); + cipher.init(Cipher.DECRYPT_MODE, SECURITY_KEY, gcmSpec); + byte[] decryptedBytes = cipher.doFinal(encryptedBytes); + decryptedString = new String(decryptedBytes); + } catch (Exception e) { + throw new RuntimeException(e); + } + return decryptedString; + } +} diff --git a/groot-bootstrap/src/main/java/com/geedgenetworks/bootstrap/command/AES256GCM96Shade.java b/groot-bootstrap/src/main/java/com/geedgenetworks/bootstrap/command/AES256GCM96Shade.java new file mode 100644 index 0000000..efee134 --- /dev/null +++ b/groot-bootstrap/src/main/java/com/geedgenetworks/bootstrap/command/AES256GCM96Shade.java @@ -0,0 +1,72 @@ +package com.geedgenetworks.bootstrap.command; + +import cn.hutool.core.util.RandomUtil; +import com.geedgenetworks.common.crypto.CryptoShade; + +import javax.crypto.Cipher; +import javax.crypto.spec.GCMParameterSpec; +import javax.crypto.spec.SecretKeySpec; +import java.nio.charset.StandardCharsets; +import java.security.*; +import java.util.Base64; + +public class AES256GCM96Shade implements CryptoShade { + private static final String IDENTIFIER = "aes-256-gcm96"; + private static final String ALGORITHM = "AES"; + private static final String TRANSFORMATION = "AES/GCM/NoPadding"; + private static final int GCM_TAG_LENGTH = 128; + private static final int GCM_NONCE_LENGTH = 12; + private static final byte[] NONCE = RandomUtil.randomBytes(GCM_NONCE_LENGTH); + + private static final String[] SENSITIVE_OPTIONS = + new String[]{"secret_key", "connection.user", "connection.password", "kafka.sasl.jaas.config", "kafka.ssl.keystore.password", "kafka.ssl.truststore.password", "kafka.ssl.key.password"}; + + private static final Key SECURITY_KEY = new SecretKeySpec(".........geedgenetworks.........".getBytes(StandardCharsets.UTF_8), ALGORITHM); + + @Override + public String[] sensitiveOptions() { + return SENSITIVE_OPTIONS; + } + + @Override + public String getIdentifier() { + return IDENTIFIER; + } + + @Override + public String encrypt(String content) { + String encryptedString = null; + try { + Cipher cipher = Cipher.getInstance(TRANSFORMATION); + GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, NONCE); + cipher.init(Cipher.ENCRYPT_MODE, SECURITY_KEY, gcmSpec); + byte[] encryptedBytes = cipher.doFinal(content.getBytes()); + byte[] combinedBytes = new byte[GCM_NONCE_LENGTH + encryptedBytes.length]; + System.arraycopy(NONCE, 0, combinedBytes, 0, GCM_NONCE_LENGTH); + System.arraycopy(encryptedBytes, 0, combinedBytes, GCM_NONCE_LENGTH, encryptedBytes.length); + encryptedString = Base64.getEncoder().encodeToString(combinedBytes); + } catch (Exception e) { + throw new RuntimeException(e); + } + return encryptedString; + } + + @Override + public String decrypt(String content) { + String decryptedString = null; + try { + byte[] combined = Base64.getDecoder().decode(content); + byte[] encryptedBytes = new byte[combined.length - GCM_NONCE_LENGTH]; + System.arraycopy(combined, 0, NONCE, 0, GCM_NONCE_LENGTH); + System.arraycopy(combined, GCM_NONCE_LENGTH, encryptedBytes, 0, encryptedBytes.length); + Cipher cipher = Cipher.getInstance(TRANSFORMATION); + GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, NONCE); + cipher.init(Cipher.DECRYPT_MODE, SECURITY_KEY, gcmSpec); + byte[] decryptedBytes = cipher.doFinal(encryptedBytes); + decryptedString = new String(decryptedBytes); + } catch (Exception e) { + throw new RuntimeException(e); + } + return decryptedString; + } +} diff --git a/groot-bootstrap/src/main/java/com/geedgenetworks/bootstrap/command/AESShade.java b/groot-bootstrap/src/main/java/com/geedgenetworks/bootstrap/command/AESShade.java index a2f4f56..37a8e5b 100644 --- a/groot-bootstrap/src/main/java/com/geedgenetworks/bootstrap/command/AESShade.java +++ b/groot-bootstrap/src/main/java/com/geedgenetworks/bootstrap/command/AESShade.java @@ -27,7 +27,7 @@ public class AESShade implements CryptoShade { @Override public String encrypt(String content) { - return SecureUtil.aes(SECURITY_KEY).encryptHex(content, StandardCharsets.UTF_8); + return SecureUtil.aes(SECURITY_KEY).encryptBase64(content, StandardCharsets.UTF_8); } @Override diff --git a/groot-bootstrap/src/main/java/com/geedgenetworks/bootstrap/command/SM4GCM96Shade.java b/groot-bootstrap/src/main/java/com/geedgenetworks/bootstrap/command/SM4GCM96Shade.java new file mode 100644 index 0000000..a6d27e4 --- /dev/null +++ b/groot-bootstrap/src/main/java/com/geedgenetworks/bootstrap/command/SM4GCM96Shade.java @@ -0,0 +1,73 @@ +package com.geedgenetworks.bootstrap.command; + +import cn.hutool.core.util.RandomUtil; +import com.geedgenetworks.common.crypto.CryptoShade; + +import javax.crypto.Cipher; +import javax.crypto.spec.GCMParameterSpec; +import javax.crypto.spec.SecretKeySpec; +import java.nio.charset.StandardCharsets; +import java.security.Key; +import java.util.Base64; + +public class SM4GCM96Shade implements CryptoShade { + private static final String IDENTIFIER = "sm4-gcm96"; + private static final String ALGORITHM = "SM4"; + private static final String TRANSFORMATION = "SM4/GCM/NoPadding"; + private static final int GCM_TAG_LENGTH = 128; + private static final int GCM_NONCE_LENGTH = 12; + private static final byte[] NONCE = RandomUtil.randomBytes(GCM_NONCE_LENGTH); + + private static final String[] SENSITIVE_OPTIONS = + new String[]{"connection.user", "connection.password", "kafka.sasl.jaas.config", "kafka.ssl.keystore.password", "kafka.ssl.truststore.password", "kafka.ssl.key.password"}; + + private static final Key SECURITY_KEY = new SecretKeySpec(".geedgenetworks.".getBytes(StandardCharsets.UTF_8), ALGORITHM); + + @Override + public String[] sensitiveOptions() { + return SENSITIVE_OPTIONS; + } + + @Override + public String getIdentifier() { + return IDENTIFIER; + } + + @Override + public String encrypt(String content) { + String encryptedString = null; + try { + + Cipher cipher = Cipher.getInstance(TRANSFORMATION); + GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, NONCE); + cipher.init(Cipher.ENCRYPT_MODE, SECURITY_KEY, gcmSpec); + byte[] encryptedBytes = cipher.doFinal(content.getBytes()); + byte[] combinedBytes = new byte[GCM_NONCE_LENGTH + encryptedBytes.length]; + System.arraycopy(NONCE, 0, combinedBytes, 0, GCM_NONCE_LENGTH); + System.arraycopy(encryptedBytes, 0, combinedBytes, GCM_NONCE_LENGTH, encryptedBytes.length); + encryptedString = Base64.getEncoder().encodeToString(combinedBytes); + } catch (Exception e) { + throw new RuntimeException(e); + } + return encryptedString; + } + + @Override + public String decrypt(String content) { + String decryptedString = null; + try { + byte[] combined = Base64.getDecoder().decode(content); + byte[] encryptedBytes = new byte[combined.length - GCM_NONCE_LENGTH]; + System.arraycopy(combined, 0, NONCE, 0, GCM_NONCE_LENGTH); + System.arraycopy(combined, GCM_NONCE_LENGTH, encryptedBytes, 0, encryptedBytes.length); + Cipher cipher = Cipher.getInstance(TRANSFORMATION); + GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, NONCE); + cipher.init(Cipher.DECRYPT_MODE, SECURITY_KEY, gcmSpec); + byte[] decryptedBytes = cipher.doFinal(encryptedBytes); + decryptedString = new String(decryptedBytes); + } catch (Exception e) { + throw new RuntimeException(e); + } + return decryptedString; + } +} diff --git a/groot-bootstrap/src/main/java/com/geedgenetworks/bootstrap/command/SM4Shade.java b/groot-bootstrap/src/main/java/com/geedgenetworks/bootstrap/command/SM4Shade.java index 6fd15bd..e274716 100644 --- a/groot-bootstrap/src/main/java/com/geedgenetworks/bootstrap/command/SM4Shade.java +++ b/groot-bootstrap/src/main/java/com/geedgenetworks/bootstrap/command/SM4Shade.java @@ -27,7 +27,7 @@ public class SM4Shade implements CryptoShade { @Override public String encrypt(String content) { - return SmUtil.sm4(SECURITY_KEY).encryptHex(content, StandardCharsets.UTF_8); + return SmUtil.sm4(SECURITY_KEY).encryptBase64(content, StandardCharsets.UTF_8); } @Override diff --git a/groot-bootstrap/src/main/resources/META-INF/services/com.geedgenetworks.common.crypto.CryptoShade b/groot-bootstrap/src/main/resources/META-INF/services/com.geedgenetworks.common.crypto.CryptoShade index 9c0d60c..273b40d 100644 --- a/groot-bootstrap/src/main/resources/META-INF/services/com.geedgenetworks.common.crypto.CryptoShade +++ b/groot-bootstrap/src/main/resources/META-INF/services/com.geedgenetworks.common.crypto.CryptoShade @@ -1,3 +1,6 @@ com.geedgenetworks.bootstrap.command.Base64Shade com.geedgenetworks.bootstrap.command.AESShade -com.geedgenetworks.bootstrap.command.SM4Shade
\ No newline at end of file +com.geedgenetworks.bootstrap.command.SM4Shade +com.geedgenetworks.bootstrap.command.AES128GCM96Shade +com.geedgenetworks.bootstrap.command.AES256GCM96Shade +com.geedgenetworks.bootstrap.command.SM4GCM96Shade
\ No newline at end of file diff --git a/groot-bootstrap/src/test/java/com/geedgenetworks/bootstrap/utils/CryptoShadeTest.java b/groot-bootstrap/src/test/java/com/geedgenetworks/bootstrap/utils/CryptoShadeTest.java index 18e84ae..f77ba44 100644 --- a/groot-bootstrap/src/test/java/com/geedgenetworks/bootstrap/utils/CryptoShadeTest.java +++ b/groot-bootstrap/src/test/java/com/geedgenetworks/bootstrap/utils/CryptoShadeTest.java @@ -11,6 +11,7 @@ import com.typesafe.config.ConfigValueFactory; import lombok.extern.slf4j.Slf4j; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; + import java.net.URISyntaxException; import java.net.URL; import java.nio.file.Paths; @@ -40,11 +41,11 @@ public class CryptoShadeTest { Assertions.assertNotNull(sinkObject.getJSONObject("clickhouse_sink") .getJSONObject("properties")); Assertions.assertEquals(sinkObject.getJSONObject("clickhouse_sink") - .getJSONObject("properties").isEmpty(), false); + .getJSONObject("properties").isEmpty(), false); Assertions.assertEquals(sinkObject.getJSONObject("clickhouse_sink") - .getJSONObject("properties").get("connection.user"),USERNAME); + .getJSONObject("properties").get("connection.user"), USERNAME); Assertions.assertNotNull(sinkObject.getJSONObject("clickhouse_sink") - .getJSONObject("properties").get("connection.password"), PASSWORD); + .getJSONObject("properties").get("connection.password"), PASSWORD); } @Test @@ -57,25 +58,49 @@ public class CryptoShadeTest { Assertions.assertEquals("Z3Jvb3RzdHJlYW1fcGFzc3dvcmQ=", encryptPassword); Assertions.assertEquals(decryptUsername, USERNAME); Assertions.assertEquals(decryptPassword, PASSWORD); + encryptUsername = CryptoShadeUtils.encryptOption("aes", USERNAME); decryptUsername = CryptoShadeUtils.decryptOption("aes", encryptUsername); encryptPassword = CryptoShadeUtils.encryptOption("aes", PASSWORD); decryptPassword = CryptoShadeUtils.decryptOption("aes", encryptPassword); - Assertions.assertEquals("ed986337dfdbe341be1d29702e6ae619", encryptUsername); - Assertions.assertEquals("159c7da83d988a9ec041d10a6bfbe221bcbaed6b62d9cc1b04ff51e633ebd105", encryptPassword); + Assertions.assertEquals("7ZhjN9/b40G+HSlwLmrmGQ==", encryptUsername); + Assertions.assertEquals("FZx9qD2Yip7AQdEKa/viIby67Wti2cwbBP9R5jPr0QU=", encryptPassword); Assertions.assertEquals(decryptUsername, USERNAME); Assertions.assertEquals(decryptPassword, PASSWORD); + encryptUsername = CryptoShadeUtils.encryptOption("sm4", USERNAME); decryptUsername = CryptoShadeUtils.decryptOption("sm4", encryptUsername); - Assertions.assertEquals("72ea74367a15cb96b0d1d42104149519", encryptUsername); + Assertions.assertEquals("cup0NnoVy5aw0dQhBBSVGQ==", encryptUsername); Assertions.assertEquals(decryptUsername, USERNAME); encryptPassword = CryptoShadeUtils.encryptOption("sm4", PASSWORD); decryptPassword = CryptoShadeUtils.decryptOption("sm4", encryptPassword); - Assertions.assertEquals("3876c7088d395bbbfa826e3648b6c9a022e7f80941c132313bde6dc8a7f2351f", encryptPassword); + Assertions.assertEquals("OHbHCI05W7v6gm42SLbJoCLn+AlBwTIxO95tyKfyNR8=", encryptPassword); + Assertions.assertEquals(decryptPassword, PASSWORD); + + System.out.println(CryptoShadeUtils.encryptOption("sm4", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"galaxy2019\";")); + System.out.println(CryptoShadeUtils.decryptOption("sm4", "f76480be84a8ee1b009504c6c56a5bed48239c348a468f94b4029a6a3148f51530b025d6dfa140af93b4c7c6fe0e3dce543773e779d272b5579555fbd3271e7fdbee088673a901b3f3b28e914a25f30a4a859d97594c5ea7d7c1dcefe8c62560baea32b6da0b767232ed8aca17af2dc6")); + System.out.println(CryptoShadeUtils.encryptOption("aes", "testuser")); + System.out.println(CryptoShadeUtils.encryptOption("aes", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"olap\" password=\"galaxy2019\";")); + + encryptUsername = CryptoShadeUtils.encryptOption("sm4-gcm96", USERNAME); + decryptUsername = CryptoShadeUtils.decryptOption("sm4-gcm96", encryptUsername); + encryptPassword = CryptoShadeUtils.encryptOption("sm4-gcm96", PASSWORD); + decryptPassword = CryptoShadeUtils.decryptOption("sm4-gcm96", encryptPassword); + Assertions.assertEquals(decryptUsername, USERNAME); + Assertions.assertEquals(decryptPassword, PASSWORD); + + encryptUsername = CryptoShadeUtils.encryptOption("aes-128-gcm96", USERNAME); + decryptUsername = CryptoShadeUtils.decryptOption("aes-128-gcm96", encryptUsername); + encryptPassword = CryptoShadeUtils.encryptOption("aes-128-gcm96", PASSWORD); + decryptPassword = CryptoShadeUtils.decryptOption("aes-128-gcm96", encryptPassword); + Assertions.assertEquals(decryptUsername, USERNAME); + Assertions.assertEquals(decryptPassword, PASSWORD); + + encryptUsername = CryptoShadeUtils.encryptOption("aes-256-gcm96", USERNAME); + decryptUsername = CryptoShadeUtils.decryptOption("aes-256-gcm96", encryptUsername); + encryptPassword = CryptoShadeUtils.encryptOption("aes-256-gcm96", PASSWORD); + decryptPassword = CryptoShadeUtils.decryptOption("aes-256-gcm96", encryptPassword); + Assertions.assertEquals(decryptUsername, USERNAME); Assertions.assertEquals(decryptPassword, PASSWORD); - System.out.println( CryptoShadeUtils.encryptOption("sm4", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"galaxy2019\";")); - System.out.println( CryptoShadeUtils.decryptOption("sm4", "f76480be84a8ee1b009504c6c56a5bed48239c348a468f94b4029a6a3148f51530b025d6dfa140af93b4c7c6fe0e3dce543773e779d272b5579555fbd3271e7fdbee088673a901b3f3b28e914a25f30a4a859d97594c5ea7d7c1dcefe8c62560baea32b6da0b767232ed8aca17af2dc6")); - System.out.println( CryptoShadeUtils.encryptOption("aes", "testuser")); - System.out.println( CryptoShadeUtils.encryptOption("aes", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"olap\" password=\"galaxy2019\";")); } } diff --git a/groot-bootstrap/src/test/resources/META-INF/services/com.geedgenetworks.common.crypto.CryptoShade b/groot-bootstrap/src/test/resources/META-INF/services/com.geedgenetworks.common.crypto.CryptoShade index 04adf41..273b40d 100644 --- a/groot-bootstrap/src/test/resources/META-INF/services/com.geedgenetworks.common.crypto.CryptoShade +++ b/groot-bootstrap/src/test/resources/META-INF/services/com.geedgenetworks.common.crypto.CryptoShade @@ -1,2 +1,6 @@ com.geedgenetworks.bootstrap.command.Base64Shade -com.geedgenetworks.bootstrap.command.AESShade
\ No newline at end of file +com.geedgenetworks.bootstrap.command.AESShade +com.geedgenetworks.bootstrap.command.SM4Shade +com.geedgenetworks.bootstrap.command.AES128GCM96Shade +com.geedgenetworks.bootstrap.command.AES256GCM96Shade +com.geedgenetworks.bootstrap.command.SM4GCM96Shade
\ No newline at end of file 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..51e2ff0 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); } @@ -118,12 +117,18 @@ public class CommonConfigDomProcessor extends AbstractDomConfigProcessor { String name = cleanNodeName(node); if (CommonConfigOptions.KMS_TYPE.key().equals(name)) { kmsConfig.setType(getTextContent(node)); + } else if (CommonConfigOptions.KMS_SECRET_KEY.key().equals(name)) { + kmsConfig.setSecretKey(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 +141,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..1c3f4d0 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."); @@ -68,47 +70,59 @@ public class CommonConfigOptions { .defaultValue("local") .withDescription("The type of KMS."); + public static final Option<String> KMS_SECRET_KEY = Options.key("secret_key") + .stringType() + .defaultValue("") + .withDescription("The type of KMS."); + public static final Option<String> KMS_URL = Options.key("url") .stringType() .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 username of KMS."); + + public static final Option<String> KMS_PASSWORD = Options.key("password") + .stringType() + .defaultValue("") + .withDescription("The access username of KMS."); + + public static final Option<String> KMS_DEFAULT_KEY_PATH = Options.key("default_key_path") .stringType() .defaultValue("") - .withDescription("The access token of KMS."); + .withDescription("The default key path of KMS."); - public static final Option<String> KMS_KEY_PATH = Options.key("key_path") + public static final Option<String> KMS_PLUGIN_KEY_PATH = Options.key("plugin_key_path") .stringType() .defaultValue("") - .withDescription("The key path of KMS."); + .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> SSL_CERT_FILE = Options.key("cert_file") + 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> 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..75a5b4c 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,11 @@ 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 secretKey = 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..d7818ab 100644 --- a/groot-common/src/main/resources/grootstream.yaml +++ b/groot-common/src/main/resources/grootstream.yaml @@ -11,6 +11,25 @@ grootstream: files: - 64af7077-eb9b-4b8f-80cf-2ceebc89bea9 - 004390bc-3135-4a6f-a492-3662ecb9e289 + + kms: +# local: +# type: local +# secret_key: .geedgenetworks. + 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 9950a64..edb1a0f 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 diff --git a/groot-core/pom.xml b/groot-core/pom.xml index 322f63d..3fb7793 100644 --- a/groot-core/pom.xml +++ b/groot-core/pom.xml @@ -119,6 +119,16 @@ <scope>provided</scope> </dependency> + <dependency> + <groupId>io.github.jopenlibs</groupId> + <artifactId>vault-java-driver</artifactId> + <version>6.2.0</version> + </dependency> + <dependency> + <groupId>org.bouncycastle</groupId> + <artifactId>bcpkix-jdk18on</artifactId> + <version>1.78.1</version> + </dependency> </dependencies> <build> diff --git a/groot-core/src/main/java/com/geedgenetworks/core/pojo/KmsKey.java b/groot-core/src/main/java/com/geedgenetworks/core/pojo/KmsKey.java new file mode 100644 index 0000000..2690254 --- /dev/null +++ b/groot-core/src/main/java/com/geedgenetworks/core/pojo/KmsKey.java @@ -0,0 +1,19 @@ +package com.geedgenetworks.core.pojo; + + +import lombok.Data; + +@Data +public class KmsKey { + + private byte[] keyData; + private int keyVersion; + + public KmsKey() { + } + + public KmsKey(byte[] keyData, int keyVersion) { + this.keyData = keyData; + this.keyVersion = keyVersion; + } +}
\ No newline at end of file diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/Encrypt.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/Encrypt.java new file mode 100644 index 0000000..cc05397 --- /dev/null +++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/Encrypt.java @@ -0,0 +1,159 @@ +package com.geedgenetworks.core.udf; + +import cn.hutool.core.util.ArrayUtil; +import com.alibaba.fastjson2.JSON; +import com.geedgenetworks.common.Constants; +import com.geedgenetworks.common.Event; +import com.geedgenetworks.common.config.CommonConfig; +import com.geedgenetworks.common.config.KmsConfig; +import com.geedgenetworks.common.config.SSLConfig; +import com.geedgenetworks.common.exception.CommonErrorCode; +import com.geedgenetworks.common.exception.GrootStreamRuntimeException; +import com.geedgenetworks.common.udf.ScalarFunction; +import com.geedgenetworks.common.udf.UDFContext; +import com.geedgenetworks.core.pojo.KmsKey; +import com.geedgenetworks.core.udf.encrypt.EncryptionAlgorithm; +import com.geedgenetworks.core.utils.*; +import com.geedgenetworks.utils.StringUtil; +import io.github.jopenlibs.vault.VaultException; +import lombok.extern.slf4j.Slf4j; +import org.apache.flink.api.common.functions.RuntimeContext; +import org.apache.flink.configuration.Configuration; + +import java.util.Arrays; +import java.util.Map; + +@Slf4j +public class Encrypt implements ScalarFunction { + + private String lookupFieldName; + private String outputFieldName; + private String identifier; + private String defaultVal; + private String type; + private transient SingleValueMap.Data<LoadIntervalDataUtil<String[]>> sensitiveFieldsData; + private transient SingleValueMap.Data<LoadIntervalDataUtil<KmsKey>> kmsKeyData; + private transient EncryptionAlgorithm encryptionAlgorithm; + + @Override + public void open(RuntimeContext runtimeContext, UDFContext udfContext) { + checkUdfContext(udfContext); + if (udfContext.getParameters().containsKey("default_val")) { + this.defaultVal = udfContext.getParameters().get("default_val").toString(); + } + this.lookupFieldName = udfContext.getLookup_fields().get(0); + this.outputFieldName = udfContext.getOutput_fields().get(0); + this.identifier = udfContext.getParameters().get("identifier").toString(); + + Configuration configuration = (Configuration) runtimeContext.getExecutionConfig().getGlobalJobParameters(); + CommonConfig commonConfig = JSON.parseObject(configuration.toMap().get(Constants.SYSPROP_GROOTSTREAM_CONFIG), CommonConfig.class); + Map<String, KmsConfig> kmsConfigs = commonConfig.getKmsConfig(); + if (kmsConfigs.isEmpty()) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Global parameter kms type is not null!"); + } else if (kmsConfigs.size() > 1) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Global parameter kms type is repeated!"); + } + KmsConfig kmsConfig = kmsConfigs.values().iterator().next(); + SSLConfig sslConfig = commonConfig.getSslConfig(); + Map<String, String> propertiesConfig = commonConfig.getPropertiesConfig(); + type = kmsConfig.getType(); + try { + encryptionAlgorithm = EncryptionAlgorithmUtils.getEncryptionAlgorithm(identifier); + if (encryptionAlgorithm == null) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Parameters identifier is illegal!"); + } + kmsKeyData = SingleValueMap.acquireData("kmsKeyData", + () -> LoadIntervalDataUtil.newInstance(() -> getKmsKey(kmsConfig, sslConfig, identifier), + LoadIntervalDataOptions.defaults("kmsKeyData", 60000)), LoadIntervalDataUtil::stop); + sensitiveFieldsData = SingleValueMap.acquireData("sensitiveFields", + () -> LoadIntervalDataUtil.newInstance(() -> getEncryptFields(propertiesConfig.get("projection.encrypt.schema.registry.uri")), + LoadIntervalDataOptions.defaults("sensitiveFields", 60000)), LoadIntervalDataUtil::stop); + KmsKey kmsKey = kmsKeyData.getData().data(); + if (encryptionAlgorithm.getSecretKeyLength() == kmsKey.getKeyData().length) { + encryptionAlgorithm.setKmsKey(kmsKey); + } else { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Global parameter kms secret Key requires " + encryptionAlgorithm.getSecretKeyLength() + " bytes!"); + } + } catch (Exception e) { + throw new GrootStreamRuntimeException(CommonErrorCode.UNSUPPORTED_OPERATION, "Initialization UDF Encrypt failed!", e); + } + } + + @Override + public Event evaluate(Event event) { + try { + KmsKey kmsKey = kmsKeyData.getData().data(); + if (kmsKey.getKeyVersion() != encryptionAlgorithm.getKmsKey().getKeyVersion() || !Arrays.equals(kmsKey.getKeyData(), encryptionAlgorithm.getKmsKey().getKeyData())) { + encryptionAlgorithm.setKmsKey(kmsKey); + } + if (ArrayUtil.contains(sensitiveFieldsData.getData().data(), lookupFieldName) && event.getExtractedFields().containsKey(lookupFieldName)) { + String value = (String) event.getExtractedFields().get(lookupFieldName); + if (StringUtil.isNotBlank(value)) { + String encryptResult = encryptionAlgorithm.encrypt(value); + if (StringUtil.isEmpty(encryptResult)) { + event.getExtractedFields().put(outputFieldName, StringUtil.isNotBlank(defaultVal) ? defaultVal : value); + } else { + if (KmsUtils.KMS_TYPE_VAULT.equals(type)) { + encryptResult = "vault:v" + encryptionAlgorithm.getKmsKey().getKeyVersion() + ":" + encryptResult; + } + event.getExtractedFields().put(outputFieldName, encryptResult); + } + } + } + } catch (Exception e) { + throw new RuntimeException(e); + } + return event; + } + + @Override + public String functionName() { + return "ENCRYPT"; + } + + @Override + public void close() { + if (sensitiveFieldsData != null) { + sensitiveFieldsData.release(); + } + if (kmsKeyData != null) { + kmsKeyData.release(); + } + } + + private void checkUdfContext(UDFContext udfContext) { + if (udfContext.getParameters() == null) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Missing required parameters"); + } + if (udfContext.getLookup_fields().size() != 1) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "The function lookup fields only support 1 value"); + } + if (udfContext.getOutput_fields().size() != 1) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "The function output fields only support 1 value"); + } + if (!udfContext.getParameters().containsKey("identifier")) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Parameters must contains identifier"); + } + } + + private KmsKey getKmsKey(KmsConfig kmsConfig, SSLConfig sslConfig, String identifier) throws VaultException { + KmsKey kmsKey = null; + if (KmsUtils.KMS_TYPE_VAULT.equals(kmsConfig.getType())) { + kmsKey = KmsUtils.getVaultKey(kmsConfig, sslConfig, identifier); + } else if (KmsUtils.KMS_TYPE_LOCAL.equals(kmsConfig.getType())) { + kmsKey = new KmsKey(kmsConfig.getSecretKey().getBytes(), 1); + } + return kmsKey; + } + + private String[] getEncryptFields(String url) { + String[] encryptFields = new String[]{"phone_number", "server_ip"}; +// try { +// String s = HttpClientPoolUtil.getInstance().httpGet(URI.create(URLUtil.normalize(url))); +// encryptFields = s.split(","); +// } catch (Exception e) { +// log.error("Get encrypt fields error! " + e.getMessage()); +// } + return encryptFields; + } +} diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/Hmac.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/Hmac.java new file mode 100644 index 0000000..0d2e1ca --- /dev/null +++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/Hmac.java @@ -0,0 +1,104 @@ +package com.geedgenetworks.core.udf; + +import cn.hutool.crypto.digest.HMac; +import cn.hutool.crypto.digest.HmacAlgorithm; +import com.geedgenetworks.common.Event; +import com.geedgenetworks.common.exception.CommonErrorCode; +import com.geedgenetworks.common.exception.GrootStreamRuntimeException; +import com.geedgenetworks.common.udf.ScalarFunction; +import com.geedgenetworks.common.udf.UDFContext; +import com.geedgenetworks.utils.StringUtil; +import lombok.extern.slf4j.Slf4j; +import org.apache.flink.api.common.functions.RuntimeContext; + +@Slf4j +public class Hmac implements ScalarFunction { + + private String lookupFieldName; + private String outputFieldName; + private String outputFormat; + private HMac hMac; + + @Override + public void open(RuntimeContext runtimeContext, UDFContext udfContext) { + checkUdfContext(udfContext); + String secretKey = udfContext.getParameters().get("secret_key").toString(); + String algorithm = "sha256"; + if (udfContext.getParameters().containsKey("algorithm")) { + algorithm = udfContext.getParameters().get("algorithm").toString(); + } + this.hMac = new HMac(getHmacAlgorithm(algorithm), secretKey.getBytes()); + this.lookupFieldName = udfContext.getLookup_fields().get(0); + this.outputFieldName = udfContext.getOutput_fields().get(0); + this.outputFormat = "base64"; + if (udfContext.getParameters().containsKey("output_format")) { + this.outputFormat = udfContext.getParameters().get("output_format").toString(); + } + } + + @Override + public Event evaluate(Event event) { + String encodeResult = ""; + String message = (String) event.getExtractedFields().get(lookupFieldName); + if (StringUtil.isNotBlank(message)) { + switch (outputFormat) { + case "hex": + encodeResult = hMac.digestHex(message); + break; + case "base64": + encodeResult = hMac.digestBase64(message, false); + break; + default: + encodeResult = hMac.digestBase64(message, false); + break; + } + } + event.getExtractedFields().put(outputFieldName, encodeResult); + return event; + } + + @Override + public String functionName() { + return "HMAC"; + } + + @Override + public void close() { + + } + + private void checkUdfContext(UDFContext udfContext) { + if (udfContext.getParameters() == null || udfContext.getOutput_fields() == null) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Missing required parameters"); + } + if (udfContext.getLookup_fields().size() != 1) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "The function lookup fields only support 1 value"); + } + if (udfContext.getOutput_fields().size() != 1) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "The function output fields only support 1 value"); + } + if (!udfContext.getParameters().containsKey("secret_key")) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "parameters must contains secret_key"); + } + } + + private String getHmacAlgorithm(String algorithm) { + if (StringUtil.containsIgnoreCase(algorithm, "sha256")) { + return HmacAlgorithm.HmacSHA256.getValue(); + } else if (StringUtil.containsIgnoreCase(algorithm, "sha1")) { + return HmacAlgorithm.HmacSHA1.getValue(); + } else if (StringUtil.containsIgnoreCase(algorithm, "md5")) { + return HmacAlgorithm.HmacMD5.getValue(); + } else if (StringUtil.containsIgnoreCase(algorithm, "sha384")) { + return HmacAlgorithm.HmacSHA384.getValue(); + } else if (StringUtil.containsIgnoreCase(algorithm, "sha512")) { + return HmacAlgorithm.HmacSHA512.getValue(); + } else if (StringUtil.containsIgnoreCase(algorithm, "sm3")) { + return HmacAlgorithm.HmacSM3.getValue(); + } else if (StringUtil.containsIgnoreCase(algorithm, "sm4")) { + return HmacAlgorithm.SM4CMAC.getValue(); + } else { + return HmacAlgorithm.HmacSHA256.getValue(); + } + } +} diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES128GCM96Algorithm.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES128GCM96Algorithm.java new file mode 100644 index 0000000..74be5a8 --- /dev/null +++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES128GCM96Algorithm.java @@ -0,0 +1,83 @@ +package com.geedgenetworks.core.udf.encrypt; + +import cn.hutool.core.util.RandomUtil; +import com.geedgenetworks.core.pojo.KmsKey; + +import javax.crypto.Cipher; +import javax.crypto.spec.GCMParameterSpec; +import javax.crypto.spec.SecretKeySpec; +import java.util.Base64; + +public class AES128GCM96Algorithm implements EncryptionAlgorithm { + private static final String IDENTIFIER = "aes-128-gcm96"; + private static final String ALGORITHM = "AES"; + private static final String TRANSFORMATION = "AES/GCM/NoPadding"; + private static final int GCM_TAG_LENGTH = 128; + private static final int GCM_96_NONCE_LENGTH = 12; + private static final int SECRET_KEY_LENGTH = 16; + + private final Cipher cipher; + private KmsKey kmsKey; + + public AES128GCM96Algorithm() throws Exception { + this.cipher = Cipher.getInstance(TRANSFORMATION); + } + + @Override + public String getIdentifier() { + return IDENTIFIER; + } + + @Override + public int getSecretKeyLength() { + return SECRET_KEY_LENGTH; + } + + @Override + public KmsKey getKmsKey() { + return kmsKey; + } + + @Override + public void setKmsKey(KmsKey kmsKey) { + this.kmsKey = kmsKey; + } + + @Override + public String encrypt(String content) { + String encryptedString = ""; + try { + byte[] nonce = RandomUtil.randomBytes(GCM_96_NONCE_LENGTH); + GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, nonce); + cipher.init(Cipher.ENCRYPT_MODE, new SecretKeySpec(kmsKey.getKeyData(), ALGORITHM), gcmSpec); + byte[] encryptedBytes = cipher.doFinal(content.getBytes()); + byte[] combinedBytes = new byte[GCM_96_NONCE_LENGTH + encryptedBytes.length]; + System.arraycopy(nonce, 0, combinedBytes, 0, GCM_96_NONCE_LENGTH); + System.arraycopy(encryptedBytes, 0, combinedBytes, GCM_96_NONCE_LENGTH, encryptedBytes.length); + encryptedString = Base64.getEncoder().encodeToString(combinedBytes); + } catch (Exception e) { + throw new RuntimeException(e); + } + return encryptedString; + } + + @Override + public String decrypt(String content) { + String decryptedString = ""; + try { + byte[] nonce = RandomUtil.randomBytes(GCM_96_NONCE_LENGTH); + Cipher cipher = Cipher.getInstance(TRANSFORMATION); + GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, nonce); + cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(kmsKey.getKeyData(), ALGORITHM), gcmSpec); + byte[] combined = Base64.getDecoder().decode(content); + byte[] encryptedBytes = new byte[combined.length - GCM_96_NONCE_LENGTH]; + System.arraycopy(combined, 0, nonce, 0, GCM_96_NONCE_LENGTH); + System.arraycopy(combined, GCM_96_NONCE_LENGTH, encryptedBytes, 0, encryptedBytes.length); + byte[] decryptedBytes = cipher.doFinal(encryptedBytes); + decryptedString = new String(decryptedBytes); + } catch (Exception e) { + throw new RuntimeException(e); + } + return decryptedString; + } +} diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES256GCM96Algorithm.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES256GCM96Algorithm.java new file mode 100644 index 0000000..64d88d9 --- /dev/null +++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES256GCM96Algorithm.java @@ -0,0 +1,83 @@ +package com.geedgenetworks.core.udf.encrypt; + +import cn.hutool.core.util.RandomUtil; +import com.geedgenetworks.core.pojo.KmsKey; + +import javax.crypto.Cipher; +import javax.crypto.spec.GCMParameterSpec; +import javax.crypto.spec.SecretKeySpec; +import java.util.Base64; + +public class AES256GCM96Algorithm implements EncryptionAlgorithm { + private static final String IDENTIFIER = "aes-256-gcm96"; + private static final String ALGORITHM = "AES"; + private static final String TRANSFORMATION = "AES/GCM/NoPadding"; + private static final int GCM_TAG_LENGTH = 128; + private static final int GCM_96_NONCE_LENGTH = 12; + private static final int SECRET_KEY_LENGTH = 32; + + private final Cipher cipher; + private KmsKey kmsKey; + + public AES256GCM96Algorithm() throws Exception { + this.cipher = Cipher.getInstance(TRANSFORMATION); + } + + @Override + public String getIdentifier() { + return IDENTIFIER; + } + + @Override + public int getSecretKeyLength() { + return SECRET_KEY_LENGTH; + } + + @Override + public KmsKey getKmsKey() { + return kmsKey; + } + + @Override + public void setKmsKey(KmsKey kmsKey) { + this.kmsKey = kmsKey; + } + + @Override + public String encrypt(String content) { + String encryptedString = ""; + try { + byte[] nonce = RandomUtil.randomBytes(GCM_96_NONCE_LENGTH); + GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, nonce); + cipher.init(Cipher.ENCRYPT_MODE, new SecretKeySpec(kmsKey.getKeyData(), ALGORITHM), gcmSpec); + byte[] encryptedBytes = cipher.doFinal(content.getBytes()); + byte[] combinedBytes = new byte[GCM_96_NONCE_LENGTH + encryptedBytes.length]; + System.arraycopy(nonce, 0, combinedBytes, 0, GCM_96_NONCE_LENGTH); + System.arraycopy(encryptedBytes, 0, combinedBytes, GCM_96_NONCE_LENGTH, encryptedBytes.length); + encryptedString = Base64.getEncoder().encodeToString(combinedBytes); + } catch (Exception e) { + throw new RuntimeException(e); + } + return encryptedString; + } + + @Override + public String decrypt(String content) { + String decryptedString = ""; + try { + byte[] nonce = RandomUtil.randomBytes(GCM_96_NONCE_LENGTH); + Cipher cipher = Cipher.getInstance(TRANSFORMATION); + GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, nonce); + cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(kmsKey.getKeyData(), ALGORITHM), gcmSpec); + byte[] combined = Base64.getDecoder().decode(content); + byte[] encryptedBytes = new byte[combined.length - GCM_96_NONCE_LENGTH]; + System.arraycopy(combined, 0, nonce, 0, GCM_96_NONCE_LENGTH); + System.arraycopy(combined, GCM_96_NONCE_LENGTH, encryptedBytes, 0, encryptedBytes.length); + byte[] decryptedBytes = cipher.doFinal(encryptedBytes); + decryptedString = new String(decryptedBytes); + } catch (Exception e) { + throw new RuntimeException(e); + } + return decryptedString; + } +} diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/SM4GCM96Algorithm.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/SM4GCM96Algorithm.java new file mode 100644 index 0000000..3c13820 --- /dev/null +++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/SM4GCM96Algorithm.java @@ -0,0 +1,83 @@ +package com.geedgenetworks.core.udf.encrypt; + +import cn.hutool.core.util.RandomUtil; +import com.geedgenetworks.core.pojo.KmsKey; + +import javax.crypto.Cipher; +import javax.crypto.spec.GCMParameterSpec; +import javax.crypto.spec.SecretKeySpec; +import java.util.Base64; + +public class SM4GCM96Algorithm implements EncryptionAlgorithm { + private static final String IDENTIFIER = "sm4-gcm96"; + private static final String ALGORITHM = "SM4"; + private static final String TRANSFORMATION = "SM4/GCM/NoPadding"; + private static final int GCM_TAG_LENGTH = 128; + private static final int GCM_96_NONCE_LENGTH = 12; + private static final int SECRET_KEY_LENGTH = 16; + + private final Cipher cipher; + private KmsKey kmsKey; + + public SM4GCM96Algorithm() throws Exception { + this.cipher = Cipher.getInstance(TRANSFORMATION); + } + + @Override + public String getIdentifier() { + return IDENTIFIER; + } + + @Override + public int getSecretKeyLength() { + return SECRET_KEY_LENGTH; + } + + @Override + public KmsKey getKmsKey() { + return kmsKey; + } + + @Override + public void setKmsKey(KmsKey kmsKey) { + this.kmsKey = kmsKey; + } + + @Override + public String encrypt(String content) { + String encryptedString = ""; + try { + byte[] nonce = RandomUtil.randomBytes(GCM_96_NONCE_LENGTH); + GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, nonce); + cipher.init(Cipher.ENCRYPT_MODE, new SecretKeySpec(kmsKey.getKeyData(), ALGORITHM), gcmSpec); + byte[] encryptedBytes = cipher.doFinal(content.getBytes()); + byte[] combinedBytes = new byte[GCM_96_NONCE_LENGTH + encryptedBytes.length]; + System.arraycopy(nonce, 0, combinedBytes, 0, GCM_96_NONCE_LENGTH); + System.arraycopy(encryptedBytes, 0, combinedBytes, GCM_96_NONCE_LENGTH, encryptedBytes.length); + encryptedString = Base64.getEncoder().encodeToString(combinedBytes); + } catch (Exception e) { + throw new RuntimeException(e); + } + return encryptedString; + } + + @Override + public String decrypt(String content) { + String decryptedString = ""; + try { + byte[] nonce = RandomUtil.randomBytes(GCM_96_NONCE_LENGTH); + Cipher cipher = Cipher.getInstance(TRANSFORMATION); + GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, nonce); + cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(kmsKey.getKeyData(), ALGORITHM), gcmSpec); + byte[] combined = Base64.getDecoder().decode(content); + byte[] encryptedBytes = new byte[combined.length - GCM_96_NONCE_LENGTH]; + System.arraycopy(combined, 0, nonce, 0, GCM_96_NONCE_LENGTH); + System.arraycopy(combined, GCM_96_NONCE_LENGTH, encryptedBytes, 0, encryptedBytes.length); + byte[] decryptedBytes = cipher.doFinal(encryptedBytes); + decryptedString = new String(decryptedBytes); + } catch (Exception e) { + throw new RuntimeException(e); + } + return decryptedString; + } +} diff --git a/groot-core/src/main/java/com/geedgenetworks/core/utils/KmsUtils.java b/groot-core/src/main/java/com/geedgenetworks/core/utils/KmsUtils.java new file mode 100644 index 0000000..8e6a345 --- /dev/null +++ b/groot-core/src/main/java/com/geedgenetworks/core/utils/KmsUtils.java @@ -0,0 +1,72 @@ +package com.geedgenetworks.core.utils; + +import cn.hutool.core.util.StrUtil; +import com.geedgenetworks.common.config.KmsConfig; +import com.geedgenetworks.common.config.SSLConfig; +import com.geedgenetworks.core.pojo.KmsKey; +import io.github.jopenlibs.vault.SslConfig; +import io.github.jopenlibs.vault.Vault; +import io.github.jopenlibs.vault.VaultConfig; +import io.github.jopenlibs.vault.VaultException; +import io.github.jopenlibs.vault.json.JsonObject; +import io.github.jopenlibs.vault.response.AuthResponse; +import io.github.jopenlibs.vault.response.LogicalResponse; +import lombok.extern.slf4j.Slf4j; + +import java.io.File; +import java.util.Base64; + +@Slf4j +public class KmsUtils { + public static final String KMS_TYPE_LOCAL = "local"; + public static final String KMS_TYPE_VAULT = "vault"; + + public static KmsKey getVaultKey(KmsConfig kmsConfig, SSLConfig sslConfig, String identifier) throws VaultException { + Vault vault = getVaultClient(kmsConfig, sslConfig); + String exportKeyPath; + if (EncryptionAlgorithmUtils.ALGORITHM_SM4_GCM96_NAME.equals(identifier)) { + exportKeyPath = kmsConfig.getPluginKeyPath() + "/export/encryption-key/" + identifier; + } else { + exportKeyPath = kmsConfig.getDefaultKeyPath() + "/export/encryption-key/" + identifier; + } + LogicalResponse exportResponse = vault.logical().read(exportKeyPath); + if (exportResponse.getRestResponse().getStatus() == 200) { + JsonObject keys = exportResponse.getDataObject().get("keys").asObject(); + return new KmsKey(Base64.getDecoder().decode(StrUtil.trim(keys.get(keys.size() + "").asString(), '"')), keys.size()); + } else { + log.error("Get kms key error! code: {} body: {}", exportResponse.getRestResponse().getStatus(), new String(exportResponse.getRestResponse().getBody())); + return null; + } + } + + public static Vault getVaultClient(KmsConfig kmsConfig, SSLConfig sslConfig) throws VaultException { + String username = kmsConfig.getUsername(); + String password = kmsConfig.getPassword(); + String url = kmsConfig.getUrl(); + boolean skipVerification = true; + String caCertificatePath = null; + String certificatePath = null; + String privateKeyPath = null; + if (sslConfig != null) { + skipVerification = sslConfig.getSkipVerification(); + caCertificatePath = sslConfig.getCaCertificatePath(); + certificatePath = sslConfig.getCertificatePath(); + privateKeyPath = sslConfig.getPrivateKeyPath(); + } + SslConfig vaultSslConfig = new SslConfig().verify(!skipVerification).build(); + if (!skipVerification) { + vaultSslConfig.pemFile(new File(caCertificatePath)) + .clientPemFile(new File(certificatePath)) + .clientKeyPemFile(new File(privateKeyPath)) + .build(); + } + VaultConfig config = new VaultConfig() + .address(url) + .engineVersion(1) + .sslConfig(vaultSslConfig) + .build(); + AuthResponse authResponse = Vault.create(config).auth().loginByUserPass(username, password); + config.token(authResponse.getAuthClientToken()); + return Vault.create(config); + } +} diff --git a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/HmacFunctionTest.java b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/HmacFunctionTest.java new file mode 100644 index 0000000..d2219d8 --- /dev/null +++ b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/HmacFunctionTest.java @@ -0,0 +1,136 @@ +package com.geedgenetworks.core.udf.test.simple; + +import com.geedgenetworks.common.Event; +import com.geedgenetworks.common.exception.GrootStreamRuntimeException; +import com.geedgenetworks.common.udf.UDFContext; +import com.geedgenetworks.core.udf.Hmac; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +public class HmacFunctionTest { + + private static final String SECRET_KEY = ".geedgenetworks."; + private static final String DATA = "13812345678"; + private static UDFContext udfContext; + + @BeforeAll + public static void setUp() { + udfContext = new UDFContext(); + udfContext.setLookup_fields(Collections.singletonList("phone_number")); + udfContext.setOutput_fields(Collections.singletonList("phone_number_mac")); + } + + @Test + public void testHmacAsBase64() { + Map<String, Object> map = new HashMap<>(); + map.put("secret_key", SECRET_KEY); + map.put("algorithm", "sha256"); + map.put("output_format", "base64"); + udfContext.setParameters(map); + Hmac hmac = new Hmac(); + hmac.open(null, udfContext); + Event event = new Event(); + Map<String, Object> extractedFields = new HashMap<>(); + extractedFields.put("phone_number", DATA); + event.setExtractedFields(extractedFields); + Event result1 = hmac.evaluate(event); + assertEquals("zaj6UKovIsDahIBeRZ2PmgPIfDEr900F2xWu+iQfFrw=", result1.getExtractedFields().get("phone_number_mac")); + } + + @Test + public void testHmacAsHex() { + Map<String, Object> map = new HashMap<>(); + map.put("secret_key", SECRET_KEY); + map.put("algorithm", "sha256"); + map.put("output_format", "hex"); + udfContext.setParameters(map); + Hmac hmac = new Hmac(); + hmac.open(null, udfContext); + Event event = new Event(); + Map<String, Object> extractedFields = new HashMap<>(); + extractedFields.put("phone_number", DATA); + event.setExtractedFields(extractedFields); + Event result1 = hmac.evaluate(event); + assertEquals("cda8fa50aa2f22c0da84805e459d8f9a03c87c312bf74d05db15aefa241f16bc", result1.getExtractedFields().get("phone_number_mac")); + } + + @Test + public void testHmacAlgorithm() { + Map<String, Object> map = new HashMap<>(); + map.put("secret_key", SECRET_KEY); + map.put("algorithm", "sm4"); + map.put("output_format", "base64"); + udfContext.setParameters(map); + Hmac hmac = new Hmac(); + hmac.open(null, udfContext); + Event event = new Event(); + Map<String, Object> extractedFields = new HashMap<>(); + extractedFields.put("phone_number", DATA); + event.setExtractedFields(extractedFields); + Event result = hmac.evaluate(event); + assertEquals("QX1q4Y7y3quYCDje9BuSjg==", result.getExtractedFields().get("phone_number_mac")); + + map = new HashMap<>(); + map.put("secret_key", SECRET_KEY); + map.put("algorithm", "sha1"); + map.put("output_format", "base64"); + udfContext.setParameters(map); + hmac = new Hmac(); + hmac.open(null, udfContext); + event.setExtractedFields(extractedFields); + result = hmac.evaluate(event); + assertEquals("NB1b1TsVZ95/0sE+d/6kdtyUFh0=", result.getExtractedFields().get("phone_number_mac")); + + map = new HashMap<>(); + map.put("secret_key", SECRET_KEY); + map.put("algorithm", "sm3"); + map.put("output_format", "base64"); + udfContext.setParameters(map); + hmac = new Hmac(); + hmac.open(null, udfContext); + event.setExtractedFields(extractedFields); + result = hmac.evaluate(event); + assertEquals("BbQNpwLWE3rkaI1WlPBJgYeD14UyL2OwTxiEoTNA3UU=", result.getExtractedFields().get("phone_number_mac")); + + map = new HashMap<>(); + map.put("secret_key", SECRET_KEY); + map.put("algorithm", "md5"); + map.put("output_format", "base64"); + udfContext.setParameters(map); + hmac = new Hmac(); + hmac.open(null, udfContext); + event.setExtractedFields(extractedFields); + result = hmac.evaluate(event); + assertEquals("BQZzRqD3ZR/nJsDIOO4dBg==", result.getExtractedFields().get("phone_number_mac")); + + map = new HashMap<>(); + map.put("secret_key", SECRET_KEY); + map.put("algorithm", "sha512"); + map.put("output_format", "base64"); + udfContext.setParameters(map); + hmac = new Hmac(); + hmac.open(null, udfContext); + event.setExtractedFields(extractedFields); + result = hmac.evaluate(event); + assertEquals("DWrndzlcqf2qvFTbuDC1gZCGmRhuAUayfsxEqr2ZlpY/QOr9HgGUZNOfytRfA4VT8OZK0BwHwcAg5pgGBvPQ4A==", result.getExtractedFields().get("phone_number_mac")); + } + + @Test + public void testHmacError() { + Map<String, Object> map = new HashMap<>(); + map.put("secret_key", SECRET_KEY); + map.put("algorithm", "sha256"); + map.put("output_format", "hex"); + udfContext.setParameters(map); + Hmac hmac = new Hmac(); + udfContext.getParameters().remove("secret_key"); + assertThrows(GrootStreamRuntimeException.class, () -> hmac.open(null, udfContext)); + } +} |