From 996a761ff08d03706f7e8e2d877fd888980ec233 Mon Sep 17 00:00:00 2001 From: Neng Lu Date: Thu, 29 Oct 2020 19:04:40 -0700 Subject: [PATCH 1/4] add e2d encryption support for pulsar functions --- .../apache/pulsar/admin/cli/CmdFunctions.java | 7 + .../apache/pulsar/admin/cli/CmdSources.java | 7 + .../common/functions/ConsumerConfig.java | 1 + .../pulsar/common/functions/CryptoConfig.java | 30 ++++ .../common/functions/ProducerConfig.java | 1 + .../pulsar/common/util/Reflections.java | 2 +- .../instance/JavaInstanceRunnable.java | 15 +- .../pulsar/functions/sink/PulsarSink.java | 91 +++++++++--- .../functions/sink/PulsarSinkConfig.java | 4 +- .../pulsar/functions/source/PulsarSource.java | 40 ++++-- .../functions/source/PulsarSourceConfig.java | 1 + .../proto/src/main/proto/Function.proto | 26 ++++ .../pulsar/functions/utils/CryptoUtils.java | 135 ++++++++++++++++++ .../functions/utils/FunctionConfigUtils.java | 67 +++++++-- .../functions/utils/SinkConfigUtils.java | 17 ++- .../functions/utils/SourceConfigUtils.java | 35 +++-- .../functions/utils/ValidatorUtils.java | 37 +++++ 17 files changed, 458 insertions(+), 58 deletions(-) create mode 100644 pulsar-common/src/main/java/org/apache/pulsar/common/functions/CryptoConfig.java create mode 100644 pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/CryptoUtils.java diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java index 76d806c85cf85..dacc8394c499f 100644 --- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java +++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java @@ -52,6 +52,7 @@ import org.apache.pulsar.client.api.SubscriptionInitialPosition; import org.apache.pulsar.common.functions.ConsumerConfig; import org.apache.pulsar.common.functions.FunctionConfig; +import org.apache.pulsar.common.functions.ProducerConfig; import org.apache.pulsar.common.functions.Resources; import org.apache.pulsar.common.functions.UpdateOptions; import org.apache.pulsar.common.functions.Utils; @@ -219,6 +220,8 @@ abstract class FunctionDetailsCommand extends BaseCommand { @Parameter(names = {"-o", "--output"}, description = "The output topic of a Pulsar Function (If none is specified, no output is written)") protected String output; + @Parameter(names = "--producer-config", description = "The custom producer configuration (as a JSON string)" ) + protected String producerConfig; // for backwards compatibility purposes @Parameter(names = "--logTopic", description = "The topic to which the logs of a Pulsar Function are produced", hidden = true) protected String DEPRECATED_logTopic; @@ -391,6 +394,10 @@ void processArguments() throws Exception { if (null != output) { functionConfig.setOutput(output); } + if (null != producerConfig) { + Type type = new TypeToken() {}.getType(); + functionConfig.setProducerConfig(new Gson().fromJson(producerConfig, type)); + } if (null != logTopic) { functionConfig.setLogTopic(logTopic); } diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java index 7ed9bc9268347..fcc580ad68ba1 100644 --- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java +++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java @@ -49,6 +49,7 @@ import org.apache.pulsar.admin.cli.utils.CmdUtils; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.admin.PulsarAdminException; +import org.apache.pulsar.common.functions.ProducerConfig; import org.apache.pulsar.common.functions.Resources; import org.apache.pulsar.common.functions.UpdateOptions; import org.apache.pulsar.common.io.ConnectorDefinition; @@ -270,6 +271,8 @@ abstract class SourceDetailsCommand extends BaseCommand { protected String DEPRECATED_destinationTopicName; @Parameter(names = "--destination-topic-name", description = "The Pulsar topic to which data is sent") protected String destinationTopicName; + @Parameter(names = "--producer-config", description = "The custom producer configuration (as a JSON string)") + protected String producerConfig; @Parameter(names = "--deserializationClassName", description = "The SerDe classname for the source", hidden = true) protected String DEPRECATED_deserializationClassName; @@ -346,6 +349,10 @@ void processArguments() throws Exception { if (null != destinationTopicName) { sourceConfig.setTopicName(destinationTopicName); } + if (null != producerConfig) { + Type type = new TypeToken() {}.getType(); + sourceConfig.setProducerConfig(new Gson().fromJson(producerConfig, type)); + } if (null != deserializationClassName) { sourceConfig.setSerdeClassName(deserializationClassName); } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/ConsumerConfig.java b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/ConsumerConfig.java index d15b41bdbe7c5..9925fe8f57ae5 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/ConsumerConfig.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/ConsumerConfig.java @@ -45,6 +45,7 @@ public class ConsumerConfig { @Builder.Default private Map consumerProperties = new HashMap<>(); private Integer receiverQueueSize; + private CryptoConfig cryptoConfig; public ConsumerConfig(String schemaType) { this.schemaType = schemaType; diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/CryptoConfig.java b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/CryptoConfig.java new file mode 100644 index 0000000000000..f40edb99e6169 --- /dev/null +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/CryptoConfig.java @@ -0,0 +1,30 @@ +package org.apache.pulsar.common.functions; + + +import java.util.Map; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.NoArgsConstructor; +import org.apache.pulsar.client.api.ConsumerCryptoFailureAction; +import org.apache.pulsar.client.api.ProducerCryptoFailureAction; + +/** + * Configuration of the producer inside the function. + */ +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +@EqualsAndHashCode +public class CryptoConfig { + private String cryptoKeyReaderClassName; + private Map cryptoKeyReaderConfig; + + private String[] encryptionKeys; + private ProducerCryptoFailureAction producerCryptoFailureAction; + + private ConsumerCryptoFailureAction consumerCryptoFailureAction; +} diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/ProducerConfig.java b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/ProducerConfig.java index 8d3dd66963d60..5b686638dbf7b 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/ProducerConfig.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/ProducerConfig.java @@ -36,4 +36,5 @@ public class ProducerConfig { private Integer maxPendingMessages; private Integer maxPendingMessagesAcrossPartitions; private Boolean useThreadLocalProducers; + private CryptoConfig cryptoConfig; } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/Reflections.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/Reflections.java index b0045d02d5185..0d4c2b49c55f3 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/Reflections.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/Reflections.java @@ -88,7 +88,7 @@ public static T createInstance(String userClassName, } catch (NoSuchMethodException e) { throw new RuntimeException("User class must have a no-arg constructor", e); } catch (IllegalAccessException e) { - throw new RuntimeException("User class must a public constructor", e); + throw new RuntimeException("User class must have a public constructor", e); } catch (InvocationTargetException e) { throw new RuntimeException("User class constructor throws exception", e); } diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java index f2cfbf35a0f6f..2b2f3601c4847 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java @@ -57,7 +57,9 @@ import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.common.functions.ConsumerConfig; +import org.apache.pulsar.common.functions.CryptoConfig; import org.apache.pulsar.common.functions.FunctionConfig; +import org.apache.pulsar.common.functions.ProducerConfig; import org.apache.pulsar.functions.api.Function; import org.apache.pulsar.functions.api.Record; import org.apache.pulsar.functions.instance.stats.ComponentStatsManager; @@ -73,6 +75,7 @@ import org.apache.pulsar.functions.source.PulsarSourceConfig; import org.apache.pulsar.common.util.Reflections; import org.apache.pulsar.functions.source.batch.BatchSourceExecutor; +import org.apache.pulsar.functions.utils.CryptoUtils; import org.apache.pulsar.functions.utils.FunctionCommon; import org.apache.pulsar.functions.utils.functioncache.FunctionCacheManager; import org.apache.pulsar.io.core.Sink; @@ -690,6 +693,10 @@ private void setupInput(ContextImpl contextImpl) throws Exception { if (conf.hasReceiverQueueSize()) { consumerConfig.setReceiverQueueSize(conf.getReceiverQueueSize().getValue()); } + if (conf.hasCryptoSpec()) { + consumerConfig.setCryptoConfig(CryptoUtils.convertFromSpec(conf.getCryptoSpec())); + } + pulsarSourceConfig.getTopicSchema().put(topic, consumerConfig); }); @@ -815,7 +822,13 @@ private void setupOutput(ContextImpl contextImpl) throws Exception { pulsarSinkConfig.setSchemaProperties(sinkSpec.getSchemaPropertiesMap()); if (this.instanceConfig.getFunctionDetails().getSink().getProducerSpec() != null) { - pulsarSinkConfig.setProducerSpec(this.instanceConfig.getFunctionDetails().getSink().getProducerSpec()); + org.apache.pulsar.functions.proto.Function.ProducerSpec conf = this.instanceConfig.getFunctionDetails().getSink().getProducerSpec(); + ProducerConfig.ProducerConfigBuilder builder = ProducerConfig.builder() + .maxPendingMessages(conf.getMaxPendingMessages()) + .maxPendingMessagesAcrossPartitions(conf.getMaxPendingMessagesAcrossPartitions()) + .useThreadLocalProducers(conf.getUseThreadLocalProducers()) + .cryptoConfig(CryptoUtils.convertFromSpec(conf.getCryptoSpec())); + pulsarSinkConfig.setProducerConfig(builder.build()); } object = new PulsarSink(this.client, pulsarSinkConfig, this.properties, this.stats, this.functionClassLoader); diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java index 00d12c12a2196..c56a838df5115 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java @@ -19,21 +19,28 @@ package org.apache.pulsar.functions.sink; import com.google.common.annotations.VisibleForTesting; +import lombok.Builder; +import lombok.Data; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; +import org.apache.commons.lang3.reflect.ConstructorUtils; import org.apache.pulsar.client.api.CompressionType; +import org.apache.pulsar.client.api.CryptoKeyReader; import org.apache.pulsar.client.api.HashingScheme; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.MessageRoutingMode; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.ProducerBuilder; +import org.apache.pulsar.client.api.ProducerCryptoFailureAction; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.TypedMessageBuilder; import org.apache.pulsar.client.impl.schema.KeyValueSchema; import org.apache.pulsar.common.functions.ConsumerConfig; +import org.apache.pulsar.common.functions.CryptoConfig; import org.apache.pulsar.common.functions.FunctionConfig; +import org.apache.pulsar.common.functions.ProducerConfig; import org.apache.pulsar.common.schema.KeyValueEncodingType; import org.apache.pulsar.functions.api.Record; import org.apache.pulsar.functions.instance.FunctionResultRouter; @@ -42,9 +49,12 @@ import org.apache.pulsar.functions.source.PulsarRecord; import org.apache.pulsar.functions.source.TopicSchema; import org.apache.pulsar.common.util.Reflections; +import org.apache.pulsar.functions.utils.CryptoUtils; import org.apache.pulsar.io.core.Sink; import org.apache.pulsar.io.core.SinkContext; +import org.bouncycastle.jce.provider.BouncyCastleProvider; +import java.security.Security; import java.util.ArrayList; import java.util.Base64; import java.util.List; @@ -55,6 +65,8 @@ import java.util.concurrent.TimeUnit; import java.util.function.Function; +import static org.apache.commons.lang.StringUtils.isEmpty; + @Slf4j public class PulsarSink implements Sink { @@ -81,9 +93,11 @@ private interface PulsarSinkProcessor { private abstract class PulsarSinkProcessorBase implements PulsarSinkProcessor { protected Map> publishProducers = new ConcurrentHashMap<>(); protected Schema schema; + protected Crypto crypto; - protected PulsarSinkProcessorBase(Schema schema) { + protected PulsarSinkProcessorBase(Schema schema, Crypto crypto) { this.schema = schema; + this.crypto = crypto; } public Producer createProducer(PulsarClient client, String topic, String producerName, Schema schema) @@ -103,15 +117,23 @@ public Producer createProducer(PulsarClient client, String topic, String prod if (producerName != null) { builder.producerName(producerName); } - if (pulsarSinkConfig.getProducerSpec() != null) { - if (pulsarSinkConfig.getProducerSpec().getMaxPendingMessages() != 0) { - builder.maxPendingMessages(pulsarSinkConfig.getProducerSpec().getMaxPendingMessages()); + if (pulsarSinkConfig.getProducerConfig() != null) { + ProducerConfig producerConfig = pulsarSinkConfig.getProducerConfig(); + if (producerConfig.getMaxPendingMessages() != 0) { + builder.maxPendingMessages(producerConfig.getMaxPendingMessages()); + } + if (producerConfig.getMaxPendingMessagesAcrossPartitions() != 0) { + builder.maxPendingMessagesAcrossPartitions(producerConfig.getMaxPendingMessagesAcrossPartitions()); } - if (pulsarSinkConfig.getProducerSpec().getMaxPendingMessagesAcrossPartitions() != 0) { - builder.maxPendingMessagesAcrossPartitions(pulsarSinkConfig.getProducerSpec().getMaxPendingMessagesAcrossPartitions()); + if (producerConfig.getCryptoConfig() != null) { + CryptoConfig cryptoConfig = producerConfig.getCryptoConfig(); + builder.cryptoKeyReader(crypto.keyReader); + builder.cryptoFailureAction(crypto.failureAction); + for (String encryptionKeyName : crypto.getEncryptionKeys()) { + builder.addEncryptionKey(encryptionKeyName); + } } } - return builder.properties(properties).create(); } @@ -175,8 +197,8 @@ public Function getPublishErrorHandler(Record record, boolea @VisibleForTesting class PulsarSinkAtMostOnceProcessor extends PulsarSinkProcessorBase { - public PulsarSinkAtMostOnceProcessor(Schema schema) { - super(schema); + public PulsarSinkAtMostOnceProcessor(Schema schema, Crypto crypto) { + super(schema, crypto); // initialize default topic try { publishProducers.put(pulsarSinkConfig.getTopic(), @@ -211,8 +233,8 @@ public void sendOutputMessage(TypedMessageBuilder msg, Record record) { @VisibleForTesting class PulsarSinkAtLeastOnceProcessor extends PulsarSinkAtMostOnceProcessor { - public PulsarSinkAtLeastOnceProcessor(Schema schema) { - super(schema); + public PulsarSinkAtLeastOnceProcessor(Schema schema, Crypto crypto) { + super(schema, crypto); } @Override @@ -226,8 +248,8 @@ public void sendOutputMessage(TypedMessageBuilder msg, Record record) { @VisibleForTesting class PulsarSinkEffectivelyOnceProcessor extends PulsarSinkProcessorBase { - public PulsarSinkEffectivelyOnceProcessor(Schema schema) { - super(schema); + public PulsarSinkEffectivelyOnceProcessor(Schema schema, Crypto crypto) { + super(schema, crypto); } @Override @@ -284,16 +306,21 @@ public void open(Map config, SinkContext sinkContext) throws Exc return; } + Crypto crypto = initializeCrypto(); + if (crypto == null) { + log.info("crypto key reader is not provided, not enabling end to end encryption"); + } + FunctionConfig.ProcessingGuarantees processingGuarantees = this.pulsarSinkConfig.getProcessingGuarantees(); switch (processingGuarantees) { case ATMOST_ONCE: - this.pulsarSinkProcessor = new PulsarSinkAtMostOnceProcessor(schema); + this.pulsarSinkProcessor = new PulsarSinkAtMostOnceProcessor(schema, crypto); break; case ATLEAST_ONCE: - this.pulsarSinkProcessor = new PulsarSinkAtLeastOnceProcessor(schema); + this.pulsarSinkProcessor = new PulsarSinkAtLeastOnceProcessor(schema, crypto); break; case EFFECTIVELY_ONCE: - this.pulsarSinkProcessor = new PulsarSinkEffectivelyOnceProcessor(schema); + this.pulsarSinkProcessor = new PulsarSinkEffectivelyOnceProcessor(schema, crypto); break; } } @@ -360,4 +387,36 @@ Schema initializeSchema() throws ClassNotFoundException { consumerConfig, false, functionClassLoader); } } + + @SuppressWarnings("unchecked") + @VisibleForTesting + Crypto initializeCrypto() throws ClassNotFoundException { + CryptoConfig cryptoConfig = pulsarSinkConfig.getProducerConfig().getCryptoConfig(); + + if (cryptoConfig == null || isEmpty(cryptoConfig.getCryptoKeyReaderClassName())) { + return null; + } + + // add provider only if it's not in the JVM +// if (Security.getProvider(BouncyCastleProvider.PROVIDER_NAME) == null) { +// Security.addProvider(new BouncyCastleProvider()); +// } + + Crypto.CryptoBuilder bldr = Crypto.builder() + .failureAction(cryptoConfig.getProducerCryptoFailureAction()) + .encryptionKeys(cryptoConfig.getEncryptionKeys()); + + bldr.keyReader(CryptoUtils.getCryptoKeyReaderInstance( + cryptoConfig.getCryptoKeyReaderClassName(), cryptoConfig.getCryptoKeyReaderConfig(), functionClassLoader)); + + return bldr.build(); + } + + @Data + @Builder + private static class Crypto { + private CryptoKeyReader keyReader; + private ProducerCryptoFailureAction failureAction; + private String[] encryptionKeys; + } } diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSinkConfig.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSinkConfig.java index a4ba7e33970ec..2999a04bb59c6 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSinkConfig.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSinkConfig.java @@ -22,7 +22,7 @@ import lombok.Setter; import lombok.ToString; import org.apache.pulsar.common.functions.FunctionConfig; -import org.apache.pulsar.functions.proto.Function; +import org.apache.pulsar.common.functions.ProducerConfig; import java.util.Map; @@ -37,5 +37,5 @@ public class PulsarSinkConfig { private Map schemaProperties; private String typeClassName; private boolean forwardSourceMessageProperty; - private Function.ProducerSpec producerSpec; + private ProducerConfig producerConfig; } diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java index d8faeb23df384..a4f4fe5427240 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java @@ -22,6 +22,7 @@ import com.google.common.annotations.VisibleForTesting; +import java.security.Security; import java.util.*; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @@ -35,8 +36,10 @@ import org.apache.pulsar.functions.api.Record; import org.apache.pulsar.common.functions.FunctionConfig; import org.apache.pulsar.common.util.Reflections; +import org.apache.pulsar.functions.utils.CryptoUtils; import org.apache.pulsar.io.core.PushSource; import org.apache.pulsar.io.core.SourceContext; +import org.bouncycastle.jce.provider.BouncyCastleProvider; @Slf4j public class PulsarSource extends PushSource implements MessageListener { @@ -71,8 +74,6 @@ public void open(Map config, SourceContext sourceContext) throws topic, conf.getSchema(), conf.getSchema().getSchemaInfo()); ConsumerBuilder cb = pulsarClient.newConsumer(conf.getSchema()) - // consume message even if can't decrypt and deliver it along with encryption-ctx - .cryptoFailureAction(ConsumerCryptoFailureAction.CONSUME) .subscriptionName(pulsarSourceConfig.getSubscriptionName()) .subscriptionInitialPosition(pulsarSourceConfig.getSubscriptionPosition()) .subscriptionType(pulsarSourceConfig.getSubscriptionType()); @@ -91,6 +92,12 @@ public void open(Map config, SourceContext sourceContext) throws if (conf.getReceiverQueueSize() != null) { cb = cb.receiverQueueSize(conf.getReceiverQueueSize()); } + if (conf.getCryptoKeyReader() != null) { + cb = cb.cryptoKeyReader(conf.getCryptoKeyReader()); + } + if (conf.getConsumerCryptoFailureAction() != null) { + cb = cb.cryptoFailureAction(conf.getConsumerCryptoFailureAction()); + } cb = cb.properties(properties); if (pulsarSourceConfig.getNegativeAckRedeliveryDelayMs() != null && pulsarSourceConfig.getNegativeAckRedeliveryDelayMs() > 0) { @@ -166,18 +173,32 @@ Map> setupConsumerConfigs() throws ClassNotFoundExcept // Check new config with schema types or classnames pulsarSourceConfig.getTopicSchema().forEach((topic, conf) -> { + ConsumerConfig.ConsumerConfigBuilder consumerConfBuilder = ConsumerConfig. builder(). + isRegexPattern(conf.isRegexPattern()). + receiverQueueSize(conf.getReceiverQueueSize()). + consumerProperties(conf.getConsumerProperties()); + Schema schema; if (conf.getSerdeClassName() != null && !conf.getSerdeClassName().isEmpty()) { schema = (Schema) topicSchema.getSchema(topic, typeArg, conf.getSerdeClassName(), true); } else { schema = (Schema) topicSchema.getSchema(topic, typeArg, conf, true); } - configs.put(topic, - ConsumerConfig. builder(). - schema(schema). - isRegexPattern(conf.isRegexPattern()). - receiverQueueSize(conf.getReceiverQueueSize()). - consumerProperties(conf.getConsumerProperties()).build()); + consumerConfBuilder.schema(schema); + + if (conf.getCryptoConfig() != null) { + // add provider only if it's not in the JVM +// if (Security.getProvider(BouncyCastleProvider.PROVIDER_NAME) == null) { +// Security.addProvider(new BouncyCastleProvider()); +// } + + consumerConfBuilder.consumerCryptoFailureAction(conf.getCryptoConfig().getConsumerCryptoFailureAction()); + consumerConfBuilder.cryptoKeyReader(CryptoUtils.getCryptoKeyReaderInstance( + conf.getCryptoConfig().getCryptoKeyReaderClassName(), + conf.getCryptoConfig().getCryptoKeyReaderConfig(), functionClassLoader)); + } + + configs.put(topic, consumerConfBuilder.build()); }); return configs; @@ -198,6 +219,7 @@ private static class ConsumerConfig { private boolean isRegexPattern; private Integer receiverQueueSize; private Map consumerProperties; + private CryptoKeyReader cryptoKeyReader; + private ConsumerCryptoFailureAction consumerCryptoFailureAction; } - } diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSourceConfig.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSourceConfig.java index c47d810e039d5..e35e7636c9233 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSourceConfig.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSourceConfig.java @@ -28,6 +28,7 @@ import org.apache.pulsar.client.api.SubscriptionInitialPosition; import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.common.functions.CryptoConfig; import org.apache.pulsar.common.util.ObjectMapperFactory; import org.apache.pulsar.common.functions.ConsumerConfig; import org.apache.pulsar.common.functions.FunctionConfig; diff --git a/pulsar-functions/proto/src/main/proto/Function.proto b/pulsar-functions/proto/src/main/proto/Function.proto index 0bc1e28c9c0d5..4243fb622272c 100644 --- a/pulsar-functions/proto/src/main/proto/Function.proto +++ b/pulsar-functions/proto/src/main/proto/Function.proto @@ -100,12 +100,38 @@ message ConsumerSpec { ReceiverQueueSize receiverQueueSize = 4; map schemaProperties = 5; map consumerProperties = 6; + CryptoSpec cryptoSpec = 7; } message ProducerSpec { int32 maxPendingMessages = 1; int32 maxPendingMessagesAcrossPartitions = 2; bool useThreadLocalProducers = 3; + CryptoSpec cryptoSpec = 4; +} + +message CryptoSpec { + enum FailureAction { + FAIL = 0; + + DISCARD = 1; + CONSUME = 2; + + SEND = 10; + } + + string cryptoKeyReaderClassName = 1; + string cryptoKeyReaderConfig = 2; + + // key names used by producer to encrypt data + repeated string producerEncryptionKeyName = 3; + // define the action if producer fail to encrypt data + // one of FAIL, SEND + FailureAction producerCryptoFailureAction = 4; + + // define the action if consumer fail to decrypt data + // one of FAIL, DISCARD, CONSUME + FailureAction consumerCryptoFailureAction = 5; } message SourceSpec { diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/CryptoUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/CryptoUtils.java new file mode 100644 index 0000000000000..9dd78beaa306b --- /dev/null +++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/CryptoUtils.java @@ -0,0 +1,135 @@ +package org.apache.pulsar.functions.utils; + +import java.lang.reflect.Constructor; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Type; +import java.util.Arrays; +import java.util.Map; + +import com.google.gson.Gson; +import com.google.gson.reflect.TypeToken; +import org.apache.pulsar.client.api.ConsumerCryptoFailureAction; +import org.apache.pulsar.client.api.CryptoKeyReader; +import org.apache.pulsar.client.api.ProducerCryptoFailureAction; +import org.apache.pulsar.common.functions.CryptoConfig; +import org.apache.pulsar.common.util.ClassLoaderUtils; +import org.apache.pulsar.functions.proto.Function; + +import static org.apache.commons.lang.StringUtils.isEmpty; + +public final class CryptoUtils { + + public static Function.CryptoSpec convert(CryptoConfig config) { + Function.CryptoSpec.Builder bldr = Function.CryptoSpec.newBuilder() + .setCryptoKeyReaderClassName(config.getCryptoKeyReaderClassName()); + + if (config.getCryptoKeyReaderConfig() != null) { + Type type = new TypeToken>() { + }.getType(); + String readerConfigString = new Gson().toJson(config.getCryptoKeyReaderConfig(), type); + bldr.setCryptoKeyReaderConfig(readerConfigString); + } + + if (config.getEncryptionKeys() != null && config.getEncryptionKeys().length > 0) { + bldr.addAllProducerEncryptionKeyName(Arrays.asList(config.getEncryptionKeys())); + } + + if (config.getProducerCryptoFailureAction() != null) { + bldr.setProducerCryptoFailureAction(getProtoFailureAction(config.getProducerCryptoFailureAction())); + } + + if (config.getConsumerCryptoFailureAction() != null) { + bldr.setConsumerCryptoFailureAction(getProtoFailureAction(config.getConsumerCryptoFailureAction())); + } + + return bldr.build(); + } + + public static CryptoConfig convertFromSpec(Function.CryptoSpec spec) { + if (spec == null || isEmpty(spec.getCryptoKeyReaderClassName())) { + return null; + } + + CryptoConfig.CryptoConfigBuilder bldr = CryptoConfig.builder(); + + Type type = new TypeToken>() { + }.getType(); + Map cryptoReaderConfig = new Gson().fromJson(spec.getCryptoKeyReaderConfig(), type); + + bldr.cryptoKeyReaderClassName(spec.getCryptoKeyReaderClassName()) + .cryptoKeyReaderConfig(cryptoReaderConfig) + .consumerCryptoFailureAction(getConsumerCryptoFailureAction(spec.getConsumerCryptoFailureAction())) + .producerCryptoFailureAction(getProducerCryptoFailureAction(spec.getProducerCryptoFailureAction())) + .encryptionKeys(spec.getProducerEncryptionKeyNameList().toArray(new String[0])); + + return bldr.build(); + } + + public static CryptoKeyReader getCryptoKeyReaderInstance(String className, Map configs, ClassLoader classLoader) { + Class cryptoClass; + try { + cryptoClass = ClassLoaderUtils.loadClass(className, classLoader); + } catch (ClassNotFoundException e) { + throw new RuntimeException( + String.format("Failed to load crypto key reader class %sx", className)); + } + + try { + Constructor ctor = cryptoClass.getConstructor(java.util.Map.class); + return (CryptoKeyReader) ctor.newInstance(configs); + } catch (NoSuchMethodException e) { + throw new RuntimeException("Key reader class does not have constructor accepts map", e); + } catch (IllegalAccessException | InstantiationException | InvocationTargetException e) { + throw new RuntimeException("Failed to create instance for key reader class", e); + } + } + + public static ProducerCryptoFailureAction getProducerCryptoFailureAction(Function.CryptoSpec.FailureAction action) { + switch (action) { + case FAIL: + return ProducerCryptoFailureAction.FAIL; + case SEND: + return ProducerCryptoFailureAction.SEND; + default: + throw new RuntimeException("Unknown producer protobuf failure action " + action.getValueDescriptor().getName()); + } + } + + public static ConsumerCryptoFailureAction getConsumerCryptoFailureAction(Function.CryptoSpec.FailureAction action) { + switch (action) { + case FAIL: + return ConsumerCryptoFailureAction.FAIL; + case DISCARD: + return ConsumerCryptoFailureAction.DISCARD; + case CONSUME: + return ConsumerCryptoFailureAction.CONSUME; + default: + throw new RuntimeException("Unknown consumer protobuf failure action " + action.getValueDescriptor().getName()); + } + } + + public static Function.CryptoSpec.FailureAction getProtoFailureAction(ProducerCryptoFailureAction action) { + switch (action) { + case FAIL: + return Function.CryptoSpec.FailureAction.FAIL; + case SEND: + return Function.CryptoSpec.FailureAction.SEND; + default: + throw new RuntimeException("Unknown producer crypto failure action " + action); + } + } + + public static Function.CryptoSpec.FailureAction getProtoFailureAction(ConsumerCryptoFailureAction action) { + switch (action) { + case FAIL: + return Function.CryptoSpec.FailureAction.FAIL; + case DISCARD: + return Function.CryptoSpec.FailureAction.DISCARD; + case CONSUME: + return Function.CryptoSpec.FailureAction.CONSUME; + default: + throw new RuntimeException("Unknown consumer crypto failure action " + action); + } + } + +} diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java index 89a17cb8401c6..bdf250a60bfeb 100644 --- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java +++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java @@ -41,6 +41,7 @@ import java.util.List; import java.util.Map; +import static org.apache.commons.lang.StringUtils.isBlank; import static org.apache.commons.lang.StringUtils.isNotBlank; import static org.apache.commons.lang.StringUtils.isNotEmpty; import static org.apache.commons.lang3.StringUtils.isEmpty; @@ -119,9 +120,9 @@ public static FunctionDetails convert(FunctionConfig functionConfig, ClassLoader functionConfig.getInputSpecs().forEach((topicName, consumerConf) -> { Function.ConsumerSpec.Builder bldr = Function.ConsumerSpec.newBuilder() .setIsRegexPattern(consumerConf.isRegexPattern()); - if (!StringUtils.isBlank(consumerConf.getSchemaType())) { + if (isNotBlank(consumerConf.getSchemaType())) { bldr.setSchemaType(consumerConf.getSchemaType()); - } else if (!StringUtils.isBlank(consumerConf.getSerdeClassName())) { + } else if (isNotBlank(consumerConf.getSerdeClassName())) { bldr.setSerdeClassName(consumerConf.getSerdeClassName()); } if (consumerConf.getReceiverQueueSize() != null) { @@ -131,6 +132,9 @@ public static FunctionDetails convert(FunctionConfig functionConfig, ClassLoader if (consumerConf.getSchemaProperties() != null) { bldr.putAllSchemaProperties(consumerConf.getSchemaProperties()); } + if (consumerConf.getCryptoConfig() != null) { + bldr.setCryptoSpec(CryptoUtils.convert(consumerConf.getCryptoConfig())); + } bldr.putAllConsumerProperties(consumerConf.getConsumerProperties()); sourceSpecBuilder.putInputSpecs(topicName, bldr.build()); }); @@ -208,15 +212,19 @@ public static FunctionDetails convert(FunctionConfig functionConfig, ClassLoader sinkSpecBuilder.setTypeClassName(typeArgs[1].getName()); } if (functionConfig.getProducerConfig() != null) { + ProducerConfig producerConf = functionConfig.getProducerConfig(); Function.ProducerSpec.Builder pbldr = Function.ProducerSpec.newBuilder(); - if (functionConfig.getProducerConfig().getMaxPendingMessages() != null) { - pbldr.setMaxPendingMessages(functionConfig.getProducerConfig().getMaxPendingMessages()); + if (producerConf.getMaxPendingMessages() != null) { + pbldr.setMaxPendingMessages(producerConf.getMaxPendingMessages()); + } + if (producerConf.getMaxPendingMessagesAcrossPartitions() != null) { + pbldr.setMaxPendingMessagesAcrossPartitions(producerConf.getMaxPendingMessagesAcrossPartitions()); } - if (functionConfig.getProducerConfig().getMaxPendingMessagesAcrossPartitions() != null) { - pbldr.setMaxPendingMessagesAcrossPartitions(functionConfig.getProducerConfig().getMaxPendingMessagesAcrossPartitions()); + if (producerConf.getUseThreadLocalProducers() != null) { + pbldr.setUseThreadLocalProducers(producerConf.getUseThreadLocalProducers()); } - if (functionConfig.getProducerConfig().getUseThreadLocalProducers() != null) { - pbldr.setUseThreadLocalProducers(functionConfig.getProducerConfig().getUseThreadLocalProducers()); + if (producerConf.getCryptoConfig() != null) { + pbldr.setCryptoSpec(CryptoUtils.convert(producerConf.getCryptoConfig())); } sinkSpecBuilder.setProducerSpec(pbldr.build()); } @@ -331,15 +339,18 @@ public static FunctionConfig convertFromDetails(FunctionDetails functionDetails) Map consumerConfigMap = new HashMap<>(); for (Map.Entry input : functionDetails.getSource().getInputSpecsMap().entrySet()) { ConsumerConfig consumerConfig = new ConsumerConfig(); - if (!isEmpty(input.getValue().getSerdeClassName())) { + if (isNotEmpty(input.getValue().getSerdeClassName())) { consumerConfig.setSerdeClassName(input.getValue().getSerdeClassName()); } - if (!isEmpty(input.getValue().getSchemaType())) { + if (isNotEmpty(input.getValue().getSchemaType())) { consumerConfig.setSchemaType(input.getValue().getSchemaType()); } if (input.getValue().hasReceiverQueueSize()) { consumerConfig.setReceiverQueueSize(input.getValue().getReceiverQueueSize().getValue()); } + if (input.getValue().hasCryptoSpec()) { + consumerConfig.setCryptoConfig(CryptoUtils.convertFromSpec(input.getValue().getCryptoSpec())); + } consumerConfig.setRegexPattern(input.getValue().getIsRegexPattern()); consumerConfig.setSchemaProperties(input.getValue().getSchemaPropertiesMap()); consumerConfigMap.put(input.getKey(), consumerConfig); @@ -366,14 +377,18 @@ public static FunctionConfig convertFromDetails(FunctionDetails functionDetails) functionConfig.setOutputSchemaType(functionDetails.getSink().getSchemaType()); } if (functionDetails.getSink().getProducerSpec() != null) { + Function.ProducerSpec spec = functionDetails.getSink().getProducerSpec(); ProducerConfig producerConfig = new ProducerConfig(); - if (functionDetails.getSink().getProducerSpec().getMaxPendingMessages() != 0) { - producerConfig.setMaxPendingMessages(functionDetails.getSink().getProducerSpec().getMaxPendingMessages()); + if (spec.getMaxPendingMessages() != 0) { + producerConfig.setMaxPendingMessages(spec.getMaxPendingMessages()); + } + if (spec.getMaxPendingMessagesAcrossPartitions() != 0) { + producerConfig.setMaxPendingMessagesAcrossPartitions(spec.getMaxPendingMessagesAcrossPartitions()); } - if (functionDetails.getSink().getProducerSpec().getMaxPendingMessagesAcrossPartitions() != 0) { - producerConfig.setMaxPendingMessagesAcrossPartitions(functionDetails.getSink().getProducerSpec().getMaxPendingMessagesAcrossPartitions()); + if (spec.hasCryptoSpec()) { + producerConfig.setCryptoConfig(CryptoUtils.convertFromSpec(spec.getCryptoSpec())); } - producerConfig.setUseThreadLocalProducers(functionDetails.getSink().getProducerSpec().getUseThreadLocalProducers()); + producerConfig.setUseThreadLocalProducers(spec.getUseThreadLocalProducers()); functionConfig.setProducerConfig(producerConfig); } if (!isEmpty(functionDetails.getLogTopic())) { @@ -544,6 +559,9 @@ private static void doJavaChecks(FunctionConfig functionConfig, ClassLoader clsL if (!isEmpty(conf.getSchemaType())) { ValidatorUtils.validateSchema(conf.getSchemaType(), typeArgs[0], clsLoader, true); } + if (conf.getCryptoConfig() != null) { + ValidatorUtils.validateCryptoKeyReader(conf.getCryptoConfig(), clsLoader, false); + } }); } @@ -565,6 +583,9 @@ private static void doJavaChecks(FunctionConfig functionConfig, ClassLoader clsL ValidatorUtils.validateSerde(functionConfig.getOutputSerdeClassName(), typeArgs[1], clsLoader, false); } + if (functionConfig.getProducerConfig() != null && functionConfig.getProducerConfig().getCryptoConfig() != null) { + ValidatorUtils.validateCryptoKeyReader(functionConfig.getProducerConfig().getCryptoConfig(), clsLoader, true); + } } private static void doPythonChecks(FunctionConfig functionConfig) { @@ -725,8 +746,24 @@ private static void doCommonChecks(FunctionConfig functionConfig) { throw new IllegalArgumentException( "Receiver queue size should be >= zero"); } + + if (conf.getCryptoConfig() != null && isBlank(conf.getCryptoConfig().getCryptoKeyReaderClassName())) { + throw new IllegalArgumentException( + "CryptoKeyReader class name required"); + } }); } + + if (functionConfig.getProducerConfig() != null && functionConfig.getProducerConfig().getCryptoConfig() != null) { + if (isBlank(functionConfig.getProducerConfig().getCryptoConfig().getCryptoKeyReaderClassName())) { + throw new IllegalArgumentException("CryptoKeyReader class name required"); + } + + if (functionConfig.getProducerConfig().getCryptoConfig().getEncryptionKeys() == null + || functionConfig.getProducerConfig().getCryptoConfig().getEncryptionKeys().length == 0) { + throw new IllegalArgumentException("Must provide encryption key name for crypto key reader"); + } + } } private static Collection collectAllInputTopics(FunctionConfig functionConfig) { diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java index 2302f4b570bd8..e046537fb0218 100644 --- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java +++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java @@ -29,6 +29,7 @@ import org.apache.commons.lang.StringUtils; import org.apache.pulsar.client.api.SubscriptionInitialPosition; import org.apache.pulsar.common.functions.ConsumerConfig; +import org.apache.pulsar.common.functions.CryptoConfig; import org.apache.pulsar.common.functions.FunctionConfig; import org.apache.pulsar.common.functions.Resources; import org.apache.pulsar.common.io.ConnectorDefinition; @@ -55,6 +56,7 @@ import static org.apache.commons.lang3.StringUtils.isEmpty; import static org.apache.commons.lang3.StringUtils.isNotBlank; +import static org.apache.commons.lang3.StringUtils.isNotEmpty; import static org.apache.pulsar.functions.utils.FunctionCommon.convertProcessingGuarantee; import static org.apache.pulsar.functions.utils.FunctionCommon.getSinkType; @@ -134,15 +136,18 @@ public static FunctionDetails convert(SinkConfig sinkConfig, ExtractedSinkDetail sinkConfig.getInputSpecs().forEach((topic, spec) -> { Function.ConsumerSpec.Builder bldr = Function.ConsumerSpec.newBuilder() .setIsRegexPattern(spec.isRegexPattern()); - if (!StringUtils.isBlank(spec.getSchemaType())) { + if (StringUtils.isNotBlank(spec.getSchemaType())) { bldr.setSchemaType(spec.getSchemaType()); - } else if (!StringUtils.isBlank(spec.getSerdeClassName())) { + } else if (StringUtils.isNotBlank(spec.getSerdeClassName())) { bldr.setSerdeClassName(spec.getSerdeClassName()); } if (spec.getReceiverQueueSize() != null) { bldr.setReceiverQueueSize(Function.ConsumerSpec.ReceiverQueueSize.newBuilder() .setValue(spec.getReceiverQueueSize()).build()); } + if (spec.getCryptoConfig() != null) { + bldr.setCryptoSpec(CryptoUtils.convert(spec.getCryptoConfig())); + } bldr.putAllConsumerProperties(spec.getConsumerProperties()); sourceSpecBuilder.putInputSpecs(topic, bldr.build()); }); @@ -259,6 +264,9 @@ public static SinkConfig convertFromDetails(FunctionDetails functionDetails) { if (input.getValue().hasReceiverQueueSize()) { consumerConfig.setReceiverQueueSize(input.getValue().getReceiverQueueSize().getValue()); } + if (input.getValue().hasCryptoSpec()) { + consumerConfig.setCryptoConfig(CryptoUtils.convertFromSpec(input.getValue().getCryptoSpec())); + } consumerConfig.setRegexPattern(input.getValue().getIsRegexPattern()); consumerConfig.setConsumerProperties(input.getValue().getConsumerPropertiesMap()); consumerConfigMap.put(input.getKey(), consumerConfig); @@ -490,6 +498,9 @@ public static ExtractedSinkDetails validate(SinkConfig sinkConfig, Path archiveP if (!isEmpty(consumerSpec.getSchemaType())) { ValidatorUtils.validateSchema(consumerSpec.getSchemaType(), typeArg, classLoader, true); } + if (consumerSpec.getCryptoConfig() != null) { + ValidatorUtils.validateCryptoKeyReader(consumerSpec.getCryptoConfig(), classLoader, false); + } } } return new ExtractedSinkDetails(sinkClassName, typeArg.getName()); @@ -637,4 +648,4 @@ public static void validateConnectorConfig(SinkConfig sinkConfig, ClassLoader cl throw new IllegalArgumentException("Could not validate sink config: " + e.getMessage()); } } -} \ No newline at end of file +} diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java index 0d31c9d3bc5af..53c481b6bf218 100644 --- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java +++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java @@ -54,6 +54,7 @@ import java.util.Map; import static org.apache.commons.lang3.StringUtils.isEmpty; +import static org.apache.commons.lang3.StringUtils.isNotEmpty; import static org.apache.pulsar.functions.utils.FunctionCommon.convertProcessingGuarantee; import static org.apache.pulsar.functions.utils.FunctionCommon.getSourceType; @@ -148,15 +149,19 @@ public static FunctionDetails convert(SourceConfig sourceConfig, ExtractedSource } if (sourceConfig.getProducerConfig() != null) { + ProducerConfig conf = sourceConfig.getProducerConfig(); Function.ProducerSpec.Builder pbldr = Function.ProducerSpec.newBuilder(); - if (sourceConfig.getProducerConfig().getMaxPendingMessages() != null) { - pbldr.setMaxPendingMessages(sourceConfig.getProducerConfig().getMaxPendingMessages()); + if (conf.getMaxPendingMessages() != null) { + pbldr.setMaxPendingMessages(conf.getMaxPendingMessages()); } - if (sourceConfig.getProducerConfig().getMaxPendingMessagesAcrossPartitions() != null) { - pbldr.setMaxPendingMessagesAcrossPartitions(sourceConfig.getProducerConfig().getMaxPendingMessagesAcrossPartitions()); + if (conf.getMaxPendingMessagesAcrossPartitions() != null) { + pbldr.setMaxPendingMessagesAcrossPartitions(conf.getMaxPendingMessagesAcrossPartitions()); } - if (sourceConfig.getProducerConfig().getUseThreadLocalProducers() != null) { - pbldr.setUseThreadLocalProducers(sourceConfig.getProducerConfig().getUseThreadLocalProducers()); + if (conf.getUseThreadLocalProducers() != null) { + pbldr.setUseThreadLocalProducers(conf.getUseThreadLocalProducers()); + } + if (conf.getCryptoConfig() != null) { + pbldr.setCryptoSpec(CryptoUtils.convert(conf.getCryptoConfig())); } sinkSpecBuilder.setProducerSpec(pbldr.build()); } @@ -231,14 +236,18 @@ public static SourceConfig convertFromDetails(FunctionDetails functionDetails) { sourceConfig.setSerdeClassName(sinkSpec.getSerDeClassName()); } if (sinkSpec.getProducerSpec() != null) { + Function.ProducerSpec spec = sinkSpec.getProducerSpec(); ProducerConfig producerConfig = new ProducerConfig(); - if (sinkSpec.getProducerSpec().getMaxPendingMessages() != 0) { - producerConfig.setMaxPendingMessages(sinkSpec.getProducerSpec().getMaxPendingMessages()); + if (spec.getMaxPendingMessages() != 0) { + producerConfig.setMaxPendingMessages(spec.getMaxPendingMessages()); + } + if (spec.getMaxPendingMessagesAcrossPartitions() != 0) { + producerConfig.setMaxPendingMessagesAcrossPartitions(spec.getMaxPendingMessagesAcrossPartitions()); } - if (sinkSpec.getProducerSpec().getMaxPendingMessagesAcrossPartitions() != 0) { - producerConfig.setMaxPendingMessagesAcrossPartitions(sinkSpec.getProducerSpec().getMaxPendingMessagesAcrossPartitions()); + if (spec.hasCryptoSpec()) { + producerConfig.setCryptoConfig(CryptoUtils.convertFromSpec(spec.getCryptoSpec())); } - producerConfig.setUseThreadLocalProducers(sinkSpec.getProducerSpec().getUseThreadLocalProducers()); + producerConfig.setUseThreadLocalProducers(spec.getUseThreadLocalProducers()); sourceConfig.setProducerConfig(producerConfig); } if (functionDetails.hasResources()) { @@ -424,6 +433,10 @@ public static ExtractedSourceDetails validate(SourceConfig sourceConfig, Path ar ValidatorUtils.validateSchema(sourceConfig.getSchemaType(), typeArg, classLoader, false); } + if (sourceConfig.getProducerConfig().getCryptoConfig() != null) { + ValidatorUtils.validateCryptoKeyReader(sourceConfig.getProducerConfig().getCryptoConfig(), classLoader, true); + } + if (typeArg.equals(TypeResolver.Unknown.class)) { throw new IllegalArgumentException( String.format("Failed to resolve type for Source class %s", sourceClassName)); diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/ValidatorUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/ValidatorUtils.java index 7eeeda00e06d5..9203cba4bf650 100644 --- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/ValidatorUtils.java +++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/ValidatorUtils.java @@ -21,7 +21,10 @@ import lombok.extern.slf4j.Slf4j; import net.jodah.typetools.TypeResolver; +import org.apache.commons.lang3.reflect.ConstructorUtils; +import org.apache.pulsar.client.api.CryptoKeyReader; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.common.functions.CryptoConfig; import org.apache.pulsar.common.schema.SchemaType; import org.apache.pulsar.common.util.ClassLoaderUtils; import org.apache.pulsar.common.util.Reflections; @@ -30,6 +33,10 @@ import org.apache.pulsar.io.core.Sink; import org.apache.pulsar.io.core.Source; +import java.lang.reflect.Constructor; +import java.lang.reflect.InvocationTargetException; +import java.util.Map; + import static org.apache.commons.lang3.StringUtils.isBlank; import static org.apache.commons.lang3.StringUtils.isEmpty; import static org.apache.commons.lang3.StringUtils.isNotBlank; @@ -59,6 +66,35 @@ private static SchemaType getBuiltinSchemaType(String schemaTypeOrClassName) { } } + + public static void validateCryptoKeyReader(CryptoConfig conf, ClassLoader classLoader, boolean isProducer) { + if (isEmpty(conf.getCryptoKeyReaderClassName())) return; + + Class cryptoClass; + try { + cryptoClass = ClassLoaderUtils.loadClass(conf.getCryptoKeyReaderClassName(), classLoader); + } catch (ClassNotFoundException | NoClassDefFoundError e) { + throw new IllegalArgumentException( + String.format("The crypto key reader class %s does not exist", conf.getCryptoKeyReaderClassName())); + } + ClassLoaderUtils.implementsClass(conf.getCryptoKeyReaderClassName(), CryptoKeyReader.class, classLoader); + + try { + cryptoClass.getConstructor(Map.class); + } catch (NoSuchMethodException ex) { + throw new IllegalArgumentException( + String.format("The crypto key reader class %s does not implement the desired constructor.", + conf.getCryptoKeyReaderClassName())); + + } catch (SecurityException e) { + throw new IllegalArgumentException("Failed to access crypto key reader class", e); + } + + if (isProducer && (conf.getEncryptionKeys() == null || conf.getEncryptionKeys().length == 0)) { + throw new IllegalArgumentException("Missing encryption key name for producer crypto key reader"); + } + } + public static void validateSerde(String inputSerializer, Class typeArg, ClassLoader clsLoader, boolean deser) { if (isEmpty(inputSerializer)) return; @@ -134,6 +170,7 @@ private static void validateSchemaType(String schemaClassName, Class typeArg, } } + public static void validateFunctionClassTypes(ClassLoader classLoader, Function.FunctionDetails.Builder functionDetailsBuilder) { // validate only if classLoader is provided From 63e012c27d9514d48ffea463efa0dc5c7c211df0 Mon Sep 17 00:00:00 2001 From: Neng Lu Date: Mon, 2 Nov 2020 10:14:11 -0800 Subject: [PATCH 2/4] setup provider in PulsarSource/Sink --- pulsar-functions/instance/pom.xml | 6 +++ .../pulsar/functions/sink/PulsarSink.java | 6 +-- .../pulsar/functions/source/PulsarSource.java | 6 +-- .../api/examples/RawFileKeyReader.java | 51 +++++++++++++++++++ 4 files changed, 63 insertions(+), 6 deletions(-) create mode 100644 pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/RawFileKeyReader.java diff --git a/pulsar-functions/instance/pom.xml b/pulsar-functions/instance/pom.xml index 859b6ecafdbc9..547700ea0781c 100644 --- a/pulsar-functions/instance/pom.xml +++ b/pulsar-functions/instance/pom.xml @@ -77,6 +77,12 @@ ${project.version} + + ${project.groupId} + pulsar-client-messagecrypto-bc + ${project.parent.version} + + org.apache.bookkeeper stream-storage-java-client diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java index c56a838df5115..77474792a9a06 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java @@ -398,9 +398,9 @@ Crypto initializeCrypto() throws ClassNotFoundException { } // add provider only if it's not in the JVM -// if (Security.getProvider(BouncyCastleProvider.PROVIDER_NAME) == null) { -// Security.addProvider(new BouncyCastleProvider()); -// } + if (Security.getProvider(BouncyCastleProvider.PROVIDER_NAME) == null) { + Security.addProvider(new BouncyCastleProvider()); + } Crypto.CryptoBuilder bldr = Crypto.builder() .failureAction(cryptoConfig.getProducerCryptoFailureAction()) diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java index a4f4fe5427240..d879ff3cfcc91 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java @@ -188,9 +188,9 @@ Map> setupConsumerConfigs() throws ClassNotFoundExcept if (conf.getCryptoConfig() != null) { // add provider only if it's not in the JVM -// if (Security.getProvider(BouncyCastleProvider.PROVIDER_NAME) == null) { -// Security.addProvider(new BouncyCastleProvider()); -// } + if (Security.getProvider(BouncyCastleProvider.PROVIDER_NAME) == null) { + Security.addProvider(new BouncyCastleProvider()); + } consumerConfBuilder.consumerCryptoFailureAction(conf.getCryptoConfig().getConsumerCryptoFailureAction()); consumerConfBuilder.cryptoKeyReader(CryptoUtils.getCryptoKeyReaderInstance( diff --git a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/RawFileKeyReader.java b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/RawFileKeyReader.java new file mode 100644 index 0000000000000..89139140224c3 --- /dev/null +++ b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/RawFileKeyReader.java @@ -0,0 +1,51 @@ +package org.apache.pulsar.functions.api.examples; + +import lombok.Data; +import org.apache.pulsar.client.api.CryptoKeyReader; +import org.apache.pulsar.client.api.EncryptionKeyInfo; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.util.Map; + +@Data +public class RawFileKeyReader implements CryptoKeyReader { + + private final String publicKeyFile; + private final String privateKeyFile; + + public RawFileKeyReader(String pubKeyFile, String privKeyFile) { + publicKeyFile = pubKeyFile; + privateKeyFile = privKeyFile; + } + + public RawFileKeyReader(Map conf) { + publicKeyFile = (String) conf.get("PUBLIC"); + privateKeyFile = (String) conf.get("PRIVATE"); + } + + @Override + public EncryptionKeyInfo getPublicKey(String keyName, Map keyMeta) { + EncryptionKeyInfo keyInfo = new EncryptionKeyInfo(); + try { + keyInfo.setKey(Files.readAllBytes(Paths.get(publicKeyFile))); + } catch (IOException e) { + System.out.println("ERROR: Failed to read public key from file " + publicKeyFile); + e.printStackTrace(); + } + return keyInfo; + } + + @Override + public EncryptionKeyInfo getPrivateKey(String keyName, Map keyMeta) { + EncryptionKeyInfo keyInfo = new EncryptionKeyInfo(); + try { + keyInfo.setKey(Files.readAllBytes(Paths.get(privateKeyFile))); + } catch (IOException e) { + System.out.println("ERROR: Failed to read private key from file " + privateKeyFile); + e.printStackTrace(); + } + return keyInfo; + } +} From fca6352f13c0ce1e549ca935892e73bae45bee21 Mon Sep 17 00:00:00 2001 From: Neng Lu Date: Tue, 3 Nov 2020 11:33:57 -0800 Subject: [PATCH 3/4] fix header and unit test --- .../pulsar/common/functions/CryptoConfig.java | 20 +++++++++++++++- .../pulsar/functions/sink/PulsarSink.java | 8 ++++--- .../api/examples/RawFileKeyReader.java | 23 +++++++++++++++++++ .../pulsar/functions/utils/CryptoUtils.java | 19 +++++++++++++++ 4 files changed, 66 insertions(+), 4 deletions(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/CryptoConfig.java b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/CryptoConfig.java index f40edb99e6169..cfcfd0d434b73 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/CryptoConfig.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/CryptoConfig.java @@ -1,5 +1,23 @@ -package org.apache.pulsar.common.functions; +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.common.functions; import java.util.Map; diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java index 77474792a9a06..f1219501f6830 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java @@ -391,12 +391,14 @@ Schema initializeSchema() throws ClassNotFoundException { @SuppressWarnings("unchecked") @VisibleForTesting Crypto initializeCrypto() throws ClassNotFoundException { - CryptoConfig cryptoConfig = pulsarSinkConfig.getProducerConfig().getCryptoConfig(); - - if (cryptoConfig == null || isEmpty(cryptoConfig.getCryptoKeyReaderClassName())) { + if (pulsarSinkConfig.getProducerConfig() == null + || pulsarSinkConfig.getProducerConfig().getCryptoConfig() == null + || isEmpty(pulsarSinkConfig.getProducerConfig().getCryptoConfig().getCryptoKeyReaderClassName())) { return null; } + CryptoConfig cryptoConfig = pulsarSinkConfig.getProducerConfig().getCryptoConfig(); + // add provider only if it's not in the JVM if (Security.getProvider(BouncyCastleProvider.PROVIDER_NAME) == null) { Security.addProvider(new BouncyCastleProvider()); diff --git a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/RawFileKeyReader.java b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/RawFileKeyReader.java index 89139140224c3..d55e3fff6a248 100644 --- a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/RawFileKeyReader.java +++ b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/RawFileKeyReader.java @@ -1,3 +1,22 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + package org.apache.pulsar.functions.api.examples; import lombok.Data; @@ -9,6 +28,10 @@ import java.nio.file.Paths; import java.util.Map; +/** + * Example function that provide a constructor with map argument + * to initialize the CryptoKeyReader class used by pulsar function + */ @Data public class RawFileKeyReader implements CryptoKeyReader { diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/CryptoUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/CryptoUtils.java index 9dd78beaa306b..1025113394bdb 100644 --- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/CryptoUtils.java +++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/CryptoUtils.java @@ -1,3 +1,22 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + package org.apache.pulsar.functions.utils; import java.lang.reflect.Constructor; From b758265f07f9dac4da8c5573fa656f66dbe4afa1 Mon Sep 17 00:00:00 2001 From: Neng Lu Date: Wed, 4 Nov 2020 14:57:07 -0800 Subject: [PATCH 4/4] fix unit test --- .../org/apache/pulsar/functions/utils/SourceConfigUtils.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java index 53c481b6bf218..aec11fee898e5 100644 --- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java +++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java @@ -433,7 +433,7 @@ public static ExtractedSourceDetails validate(SourceConfig sourceConfig, Path ar ValidatorUtils.validateSchema(sourceConfig.getSchemaType(), typeArg, classLoader, false); } - if (sourceConfig.getProducerConfig().getCryptoConfig() != null) { + if (sourceConfig.getProducerConfig() != null && sourceConfig.getProducerConfig().getCryptoConfig() != null) { ValidatorUtils.validateCryptoKeyReader(sourceConfig.getProducerConfig().getCryptoConfig(), classLoader, true); }