[SDC-DISTRO-CLIENT] Add TLS config to Kafka
Change-Id: I5fe28be288a042d55b53d32008a144bb5cd6e5c4
Signed-off-by: david.mcweeney <david.mcweeney@est.tech>
Issue-ID: SDC-4476
diff --git a/sdc-distribution-client/src/main/java/org/onap/sdc/utils/kafka/KafkaCommonConfig.java b/sdc-distribution-client/src/main/java/org/onap/sdc/utils/kafka/KafkaCommonConfig.java
new file mode 100644
index 0000000..477e677
--- /dev/null
+++ b/sdc-distribution-client/src/main/java/org/onap/sdc/utils/kafka/KafkaCommonConfig.java
@@ -0,0 +1,82 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * sdc-distribution-client
+ * ================================================================================
+ * Copyright (C) 2023 Nordix Foundation. All rights reserved.
+ * ================================================================================
+ * Licensed 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.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.sdc.utils.kafka;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.SslConfigs;
+import org.onap.sdc.impl.Configuration;
+import java.util.Properties;
+import java.util.UUID;
+
+public class KafkaCommonConfig {
+ private final Configuration configuration;
+ public KafkaCommonConfig(Configuration configuration){
+ this.configuration = configuration;
+ }
+
+ public Properties getConsumerProperties(){
+ Properties props = new Properties();
+ setCommonProperties(props);
+
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
+ props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
+ props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, configuration.getKafkaConsumerMaxPollInterval() * 1000);
+ props.put(ConsumerConfig.CLIENT_ID_CONFIG, configuration.getConsumerID() + "-consumer-" + UUID.randomUUID());
+ props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, configuration.getKafkaConsumerSessionTimeout() * 1000);
+ props.put(ConsumerConfig.GROUP_ID_CONFIG, configuration.getConsumerGroup());
+ props.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, false);
+ props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
+
+
+ return props;
+ }
+
+ public Properties getProducerProperties(){
+ Properties props = new Properties();
+ setCommonProperties(props);
+ props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+ props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+ props.put(ProducerConfig.CLIENT_ID_CONFIG, configuration.getConsumerID() + "-producer-" + UUID.randomUUID());
+
+ return props;
+ }
+
+ private void setCommonProperties(Properties props) {
+ String securityProtocolConfig = configuration.getKafkaSecurityProtocolConfig();
+ props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocolConfig);
+ props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, configuration.getMsgBusAddress());
+
+ if("SSL".equals(securityProtocolConfig)) {
+ props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, configuration.getTrustStorePassword());
+ props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, configuration.getTrustStorePath());
+ props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, configuration.getKeyStorePassword());
+ props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, configuration.getKeyStorePath());
+ props.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, configuration.getKeyStorePassword());
+ }
+ else{
+ props.put(SaslConfigs.SASL_JAAS_CONFIG, configuration.getKafkaSaslJaasConfig());
+ props.put(SaslConfigs.SASL_MECHANISM, configuration.getKafkaSaslMechanism());
+ }
+ }
+
+}
diff --git a/sdc-distribution-client/src/main/java/org/onap/sdc/utils/kafka/SdcKafkaConsumer.java b/sdc-distribution-client/src/main/java/org/onap/sdc/utils/kafka/SdcKafkaConsumer.java
index 982ba5d..c8c92bb 100644
--- a/sdc-distribution-client/src/main/java/org/onap/sdc/utils/kafka/SdcKafkaConsumer.java
+++ b/sdc-distribution-client/src/main/java/org/onap/sdc/utils/kafka/SdcKafkaConsumer.java
@@ -53,21 +53,10 @@
* @param configuration The config provided to the client
*/
public SdcKafkaConsumer(Configuration configuration) {
- Properties props = new Properties();
- props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, configuration.getMsgBusAddress());
- props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, configuration.getKafkaSecurityProtocolConfig());
- props.put(SaslConfigs.SASL_MECHANISM, configuration.getKafkaSaslMechanism());
- props.put(SaslConfigs.SASL_JAAS_CONFIG, configuration.getKafkaSaslJaasConfig());
- props.put(ConsumerConfig.GROUP_ID_CONFIG, configuration.getConsumerGroup());
- props.put(ConsumerConfig.CLIENT_ID_CONFIG, configuration.getConsumerID() + "-consumer-" + UUID.randomUUID());
- props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
- props.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, false);
- props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, configuration.getKafkaConsumerMaxPollInterval() * 1000);
- props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, configuration.getKafkaConsumerSessionTimeout() * 1000);
- props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
- props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
- consumer = new KafkaConsumer<>(props);
+ KafkaCommonConfig kafkaCommonConfig = new KafkaCommonConfig(configuration);
+ Properties props = kafkaCommonConfig.getConsumerProperties();
pollTimeout = configuration.getPollingTimeout();
+ consumer = new KafkaConsumer<>(props);
}
/**
diff --git a/sdc-distribution-client/src/main/java/org/onap/sdc/utils/kafka/SdcKafkaProducer.java b/sdc-distribution-client/src/main/java/org/onap/sdc/utils/kafka/SdcKafkaProducer.java
index 19338fc..b151b23 100644
--- a/sdc-distribution-client/src/main/java/org/onap/sdc/utils/kafka/SdcKafkaProducer.java
+++ b/sdc-distribution-client/src/main/java/org/onap/sdc/utils/kafka/SdcKafkaProducer.java
@@ -31,6 +31,7 @@
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.SslConfigs;
import org.onap.sdc.impl.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,14 +51,8 @@
* @param configuration The config provided to the client
*/
public SdcKafkaProducer(Configuration configuration) {
- Properties props = new Properties();
- props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, configuration.getMsgBusAddress());
- props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, configuration.getKafkaSecurityProtocolConfig());
- props.put(SaslConfigs.SASL_MECHANISM, configuration.getKafkaSaslMechanism());
- props.put(SaslConfigs.SASL_JAAS_CONFIG, configuration.getKafkaSaslJaasConfig());
- props.put(ProducerConfig.CLIENT_ID_CONFIG, configuration.getConsumerID() + "-producer-" + UUID.randomUUID());
- props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
- props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+ KafkaCommonConfig kafkaCommonConfig = new KafkaCommonConfig(configuration);
+ Properties props = kafkaCommonConfig.getProducerProperties();
producer = new KafkaProducer<>(props);
msgBusAddresses = configuration.getMsgBusAddress();
topicName = configuration.getStatusTopicName();
diff --git a/sdc-distribution-client/src/test/java/org/onap/sdc/utils/KafkaCommonConfigTest.java b/sdc-distribution-client/src/test/java/org/onap/sdc/utils/KafkaCommonConfigTest.java
new file mode 100644
index 0000000..36730b5
--- /dev/null
+++ b/sdc-distribution-client/src/test/java/org/onap/sdc/utils/KafkaCommonConfigTest.java
@@ -0,0 +1,57 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * sdc-distribution-client
+ * ================================================================================
+ * Copyright (C) 2023 Nordix Foundation. All rights reserved.
+ * ================================================================================
+ * Licensed 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.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.sdc.utils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.SslConfigs;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.onap.sdc.impl.Configuration;
+import org.onap.sdc.utils.kafka.KafkaCommonConfig;
+
+public class KafkaCommonConfigTest {
+ private static final Configuration testConfigNoSSL = new Configuration(new TestConfiguration());
+ private static final Configuration testConfigWithSSL = new Configuration(new TestConfigurationSSLProtocol());
+
+ @Test
+ public void testConsumerPropertiesNoSSL(){
+ List<String> msgBusAddress = new ArrayList<>();
+ msgBusAddress.add("address1");
+ testConfigNoSSL.setMsgBusAddress(msgBusAddress);
+ KafkaCommonConfig kafkaCommonConfig = new KafkaCommonConfig(testConfigNoSSL);
+ Properties consumerProperties = kafkaCommonConfig.getConsumerProperties();
+ Assertions.assertEquals(consumerProperties.getProperty(SaslConfigs.SASL_JAAS_CONFIG), testConfigNoSSL.getKafkaSaslJaasConfig());
+ }
+
+ @Test
+ public void testProducerPropertiesWithSSL(){
+ List<String> msgBusAddress = new ArrayList<>();
+ msgBusAddress.add("address1");
+ testConfigWithSSL.setMsgBusAddress(msgBusAddress);
+ KafkaCommonConfig kafkaCommonConfig = new KafkaCommonConfig(testConfigWithSSL);
+ Properties consumerProperties = kafkaCommonConfig.getProducerProperties();
+
+ Assertions.assertNull(consumerProperties.getProperty(SaslConfigs.SASL_JAAS_CONFIG));
+ Assertions.assertEquals(consumerProperties.getProperty(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG), testConfigNoSSL.getTrustStorePassword());
+ }
+}
diff --git a/sdc-distribution-client/src/test/java/org/onap/sdc/utils/SdcKafkaTest.java b/sdc-distribution-client/src/test/java/org/onap/sdc/utils/SdcKafkaTest.java
index 2037be6..c0c60a8 100644
--- a/sdc-distribution-client/src/test/java/org/onap/sdc/utils/SdcKafkaTest.java
+++ b/sdc-distribution-client/src/test/java/org/onap/sdc/utils/SdcKafkaTest.java
@@ -34,14 +34,12 @@
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
-import org.junitpioneer.jupiter.SetEnvironmentVariable;
import org.onap.sdc.impl.Configuration;
import org.onap.sdc.utils.kafka.SdcKafkaConsumer;
import org.onap.sdc.utils.kafka.SdcKafkaProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-@SetEnvironmentVariable(key = "SASL_JAAS_CONFIG", value = "org.apache.kafka.common.security.scram.ScramLoginModule required username=admin password=admin-secret;")
class SdcKafkaTest {
private static final Logger logger = LoggerFactory.getLogger(SdcKafkaTest.class);
diff --git a/sdc-distribution-client/src/test/java/org/onap/sdc/utils/TestConfigurationSSLProtocol.java b/sdc-distribution-client/src/test/java/org/onap/sdc/utils/TestConfigurationSSLProtocol.java
new file mode 100644
index 0000000..466bed9
--- /dev/null
+++ b/sdc-distribution-client/src/test/java/org/onap/sdc/utils/TestConfigurationSSLProtocol.java
@@ -0,0 +1,176 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * sdc-distribution-client
+ * ================================================================================
+ * Copyright (C) 2023 Nordix Foundation. All rights reserved.
+ * ================================================================================
+ * Licensed 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.sdc.utils;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.onap.sdc.api.consumer.IConfiguration;
+
+@Getter
+@Setter
+public class TestConfigurationSSLProtocol implements IConfiguration {
+
+ private String user;
+ private String password;
+ private int httpProxyPort;
+ private String sdcAddress;
+ private String consumerID;
+ private int httpsProxyPort;
+ private String keyStorePath;
+ private String httpProxyHost;
+ private String consumerGroup;
+ private String httpsProxyHost;
+ private Boolean useSystemProxy;
+ private String environmentName;
+ private String keyStorePassword;
+ private Boolean useHttpsWithSDC;
+ private String sdcStatusTopicName;
+ private boolean activateServerTLSAuth;
+ private String sdcNotificationTopicName;
+ private final String kafkaSaslMechanism;
+ private boolean isFilterInEmptyResources;
+ private final String kafkaSaslJaasConfig;
+ private List<String> relevantArtifactTypes;
+ private final int kafkaConsumerSessionTimeout;
+ private final int kafkaConsumerMaxPollInterval;
+ private final String kafkaSecurityProtocolConfig;
+ private int pollingTimeout = DistributionClientConstants.POLLING_TIMEOUT_SEC;
+ private int pollingInterval = DistributionClientConstants.MIN_POLLING_INTERVAL_SEC;
+
+ public TestConfigurationSSLProtocol() {
+ this.user = "mso-user";
+ this.pollingTimeout = 20;
+ this.httpProxyPort = 8080;
+ this.pollingInterval = 20;
+ this.password = "password";
+ this.useHttpsWithSDC = true;
+ this.httpProxyHost = "proxy";
+ this.environmentName = "PROD";
+ this.consumerID = "mso-123456";
+ this.consumerGroup = "mso-group";
+ this.activateServerTLSAuth = true;
+ this.kafkaSaslMechanism = "PLAIN";
+ this.sdcAddress = "localhost:8443";
+ this.keyStorePassword = "Aa123456";
+ this.isFilterInEmptyResources = false;
+ this.kafkaConsumerSessionTimeout = 50;
+ this.kafkaConsumerMaxPollInterval = 600;
+ this.keyStorePath = "etc/sdc-client.jks";
+ this.kafkaSecurityProtocolConfig = "SSL";
+ this.setSdcStatusTopicName("SDC-STATUS-TOPIC");
+ this.relevantArtifactTypes = new ArrayList<>();
+ this.setSdcNotificationTopicName("SDC-NOTIF-TOPIC");
+ this.relevantArtifactTypes.add(ArtifactTypeEnum.HEAT.name());
+ this.kafkaSaslJaasConfig = "org.apache.kafka.common.security.scram.ScramLoginModule required username=admin password=admin-secret;";
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + ((sdcAddress == null) ? 0 : sdcAddress.hashCode());
+ result = prime * result + ((consumerID == null) ? 0 : consumerID.hashCode());
+ result = prime * result + ((consumerGroup == null) ? 0 : consumerGroup.hashCode());
+ result = prime * result + ((environmentName == null) ? 0 : environmentName.hashCode());
+ result = prime * result + ((password == null) ? 0 : password.hashCode());
+ result = prime * result + pollingInterval;
+ result = prime * result + pollingTimeout;
+ result = prime * result + ((relevantArtifactTypes == null) ? 0 : relevantArtifactTypes.hashCode());
+ result = prime * result + ((user == null) ? 0 : user.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean activateServerTLSAuth() {
+
+ return activateServerTLSAuth;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj)
+ return true;
+ if (obj == null)
+ return false;
+ if (getClass() != obj.getClass())
+ return false;
+ TestConfigurationSSLProtocol other = (TestConfigurationSSLProtocol) obj;
+ if (sdcAddress == null) {
+ if (other.sdcAddress != null)
+ return false;
+ } else if (!sdcAddress.equals(other.sdcAddress))
+ return false;
+ if (consumerID == null) {
+ if (other.consumerID != null)
+ return false;
+ } else if (!consumerID.equals(other.consumerID))
+ return false;
+ if (consumerGroup == null) {
+ if (other.consumerGroup != null)
+ return false;
+ } else if (!consumerGroup.equals(other.consumerGroup))
+ return false;
+ if (environmentName == null) {
+ if (other.environmentName != null)
+ return false;
+ } else if (!environmentName.equals(other.environmentName))
+ return false;
+ if (password == null) {
+ if (other.password != null)
+ return false;
+ } else if (!password.equals(other.password))
+ return false;
+ if (pollingInterval != other.pollingInterval)
+ return false;
+ if (pollingTimeout != other.pollingTimeout)
+ return false;
+ if (relevantArtifactTypes == null) {
+ if (other.relevantArtifactTypes != null)
+ return false;
+ } else if (!relevantArtifactTypes.equals(other.relevantArtifactTypes))
+ return false;
+ if (user == null) {
+ if (other.user != null)
+ return false;
+ } else if (!user.equals(other.user))
+ return false;
+ if (keyStorePath == null) {
+ if (other.keyStorePath != null)
+ return false;
+ } else if (!keyStorePath.equals(other.keyStorePath))
+ return false;
+ if (keyStorePassword == null) {
+ return other.keyStorePassword == null;
+ } else
+ return keyStorePassword.equals(other.keyStorePassword);
+ }
+
+ @Override
+ public String toString() {
+ return "TestConfiguration [sdcAddress=" + sdcAddress + ", user=" + user + ", password=" + password
+ + ", pollingInterval=" + pollingInterval + ", pollingTimeout=" + pollingTimeout
+ + ", relevantArtifactTypes=" + relevantArtifactTypes + ", consumerGroup=" + consumerGroup
+ + ", environmentName=" + environmentName + ", consumerID=" + consumerID + "]";
+ }
+}