[SDC-DISTRO-CLIENT] Add TLS config to Kafka

Change-Id: I5fe28be288a042d55b53d32008a144bb5cd6e5c4
Signed-off-by: david.mcweeney <david.mcweeney@est.tech>
Issue-ID: SDC-4476

6 files changed, 321 insertions, 24 deletions

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 @@ public class SdcKafkaConsumer {
      * @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.ProducerRecord;
 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 @@ public class SdcKafkaProducer {
      * @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.AfterAll;
 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 + "]";
+	}
+}