diff options
Diffstat (limited to 'groot-core')
20 files changed, 753 insertions, 322 deletions
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/pojo/DataEncryptionKey.java b/groot-core/src/main/java/com/geedgenetworks/core/pojo/DataEncryptionKey.java new file mode 100644 index 0000000..8c0d516 --- /dev/null +++ b/groot-core/src/main/java/com/geedgenetworks/core/pojo/DataEncryptionKey.java @@ -0,0 +1,19 @@ +package com.geedgenetworks.core.pojo; + + +import lombok.Data; + +@Data +public class DataEncryptionKey { + + private byte[] data; + private int version; + + public DataEncryptionKey() { + } + + public DataEncryptionKey(byte[] data, int version) { + this.data = data; + this.version = version; + } +}
\ No newline at end of file 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 deleted file mode 100644 index 2690254..0000000 --- a/groot-core/src/main/java/com/geedgenetworks/core/pojo/KmsKey.java +++ /dev/null @@ -1,19 +0,0 @@ -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/Domain.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/Domain.java index 74816f5..9046472 100644 --- a/groot-core/src/main/java/com/geedgenetworks/core/udf/Domain.java +++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/Domain.java @@ -100,10 +100,9 @@ public class Domain implements ScalarFunction { throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, result.getMsg()); } - if(!udfContext.getParameters().containsKey(UDFContextConfigOptions.PARAMETERS_OPTION.key())){ - throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, String.format( - "UDF: %s, [%s] Option should be specified.", - udfContext.getFunction(), UDFContextConfigOptions.PARAMETERS.key())); + result = CheckUDFContextUtil.checkParametersContainsKeys(udfContext, UDFContextConfigOptions.PARAMETERS_OPTION.key()); + if (!result.isSuccess()) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, result.getMsg()); } String optionValue = udfContext.getParameters().get(UDFContextConfigOptions.PARAMETERS_OPTION.key()).toString(); 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 index c8e21b2..2fa0804 100644 --- a/groot-core/src/main/java/com/geedgenetworks/core/udf/Encrypt.java +++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/Encrypt.java @@ -1,35 +1,28 @@ package com.geedgenetworks.core.udf; import cn.hutool.core.util.URLUtil; -import cn.hutool.json.JSONArray; import cn.hutool.json.JSONObject; import cn.hutool.json.JSONUtil; 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.config.*; 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.pojo.DataEncryptionKey; +import com.geedgenetworks.core.udf.encrypt.Crypto; import com.geedgenetworks.core.utils.*; import com.geedgenetworks.shaded.org.apache.http.HttpHeaders; import com.geedgenetworks.shaded.org.apache.http.HttpStatus; import com.geedgenetworks.shaded.org.apache.http.message.BasicHeader; -import com.geedgenetworks.utils.StringUtil; import lombok.extern.slf4j.Slf4j; import org.apache.flink.api.common.functions.RuntimeContext; import org.apache.flink.configuration.Configuration; -import java.io.IOException; import java.net.URI; -import java.util.Arrays; -import java.util.Map; -import java.util.Set; +import java.util.*; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -40,86 +33,122 @@ public class Encrypt implements ScalarFunction { private String outputFieldName; private String identifier; private String defaultVal; - private String type; - private transient SingleValueMap.Data<LoadIntervalDataUtil<Set<String>>> sensitiveFieldsData; - private transient SingleValueMap.Data<LoadIntervalDataUtil<KmsKey>> kmsKeyData; - private transient EncryptionAlgorithm encryptionAlgorithm; + private String kmsType; + private transient SingleValueMap.Data<LoadIntervalDataUtil<Set<String>>> sensitiveFieldState; + private transient SingleValueMap.Data<LoadIntervalDataUtil<DataEncryptionKey>> dekState; + private transient Crypto crypto; @Override public void open(RuntimeContext runtimeContext, UDFContext udfContext) { - checkUdfContext(udfContext); - if (udfContext.getParameters().containsKey("default_val")) { - this.defaultVal = udfContext.getParameters().get("default_val").toString(); - } + checkConfig(udfContext); + configureParameters(udfContext); + initializeCrypto(runtimeContext); + } + + private void configureParameters(UDFContext udfContext) { + this.defaultVal = Optional.ofNullable(udfContext.getParameters().get("default_val")) + .map(Object::toString).orElse(""); this.lookupFieldName = udfContext.getLookupFields().get(0); this.outputFieldName = udfContext.getOutputFields().get(0); - this.identifier = udfContext.getParameters().get("identifier").toString(); + this.identifier = udfContext.getParameters().get(UDFContextConfigOptions.PARAMETERS_IDENTIFIER.key()).toString(); + } + + private void initializeCrypto(RuntimeContext runtimeContext) { Configuration configuration = (Configuration) runtimeContext.getExecutionConfig().getGlobalJobParameters(); CommonConfig commonConfig = JSON.parseObject(configuration.toMap().get(Constants.SYSPROP_GROOTSTREAM_CONFIG), CommonConfig.class); KmsConfig kmsConfig = commonConfig.getKmsConfig().get(configuration.toMap().get(Constants.SYSPROP_KMS_TYPE_CONFIG)); - SSLConfig sslConfig = commonConfig.getSslConfig(); + ClientSSLConfig sslConfig = commonConfig.getSslConfig(); Map<String, String> propertiesConfig = commonConfig.getPropertiesConfig(); - type = kmsConfig.getType(); + this.kmsType = kmsConfig.getType(); try { - encryptionAlgorithm = EncryptionAlgorithmUtils.createEncryptionAlgorithm(identifier); - if (encryptionAlgorithm == null) { - throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Parameters identifier is illegal!"); + this.crypto = CryptoProvider.createEncryptionAlgorithm(identifier); + initializeDataEncryptionKeyIfNeeded(kmsConfig, sslConfig, propertiesConfig); + initializeSensitiveFieldsIfNeeded(propertiesConfig); + } catch (Exception e) { + throw new GrootStreamRuntimeException(CommonErrorCode.UNSUPPORTED_OPERATION, "Initialization UDF Encrypt failed!", e); + } + } + + private void initializeDataEncryptionKeyIfNeeded(KmsConfig kmsConfig, ClientSSLConfig sslConfig, Map<String, String> propertiesConfig) throws Exception { + if (KMSUtils.KMS_TYPE_VAULT.equals(kmsType)) { + this.dekState = SingleValueMap.acquireData("dekState", + () -> LoadIntervalDataUtil.newInstance( + () -> KMSUtils.getVaultKey(kmsConfig, sslConfig, identifier), + LoadIntervalDataOptions.defaults("dekState", + Integer.parseInt(propertiesConfig.getOrDefault(Constants.SYSPROP_ENCRYPT_KMS_KEY_SCHEDULER_INTERVAL_NAME, "1")) * 60000L)), + LoadIntervalDataUtil::stop); + + DataEncryptionKey dek = dekState.getData().data(); + + if (dek == null) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, String.format("URI: %s, Unable to get KMS Secret Key!", kmsConfig.getUrl())); } - if (!type.equals(KmsUtils.KMS_TYPE_LOCAL)) { - kmsKeyData = SingleValueMap.acquireData("kmsKeyData", - () -> LoadIntervalDataUtil.newInstance(() -> KmsUtils.getVaultKey(kmsConfig, sslConfig, identifier), - LoadIntervalDataOptions.defaults("kmsKeyData", Integer.parseInt(propertiesConfig.getOrDefault(Constants.SYSPROP_ENCRYPT_KMS_KEY_SCHEDULER_INTERVAL_NAME, "5")) * 60000L)), - LoadIntervalDataUtil::stop); - KmsKey kmsKey = kmsKeyData.getData().data(); - if (kmsKey == null) { - throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Initialization UDF Encrypt failed!"); - } - if (encryptionAlgorithm.getSecretKeyLength() != kmsKey.getKeyData().length) { - throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Global parameter kms secret Key requires " + encryptionAlgorithm.getSecretKeyLength() + " bytes!"); - } - encryptionAlgorithm.setKmsKey(kmsKey); + + if (crypto.getSecretKeyLength() != dek.getData().length) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, String.format("Version: %s, [%s] KMS Secret Key requires %s bytes!", + dek.getVersion(), crypto.getIdentifier(), crypto.getSecretKeyLength())); } - sensitiveFieldsData = SingleValueMap.acquireData("sensitiveFields", - () -> LoadIntervalDataUtil.newInstance(() -> getSensitiveFields(propertiesConfig.get("projection.encrypt.schema.registry.uri")), - LoadIntervalDataOptions.defaults("sensitiveFields", Integer.parseInt(propertiesConfig.getOrDefault(Constants.SYSPROP_ENCRYPT_SENSITIVE_FIELDS_SCHEDULER_INTERVAL_NAME, "5")) * 60000L)), + + crypto.setDataEncryptionKey(dek); + } + } + + private void initializeSensitiveFieldsIfNeeded(Map<String, String> propertiesConfig) throws Exception { + String schemaUri = propertiesConfig.get("projection.encrypt.schema.registry.uri"); + if (schemaUri != null) { + this.sensitiveFieldState = SingleValueMap.acquireData("sensitiveFieldState", + () -> LoadIntervalDataUtil.newInstance( + () -> getSensitiveFields(schemaUri), + LoadIntervalDataOptions.defaults("sensitiveFieldState", + Integer.parseInt(propertiesConfig.getOrDefault(Constants.SYSPROP_ENCRYPT_SENSITIVE_FIELDS_SCHEDULER_INTERVAL_NAME, "1")) * 60000L)), LoadIntervalDataUtil::stop); - if (sensitiveFieldsData.getData().data() == null) { - throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Initialization UDF Encrypt failed!"); + if (sensitiveFieldState.getData().data() == null) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, String.format("URI: %s, Unable to get encrypted fields!", schemaUri)); } - } catch (Exception e) { - throw new GrootStreamRuntimeException(CommonErrorCode.UNSUPPORTED_OPERATION, "Initialization UDF Encrypt failed!", e); } } + @Override public Event evaluate(Event event) { try { - if (!type.equals(KmsUtils.KMS_TYPE_LOCAL)) { - KmsKey kmsKey = kmsKeyData.getData().data(); - if (kmsKey.getKeyVersion() != encryptionAlgorithm.getKmsKey().getKeyVersion() || !Arrays.equals(kmsKey.getKeyData(), encryptionAlgorithm.getKmsKey().getKeyData())) { - encryptionAlgorithm.setKmsKey(kmsKey); - } - } - if (sensitiveFieldsData.getData().data().contains(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); - } - } + refreshDataEncryptionKey(); + + if (isSensitiveField(lookupFieldName)) { + executeEncrypt(event); } } catch (Exception e) { - throw new RuntimeException(e); + log.error("Encrypt failed!", e); } return event; } + private void refreshDataEncryptionKey() throws Exception { + if (dekState == null || dekState.getData().data() == null) { + return; + } + DataEncryptionKey dek = dekState.getData().data(); + if (!Arrays.equals(dek.getData(), crypto.getDataEncryptionKey().getData())) { + crypto.setDataEncryptionKey(dek); + } + + } + + private boolean isSensitiveField(String fieldName) throws Exception { + return sensitiveFieldState == null || sensitiveFieldState.getData().data().contains(fieldName); + } + + private void executeEncrypt(Event event) { + Object value = event.getExtractedFields().get(lookupFieldName); + if (value != null) { + String encryptResult = Optional.ofNullable(crypto.encrypt(value.toString())).orElse(defaultVal); + if (KMSUtils.KMS_TYPE_VAULT.equals(kmsType)) { + encryptResult = "vault:v" + crypto.getDataEncryptionKey().getVersion() + ":" + encryptResult; + } + event.getExtractedFields().put(outputFieldName, encryptResult); + } + } + @Override public String functionName() { return "ENCRYPT"; @@ -127,41 +156,65 @@ public class Encrypt implements ScalarFunction { @Override public void close() { - if (sensitiveFieldsData != null) { - sensitiveFieldsData.release(); + if (sensitiveFieldState != null) { + sensitiveFieldState.release(); } - if (kmsKeyData != null) { - kmsKeyData.release(); + if (dekState != null) { + dekState.release(); } } - private void checkUdfContext(UDFContext udfContext) { - if (udfContext.getParameters() == null) { - throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Missing required parameters"); + @Override + public void checkConfig(UDFContext udfContext) { + + CheckResult result = CheckUDFContextUtil.checkAllExists(udfContext, + UDFContextConfigOptions.LOOKUP_FIELDS.key(), + UDFContextConfigOptions.OUTPUT_FIELDS.key(), + UDFContextConfigOptions.PARAMETERS.key()); + + if (!result.isSuccess()) { + throw new GrootStreamRuntimeException(CommonErrorCode.CONFIG_VALIDATION_FAILED, result.getMsg()); } - if (udfContext.getLookupFields().size() != 1) { - throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "The function lookup fields only support 1 value"); + + result = CheckUDFContextUtil.checkCollectionSingleItemExists(udfContext, UDFContextConfigOptions.LOOKUP_FIELDS.key()); + if (!result.isSuccess()) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, result.getMsg()); } - if (udfContext.getOutputFields().size() != 1) { - throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "The function output fields only support 1 value"); + + result = CheckUDFContextUtil.checkCollectionSingleItemExists(udfContext, UDFContextConfigOptions.OUTPUT_FIELDS.key()); + if (!result.isSuccess()) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, result.getMsg()); } - if (!udfContext.getParameters().containsKey("identifier")) { - throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Parameters must contains identifier"); + + result = CheckUDFContextUtil.checkParametersContainsKeys(udfContext, UDFContextConfigOptions.PARAMETERS_IDENTIFIER.key()); + if (!result.isSuccess()) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, result.getMsg()); } + + } - public Set<String> getSensitiveFields(String url) throws IOException { - Set<String> sensitiveFieldsSet; - String sensitiveFieldsStr = HttpClientPoolUtil.getInstance().httpGet(URI.create(URLUtil.normalize(url)), new BasicHeader(HttpHeaders.CONTENT_TYPE, "application/x-www-form-urlencoded")); - JSONObject sensitiveFieldsJson = JSONUtil.parseObj(sensitiveFieldsStr); - if (sensitiveFieldsJson.getInt("status", HttpStatus.SC_INTERNAL_SERVER_ERROR) == HttpStatus.SC_OK) { - JSONArray sensitiveFieldsJsonArr = sensitiveFieldsJson.getJSONArray("data"); - sensitiveFieldsSet = IntStream.range(0, sensitiveFieldsJsonArr.size()) - .mapToObj(sensitiveFieldsJsonArr::getStr) - .collect(Collectors.toSet()); - } else { - throw new GrootStreamRuntimeException(CommonErrorCode.UNSUPPORTED_OPERATION, "Get encrypt fields error! Error message: " + sensitiveFieldsStr); + // If accessing the schema URI fails, the last known state of sensitiveFieldState will be used + private Set<String> getSensitiveFields(String url) throws Exception { + try { + String result = HttpClientPoolUtil.getInstance() + .httpGet(URI.create(URLUtil.normalize(url)), new BasicHeader(HttpHeaders.CONTENT_TYPE, "application/x-www-form-urlencoded")); + JSONObject resultJson = JSONUtil.parseObj(result); + int statusCode = resultJson.getInt("status", HttpStatus.SC_INTERNAL_SERVER_ERROR); + if (statusCode == HttpStatus.SC_OK) { + return Optional.ofNullable(resultJson.getJSONArray("data")) + .map(jsonArray -> IntStream.range(0, jsonArray.size()) + .mapToObj(jsonArray::getStr) + .collect(Collectors.toSet())) + .orElseGet(Collections::emptySet); + } else { + log.error("Get sensitive fields error! Error message: {}", result); + return sensitiveFieldState.getData().data(); + } + } catch (Exception e) { + log.error("Get sensitive fields error! Error message: {}", e.getMessage()); + return sensitiveFieldState.getData().data(); // return the previous value } - return sensitiveFieldsSet; } + } 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 index 098cdef..a18d361 100644 --- a/groot-core/src/main/java/com/geedgenetworks/core/udf/Hmac.java +++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/Hmac.java @@ -3,6 +3,9 @@ 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.config.CheckResult; +import com.geedgenetworks.common.config.CheckUDFContextUtil; +import com.geedgenetworks.common.config.UDFContextConfigOptions; import com.geedgenetworks.common.exception.CommonErrorCode; import com.geedgenetworks.common.exception.GrootStreamRuntimeException; import com.geedgenetworks.common.udf.ScalarFunction; @@ -21,7 +24,7 @@ public class Hmac implements ScalarFunction { @Override public void open(RuntimeContext runtimeContext, UDFContext udfContext) { - checkUdfContext(udfContext); + checkConfig(udfContext); String secretKey = udfContext.getParameters().get("secret_key").toString(); String algorithm = "sha256"; if (udfContext.getParameters().containsKey("algorithm")) { @@ -39,21 +42,22 @@ public class Hmac implements ScalarFunction { @Override public Event evaluate(Event event) { String encodeResult = ""; - String message = (String) event.getExtractedFields().get(lookupFieldName); - if (StringUtil.isNotBlank(message)) { + Object value = event.getExtractedFields().get(lookupFieldName); + if (StringUtil.isNotEmpty(value)) { switch (outputFormat) { case "hex": - encodeResult = hMac.digestHex(message); + encodeResult = hMac.digestHex(value.toString()); break; case "base64": - encodeResult = hMac.digestBase64(message, false); + encodeResult = hMac.digestBase64(value.toString(), false); break; default: - encodeResult = hMac.digestBase64(message, false); + encodeResult = hMac.digestBase64(value.toString(), false); break; } + event.getExtractedFields().put(outputFieldName, encodeResult); } - event.getExtractedFields().put(outputFieldName, encodeResult); + return event; } @@ -67,21 +71,25 @@ public class Hmac implements ScalarFunction { } - private void checkUdfContext(UDFContext udfContext) { - if (udfContext.getParameters() == null || udfContext.getOutputFields() == null) { - throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Missing required parameters"); - } - if (udfContext.getLookupFields().size() != 1) { - throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "The function lookup fields only support 1 value"); - } - if (udfContext.getOutputFields().size() != 1) { - throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "The function output fields only support 1 value"); + @Override + public void checkConfig(UDFContext udfContext) { + + CheckResult result = CheckUDFContextUtil.checkAllExists(udfContext, + UDFContextConfigOptions.LOOKUP_FIELDS.key(), + UDFContextConfigOptions.OUTPUT_FIELDS.key(), + UDFContextConfigOptions.PARAMETERS.key()); + + if (!result.isSuccess()) { + throw new GrootStreamRuntimeException(CommonErrorCode.CONFIG_VALIDATION_FAILED, result.getMsg()); } - if (!udfContext.getParameters().containsKey("secret_key")) { - throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "parameters must contains secret_key"); + result = CheckUDFContextUtil.checkParametersContainsKeys(udfContext, UDFContextConfigOptions.PARAMETERS_SECRET_KEY.key()); + if (!result.isSuccess()) { + throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, result.getMsg()); } + } + private String getHmacAlgorithm(String algorithm) { if (StringUtil.containsIgnoreCase(algorithm, "sha256")) { return HmacAlgorithm.HmacSHA256.getValue(); diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES128GCM.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES128GCM.java new file mode 100644 index 0000000..f08383a --- /dev/null +++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES128GCM.java @@ -0,0 +1,81 @@ +package com.geedgenetworks.core.udf.encrypt; + +import com.geedgenetworks.core.pojo.DataEncryptionKey; + +import javax.crypto.Cipher; +import javax.crypto.spec.GCMParameterSpec; +import javax.crypto.spec.SecretKeySpec; +import java.util.Base64; + +public class AES128GCM implements Crypto { + private static final String IDENTIFIER = "aes-128-gcm"; + 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 SECRET_KEY_LENGTH = 16; + private static final byte[] DEFAULT_SECRET_KEY = ".geedgenetworks.".getBytes(); + private static final byte[] NONCE = "Galaxy2019#*".getBytes(); + + private DataEncryptionKey dek; + + public AES128GCM() { + this.dek = new DataEncryptionKey(DEFAULT_SECRET_KEY, 1); + } + + @Override + public String getIdentifier() { + return IDENTIFIER; + } + + @Override + public int getSecretKeyLength() { + return SECRET_KEY_LENGTH; + } + + @Override + public DataEncryptionKey getDataEncryptionKey() { + return dek; + } + + @Override + public void setDataEncryptionKey(DataEncryptionKey dek) { + this.dek = dek; + } + + @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, new SecretKeySpec(dek.getData(), ALGORITHM), gcmSpec); + byte[] encryptedBytes = cipher.doFinal(content.getBytes()); + byte[] combinedBytes = new byte[NONCE.length + encryptedBytes.length]; + System.arraycopy(NONCE, 0, combinedBytes, 0, NONCE.length); + System.arraycopy(encryptedBytes, 0, combinedBytes, 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 - NONCE.length]; + System.arraycopy(combined, 0, NONCE, 0, NONCE.length); + System.arraycopy(combined, NONCE.length, encryptedBytes, 0, encryptedBytes.length); + GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, NONCE); + Cipher cipher = Cipher.getInstance(TRANSFORMATION); + cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(dek.getData(), ALGORITHM), gcmSpec); + 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/AES128GCM96.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES128GCM96.java index 90669b3..6ca1af2 100644 --- a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES128GCM96.java +++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES128GCM96.java @@ -1,14 +1,18 @@ package com.geedgenetworks.core.udf.encrypt; import cn.hutool.core.util.RandomUtil; -import com.geedgenetworks.core.pojo.KmsKey; +import com.geedgenetworks.common.exception.CommonErrorCode; +import com.geedgenetworks.common.exception.GrootStreamRuntimeException; +import com.geedgenetworks.core.pojo.DataEncryptionKey; import javax.crypto.Cipher; +import javax.crypto.NoSuchPaddingException; import javax.crypto.spec.GCMParameterSpec; import javax.crypto.spec.SecretKeySpec; +import java.security.NoSuchAlgorithmException; import java.util.Base64; -public class AES128GCM96 implements EncryptionAlgorithm { +public class AES128GCM96 implements Crypto { private static final String IDENTIFIER = "aes-128-gcm96"; private static final String ALGORITHM = "AES"; private static final String TRANSFORMATION = "AES/GCM/NoPadding"; @@ -18,11 +22,15 @@ public class AES128GCM96 implements EncryptionAlgorithm { private static final byte[] DEFAULT_SECRET_KEY = ".geedgenetworks.".getBytes(); private final Cipher cipher; - private KmsKey kmsKey; + private DataEncryptionKey dek; - public AES128GCM96() throws Exception { - this.cipher = Cipher.getInstance(TRANSFORMATION); - this.kmsKey = new KmsKey(DEFAULT_SECRET_KEY, 1); + public AES128GCM96() { + try { + this.cipher = Cipher.getInstance(TRANSFORMATION); + this.dek = new DataEncryptionKey(DEFAULT_SECRET_KEY, 1); + } catch (NoSuchAlgorithmException | NoSuchPaddingException e) { + throw new GrootStreamRuntimeException(CommonErrorCode.UNSUPPORTED_OPERATION, "Failed to initialize AES128GCM96", e); + } } @Override @@ -36,13 +44,13 @@ public class AES128GCM96 implements EncryptionAlgorithm { } @Override - public KmsKey getKmsKey() { - return kmsKey; + public DataEncryptionKey getDataEncryptionKey() { + return dek; } @Override - public void setKmsKey(KmsKey kmsKey) { - this.kmsKey = kmsKey; + public void setDataEncryptionKey(DataEncryptionKey dek) { + this.dek = dek; } @Override @@ -51,7 +59,7 @@ public class AES128GCM96 implements EncryptionAlgorithm { 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); + cipher.init(Cipher.ENCRYPT_MODE, new SecretKeySpec(dek.getData(), 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); @@ -73,7 +81,7 @@ public class AES128GCM96 implements EncryptionAlgorithm { System.arraycopy(combined, 0, nonce, 0, GCM_96_NONCE_LENGTH); System.arraycopy(combined, GCM_96_NONCE_LENGTH, encryptedBytes, 0, encryptedBytes.length); GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, nonce); - cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(kmsKey.getKeyData(), ALGORITHM), gcmSpec); + cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(dek.getData(), ALGORITHM), gcmSpec); byte[] decryptedBytes = cipher.doFinal(encryptedBytes); decryptedString = new String(decryptedBytes); } catch (Exception e) { diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES256GCM96.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES256GCM96.java index 0306616..c3dd83c 100644 --- a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES256GCM96.java +++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES256GCM96.java @@ -1,14 +1,18 @@ package com.geedgenetworks.core.udf.encrypt; import cn.hutool.core.util.RandomUtil; -import com.geedgenetworks.core.pojo.KmsKey; +import com.geedgenetworks.common.exception.CommonErrorCode; +import com.geedgenetworks.common.exception.GrootStreamRuntimeException; +import com.geedgenetworks.core.pojo.DataEncryptionKey; import javax.crypto.Cipher; +import javax.crypto.NoSuchPaddingException; import javax.crypto.spec.GCMParameterSpec; import javax.crypto.spec.SecretKeySpec; +import java.security.NoSuchAlgorithmException; import java.util.Base64; -public class AES256GCM96 implements EncryptionAlgorithm { +public class AES256GCM96 implements Crypto { private static final String IDENTIFIER = "aes-256-gcm96"; private static final String ALGORITHM = "AES"; private static final String TRANSFORMATION = "AES/GCM/NoPadding"; @@ -18,11 +22,15 @@ public class AES256GCM96 implements EncryptionAlgorithm { private static final byte[] DEFAULT_SECRET_KEY = ".........geedgenetworks.........".getBytes(); private final Cipher cipher; - private KmsKey kmsKey; + private DataEncryptionKey dek; - public AES256GCM96() throws Exception { - this.cipher = Cipher.getInstance(TRANSFORMATION); - this.kmsKey = new KmsKey(DEFAULT_SECRET_KEY, 1); + public AES256GCM96() { + try { + this.cipher = Cipher.getInstance(TRANSFORMATION); + this.dek = new DataEncryptionKey(DEFAULT_SECRET_KEY, 1); + } catch (NoSuchAlgorithmException | NoSuchPaddingException e) { + throw new GrootStreamRuntimeException(CommonErrorCode.UNSUPPORTED_OPERATION, "Failed to initialize AES128GCM96", e); + } } @Override @@ -36,13 +44,13 @@ public class AES256GCM96 implements EncryptionAlgorithm { } @Override - public KmsKey getKmsKey() { - return kmsKey; + public DataEncryptionKey getDataEncryptionKey() { + return dek; } @Override - public void setKmsKey(KmsKey kmsKey) { - this.kmsKey = kmsKey; + public void setDataEncryptionKey(DataEncryptionKey dek) { + this.dek = dek; } @Override @@ -51,7 +59,7 @@ public class AES256GCM96 implements EncryptionAlgorithm { 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); + cipher.init(Cipher.ENCRYPT_MODE, new SecretKeySpec(dek.getData(), 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); @@ -73,7 +81,7 @@ public class AES256GCM96 implements EncryptionAlgorithm { System.arraycopy(combined, 0, nonce, 0, GCM_96_NONCE_LENGTH); System.arraycopy(combined, GCM_96_NONCE_LENGTH, encryptedBytes, 0, encryptedBytes.length); GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, nonce); - cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(kmsKey.getKeyData(), ALGORITHM), gcmSpec); + cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(dek.getData(), ALGORITHM), gcmSpec); byte[] decryptedBytes = cipher.doFinal(encryptedBytes); decryptedString = new String(decryptedBytes); } catch (Exception e) { diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/EncryptionAlgorithm.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/Crypto.java index 3fc4e74..43d7dce 100644 --- a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/EncryptionAlgorithm.java +++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/Crypto.java @@ -1,15 +1,15 @@ package com.geedgenetworks.core.udf.encrypt; -import com.geedgenetworks.core.pojo.KmsKey; +import com.geedgenetworks.core.pojo.DataEncryptionKey; -public interface EncryptionAlgorithm { +public interface Crypto { String getIdentifier(); int getSecretKeyLength(); - KmsKey getKmsKey(); + DataEncryptionKey getDataEncryptionKey(); - void setKmsKey(KmsKey kmsKey); + void setDataEncryptionKey(DataEncryptionKey dek); String encrypt(String content); diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/SM4GCM96.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/SM4GCM96.java index f4ad0a2..d6d701c 100644 --- a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/SM4GCM96.java +++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/SM4GCM96.java @@ -1,14 +1,18 @@ package com.geedgenetworks.core.udf.encrypt; import cn.hutool.core.util.RandomUtil; -import com.geedgenetworks.core.pojo.KmsKey; +import com.geedgenetworks.common.exception.CommonErrorCode; +import com.geedgenetworks.common.exception.GrootStreamRuntimeException; +import com.geedgenetworks.core.pojo.DataEncryptionKey; import javax.crypto.Cipher; +import javax.crypto.NoSuchPaddingException; import javax.crypto.spec.GCMParameterSpec; import javax.crypto.spec.SecretKeySpec; +import java.security.NoSuchAlgorithmException; import java.util.Base64; -public class SM4GCM96 implements EncryptionAlgorithm { +public class SM4GCM96 implements Crypto { private static final String IDENTIFIER = "sm4-gcm96"; private static final String ALGORITHM = "SM4"; private static final String TRANSFORMATION = "SM4/GCM/NoPadding"; @@ -18,11 +22,15 @@ public class SM4GCM96 implements EncryptionAlgorithm { private static final byte[] DEFAULT_SECRET_KEY = ".geedgenetworks.".getBytes(); private final Cipher cipher; - private KmsKey kmsKey; + private DataEncryptionKey dek; - public SM4GCM96() throws Exception { - this.cipher = Cipher.getInstance(TRANSFORMATION); - this.kmsKey = new KmsKey(DEFAULT_SECRET_KEY, 1); + public SM4GCM96() { + try { + this.cipher = Cipher.getInstance(TRANSFORMATION); + this.dek = new DataEncryptionKey(DEFAULT_SECRET_KEY, 1); + } catch (NoSuchAlgorithmException | NoSuchPaddingException e) { + throw new GrootStreamRuntimeException (CommonErrorCode.UNSUPPORTED_OPERATION, "Failed to initialize SM4GCM96", e); + } } @Override @@ -36,13 +44,13 @@ public class SM4GCM96 implements EncryptionAlgorithm { } @Override - public KmsKey getKmsKey() { - return kmsKey; + public DataEncryptionKey getDataEncryptionKey() { + return dek; } @Override - public void setKmsKey(KmsKey kmsKey) { - this.kmsKey = kmsKey; + public void setDataEncryptionKey(DataEncryptionKey dek) { + this.dek = dek; } @Override @@ -51,7 +59,7 @@ public class SM4GCM96 implements EncryptionAlgorithm { 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); + cipher.init(Cipher.ENCRYPT_MODE, new SecretKeySpec(dek.getData(), 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); @@ -73,7 +81,7 @@ public class SM4GCM96 implements EncryptionAlgorithm { System.arraycopy(combined, 0, nonce, 0, GCM_96_NONCE_LENGTH); System.arraycopy(combined, GCM_96_NONCE_LENGTH, encryptedBytes, 0, encryptedBytes.length); GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, nonce); - cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(kmsKey.getKeyData(), ALGORITHM), gcmSpec); + cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(dek.getData(), ALGORITHM), gcmSpec); byte[] decryptedBytes = cipher.doFinal(encryptedBytes); decryptedString = new String(decryptedBytes); } catch (Exception e) { diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/UUIDNameSpace.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/UUIDNameSpace.java index a8941e2..0f8c851 100644 --- a/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/UUIDNameSpace.java +++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/UUIDNameSpace.java @@ -13,8 +13,8 @@ public enum UUIDNameSpace { NAMESPACE_IP(UUID.fromString("6ba7b890-9dad-11d1-80b4-00c04fd430c8")), NAMESPACE_DOMAIN(UUID.fromString("6ba7b891-9dad-11d1-80b4-00c04fd430c8")), NAMESPACE_APP(UUID.fromString("6ba7b892-9dad-11d1-80b4-00c04fd430c8")), - NAMESPACE_SUBSCRIBER(UUID.fromString("6ba7b893-9dad-11d1-80b4-00c04fd430c8")); - + NAMESPACE_SUBSCRIBER(UUID.fromString("6ba7b893-9dad-11d1-80b4-00c04fd430c8")), + NAMESPACE_CELL(UUID.fromString("6ba7b894-9dad-11d1-80b4-00c04fd430c8")); private final UUID uuid; // Static map to hold the mapping from name to UUID diff --git a/groot-core/src/main/java/com/geedgenetworks/core/utils/CryptoProvider.java b/groot-core/src/main/java/com/geedgenetworks/core/utils/CryptoProvider.java new file mode 100644 index 0000000..4cc3c25 --- /dev/null +++ b/groot-core/src/main/java/com/geedgenetworks/core/utils/CryptoProvider.java @@ -0,0 +1,61 @@ +package com.geedgenetworks.core.utils; + +import com.geedgenetworks.core.udf.encrypt.Crypto; +import com.geedgenetworks.core.udf.encrypt.AES128GCM; +import com.geedgenetworks.core.udf.encrypt.AES128GCM96; +import com.geedgenetworks.core.udf.encrypt.AES256GCM96; +import com.geedgenetworks.core.udf.encrypt.SM4GCM96; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; + +import java.util.HashMap; +import java.util.Map; +import java.util.function.Supplier; + +@Slf4j +public final class CryptoProvider { + + @Getter + public enum Algorithm { + AES_128_GCM("aes-128-gcm"), + AES_128_GCM96("aes-128-gcm96"), + AES_256_GCM96("aes-256-gcm96"), + SM4_GCM96("sm4-gcm96"); + + private final String identifier; + + Algorithm(String identifier) { + this.identifier = identifier; + } + + public static Algorithm fromIdentifier(String identifier) { + for (Algorithm algorithm : values()) { + if (algorithm.identifier.equalsIgnoreCase(identifier)) { + return algorithm; + } + } + throw new IllegalArgumentException("Unsupported algorithm identifier: " + identifier); + } + } + + private static final Map<String, Supplier<Crypto>> ALGORITHM_REGISTRY = new HashMap<>(); + + static { + ALGORITHM_REGISTRY.put(Algorithm.AES_128_GCM.getIdentifier(), AES128GCM::new); + ALGORITHM_REGISTRY.put(Algorithm.AES_128_GCM96.getIdentifier(), AES128GCM96::new); + ALGORITHM_REGISTRY.put(Algorithm.AES_256_GCM96.getIdentifier(), AES256GCM96::new); + ALGORITHM_REGISTRY.put(Algorithm.SM4_GCM96.getIdentifier(), SM4GCM96::new); + } + + private CryptoProvider() { + // Private constructor to prevent instantiation + } + + public static Crypto createEncryptionAlgorithm(String identifier) { + Supplier<Crypto> supplier = ALGORITHM_REGISTRY.get(identifier.toLowerCase()); + if (supplier == null) { + throw new IllegalArgumentException("Unsupported algorithm identifier: " + identifier); + } + return supplier.get(); // Lazily creates a new instance when called + } +} diff --git a/groot-core/src/main/java/com/geedgenetworks/core/utils/EncryptionAlgorithmUtils.java b/groot-core/src/main/java/com/geedgenetworks/core/utils/EncryptionAlgorithmUtils.java deleted file mode 100644 index 0327e49..0000000 --- a/groot-core/src/main/java/com/geedgenetworks/core/utils/EncryptionAlgorithmUtils.java +++ /dev/null @@ -1,30 +0,0 @@ -package com.geedgenetworks.core.utils; - -import com.geedgenetworks.core.udf.encrypt.EncryptionAlgorithm; -import com.geedgenetworks.core.udf.encrypt.AES128GCM96; -import com.geedgenetworks.core.udf.encrypt.AES256GCM96; -import com.geedgenetworks.core.udf.encrypt.SM4GCM96; -import lombok.extern.slf4j.Slf4j; - -/** - * Crypto shade utilities - */ -@Slf4j -public final class EncryptionAlgorithmUtils { - public static final String ALGORITHM_AES_128_GCM96_NAME = "aes-128-gcm96"; - public static final String ALGORITHM_AES_256_GCM96_NAME = "aes-256-gcm96"; - public static final String ALGORITHM_SM4_GCM96_NAME = "sm4-gcm96"; - - public static EncryptionAlgorithm createEncryptionAlgorithm(String identifier) throws Exception { - switch (identifier) { - case ALGORITHM_AES_128_GCM96_NAME: - return new AES128GCM96(); - case ALGORITHM_AES_256_GCM96_NAME: - return new AES256GCM96(); - case ALGORITHM_SM4_GCM96_NAME: - return new SM4GCM96(); - default: - return null; - } - } -} 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..f81aae6 --- /dev/null +++ b/groot-core/src/main/java/com/geedgenetworks/core/utils/KMSUtils.java @@ -0,0 +1,69 @@ +package com.geedgenetworks.core.utils; + +import cn.hutool.core.util.StrUtil; +import com.geedgenetworks.common.config.KmsConfig; +import com.geedgenetworks.common.config.ClientSSLConfig; +import com.geedgenetworks.core.pojo.DataEncryptionKey; +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 DataEncryptionKey getVaultKey(KmsConfig kmsConfig, ClientSSLConfig sslConfig, String identifier) { + try { + Vault vault = getVaultClient(kmsConfig, sslConfig); + String exportKeyPath = constructKeyPath(kmsConfig, identifier); + LogicalResponse exportResponse = vault.logical().read(exportKeyPath); + if (exportResponse.getRestResponse().getStatus() == 200) { + JsonObject keys = exportResponse.getDataObject().get("keys").asObject(); + return new DataEncryptionKey(Base64.getDecoder().decode(StrUtil.trim(keys.get(keys.size() + "").asString(), '"')), keys.size()); + } + } catch (VaultException e) { + log.error("Error retrieving vault key from path: {}. ", identifier, e); + } catch (Exception e) { + log.error("Unexpected error retrieving vault key.", e); + } + return null; + } + + private static String constructKeyPath(KmsConfig kmsConfig, String identifier) { + return (CryptoProvider.Algorithm.SM4_GCM96.getIdentifier().equals(identifier) + ? kmsConfig.getPluginKeyPath() + : kmsConfig.getDefaultKeyPath()) + "/export/encryption-key/" + identifier; + } + + public static Vault getVaultClient(KmsConfig kmsConfig, ClientSSLConfig sslConfig) throws VaultException { + VaultConfig config = new VaultConfig() + .address(kmsConfig.getUrl()) + .engineVersion(1) + .sslConfig(configureSSL(sslConfig)) + .build(); + AuthResponse authResponse = Vault.create(config).auth().loginByUserPass(kmsConfig.getUsername(), kmsConfig.getPassword()); + config.token(authResponse.getAuthClientToken()); + return Vault.create(config); + } + + public static SslConfig configureSSL(ClientSSLConfig clientSSLConfig) throws VaultException { + boolean verifySSL = clientSSLConfig != null && !clientSSLConfig.getSkipVerification(); + SslConfig sslConfig = new SslConfig().verify(verifySSL).build(); + if (verifySSL) { + sslConfig.pemFile(new File(clientSSLConfig.getCaCertificatePath())) + .clientPemFile(new File(clientSSLConfig.getCertificatePath())) + .clientKeyPemFile(new File(clientSSLConfig.getPrivateKeyPath())) + .build(); + } + return sslConfig; + } +} 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 deleted file mode 100644 index 9519dd5..0000000 --- a/groot-core/src/main/java/com/geedgenetworks/core/utils/KmsUtils.java +++ /dev/null @@ -1,71 +0,0 @@ -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 Exception { - 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 { - throw new RuntimeException("Get vault key error! code: " + exportResponse.getRestResponse().getStatus() + " body: " + new String(exportResponse.getRestResponse().getBody())); - } - } - - 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/EncryptFunctionTest.java b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/EncryptFunctionTest.java index e9f1698..20f3c0d 100644 --- a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/EncryptFunctionTest.java +++ b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/EncryptFunctionTest.java @@ -6,15 +6,16 @@ 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.config.ClientSSLConfig; import com.geedgenetworks.common.exception.GrootStreamRuntimeException; import com.geedgenetworks.common.udf.UDFContext; -import com.geedgenetworks.core.pojo.KmsKey; +import com.geedgenetworks.core.pojo.DataEncryptionKey; import com.geedgenetworks.core.udf.Encrypt; -import com.geedgenetworks.core.udf.encrypt.EncryptionAlgorithm; -import com.geedgenetworks.core.utils.EncryptionAlgorithmUtils; +import com.geedgenetworks.core.udf.encrypt.Crypto; +import com.geedgenetworks.core.utils.CryptoProvider; import com.geedgenetworks.core.utils.HttpClientPoolUtil; -import com.geedgenetworks.core.utils.KmsUtils; +import com.geedgenetworks.core.utils.KMSUtils; +import io.github.jopenlibs.vault.VaultException; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.functions.RuntimeContext; import org.apache.flink.configuration.Configuration; @@ -55,11 +56,11 @@ public class EncryptFunctionTest { @Test public void testEncryptByVault() throws Exception { String secretKey = RandomUtil.randomString(32); - MockedStatic<KmsUtils> kmsUtilsMockedStatic = Mockito.mockStatic(KmsUtils.class); - Mockito.when(KmsUtils.getVaultKey(ArgumentMatchers.any(), ArgumentMatchers.any(), ArgumentMatchers.any())).thenReturn(new KmsKey(secretKey.getBytes(), 1)); + MockedStatic<KMSUtils> kmsUtilsMockedStatic = Mockito.mockStatic(KMSUtils.class); + Mockito.when(KMSUtils.getVaultKey(ArgumentMatchers.any(), ArgumentMatchers.any(), ArgumentMatchers.any())).thenReturn(new DataEncryptionKey(secretKey.getBytes(), 1)); RuntimeContext runtimeContext = mockVaultRuntimeContext(); Map<String, Object> map = new HashMap<>(); - map.put("identifier", EncryptionAlgorithmUtils.ALGORITHM_AES_256_GCM96_NAME); + map.put("identifier", CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier()); udfContext.setParameters(map); Encrypt encrypt = new Encrypt(); encrypt.open(runtimeContext, udfContext); @@ -68,12 +69,12 @@ public class EncryptFunctionTest { extractedFields.put("phone_number", DATA); event.setExtractedFields(extractedFields); Event result = encrypt.evaluate(event); - EncryptionAlgorithm encryptionAlgorithm = EncryptionAlgorithmUtils.createEncryptionAlgorithm(EncryptionAlgorithmUtils.ALGORITHM_AES_256_GCM96_NAME); - assertNotNull(encryptionAlgorithm); - encryptionAlgorithm.setKmsKey(new KmsKey(secretKey.getBytes(), 1)); + Crypto crypto = CryptoProvider.createEncryptionAlgorithm(CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier()); + assertNotNull(crypto); + crypto.setDataEncryptionKey(new DataEncryptionKey(secretKey.getBytes(), 1)); String encrypted = result.getExtractedFields().get("phone_number").toString(); assertTrue(encrypted.contains("vault:v1:")); - String decrypted = encryptionAlgorithm.decrypt(encrypted.split(":")[2]); + String decrypted = crypto.decrypt(encrypted.split(":")[2]); assertEquals(DATA, decrypted); encrypt.close(); kmsUtilsMockedStatic.close(); @@ -81,10 +82,9 @@ public class EncryptFunctionTest { @Test public void testEncryptByLocal() throws Exception { - byte[] secretKey = ".........geedgenetworks.........".getBytes(); RuntimeContext runtimeContext = mockLocalRuntimeContext(); Map<String, Object> map = new HashMap<>(); - map.put("identifier", EncryptionAlgorithmUtils.ALGORITHM_AES_256_GCM96_NAME); + map.put("identifier", CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier()); udfContext.setParameters(map); Encrypt encrypt = new Encrypt(); encrypt.open(runtimeContext, udfContext); @@ -93,61 +93,182 @@ public class EncryptFunctionTest { extractedFields.put("phone_number", DATA); event.setExtractedFields(extractedFields); Event result = encrypt.evaluate(event); - EncryptionAlgorithm encryptionAlgorithm = EncryptionAlgorithmUtils.createEncryptionAlgorithm(EncryptionAlgorithmUtils.ALGORITHM_AES_256_GCM96_NAME); + Crypto encryptionAlgorithm = CryptoProvider.createEncryptionAlgorithm(CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier()); assertNotNull(encryptionAlgorithm); - encryptionAlgorithm.setKmsKey(new KmsKey(secretKey, 1)); String decrypted = encryptionAlgorithm.decrypt((String) result.getExtractedFields().get("phone_number")); assertEquals(DATA, decrypted); encrypt.close(); } @Test + public void testEncryptGetKeyAndSensitiveFieldsError() throws Exception { + String secretKey = RandomUtil.randomString(32); + MockedStatic<KMSUtils> kmsUtilsMockedStatic = Mockito.mockStatic(KMSUtils.class); + Mockito.when(KMSUtils.getVaultKey(ArgumentMatchers.any(), ArgumentMatchers.any(), ArgumentMatchers.any())).thenReturn(new DataEncryptionKey(secretKey.getBytes(), 1)); + RuntimeContext runtimeContext = mockVaultRuntimeContext(); + Map<String, Object> map = new HashMap<>(); + map.put("identifier", CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier()); + udfContext.setParameters(map); + Encrypt encrypt = new Encrypt(); + encrypt.open(runtimeContext, udfContext); + Event event = new Event(); + Map<String, Object> extractedFields = new HashMap<>(); + extractedFields.put("phone_number", DATA); + event.setExtractedFields(extractedFields); + Event result = encrypt.evaluate(event); + Crypto crypto = CryptoProvider.createEncryptionAlgorithm(CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier()); + assertNotNull(crypto); + crypto.setDataEncryptionKey(new DataEncryptionKey(secretKey.getBytes(), 1)); + String encrypted = result.getExtractedFields().get("phone_number").toString(); + assertTrue(encrypted.contains("vault:v1:")); + String decrypted = crypto.decrypt(encrypted.split(":")[2]); + assertEquals(DATA, decrypted); + + Thread.sleep(90000); + event = new Event(); + extractedFields = new HashMap<>(); + extractedFields.put("phone_number", DATA); + event.setExtractedFields(extractedFields); + result = encrypt.evaluate(event); + encrypted = result.getExtractedFields().get("phone_number").toString(); + assertTrue(encrypted.contains("vault:v1:")); + decrypted = crypto.decrypt(encrypted.split(":")[2]); + assertEquals(DATA, decrypted); + encrypt.close(); + kmsUtilsMockedStatic.close(); + } + + @Test + public void testEncryptNoSensitiveFields() throws Exception { + String secretKey = RandomUtil.randomString(32); + MockedStatic<KMSUtils> kmsUtilsMockedStatic = Mockito.mockStatic(KMSUtils.class); + Mockito.when(KMSUtils.getVaultKey(ArgumentMatchers.any(), ArgumentMatchers.any(), ArgumentMatchers.any())).thenReturn(new DataEncryptionKey(secretKey.getBytes(), 1)); + RuntimeContext runtimeContext = mockVaultNoSensitiveFieldsRuntimeContext(); + Map<String, Object> map = new HashMap<>(); + map.put("identifier", CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier()); + udfContext.setParameters(map); + Encrypt encrypt = new Encrypt(); + encrypt.open(runtimeContext, udfContext); + Event event = new Event(); + Map<String, Object> extractedFields = new HashMap<>(); + extractedFields.put("phone_number", DATA); + event.setExtractedFields(extractedFields); + Event result = encrypt.evaluate(event); + Crypto crypto = CryptoProvider.createEncryptionAlgorithm(CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier()); + assertNotNull(crypto); + crypto.setDataEncryptionKey(new DataEncryptionKey(secretKey.getBytes(), 1)); + String encrypted = result.getExtractedFields().get("phone_number").toString(); + assertTrue(encrypted.contains("vault:v1:")); + String decrypted = crypto.decrypt(encrypted.split(":")[2]); + assertEquals(DATA, decrypted); + encrypt.close(); + kmsUtilsMockedStatic.close(); + } + + @Test + public void testEncryptFirstGetKeyError() throws Exception { + RuntimeContext runtimeContext = mockVaultRuntimeContext(); + Map<String, Object> map = new HashMap<>(); + map.put("identifier", CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier()); + udfContext.setParameters(map); + Encrypt encrypt = new Encrypt(); + assertThrows(GrootStreamRuntimeException.class, () -> encrypt.open(runtimeContext, udfContext)); + encrypt.close(); + } + + @Test + public void testEncryptFirstGetSensitiveFieldsError() throws Exception { + httpClientPoolUtilMockedStatic.close(); + String secretKey = RandomUtil.randomString(32); + MockedStatic<KMSUtils> kmsUtilsMockedStatic = Mockito.mockStatic(KMSUtils.class); + Mockito.when(KMSUtils.getVaultKey(ArgumentMatchers.any(), ArgumentMatchers.any(), ArgumentMatchers.any())).thenReturn(new DataEncryptionKey(secretKey.getBytes(), 1)); + RuntimeContext runtimeContext = mockVaultRuntimeContext(); + Map<String, Object> map = new HashMap<>(); + map.put("identifier", CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier()); + udfContext.setParameters(map); + Encrypt encrypt = new Encrypt(); + assertThrows(GrootStreamRuntimeException.class, () -> encrypt.open(runtimeContext, udfContext)); + encrypt.close(); + kmsUtilsMockedStatic.close(); + httpClientPoolUtilMockedStatic = mockSensitiveFields(); + } + + @Test + public void testEncryptByVaultBySSL() throws Exception { + ClientSSLConfig sslConfig = new ClientSSLConfig(); + sslConfig.setSkipVerification(true); + assertDoesNotThrow(() -> KMSUtils.configureSSL(sslConfig)); + + sslConfig.setSkipVerification(false); + assertThrows(VaultException.class, () -> KMSUtils.configureSSL(sslConfig)); + + sslConfig.setSkipVerification(false); + sslConfig.setCaCertificatePath("src/test/resources/ssl/ca.crt"); + sslConfig.setCertificatePath("src/test/resources/ssl/server.crt"); + sslConfig.setPrivateKeyPath("src/test/resources/ssl/server.key"); + assertDoesNotThrow(() -> KMSUtils.configureSSL(sslConfig)); + } + + @Test public void testEncryptByIdentifier() { Map<String, Object> map = new HashMap<>(); - map.put("identifier", EncryptionAlgorithmUtils.ALGORITHM_AES_256_GCM96_NAME); + map.put("identifier", CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier()); udfContext.setParameters(map); Encrypt encrypt1 = new Encrypt(); assertDoesNotThrow(() -> encrypt1.open(mockLocalRuntimeContext(), udfContext)); encrypt1.close(); Encrypt encrypt2 = new Encrypt(); - map.put("identifier", EncryptionAlgorithmUtils.ALGORITHM_AES_128_GCM96_NAME); + map.put("identifier", CryptoProvider.Algorithm.AES_128_GCM96.getIdentifier()); udfContext.setParameters(map); assertDoesNotThrow(() -> encrypt2.open(mockLocalRuntimeContext(), udfContext)); encrypt2.close(); Encrypt encrypt3 = new Encrypt(); - map.put("identifier", EncryptionAlgorithmUtils.ALGORITHM_SM4_GCM96_NAME); + map.put("identifier", CryptoProvider.Algorithm.SM4_GCM96.getIdentifier()); udfContext.setParameters(map); assertDoesNotThrow(() -> encrypt3.open(mockLocalRuntimeContext(), udfContext)); encrypt3.close(); + + Encrypt encrypt4 = new Encrypt(); + map.put("identifier", CryptoProvider.Algorithm.AES_128_GCM.getIdentifier()); + udfContext.setParameters(map); + assertDoesNotThrow(() -> encrypt4.open(mockLocalRuntimeContext(), udfContext)); + encrypt4.close(); } @Test public void testEncryptionAlgorithm() throws Exception { - EncryptionAlgorithm encryptionAlgorithm = EncryptionAlgorithmUtils.createEncryptionAlgorithm(EncryptionAlgorithmUtils.ALGORITHM_AES_128_GCM96_NAME); - assertNotNull(encryptionAlgorithm); - encryptionAlgorithm.setKmsKey(new KmsKey("aaaaaaaaaaaaaaaa".getBytes(), 1)); - String encryptData = encryptionAlgorithm.encrypt(DATA); - String decryptData = encryptionAlgorithm.decrypt(encryptData); + Crypto crypto = CryptoProvider.createEncryptionAlgorithm(CryptoProvider.Algorithm.AES_128_GCM96.getIdentifier()); + assertNotNull(crypto); + crypto.setDataEncryptionKey(new DataEncryptionKey("aaaaaaaaaaaaaaaa".getBytes(), 1)); + String encryptData = crypto.encrypt(DATA); + String decryptData = crypto.decrypt(encryptData); assertEquals(DATA, decryptData); - encryptionAlgorithm = EncryptionAlgorithmUtils.createEncryptionAlgorithm(EncryptionAlgorithmUtils.ALGORITHM_AES_256_GCM96_NAME); - assertNotNull(encryptionAlgorithm); - encryptionAlgorithm.setKmsKey(new KmsKey("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".getBytes(), 1)); - encryptData = encryptionAlgorithm.encrypt(DATA); - decryptData = encryptionAlgorithm.decrypt(encryptData); + crypto = CryptoProvider.createEncryptionAlgorithm(CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier()); + assertNotNull(crypto); + crypto.setDataEncryptionKey(new DataEncryptionKey("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".getBytes(), 1)); + encryptData = crypto.encrypt(DATA); + decryptData = crypto.decrypt(encryptData); assertEquals(DATA, decryptData); - encryptionAlgorithm = EncryptionAlgorithmUtils.createEncryptionAlgorithm(EncryptionAlgorithmUtils.ALGORITHM_SM4_GCM96_NAME); - assertNotNull(encryptionAlgorithm); - encryptionAlgorithm.setKmsKey(new KmsKey("aaaaaaaaaaaaaaaa".getBytes(), 1)); - encryptData = encryptionAlgorithm.encrypt(DATA); - decryptData = encryptionAlgorithm.decrypt(encryptData); + crypto = CryptoProvider.createEncryptionAlgorithm(CryptoProvider.Algorithm.SM4_GCM96.getIdentifier()); + assertNotNull(crypto); + crypto.setDataEncryptionKey(new DataEncryptionKey("aaaaaaaaaaaaaaaa".getBytes(), 1)); + encryptData = crypto.encrypt(DATA); + decryptData = crypto.decrypt(encryptData); + assertEquals(DATA, decryptData); + + crypto = CryptoProvider.createEncryptionAlgorithm(CryptoProvider.Algorithm.AES_128_GCM.getIdentifier()); + assertNotNull(crypto); + crypto.setDataEncryptionKey(new DataEncryptionKey("aaaaaaaaaaaaaaaa".getBytes(), 1)); + encryptData = crypto.encrypt(DATA); + assertEquals("R2FsYXh5MjAxOSMq6Q4PFGRvBmtSQ36Ug9XDHyMXB7Oye/OPITNW", encryptData); + decryptData = crypto.decrypt(encryptData); assertEquals(DATA, decryptData); - encryptionAlgorithm = EncryptionAlgorithmUtils.createEncryptionAlgorithm("sm4"); - assertNull(encryptionAlgorithm); + assertThrows(IllegalArgumentException.class, () -> CryptoProvider.createEncryptionAlgorithm("sm4")); } @Test @@ -178,12 +299,12 @@ public class EncryptFunctionTest { CommonConfig commonConfig = new CommonConfig(); Map<String, KmsConfig> kmsConfigs = new HashMap<>(); KmsConfig kmsConfig = new KmsConfig(); - kmsConfig.setType(KmsUtils.KMS_TYPE_LOCAL); - kmsConfigs.put(KmsUtils.KMS_TYPE_LOCAL, kmsConfig); + kmsConfig.setType(KMSUtils.KMS_TYPE_LOCAL); + kmsConfigs.put(KMSUtils.KMS_TYPE_LOCAL, kmsConfig); kmsConfig = new KmsConfig(); - kmsConfig.setType(KmsUtils.KMS_TYPE_VAULT); - kmsConfigs.put(KmsUtils.KMS_TYPE_VAULT, kmsConfig); - SSLConfig sslConfig = new SSLConfig(); + kmsConfig.setType(KMSUtils.KMS_TYPE_VAULT); + kmsConfigs.put(KMSUtils.KMS_TYPE_VAULT, kmsConfig); + ClientSSLConfig sslConfig = new ClientSSLConfig(); sslConfig.setSkipVerification(true); Map<String, String> propertiesConfig = new HashMap<>(); propertiesConfig.put("projection.encrypt.schema.registry.uri", "127.0.0.1:9999/v1/schema/session_record?option=encrypt_fields"); @@ -191,7 +312,7 @@ public class EncryptFunctionTest { commonConfig.setSslConfig(sslConfig); commonConfig.setPropertiesConfig(propertiesConfig); configuration.setString(Constants.SYSPROP_GROOTSTREAM_CONFIG, JSON.toJSONString(commonConfig)); - configuration.setString(Constants.SYSPROP_KMS_TYPE_CONFIG, KmsUtils.KMS_TYPE_LOCAL); + configuration.setString(Constants.SYSPROP_KMS_TYPE_CONFIG, KMSUtils.KMS_TYPE_LOCAL); Mockito.when(executionConfig.getGlobalJobParameters()).thenReturn(configuration); return runtimeContext; } @@ -208,20 +329,53 @@ public class EncryptFunctionTest { CommonConfig commonConfig = new CommonConfig(); Map<String, KmsConfig> kmsConfigs = new HashMap<>(); KmsConfig kmsConfig = new KmsConfig(); - kmsConfig.setType(KmsUtils.KMS_TYPE_VAULT); - kmsConfigs.put(KmsUtils.KMS_TYPE_VAULT, kmsConfig); + kmsConfig.setType(KMSUtils.KMS_TYPE_VAULT); + kmsConfigs.put(KMSUtils.KMS_TYPE_VAULT, kmsConfig); kmsConfig = new KmsConfig(); - kmsConfig.setType(KmsUtils.KMS_TYPE_LOCAL); - kmsConfigs.put(KmsUtils.KMS_TYPE_LOCAL, kmsConfig); - SSLConfig sslConfig = new SSLConfig(); + kmsConfig.setType(KMSUtils.KMS_TYPE_LOCAL); + kmsConfigs.put(KMSUtils.KMS_TYPE_LOCAL, kmsConfig); + ClientSSLConfig sslConfig = new ClientSSLConfig(); sslConfig.setSkipVerification(true); Map<String, String> propertiesConfig = new HashMap<>(); propertiesConfig.put("projection.encrypt.schema.registry.uri", "127.0.0.1:9999/v1/schema/session_record?option=encrypt_fields"); + propertiesConfig.put(Constants.SYSPROP_ENCRYPT_KMS_KEY_SCHEDULER_INTERVAL_NAME, "1"); + propertiesConfig.put(Constants.SYSPROP_ENCRYPT_SENSITIVE_FIELDS_SCHEDULER_INTERVAL_NAME, "1"); + commonConfig.setKmsConfig(kmsConfigs); + commonConfig.setSslConfig(sslConfig); + commonConfig.setPropertiesConfig(propertiesConfig); + configuration.setString(Constants.SYSPROP_GROOTSTREAM_CONFIG, JSON.toJSONString(commonConfig)); + configuration.setString(Constants.SYSPROP_KMS_TYPE_CONFIG, KMSUtils.KMS_TYPE_VAULT); + Mockito.when(executionConfig.getGlobalJobParameters()).thenReturn(configuration); + return runtimeContext; + } + + static RuntimeContext mockVaultNoSensitiveFieldsRuntimeContext() { + RuntimeContext runtimeContext = Mockito.mock(RuntimeContext.class); + ExecutionConfig executionConfig = Mockito.mock(ExecutionConfig.class); + Mockito.when(runtimeContext.getExecutionConfig()).thenReturn(executionConfig); + MetricGroup metricGroup = Mockito.mock(OperatorMetricGroup.class); + Mockito.when(runtimeContext.getMetricGroup()).thenReturn(metricGroup); + Mockito.when(metricGroup.addGroup(Mockito.anyString())).thenReturn(metricGroup); + Mockito.when(metricGroup.counter(Mockito.anyString())).thenReturn(new SimpleCounter()); + Configuration configuration = new Configuration(); + CommonConfig commonConfig = new CommonConfig(); + Map<String, KmsConfig> kmsConfigs = new HashMap<>(); + KmsConfig kmsConfig = new KmsConfig(); + kmsConfig.setType(KMSUtils.KMS_TYPE_VAULT); + kmsConfigs.put(KMSUtils.KMS_TYPE_VAULT, kmsConfig); + kmsConfig = new KmsConfig(); + kmsConfig.setType(KMSUtils.KMS_TYPE_LOCAL); + kmsConfigs.put(KMSUtils.KMS_TYPE_LOCAL, kmsConfig); + ClientSSLConfig sslConfig = new ClientSSLConfig(); + sslConfig.setSkipVerification(true); + Map<String, String> propertiesConfig = new HashMap<>(); + propertiesConfig.put(Constants.SYSPROP_ENCRYPT_KMS_KEY_SCHEDULER_INTERVAL_NAME, "1"); + propertiesConfig.put(Constants.SYSPROP_ENCRYPT_SENSITIVE_FIELDS_SCHEDULER_INTERVAL_NAME, "1"); commonConfig.setKmsConfig(kmsConfigs); commonConfig.setSslConfig(sslConfig); commonConfig.setPropertiesConfig(propertiesConfig); configuration.setString(Constants.SYSPROP_GROOTSTREAM_CONFIG, JSON.toJSONString(commonConfig)); - configuration.setString(Constants.SYSPROP_KMS_TYPE_CONFIG, KmsUtils.KMS_TYPE_VAULT); + configuration.setString(Constants.SYSPROP_KMS_TYPE_CONFIG, KMSUtils.KMS_TYPE_VAULT); Mockito.when(executionConfig.getGlobalJobParameters()).thenReturn(configuration); return runtimeContext; } diff --git a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/UUIDTest.java b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/UUIDTest.java index 534569b..70216b4 100644 --- a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/UUIDTest.java +++ b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/UUIDTest.java @@ -136,4 +136,21 @@ public class UUIDTest { Event result1 = uuidv5.evaluate(event); assertEquals("9b154520-3c29-541c-bb81-f649354dae67", result1.getExtractedFields().get("uuid").toString()); } + @Test + public void testUuidV5ForCell() { + udfContext = new UDFContext(); + UUIDv5 uuidv5 = new UUIDv5(); + parameters = new HashMap<>(); + udfContext.setParameters(parameters); + udfContext.setLookupFields(List.of("cell")); + udfContext.setOutputFields(Collections.singletonList("uuid")); + parameters.put("namespace","NAMESPACE_CELL"); + uuidv5.open(null, udfContext); + Event event = new Event(); + Map<String, Object> extractedFields = new HashMap<>(); + extractedFields.put("cell", "test1"); + event.setExtractedFields(extractedFields); + Event result1 = uuidv5.evaluate(event); + assertEquals("4693c13f-e97f-5846-bdb0-f80ce2d2869d", result1.getExtractedFields().get("uuid").toString()); + } } diff --git a/groot-core/src/test/resources/ssl/ca.crt b/groot-core/src/test/resources/ssl/ca.crt new file mode 100644 index 0000000..91201f9 --- /dev/null +++ b/groot-core/src/test/resources/ssl/ca.crt @@ -0,0 +1,18 @@ +-----BEGIN CERTIFICATE----- +MIIC8TCCAdmgAwIBAgIJAOwjrQQ/LLx9MA0GCSqGSIb3DQEBCwUAMA8xDTALBgNV +BAMMBE15Q0EwHhcNMjQxMDEyMDMxNzU3WhcNMzQxMDEwMDMxNzU3WjAPMQ0wCwYD +VQQDDARNeUNBMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAqD/Yapxq +Nm9+4+JccB5XPJ5dTeu7qQORFHBVpPClnLYXsNdmYq+N7ixhsXBKbHdzlB15dZI4 +lTvvGnf7sW4ij1wz42HkF5ZnalP6b7FjxfDqAavFEDtdfOT8f5hkbGrrC2F5GogQ +BoqwFxqHx19UxO4OU4NHB8pSGpFhFgZ0xDkd0gRE6WKXOoWfDLzalFlKI8ISePjq +JD4RbsjsM2PF7sQNWl+Vf8JX+BwV8u7IFuN/qdakDIPn9KQ2KBhHYCdOqK8SSoKz +Pf5TvoaiNGjCLeDG2NuRQV4WmA1Gcs0az4rhnYRk2azlWBJld/yigRE5+xsGl7fK +gqVA7v6kR+GbswIDAQABo1AwTjAdBgNVHQ4EFgQU1K3YFle/hPPdpMjVLnHitVgr +v50wHwYDVR0jBBgwFoAU1K3YFle/hPPdpMjVLnHitVgrv50wDAYDVR0TBAUwAwEB +/zANBgkqhkiG9w0BAQsFAAOCAQEAf7MkFkypRF+e7qt1DUkwS3zPJUYXJOriHp3F +W+gQs0kDTEF3TRVJAITU4nzveWO77KVsPd8zs1W5sDxGnsIkhlJ3NVsYKone28Rz +AWaVy48J9Pj4O1hk9GLH0F6vkByRbXC6Cmcu3C7Tvxnxmegni98/Ja/ASAlBMLrE +YMl5kG87vgAMgYB7RETA9KmzNkTGIz4UcvqN/7RGxnj7bJP0fe/kwlZyliT4wHbQ +s7+9Jt1kE+fFgpz4q3gq2DEcRenFS43jb53WFPHKD8E1fXhk32pqr9aU+v7wHhKo +lX9K1dXWvkw+NnMZIFV9VurfuRlngnfd7mD8yEoMK50VR01JSQ== +-----END CERTIFICATE----- diff --git a/groot-core/src/test/resources/ssl/server.crt b/groot-core/src/test/resources/ssl/server.crt new file mode 100644 index 0000000..19d40c7 --- /dev/null +++ b/groot-core/src/test/resources/ssl/server.crt @@ -0,0 +1,20 @@ +-----BEGIN CERTIFICATE----- +MIIDUDCCAjigAwIBAgIJAPO4WswvRR16MA0GCSqGSIb3DQEBCwUAMA8xDTALBgNV +BAMMBE15Q0EwHhcNMjQxMDEyMDMxNzU3WhcNMzQxMDEwMDMxNzU3WjBhMQswCQYD +VQQGEwJVUzEOMAwGA1UECAwFU3RhdGUxDTALBgNVBAcMBENpdHkxFTATBgNVBAoM +DE9yZ2FuaXphdGlvbjENMAsGA1UECwwEVW5pdDENMAsGA1UEAwwETXlDQTCCASIw +DQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAKrUoeeyulohpHkmawWMJ6j44hES +SzxS83cmqPpwBMksW60HQ+lQXalDYERdwGbgrN/v2YcYQx3NqfnE9UDXal7QQYp6 +d+TpVw+LqQOzinzxsC/kJp4RRcIkiGYyE6EedYFVzVOeQiB5HogG3LPU/CdCjlXq +WhoqAF0jPoGmkUQZiGNfXJx7eWK40C4Bw2q+Q8sKxlLuqHWOMF/dFWWCmFyDMdZ9 +LdQjgYy8s6Bq73FhU5n4fQ+ivGt8iNvJAqJP7wgPGt6UEi+i7hIoWIqeZPFnqj5w +t3FtLfpK7NBua7YLfIE0J8oA42QplmEh1Mn+l6/tj0K0cYUaDHSbm6ccITcCAwEA +AaNdMFswCQYDVR0TBAIwADALBgNVHQ8EBAMCBLAwHQYDVR0lBBYwFAYIKwYBBQUH +AwEGCCsGAQUFBwMCMCIGA1UdEQQbMBmCEXNlcnZlci5kb21haW4uY29thwTAqCjf +MA0GCSqGSIb3DQEBCwUAA4IBAQBYo2N6Ta93LM8RatQz1fPcD8qnwod+JTINGSYs +nC4qqMtc3XtMN7ZrZr0IQAS2s7LiDsoIzNDIG3sJTnCKAYrE0rQcTtMHK2uLtntd +8L4Vjj9upJG0faOtpqgBbTD9BwVXIhhjgttoIMh90pyr62KsK/KYDumWd+yikkkm +PazsSfm2AURfiANLKk+eq/N4WWPLMJN6HbEzJtnuxYtIRj7VHNjnKD5p8+ulYCzQ +pehmudmNu6UfzwsuAR/HI8tulnw6v8GS3lz/9yCf8W+DLG0dWE1i5/A5Hjcu78mW +NDzarO9lXwJhgeXqn4fxq81b7nGIneTvkpR8vhOG1P+HZiwd +-----END CERTIFICATE----- diff --git a/groot-core/src/test/resources/ssl/server.key b/groot-core/src/test/resources/ssl/server.key new file mode 100644 index 0000000..a2efffe --- /dev/null +++ b/groot-core/src/test/resources/ssl/server.key @@ -0,0 +1,28 @@ +-----BEGIN PRIVATE KEY----- +MIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQCq1KHnsrpaIaR5 +JmsFjCeo+OIREks8UvN3Jqj6cATJLFutB0PpUF2pQ2BEXcBm4Kzf79mHGEMdzan5 +xPVA12pe0EGKenfk6VcPi6kDs4p88bAv5CaeEUXCJIhmMhOhHnWBVc1TnkIgeR6I +Btyz1PwnQo5V6loaKgBdIz6BppFEGYhjX1yce3liuNAuAcNqvkPLCsZS7qh1jjBf +3RVlgphcgzHWfS3UI4GMvLOgau9xYVOZ+H0PorxrfIjbyQKiT+8IDxrelBIvou4S +KFiKnmTxZ6o+cLdxbS36SuzQbmu2C3yBNCfKAONkKZZhIdTJ/pev7Y9CtHGFGgx0 +m5unHCE3AgMBAAECggEAJndMoaSAC62JoHIDJTOi9oxcMyXgZQv0oH7HC+VPXpEr +b3G0XAPpoyR1t884dLHgm2ghnibbbgmSXZh49QnMfN798xWSi6vzG6ACBcBWzb2K +Q65m967B+25IfGKIQv5dzSqp2ktHbpJ3Sn/pEGFECf8Vl8j2Uu/kNxSpjX4ZNbD/ +5Nz+gtrqvlOBhr7usuIS8SYEuc3/wpYDmL9cR0ws+Uuc/kjecH/UzuGtuFeai70h +9/whiM+q5poBCH6Wjp0rYUlPa68O/yROtrJsB7a7BXR1nvFxtn5G6T2eKjQac9xs +JI+ruUw0RbFGyzRbtKUdBhWRUeYYjBYoq+CcB3odGQKBgQDYwzmQV64Xyfwx/fIw +wepLlAKNAhqY8z6+1Pg7qhutaeFYG5uv3jWYWpmgWO+WyKefjyPdIzNyU/36OO7/ +s5r3ZryA/15DRlvEDbSDAeVyjQEDNdDIzfNqQ0h+8j7vBJlykGgCekMITjiJUv8a +2yUsHyBevI3WFpUDaAMUtUmK3QKBgQDJwOtdR+H62/1PX23vyP6c/zU520j92zQv +5yrHVQINtFOSmzBgdpM/G0yLEJIbw0Fvz8/cx8IYUQnQQPuCS7ZxJJSqiwxV33Qz +hhtZJhH61lmE2guHURdytaE9heGMwfrgnmG64kdlVNFHZ492ltSIq/C+zUB2/5YT +Mm0f8yfpIwKBgQDObTzIpXd52DWANmMK4+EIkK/NMY+60QuUGKU9zMYG46piigg9 +99P6f22GMqwYYIahgWOaGQfJfQuF2+pfQN/3c7NY9dkDIGIL1zFtAcVMzdOFBx8J +3HhPXjwQCQq9/RdU7wjeMyjbJALbZFrlbIV9+zaMgexhUagfUlJ8yhh7UQKBgQCt +npVtSsTPqq0MtyTWavOhi4X0ah8gRplcd+S6cQ85V+triJ1TBfelIQr3yaTSu27+ +l6lbZ5RCdMqrKqDF+f3g1AgT02EkLQ3EoS27xCVI5VlYGIQ/SKuTDXbaiPIWvX/1 ++JZFyyCBtUH73sT42se/bafZqqxFO6Gcl5KNIiVAXQKBgQDYigejeQ81cL72lf7d +P/n5HU92cDBAt+sF8CJoyaHqcInn0sm2t9Eix/rcMY6da7cyeA8HSdiwPjmFRkn9 +jJrrJZGkjm9zE2+QX7/nFCV5MXCf4216gjkuAIdcWGQFiNNrw8eokxQKn0Y7N5Jh +AaOyISDAAaqdFO6Nwsixscvd0A== +-----END PRIVATE KEY----- |