Support TLS for cproto connector

Closes gh-122

Author: evgeniycheban

builtin-adapter/CMakeLists.txt

@@ -1,4 +1,4 @@
-cmake_minimum_required(VERSION 3.0)
+cmake_minimum_required(VERSION 3.5)
 
 project(builtin-adapter)
 
@@ -17,3 +17,18 @@ add_library(${TARGET} SHARED BuiltinAdapter.cpp)
 include_directories(${JNI_INCLUDE_DIRS} ${REINDEXER_INCLUDE_DIRS})
 
 target_link_libraries(${TARGET} reindexer_server_library reindexer_server_resources ${REINDEXER_LIBRARIES})
+
+# workaround for MacOS to propagate correct openssl path to Reindexer.
+if(APPLE)
+  find_package(OpenSSL)
+  if(OpenSSL_FOUND)
+    if(NOT DEFINED OPENSSL_LIB_PATH)
+      get_filename_component(OPENSSL_LIB_PATH ${OPENSSL_CRYPTO_LIBRARY} DIRECTORY)
+    endif()
+    message("-- Found package: OpenSSL@${OPENSSL_VERSION} - lib path: ${OPENSSL_LIB_PATH}")
+    set_target_properties(${TARGET} PROPERTIES
+      BUILD_WITH_INSTALL_RPATH TRUE
+      INSTALL_RPATH ${OPENSSL_LIB_PATH}
+    )
+  endif()
+endif()

pom.xml

@@ -139,6 +139,10 @@
                     <!-- Activate the use of TCP to transmit events to the plugin. -->
                     <forkNode implementation="org.apache.maven.plugin.surefire.extensions.SurefireForkNodeFactory" />
                     <groups>${tests}</groups>
+                    <systemPropertyVariables>
+                        <javax.net.ssl.trustStore>${project.basedir}/src/test/resources/builtin-server.jks</javax.net.ssl.trustStore>
+                        <javax.net.ssl.trustStorePassword>password</javax.net.ssl.trustStorePassword>
+                    </systemPropertyVariables>
                 </configuration>
             </plugin>
 

src/main/java/ru/rt/restream/reindexer/ReindexerConfiguration.java

@@ -33,6 +33,8 @@
 import java.util.Objects;
 import java.util.function.Consumer;
 
+import javax.net.ssl.SSLSocketFactory;
+
 /**
  * Represents approach for bootstrapping Reindexer.
  */
@@ -54,6 +56,8 @@ public final class ReindexerConfiguration {
 
     private String serverConfigFile = "default-builtin-server-config.yml";
 
+    private SSLSocketFactory sslSocketFactory;
+
     private ReindexerConfiguration() {
 
     }
@@ -161,6 +165,17 @@ public ReindexerConfiguration serverConfigFile(String serverConfigFile) {
         return this;
     }
 
+    /**
+     * Configure an {@link SSLSocketFactory}.
+     *
+     * @param sslSocketFactory the {@link SSLSocketFactory} to use
+     * @return the {@link ReindexerConfiguration} for further customizations
+     */
+    public ReindexerConfiguration sslSocketFactory(SSLSocketFactory sslSocketFactory) {
+        this.sslSocketFactory = sslSocketFactory;
+        return this;
+    }
+
     /**
      * Build and return reindexer connector instance.
      *
@@ -186,18 +201,23 @@ public Reindexer getReindexer() {
 
     private Binding getBinding(String protocol, List<URI> uris) {
         switch (protocol) {
+            case "cprotos":
+                if (sslSocketFactory == null) {
+                    sslSocketFactory = (SSLSocketFactory) SSLSocketFactory.getDefault();
+                }
             case "cproto":
                 DataSourceConfiguration dataSourceConfig = DataSourceConfiguration.builder()
                         .urls(urls)
                         .allowUnlistedDataSource(allowUnlistedDataSource)
+                        .sslSocketFactory(sslSocketFactory)
                         .build();
                 return new Cproto(dataSourceFactory, dataSourceConfig, connectionPoolSize, requestTimeout);
             case "builtin":
                 return new Builtin(uris.get(0), requestTimeout);
             case "builtinserver":
                 return new BuiltinServer(uris.get(0), serverConfigFile, serverStartupTimeout, requestTimeout);
             default:
-                throw new UnimplementedException("Protocol: '" + protocol + "' is not suppored");
+                throw new UnimplementedException("Protocol: '" + protocol + "' is not supported");
         }
     }
 

src/main/java/ru/rt/restream/reindexer/binding/cproto/DataSourceConfiguration.java

@@ -22,6 +22,8 @@
 import java.util.List;
 import java.util.Objects;
 
+import javax.net.ssl.SSLSocketFactory;
+
 /**
  * A {@link DataSource} configuration.
  */
@@ -37,6 +39,11 @@ public class DataSourceConfiguration {
      */
     private final boolean allowUnlistedDataSource;
 
+    /**
+     * An {@link SSLSocketFactory} to connect to Reindexer using cprotos (SSL/TLS) protocol.
+     */
+    private final SSLSocketFactory sslSocketFactory;
+
     /**
      * An index of the current active data source.
      */
@@ -46,6 +53,7 @@ private DataSourceConfiguration(Builder builder) {
         urls = builder.urls;
         allowUnlistedDataSource = builder.allowUnlistedDataSource;
         active = builder.active;
+		sslSocketFactory = builder.sslSocketFactory;
     }
 
     public static Builder builder() {
@@ -69,6 +77,15 @@ public boolean isAllowUnlistedDataSource() {
         return allowUnlistedDataSource;
     }
 
+    /**
+     * Returns an {@link SSLSocketFactory} to connect to Reindexer using cprotos (SSL/TLS) protocol.
+     *
+     * @return the {@link SSLSocketFactory} to use
+     */
+    public SSLSocketFactory getSslSocketFactory() {
+        return sslSocketFactory;
+    }
+
     /**
      * Returns the index of the current active data source.
      *
@@ -102,6 +119,11 @@ public static class Builder {
          */
         private boolean allowUnlistedDataSource = true;
 
+        /**
+         * An {@link SSLSocketFactory} to connect to Reindexer using cprotos (SSL/TLS) protocol.
+         */
+        private SSLSocketFactory sslSocketFactory;
+
         /**
          * An index of the current active data source.
          */
@@ -157,6 +179,17 @@ public Builder allowUnlistedDataSource(boolean allowUnlistedDataSource) {
             return this;
         }
 
+        /**
+         * Configure an {@link SSLSocketFactory} to connect to Reindexer using cprotos (TLS) protocol.
+         *
+         * @param sslSocketFactory the {@link SSLSocketFactory} to use
+         * @return the {@link Builder} for further customizations
+         */
+        public Builder sslSocketFactory(SSLSocketFactory sslSocketFactory) {
+            this.sslSocketFactory = sslSocketFactory;
+            return this;
+        }
+
         /**
          * Build and return a {@link DataSource} configuration.
          *

src/main/java/ru/rt/restream/reindexer/binding/cproto/DataSourceFactoryStrategy.java

@@ -43,7 +43,7 @@ public enum DataSourceFactoryStrategy implements DataSourceFactory {
         public DataSource getDataSource(DataSourceConfiguration configuration) {
             List<String> urls = configuration.getUrls();
             configuration.setActive((configuration.getActive() + 1) % urls.size());
-            return new PhysicalDataSource(urls.get(configuration.getActive()));
+            return new PhysicalDataSource(urls.get(configuration.getActive()), configuration.getSslSocketFactory());
         }
     },
 
@@ -55,7 +55,7 @@ public DataSource getDataSource(DataSourceConfiguration configuration) {
         public DataSource getDataSource(DataSourceConfiguration configuration) {
             List<String> urls = configuration.getUrls();
             configuration.setActive(ThreadLocalRandom.current().nextInt(urls.size()));
-            return new PhysicalDataSource(urls.get(configuration.getActive()));
+            return new PhysicalDataSource(urls.get(configuration.getActive()), configuration.getSslSocketFactory());
         }
     },
 

src/main/java/ru/rt/restream/reindexer/binding/cproto/PhysicalConnection.java

@@ -46,6 +46,9 @@
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
+import javax.net.ssl.SSLSocket;
+import javax.net.ssl.SSLSocketFactory;
+
 import static ru.rt.restream.reindexer.binding.Consts.APP_PROPERTY_NAME;
 import static ru.rt.restream.reindexer.binding.Consts.BINDING_CAPABILITY_COMPLEX_RANK;
 import static ru.rt.restream.reindexer.binding.Consts.BINDING_CAPABILITY_NAMESPACE_INCARNATIONS;
@@ -106,9 +109,18 @@ public class PhysicalConnection implements Connection {
     private final ScheduledFuture<?> writeTaskFuture;
 
     public PhysicalConnection(String host, int port, String user, String password, String database,
+                              SSLSocketFactory sslSocketFactory,
                               Duration requestTimeout, ScheduledExecutorService scheduler) {
         try {
-            clientSocket = new Socket(host, port);
+            if (sslSocketFactory != null) {
+                LOGGER.debug("rx: using SSL/TLS connection to {}:{}", host, port);
+                SSLSocket sslSocket = (SSLSocket) sslSocketFactory.createSocket(host, port);
+                // Fail fast if SSL/TLS handshake fails.
+                sslSocket.startHandshake();
+                clientSocket = sslSocket;
+            } else {
+                clientSocket = new Socket(host, port);
+            }
             output = new DataOutputStream(clientSocket.getOutputStream());
             input = new DataInputStream(clientSocket.getInputStream());
             timeout = requestTimeout;

src/main/java/ru/rt/restream/reindexer/binding/cproto/PhysicalDataSource.java

@@ -20,6 +20,8 @@
 import java.time.Duration;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 
+import javax.net.ssl.SSLSocketFactory;
+
 /**
  * A {@link DataSource} that creates a {@link PhysicalConnection}.
  */
@@ -35,12 +37,27 @@ public class PhysicalDataSource implements DataSource {
 
     private final String database;
 
+    private final SSLSocketFactory sslSocketFactory;
+
     /**
      * Creates an instance.
      *
      * @param url the URL to use
+     * @deprecated Use {@link #PhysicalDataSource(String, SSLSocketFactory)}
+     * to connect to Reindexer using cprotos (SSL/TLS) protocol.
      */
+    @Deprecated
     public PhysicalDataSource(String url) {
+        this(url, null);
+    }
+
+    /**
+     * Creates an instance.
+     *
+     * @param url              the URL to use
+     * @param sslSocketFactory the {@link SSLSocketFactory} socket factory to use
+     */
+    public PhysicalDataSource(String url, SSLSocketFactory sslSocketFactory) {
         URI uri = URI.create(url);
         host = uri.getHost();
         port = uri.getPort();
@@ -58,11 +75,12 @@ public PhysicalDataSource(String url) {
             password = "";
         }
         database = uri.getPath().substring(1);
+        this.sslSocketFactory = sslSocketFactory;
     }
 
     @Override
     public Connection getConnection(Duration timeout, ScheduledThreadPoolExecutor scheduler) {
-        return new PhysicalConnection(host, port, user, password, database, timeout, scheduler);
+        return new PhysicalConnection(host, port, user, password, database, sslSocketFactory, timeout, scheduler);
     }
 
     @Override

src/test/java/ru/rt/restream/reindexer/connector/CprotosReindexerTest.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2020 Restream
+ *
+ * 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.
+ */
+package ru.rt.restream.reindexer.connector;
+
+import org.junit.jupiter.api.BeforeAll;
+import ru.rt.restream.category.CprotoTest;
+import ru.rt.restream.reindexer.db.DbLocator;
+import ru.rt.restream.reindexer.db.DbLocator.Type;
+
+/**
+ * Tests for Cprotos (SSL/TLS) protocol implementation.
+ */
+@CprotoTest
+public class CprotosReindexerTest extends ReindexerTest {
+
+    @BeforeAll
+    @Override
+    protected void initDb() {
+        db = DbLocator.getDb(Type.CPROTOS);
+    }
+
+}

src/test/java/ru/rt/restream/reindexer/db/DbBaseTest.java

@@ -41,7 +41,7 @@ public abstract class DbBaseTest {
      * Initializes the Reindexer instance before all tests.
      */
     @BeforeAll
-    void initDb() {
+    protected void initDb() {
         if (this.getClass().isAnnotationPresent(CprotoTest.class)) {
             db = DbLocator.getDb(CPROTO);
         } else if (this.getClass().isAnnotationPresent(BuiltinTest.class)) {