Upgrade CBS java SDK to support SSL

 - add TrustStoreKeys class for one-way TLS for CBS client
 - use trust.jks & trust.pass
 - add unit test
 - top up version of Vavr lib (due to bug)

Issue-ID: DCAEGEN2-1552
Signed-off-by: Piotr Wielebski <piotr.wielebski@nokia.com>
Change-Id: I372c559cce5db8eba5448d99e12cdf6609c40d00

19 files changed, 399 insertions, 60 deletions

diff --git a/pom.xml b/pom.xml
index cd73fdae..79fef86e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -70,12 +70,14 @@
         <logback.version>1.2.3</logback.version>
         <mockito.version>2.28.2</mockito.version>
         <protobuf.version>3.6.1</protobuf.version>
-        <vavr.version>0.10.0</vavr.version>
+        <vavr.version>0.10.2</vavr.version>
         <commons-text.version>1.6</commons-text.version>
         <jetbrains-annotations.version>16.0.3</jetbrains-annotations.version>
         <protoc-jar-maven-plugin.version>3.6.0.2</protoc-jar-maven-plugin.version>
         <testcontainers.version>1.12.0</testcontainers.version>
         <spring.boot.version>2.2.1.RELEASE</spring.boot.version>
+        <spring.boot.version>2.1.5.RELEASE</spring.boot.version>
+        <system.rules.version>1.17.2</system.rules.version>
     </properties>
 
     <modules>
@@ -264,6 +266,12 @@
             </dependency>
 
             <dependency>
+                <groupId>com.github.stefanbirkner</groupId>
+                <artifactId>system-rules</artifactId>
+                <version>${system.rules.version}</version>
+                <scope>test</scope>
+            </dependency>
+            <dependency>
                 <groupId>org.mockito</groupId>
                 <artifactId>mockito-core</artifactId>
                 <version>${mockito.version}</version>
diff --git a/rest-services/cbs-client/pom.xml b/rest-services/cbs-client/pom.xml
index a9aace8a..d1f8aa0c 100644
--- a/rest-services/cbs-client/pom.xml
+++ b/rest-services/cbs-client/pom.xml
@@ -34,6 +34,11 @@
         </dependency>
 
         <dependency>
+            <groupId>com.github.stefanbirkner</groupId>
+            <artifactId>system-rules</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
             <groupId>org.mockito</groupId>
             <artifactId>mockito-core</artifactId>
             <scope>test</scope>
diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientFactory.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientFactory.java
index 821805fc..00aad603 100644
--- a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientFactory.java
+++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientFactory.java
@@ -25,6 +25,7 @@ import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.RxHttpClientFa
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.CbsClientImpl;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.CbsLookup;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
+import org.onap.dcaegen2.services.sdk.security.ssl.TrustStoreKeys;
 import reactor.core.publisher.Mono;
 
 /**
@@ -54,10 +55,16 @@ public class CbsClientFactory {
      */
     public static @NotNull Mono<CbsClient> createCbsClient(CbsClientConfiguration configuration) {
         return Mono.defer(() -> {
-            final RxHttpClient httpClient = RxHttpClientFactory.create();
-            final CbsLookup lookup = new CbsLookup();
-            return lookup.lookup(configuration)
-                    .map(addr -> new CbsClientImpl(httpClient, configuration.appName(), addr));
+            final RxHttpClient httpClient = buildHttpClient(configuration.trustStoreKeys());
+            final CbsLookup cbsLookup = new CbsLookup();
+            return cbsLookup.lookup(configuration)
+                    .map(addr -> new CbsClientImpl(httpClient, configuration.appName(), addr, configuration.protocol()));
         });
     }
+
+    private static RxHttpClient buildHttpClient(TrustStoreKeys trustStoreKeys) {
+        return trustStoreKeys != null
+                ? RxHttpClientFactory.create(trustStoreKeys)
+                : RxHttpClientFactory.create();
+    }
 }
diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/exceptions/CbsClientConfigurationException.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/exceptions/CbsClientConfigurationException.java
new file mode 100644
index 00000000..a7c88a3d
--- /dev/null
+++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/exceptions/CbsClientConfigurationException.java
@@ -0,0 +1,29 @@
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 Nokia. 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.dcaegen2.services.sdk.rest.services.cbs.client.api.exceptions;
+
+import org.jetbrains.annotations.NotNull;
+
+public class CbsClientConfigurationException extends RuntimeException {
+    public CbsClientConfigurationException(final @NotNull String message) {
+        super(message);
+    }
+}
\ No newline at end of file
diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImpl.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImpl.java
index 6f37cd2b..a895f3a1 100644
--- a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImpl.java
+++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImpl.java
@@ -20,9 +20,6 @@
 package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl;
 
 import com.google.gson.JsonObject;
-import java.net.InetSocketAddress;
-import java.net.MalformedURLException;
-import java.net.URL;
 import org.jetbrains.annotations.NotNull;
 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpMethod;
 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
@@ -34,17 +31,23 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Mono;
 
+import java.net.InetSocketAddress;
+import java.net.MalformedURLException;
+import java.net.URL;
+
 public class CbsClientImpl implements CbsClient {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(CbsClientImpl.class);
     private final RxHttpClient httpClient;
     private final String serviceName;
     private final InetSocketAddress cbsAddress;
+    private final String protocol;
 
-    public CbsClientImpl(RxHttpClient httpClient, String serviceName, InetSocketAddress cbsAddress) {
+    public CbsClientImpl(RxHttpClient httpClient, String serviceName, InetSocketAddress cbsAddress, String protocol) {
         this.httpClient = httpClient;
         this.serviceName = serviceName;
         this.cbsAddress = cbsAddress;
+        this.protocol = protocol;
     }
 
     @Override
@@ -66,7 +69,7 @@ public class CbsClientImpl implements CbsClient {
     private URL constructUrl(CbsRequest request) {
         try {
             return new URL(
-                    "http",
+                    this.protocol,
                     cbsAddress.getHostString(),
                     cbsAddress.getPort(),
                     request.requestPath().getForService(serviceName));
diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/model/CbsClientConfiguration.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/model/CbsClientConfiguration.java
index e3c7d2ea..2fb07501 100644
--- a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/model/CbsClientConfiguration.java
+++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/model/CbsClientConfiguration.java
@@ -22,6 +22,17 @@ package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model;
 
 import org.immutables.value.Value;
 import org.jetbrains.annotations.Nullable;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.exceptions.CbsClientConfigurationException;
+import org.onap.dcaegen2.services.sdk.security.ssl.ImmutableTrustStoreKeys;
+import org.onap.dcaegen2.services.sdk.security.ssl.Passwords;
+import org.onap.dcaegen2.services.sdk.security.ssl.SecurityKeysStore;
+import org.onap.dcaegen2.services.sdk.security.ssl.TrustStoreKeys;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Optional;
 
 /**
  * Immutable object which helps with construction of cloudRequestObject for specified Client. For usage take a look in
@@ -34,6 +45,16 @@ import org.jetbrains.annotations.Nullable;
  */
 @Value.Immutable(prehash = true)
 public interface CbsClientConfiguration {
+    Logger LOGGER = LoggerFactory.getLogger(CbsClientConfiguration.class);
+
+    String TRUST_JKS = "trust.jks";
+    String TRUST_PASS = "trust.pass";
+    Integer PORT_FOR_CBS_OVER_TLS = 10443;
+
+    /**
+     * Name of environment variable containing path to the cacert.pem file.
+     */
+    String DCAE_CA_CERT_PATH = "DCAE_CA_CERTPATH";
 
     /**
      * Name of environment variable containing Config Binding Service network hostname.
@@ -50,6 +71,7 @@ public interface CbsClientConfiguration {
      */
     String ENV_APP_NAME = "HOSTNAME";
 
+
     /**
      * Name of environment variable containing Consul host name.
      *
@@ -80,18 +102,25 @@ public interface CbsClientConfiguration {
     @Value.Parameter
     String appName();
 
+    @Value.Parameter
+    @Nullable
+    String protocol();
+
+    @Value.Default
+    default @Nullable TrustStoreKeys trustStoreKeys() {
+        return null;
+    }
+
     @Value.Default
     @Deprecated
     default String consulHost() {
         return "consul-server";
     }
-
     @Value.Default
     @Deprecated
     default Integer consulPort() {
         return 8500;
     }
-
     @Value.Default
     @Deprecated
     default String cbsName() {
@@ -102,14 +131,68 @@ public interface CbsClientConfiguration {
      * Creates CbsClientConfiguration from system environment variables.
      *
      * @return an instance of CbsClientConfiguration
-     * @throws NullPointerException when at least one of required parameters is missing
+     * @throws CbsClientConfigurationException when at least one of required parameters is missing
      */
     static CbsClientConfiguration fromEnvironment() {
-        return ImmutableCbsClientConfiguration.builder()
-                .consulHost(System.getenv(ENV_CONSUL_HOST))
-                .hostname(System.getenv(ENV_CBS_HOSTNAME))
-                .port(Integer.valueOf(System.getenv(ENV_CBS_PORT)))
-                .appName(System.getenv(ENV_APP_NAME))
+        String pathToCaCert = System.getenv(DCAE_CA_CERT_PATH);
+
+        ImmutableCbsClientConfiguration.Builder configBuilder = ImmutableCbsClientConfiguration.builder()
+                .hostname(getEnv(ENV_CBS_HOSTNAME))
+                .appName(getEnv(ENV_APP_NAME));
+        return Optional.ofNullable(pathToCaCert).filter(certPath -> !"".equals(certPath))
+                .map(certPath -> createSslHttpConfig(configBuilder, certPath))
+                .orElse(createPlainHttpConfig(configBuilder));
+    }
+
+    static CbsClientConfiguration createPlainHttpConfig(ImmutableCbsClientConfiguration.Builder configBuilder) {
+        LOGGER.info("CBS client will use plain http protocol.");
+        return configBuilder
+                .protocol("http")
+                .port(Integer.valueOf(getEnv(ENV_CBS_PORT)))
+                .build();
+    }
+
+    static CbsClientConfiguration createSslHttpConfig(ImmutableCbsClientConfiguration.Builder configBuilder,
+                                                      String pathToCaCert) {
+        LOGGER.info("CBS client will use http over TLS.");
+        return configBuilder
+                .trustStoreKeys(crateSecurityKeysFromEnvironment(createPathToJksFile(pathToCaCert)))
+                .port(PORT_FOR_CBS_OVER_TLS)
+                .protocol("https")
                 .build();
     }
+
+    static TrustStoreKeys crateSecurityKeysFromEnvironment(String pathToCerts) {
+        LOGGER.info("Path to cert files: {}", pathToCerts + "/");
+        validateIfFilesExist(pathToCerts);
+        return ImmutableTrustStoreKeys.builder()
+                .trustStore(SecurityKeysStore.fromPath(Paths.get(pathToCerts + "/" + TRUST_JKS)))
+                .trustStorePassword(Passwords.fromPath(Paths.get(pathToCerts + "/" + TRUST_PASS)))
+                .build();
+    }
+
+    static String createPathToJksFile(String pathToCaCertPemFile) {
+        return pathToCaCertPemFile.substring(0, pathToCaCertPemFile.lastIndexOf("/"));
+    }
+
+    static String getEnv(String envName) {
+        String envValue = System.getenv(envName);
+        validateEnv(envName, envValue);
+        return envValue;
+    }
+
+    static void validateEnv(String envName, String envValue) {
+        if (envValue == null || "".equals(envValue)) {
+            throw new CbsClientConfigurationException("Cannot read " + envName + " from environment.");
+        }
+    }
+
+    static void validateIfFilesExist(String pathToFile) {
+        boolean areFilesExist = Files.exists(Paths.get(pathToFile + "/" + TRUST_JKS)) &&
+                Files.exists(Paths.get(pathToFile + "/" + TRUST_PASS));
+
+        if (!areFilesExist) {
+            throw new CbsClientConfigurationException("Required files do not exist in " + pathToFile + " directory.");
+        }
+    }
 }
diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientConfigurationTest.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientConfigurationTest.java
index e00fd6bd..d0df0b6c 100644
--- a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientConfigurationTest.java
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClientConfigurationTest.java
@@ -21,18 +21,132 @@
 package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api;
 
 
-import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
-
+import org.junit.Rule;
+import org.junit.contrib.java.lang.system.EnvironmentVariables;
 import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.exceptions.CbsClientConfigurationException;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
+import org.onap.dcaegen2.services.sdk.security.ssl.Passwords;
+
+import java.net.URISyntaxException;
+import java.nio.file.Paths;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 
 /**
  * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
  * @since February 2019
  */
 class CbsClientConfigurationTest {
+
+    @Rule
+    public final EnvironmentVariables envs = new EnvironmentVariables();
+
+    @Test
+    void fromEnvironment_shouldReturnConfigurationForConnectionWithoutTls_when_DCAE_CA_CERTPATH_isEmpty() {
+        // given
+        envs.set("DCAE_CA_CERTPATH", "");
+        envs.set("CONFIG_BINDING_SERVICE", "config-binding-service");
+        envs.set("CONFIG_BINDING_SERVICE_SERVICE_PORT", "10000");
+        envs.set("HOSTNAME", "dcae-prh");
+        envs.set("CONSUL_HOST", "consul-server.onap");
+
+        // when
+        CbsClientConfiguration configuration = CbsClientConfiguration.fromEnvironment();
+
+        // then
+        assertThat(configuration.trustStoreKeys()).isEqualTo(null);
+        assertThat(configuration.protocol()).isEqualTo("http");
+    }
+
+    @Test
+    void fromEnvironment_shouldReturnConfigurationForConnectionOverTls_when_DCAE_CA_CERTPATH_isSet() throws URISyntaxException {
+        // given
+        envs.set("DCAE_CA_CERTPATH", preparePathToCertFile());
+        envs.set("CONFIG_BINDING_SERVICE", "config-binding-service");
+        envs.set("CONFIG_BINDING_SERVICE_PORT_10443_TCP_PORT", "10443");
+        envs.set("HOSTNAME", "dcae-prh");
+        envs.set("CONSUL_HOST", "consul-server.onap");
+
+        // when
+        CbsClientConfiguration configuration = CbsClientConfiguration.fromEnvironment();
+
+        // then
+        assertThat(configuration.trustStoreKeys()).isNotNull();
+        assertThat(configuration.protocol()).isEqualTo("https");
+    }
+
+    @Test
+    void fromEnvironment_shouldReturn_CbsClientConfigurationException_When_DCAE_CA_CERTPATH_is_Null() {
+        // given
+        envs.set("DCAE_CA_CERTPATH", null);
+        envs.set("CONFIG_BINDING_SERVICE_SERVICE_PORT", "9090");
+        envs.set("CONFIG_BINDING_SERVICE", "config-binding-service");
+        envs.set("CONFIG_BINDING_SERVICE_PORT_10443_TCP_PORT", "10443");
+        envs.set("HOSTNAME", "dcae-prh");
+        envs.set("CONSUL_HOST", "consul-server.onap");
+
+        // when
+        CbsClientConfiguration configuration = CbsClientConfiguration.fromEnvironment();
+
+        // then
+        assertThat(configuration.trustStoreKeys()).isNull();
+        assertThat(configuration.protocol()).isEqualTo("http");
+    }
+
+    @Test
+    void fromEnvironment_shouldReturn_CbsClientConfigurationException_WhenAllEnvVariablesAreMissing() {
+        assertThatExceptionOfType(CbsClientConfigurationException.class)
+                .isThrownBy(CbsClientConfiguration::fromEnvironment);
+    }
+
+    @Test
+    void fromEnvironment_shouldReturn_CbsClientConfigurationException_When_DCAE_CA_CERTPATH_isWrong() {
+        // given
+        envs.set("DCAE_CA_CERTPATH", "/home/cacert.pem");
+        envs.set("HOSTNAME", "dcae-prh");
+        envs.set("CONFIG_BINDING_SERVICE", "config-binding-service");
+        envs.set("CONFIG_BINDING_SERVICE_PORT_10443_TCP_PORT", "10443");
+        envs.set("CONSUL_HOST", "consul-server.onap");
+
+        // then
+        assertThatExceptionOfType(CbsClientConfigurationException.class)
+                .isThrownBy(CbsClientConfiguration::fromEnvironment)
+                .withMessageContaining("Required files do not exist in /home directory");
+    }
+
     @Test
-    void fromEnvironmentShouldFailWhenEnvVariablesAreMissing() {
-        assertThatExceptionOfType(NullPointerException.class).isThrownBy(CbsClientConfiguration::fromEnvironment);
+    void fromEnvironment_shouldReturn_CbsClientConfigurationException_When_HOSTNAME_isMissing() throws URISyntaxException {
+        // given
+        envs.set("HOSTNAME", "");
+        envs.set("DCAE_CA_CERTPATH", preparePathToCertFile());
+        envs.set("CONFIG_BINDING_SERVICE", "config-binding-service");
+        envs.set("CONFIG_BINDING_SERVICE_PORT_10443_TCP_PORT", "10443");
+        envs.set("CONSUL_HOST", "consul-server.onap");
+
+        // then
+        assertThatExceptionOfType(CbsClientConfigurationException.class)
+                .isThrownBy(CbsClientConfiguration::fromEnvironment)
+                .withMessageContaining("Cannot read HOSTNAME from environment.");
+    }
+
+    @Test
+    void fromEnvironment_shouldReturn_CbsClientConfigurationException_When_CONFIG_BINDING_SERVICE_SERVICE_PORT_isEmpty() {
+        // given
+        envs.set("CONFIG_BINDING_SERVICE_SERVICE_PORT", "");
+        envs.set("DCAE_CA_CERTPATH", "");
+        envs.set("HOSTNAME", "dcae-prh");
+        envs.set("CONFIG_BINDING_SERVICE", "config-binding-service");
+        envs.set("CONSUL_HOST", "consul-server.onap");
+
+        // then
+        assertThatExceptionOfType(CbsClientConfigurationException.class)
+                .isThrownBy(CbsClientConfiguration::fromEnvironment)
+                .withMessageContaining("Cannot read CONFIG_BINDING_SERVICE_SERVICE_PORT from environment.");
+    }
+
+    private String preparePathToCertFile() throws URISyntaxException {
+        return Paths.get(Passwords.class.getResource("/test-certs/cacert.pem").toURI()) + "";
     }
 }
\ No newline at end of file
diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImplIT.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImplIT.java
index 43b2a7bb..5804c165 100644
--- a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImplIT.java
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImplIT.java
@@ -20,17 +20,8 @@
 
 package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.onap.dcaegen2.services.sdk.model.streams.StreamType.KAFKA;
-import static org.onap.dcaegen2.services.sdk.model.streams.StreamType.MESSAGE_ROUTER;
-import static org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer.sendResource;
-import static org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamPredicates.streamOfType;
-
 import com.google.gson.JsonObject;
 import io.vavr.collection.Stream;
-
-import java.time.Duration;
-
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
@@ -47,14 +38,22 @@ import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.exceptions.St
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.DataStreams;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParser;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParsers;
-import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsRequest;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsRequest;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.ImmutableCbsClientConfiguration;
 import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
+import java.time.Duration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.onap.dcaegen2.services.sdk.model.streams.StreamType.KAFKA;
+import static org.onap.dcaegen2.services.sdk.model.streams.StreamType.MESSAGE_ROUTER;
+import static org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer.sendResource;
+import static org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamPredicates.streamOfType;
+
 /**
  * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
  * @since February 2019
@@ -77,6 +76,7 @@ class CbsClientImplIT {
                         .get("/sampleKey/dcae-component", (req, resp) -> sendResource(resp, SAMPLE_KEY))
         );
         sampleConfiguration = ImmutableCbsClientConfiguration.builder()
+                .protocol("http")
                 .appName("dcae-component")
                 .hostname(server.host())
                 .port(server.port())
diff --git a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImplTest.java b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImplTest.java
index 78b79f9d..40cf7100 100644
--- a/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImplTest.java
+++ b/rest-services/cbs-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImplTest.java
@@ -20,14 +20,7 @@
 
 package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.BDDMockito.given;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-
 import com.google.gson.JsonObject;
-import java.net.InetSocketAddress;
 import org.junit.jupiter.api.Test;
 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpMethod;
 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpRequest;
@@ -40,6 +33,14 @@ import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsRequests;
 import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
 import reactor.core.publisher.Mono;
 
+import java.net.InetSocketAddress;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
 /**
  * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
  * @since February 2019
@@ -52,7 +53,7 @@ class CbsClientImplTest {
         // given
         InetSocketAddress cbsAddress = InetSocketAddress.createUnresolved("cbshost", 6969);
         String serviceName = "dcaegen2-ves-collector";
-        final CbsClient cut = new CbsClientImpl(httpClient, serviceName, cbsAddress);
+        final CbsClient cut = new CbsClientImpl(httpClient, serviceName, cbsAddress, "http");
         final HttpResponse httpResponse = ImmutableHttpResponse.builder()
                 .url("http://xxx")
                 .statusCode(200)
diff --git a/rest-services/cbs-client/src/test/resources/test-certs/cacert.pem b/rest-services/cbs-client/src/test/resources/test-certs/cacert.pem
new file mode 100644
index 00000000..897c8ae4
--- /dev/null
+++ b/rest-services/cbs-client/src/test/resources/test-certs/cacert.pem
@@ -0,0 +1,31 @@
+-----BEGIN CERTIFICATE-----
+MIIFPjCCAyagAwIBAgIJAJ6u7cCnzrWdMA0GCSqGSIb3DQEBCwUAMCwxDjAMBgNV
+BAsMBU9TQUFGMQ0wCwYDVQQKDARPTkFQMQswCQYDVQQGEwJVUzAeFw0xODA0MDUx
+NDE1MjhaFw0zODAzMzExNDE1MjhaMCwxDjAMBgNVBAsMBU9TQUFGMQ0wCwYDVQQK
+DARPTkFQMQswCQYDVQQGEwJVUzCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoC
+ggIBAMA5pkgRs7NhGG4ew5JouhyYakgYUyFaG121+/h8qbSdt0hVQv56+EA41Yq7
+XGie7RYDQK9NmAFF3gruE+6X7wvJiChp+Cyd7sFMnb65uWhxEdxWTM2BJFrgfzUn
+H8ZCxgaCo3XH4PzlKRy2LQQJEJECwl/RZmRCXijMt5e9h8XoZY/fKkKcZZUsWNCM
+pTo266wjvA9MXLmdgReRj0+vrCjrNqy+htwJDztoiHWiYPqT6o8EvGcgjNqjlZx7
+NUNf8MfLDByqKF6+wRbHv1GKjn3/Vijd45Fv8riyRYROiFanvbV6jIfBkv8PZbXg
+2VDWsYsgp8NAvMxK+iV8cO+Ck3lBI2GOPZbCEqpPVTYbLUz6sczAlCXwQoPzDIZY
+wYa3eR/gYLY1gP2iEVHORag3bLPap9ZX5E8DZkzTNTjovvLk8KaCmfcaUMJsBtDd
+ApcUitz10cnRyZc1sX3gE1f3DpzQM6t9C5sOVyRhDcSrKqqwb9m0Ss04XAS9FsqM
+P3UWYQyqDXSxlUAYaX892u8mV1hxnt2gjb22RloXMM6TovM3sSrJS0wH+l1nznd6
+aFXftS/G4ZVIVZ/LfT1is4StoyPWZCwwwly1z8qJQ/zhip5NgZTxQw4mi7ww35DY
+PdAQOCoajfSvFjqslQ/cPRi/MRCu079heVb5fQnnzVtnpFQRAgMBAAGjYzBhMB0G
+A1UdDgQWBBRTVTPyS+vQUbHBeJrBKDF77+rtSTAfBgNVHSMEGDAWgBRTVTPyS+vQ
+UbHBeJrBKDF77+rtSTAPBgNVHRMBAf8EBTADAQH/MA4GA1UdDwEB/wQEAwIBhjAN
+BgkqhkiG9w0BAQsFAAOCAgEAPx/IaK94n02wPxpnYTy+LVLIxwdq/kawNd6IbiMz
+L87zmNMDmHcGbfoRCj8OkhuggX9Lx1/CkhpXimuYsZOFQi5blr/u+v4mIbsgbmi9
+7j+cUHDP0zLycvSvxKHty51LwmaX9a4wkJl5zBU4O1sd/H9tWcEmwJ39ltKoBKBx
+c94Zc3iMm5ytRWGj+0rKzLDAXEWpoZ5bE5PLJauA6UDCxDLfs3FwhbS7uDggxYvf
+jySF5FCNET94oJ+m8s7VeHvoa8iPGKvXrIqdd7XDHnqJJlVKr7m9S0fMbyEB8ci2
+RtOXDt93ifY1uhoEtEykn4dqBSp8ezvNMnwoXdYPDvTd9uCAFeWFLVreBAWxd25h
+PsBTkZA5hpa/rA+mKv6Af4VBViYr8cz4dZCsFChuioVebe9ighrfjB//qKepFjPF
+CyjzKN1u0JKm/2x/ORqxkTONG8p3uDwoIOyimUcTtTMv42bfYD88RKakqSFXE9G+
+Z0LlaKABqfjK49o/tsAp+c5LoNlYllKhnetO3QAdraHwdmC36BhoghzR1jpX751A
+cZn2VH3Q4XKyp01cJNCJIrua+A+bx6zh3RyW6zIIkbRCbET+UD+4mr8WIcSE3mtR
+ZVlnhUDO4z9//WKMVzwS9Rh8/kuszrGFI1KQozXCHLrce3YP6RYZfOed79LXaRwX
+dYY=
+-----END CERTIFICATE-----
\ No newline at end of file
diff --git a/rest-services/cbs-client/src/test/resources/test-certs/cert.jks b/rest-services/cbs-client/src/test/resources/test-certs/cert.jks
new file mode 100644
index 00000000..e74ce64f
--- /dev/null
+++ b/rest-services/cbs-client/src/test/resources/test-certs/cert.jks
diff --git a/rest-services/cbs-client/src/test/resources/test-certs/jks.pass b/rest-services/cbs-client/src/test/resources/test-certs/jks.pass
new file mode 100644
index 00000000..39823872
--- /dev/null
+++ b/rest-services/cbs-client/src/test/resources/test-certs/jks.pass
@@ -0,0 +1 @@
+mYHC98!qX}7h?W}jRv}MIXTJ
\ No newline at end of file
diff --git a/rest-services/cbs-client/src/test/resources/test-certs/trust.jks b/rest-services/cbs-client/src/test/resources/test-certs/trust.jks
new file mode 100644
index 00000000..10103cfb
--- /dev/null
+++ b/rest-services/cbs-client/src/test/resources/test-certs/trust.jks
diff --git a/rest-services/cbs-client/src/test/resources/test-certs/trust.pass b/rest-services/cbs-client/src/test/resources/test-certs/trust.pass
new file mode 100644
index 00000000..168e64bd
--- /dev/null
+++ b/rest-services/cbs-client/src/test/resources/test-certs/trust.pass
@@ -0,0 +1 @@
+*TQH?Lnszprs4LmlAj38yds(
\ No newline at end of file
diff --git a/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClientFactory.java b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClientFactory.java
index 1453adb9..9b23f1d9 100644
--- a/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClientFactory.java
+++ b/rest-services/http-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/adapters/http/RxHttpClientFactory.java
@@ -24,6 +24,7 @@ import io.netty.handler.ssl.SslContext;
 import org.jetbrains.annotations.NotNull;
 import org.onap.dcaegen2.services.sdk.security.ssl.SecurityKeys;
 import org.onap.dcaegen2.services.sdk.security.ssl.SslFactory;
+import org.onap.dcaegen2.services.sdk.security.ssl.TrustStoreKeys;
 import reactor.netty.http.client.HttpClient;
 
 /**
@@ -47,6 +48,11 @@ public final class RxHttpClientFactory {
         return create(context);
     }
 
+    public static RxHttpClient create(TrustStoreKeys trustStoreKeys) {
+        final SslContext context = SSL_FACTORY.createSecureClientContext(trustStoreKeys);
+        return create(context);
+    }
+
     public static RxHttpClient createInsecure() {
         final SslContext context = SSL_FACTORY.createInsecureClientContext();
         return create(context);
diff --git a/security/ssl/src/main/java/org/onap/dcaegen2/services/sdk/security/ssl/SslFactory.java b/security/ssl/src/main/java/org/onap/dcaegen2/services/sdk/security/ssl/SslFactory.java
index 963484a1..bdc55542 100644
--- a/security/ssl/src/main/java/org/onap/dcaegen2/services/sdk/security/ssl/SslFactory.java
+++ b/security/ssl/src/main/java/org/onap/dcaegen2/services/sdk/security/ssl/SslFactory.java
@@ -24,6 +24,12 @@ import io.netty.handler.ssl.ClientAuth;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
 import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
+import org.onap.dcaegen2.services.sdk.security.ssl.exceptions.ReadingSecurityKeysStoreException;
+import org.onap.dcaegen2.services.sdk.security.ssl.exceptions.SecurityConfigurationException;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLException;
+import javax.net.ssl.TrustManagerFactory;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.StandardOpenOption;
@@ -32,11 +38,6 @@ import java.security.KeyStore;
 import java.security.KeyStoreException;
 import java.security.NoSuchAlgorithmException;
 import java.security.cert.CertificateException;
-import javax.net.ssl.KeyManagerFactory;
-import javax.net.ssl.SSLException;
-import javax.net.ssl.TrustManagerFactory;
-import org.onap.dcaegen2.services.sdk.security.ssl.exceptions.ReadingSecurityKeysStoreException;
-import org.onap.dcaegen2.services.sdk.security.ssl.exceptions.SecurityConfigurationException;
 
 /**
  * @since 1.1.1
@@ -63,6 +64,22 @@ public class SslFactory {
     }
 
     /**
+     * Creates Netty SSL <em>client</em> context using provided TrustStore keys.
+     *
+     * @param keys - TrustStore keys to be used
+     * @return configured SSL context
+     */
+    public SslContext createSecureClientContext(final TrustStoreKeys keys) {
+        try {
+            return SslContextBuilder.forClient()
+                    .trustManager(trustManagerFactory(keys))
+                    .build();
+        } catch (SSLException e) {
+            throw new SecurityConfigurationException(EXCEPTION_MESSAGE, e);
+        }
+    }
+
+    /**
      * Creates Netty SSL <em>server</em> context using provided security keys. Will require client authentication.
      *
      * @param keys - security keys to be used
@@ -111,6 +128,10 @@ public class SslFactory {
         return trustManagerFactory(keys.trustStore(), keys.trustStorePassword());
     }
 
+    private TrustManagerFactory trustManagerFactory(TrustStoreKeys keys) {
+        return trustManagerFactory(keys.trustStore(), keys.trustStorePassword());
+    }
+
     private KeyManagerFactory keyManagerFactory(SecurityKeys keys) {
         return keyManagerFactory(keys.keyStore(), keys.keyStorePassword());
     }
diff --git a/security/ssl/src/main/java/org/onap/dcaegen2/services/sdk/security/ssl/TrustStoreKeys.java b/security/ssl/src/main/java/org/onap/dcaegen2/services/sdk/security/ssl/TrustStoreKeys.java
new file mode 100644
index 00000000..99b38e3b
--- /dev/null
+++ b/security/ssl/src/main/java/org/onap/dcaegen2/services/sdk/security/ssl/TrustStoreKeys.java
@@ -0,0 +1,31 @@
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 Nokia. 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.dcaegen2.services.sdk.security.ssl;
+
+import org.immutables.value.Value;
+
+
+@Value.Immutable
+public interface TrustStoreKeys {
+    SecurityKeysStore trustStore();
+
+    Password trustStorePassword();
+}
\ No newline at end of file
diff --git a/security/ssl/src/test/java/org/onap/dcaegen2/services/sdk/security/ssl/SslFactoryIT.java b/security/ssl/src/test/java/org/onap/dcaegen2/services/sdk/security/ssl/SslFactoryIT.java
index 966aa5cb..0bd57a40 100644
--- a/security/ssl/src/test/java/org/onap/dcaegen2/services/sdk/security/ssl/SslFactoryIT.java
+++ b/security/ssl/src/test/java/org/onap/dcaegen2/services/sdk/security/ssl/SslFactoryIT.java
@@ -19,18 +19,18 @@
  */
 package org.onap.dcaegen2.services.sdk.security.ssl;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.onap.dcaegen2.services.sdk.security.ssl.Passwords.fromResource;
-
 import io.netty.handler.ssl.SslContext;
-import java.net.URISyntaxException;
-import java.nio.file.Paths;
-import org.assertj.core.api.Assertions;
 import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.Test;
 import org.onap.dcaegen2.services.sdk.security.ssl.exceptions.ReadingSecurityKeysStoreException;
 
+import java.net.URISyntaxException;
+import java.nio.file.Paths;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.onap.dcaegen2.services.sdk.security.ssl.Passwords.fromResource;
+
 /**
  * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
  * @since April 2019
diff --git a/services/hv-ves-client/producer/ct/src/test/java/org/onap/dcaegen2/services/sdk/services/hvves/client/producer/ct/SystemUnderTestWrapper.java b/services/hv-ves-client/producer/ct/src/test/java/org/onap/dcaegen2/services/sdk/services/hvves/client/producer/ct/SystemUnderTestWrapper.java
index 45511d7f..361a92a0 100644
--- a/services/hv-ves-client/producer/ct/src/test/java/org/onap/dcaegen2/services/sdk/services/hvves/client/producer/ct/SystemUnderTestWrapper.java
+++ b/services/hv-ves-client/producer/ct/src/test/java/org/onap/dcaegen2/services/sdk/services/hvves/client/producer/ct/SystemUnderTestWrapper.java
@@ -23,13 +23,6 @@ import io.netty.buffer.ByteBuf;
 import io.netty.handler.ssl.SslContext;
 import io.vavr.collection.HashSet;
 import io.vavr.control.Try;
-
-import java.net.InetSocketAddress;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.time.Duration;
-import java.util.Optional;
-
 import org.onap.dcaegen2.services.sdk.security.ssl.*;
 import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.HvVesProducer;
 import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.HvVesProducerFactory;
@@ -40,6 +33,12 @@ import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.options
 import org.onap.ves.VesEventOuterClass.VesEvent;
 import reactor.core.publisher.Flux;
 
+import java.net.InetSocketAddress;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.time.Duration;
+import java.util.Optional;
+
 /**
  * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
  */
@@ -119,5 +118,4 @@ public class SystemUnderTestWrapper {
     private Try<Path> resource(String resource) {
         return Try.of(() -> Paths.get(Passwords.class.getResource(resource).toURI()));
     }
-
 }