Allow to set SSLParameter in TLS configuration

Fixes #11

Author: acogoluegnes

src/docs/asciidoc/api.adoc

@@ -201,6 +201,10 @@ The server certificate chain and the client private key are the typical
 elements that need to be configured.
 |The JDK trust manager and no client private key.
 
+|`tls#sslParameters`
+|Set the `SSLParameters` for the `SSLEngine`. The provided parameters will be merged into the parameters returned by `SSLEngine#getSSLParameters()`. Can be used to set SNI information with `SSLParameters#setServerNames(List)`.
+|`null`
+
 |`tls#trustEverything`
 |Helper to configure a `SslContext` that trusts all server certificates
 and does not use a client private key. **Only for development**.

src/main/java/com/rabbitmq/stream/EnvironmentBuilder.java

@@ -25,6 +25,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ScheduledExecutorService;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLParameters;
 
 /**
  * API to configure and create an {@link Environment}.
@@ -332,6 +334,21 @@ interface TlsConfiguration {
      */
     TlsConfiguration sslContext(SslContext sslContext);
 
+    /**
+     * Set {@link SSLParameters} for the {@link javax.net.ssl.SSLEngine}.
+     *
+     * <p>Provided {@link SSLParameters} will be merged into the {@link SSLParameters} returned by
+     * {@link SSLEngine#getSSLParameters()}, that is non-null property values from the provided
+     * instance will override those in the original instance.
+     *
+     * <p>This is typically use to provide SNI information with {@link
+     * SSLParameters#setServerNames(List)}.
+     *
+     * @param sslParameters
+     * @return the TLS configuration helper
+     */
+    TlsConfiguration sslParameters(SSLParameters sslParameters);
+
     /**
      * Convenience method to set a {@link SslContext} that trusts all servers.
      *

src/main/java/com/rabbitmq/stream/impl/Client.java

@@ -285,12 +285,16 @@ public void write(
             if (parameters.sslContext != null) {
               SslHandler sslHandler =
                   parameters.sslContext.newHandler(ch.alloc(), parameters.host, parameters.port);
+              SSLEngine sslEngine = sslHandler.engine();
+              SSLParameters sslParameters = sslEngine.getSSLParameters();
               if (parameters.tlsHostnameVerification) {
-                SSLEngine sslEngine = sslHandler.engine();
-                SSLParameters sslParameters = sslEngine.getSSLParameters();
                 sslParameters.setEndpointIdentificationAlgorithm("HTTPS");
-                sslEngine.setSSLParameters(sslParameters);
               }
+              if (parameters.sslParameters != null) {
+                Utils.mergeSslParameters(sslParameters, parameters.sslParameters);
+              }
+              sslEngine.setSSLParameters(sslParameters);
+
               ch.pipeline().addFirst("ssl", sslHandler);
             }
             channelCustomizer.customize(ch);
@@ -1919,6 +1923,7 @@ public static class ClientParameters {
     private ChunkChecksum chunkChecksum = JdkChunkChecksum.CRC32_SINGLETON;
     private MetricsCollector metricsCollector = NoOpMetricsCollector.SINGLETON;
     private SslContext sslContext;
+    private SSLParameters sslParameters;
     private boolean tlsHostnameVerification = true;
     private ByteBufAllocator byteBufAllocator;
     private Duration rpcTimeout;
@@ -2065,6 +2070,11 @@ public ClientParameters sslContext(SslContext sslContext) {
       return this;
     }
 
+    public ClientParameters sslParameters(SSLParameters sslParameters) {
+      this.sslParameters = sslParameters;
+      return this;
+    }
+
     public ClientParameters tlsHostnameVerification(boolean tlsHostnameVerification) {
       this.tlsHostnameVerification = tlsHostnameVerification;
       return this;

src/main/java/com/rabbitmq/stream/impl/StreamEnvironment.java

@@ -145,6 +145,7 @@ class StreamEnvironment implements Environment {
         clientParametersPrototype.sslContext(sslContext);
         clientParametersPrototype.tlsHostnameVerification(
             tlsConfiguration.hostnameVerificationEnabled());
+        clientParametersPrototype.sslParameters(tlsConfiguration.sslParameters());
 
       } catch (SSLException e) {
         throw new StreamException("Error while creating Netty SSL context", e);

src/main/java/com/rabbitmq/stream/impl/StreamEnvironmentBuilder.java

@@ -38,6 +38,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.stream.Collectors;
 import javax.net.ssl.SSLException;
+import javax.net.ssl.SSLParameters;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -300,6 +301,7 @@ static final class DefaultTlsConfiguration implements TlsConfiguration {
     private boolean enabled = false;
     private boolean hostnameVerification = true;
     private SslContext sslContext;
+    private SSLParameters sslParameters;
 
     private DefaultTlsConfiguration(EnvironmentBuilder environmentBuilder) {
       this.environmentBuilder = environmentBuilder;
@@ -323,6 +325,12 @@ public TlsConfiguration sslContext(SslContext sslContext) {
       return this;
     }
 
+    @Override
+    public TlsConfiguration sslParameters(SSLParameters sslParameters) {
+      this.sslParameters = sslParameters;
+      return this;
+    }
+
     @Override
     public TlsConfiguration trustEverything() {
       LOGGER.warn(
@@ -360,5 +368,9 @@ public boolean hostnameVerificationEnabled() {
     public SslContext sslContext() {
       return sslContext;
     }
+
+    public SSLParameters sslParameters() {
+      return sslParameters;
+    }
   }
 }

src/main/java/com/rabbitmq/stream/impl/Utils.java

@@ -26,6 +26,8 @@
 import java.util.function.Consumer;
 import java.util.function.LongConsumer;
 import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import javax.net.ssl.SSLParameters;
 import javax.net.ssl.X509TrustManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -213,4 +215,92 @@ public X509Certificate[] getAcceptedIssuers() {
       return new X509Certificate[0];
     }
   }
+
+  static void mergeSslParameters(SSLParameters original, SSLParameters provided) {
+    if (notEmptyArray(provided.getCipherSuites())) {
+      LOGGER.debug(
+          "Setting SSLParameters cipherSuites from {} to {}",
+          arrayToString(original.getCipherSuites()),
+          arrayToString(provided.getCipherSuites()));
+      original.setCipherSuites(provided.getCipherSuites());
+    }
+    if (notEmptyArray(provided.getProtocols())) {
+      LOGGER.debug(
+          "Setting SSLParameters protocols from {} to {}",
+          arrayToString(original.getProtocols()),
+          arrayToString(provided.getProtocols()));
+      original.setProtocols(provided.getProtocols());
+    }
+    if (original.getWantClientAuth() != provided.getWantClientAuth()) {
+      LOGGER.debug(
+          "Setting SSLParameters wantClientAuth from {} to {}",
+          original.getWantClientAuth(),
+          provided.getWantClientAuth());
+      original.setWantClientAuth(provided.getWantClientAuth());
+    }
+    if (original.getNeedClientAuth() != provided.getNeedClientAuth()) {
+      LOGGER.debug(
+          "Setting SSLParameters needClientAuth from {} to {}",
+          original.getNeedClientAuth(),
+          provided.getNeedClientAuth());
+      original.setNeedClientAuth(provided.getNeedClientAuth());
+    }
+    if (notNullOrBlank(provided.getEndpointIdentificationAlgorithm())) {
+      LOGGER.debug(
+          "Setting SSLParameters endpointIdentificationAlgorithm from {} to {}",
+          original.getEndpointIdentificationAlgorithm(),
+          provided.getEndpointIdentificationAlgorithm());
+      original.setEndpointIdentificationAlgorithm(provided.getEndpointIdentificationAlgorithm());
+    }
+    if (provided.getAlgorithmConstraints() != null) {
+      LOGGER.debug(
+          "Setting SSLParameters algorithmConstraints from {} to {}",
+          original.getAlgorithmConstraints(),
+          provided.getAlgorithmConstraints());
+      original.setAlgorithmConstraints(provided.getAlgorithmConstraints());
+    }
+    if (provided.getServerNames() != null) {
+      LOGGER.debug(
+          "Setting SSLParameters serverNames from {} to {}",
+          original.getServerNames(),
+          provided.getServerNames());
+      original.setServerNames(provided.getServerNames());
+    }
+    if (provided.getSNIMatchers() != null) {
+      LOGGER.debug(
+          "Setting SSLParameters SNIMatchers from {} to {}",
+          original.getSNIMatchers(),
+          provided.getSNIMatchers());
+      original.setSNIMatchers(provided.getSNIMatchers());
+    }
+    if (original.getUseCipherSuitesOrder() != provided.getUseCipherSuitesOrder()) {
+      LOGGER.debug(
+          "Setting SSLParameters useCipherSuitesOrder from {} to {}",
+          original.getUseCipherSuitesOrder(),
+          provided.getUseCipherSuitesOrder());
+      original.setUseCipherSuitesOrder(provided.getUseCipherSuitesOrder());
+    }
+  }
+
+  private static boolean notNullOrBlank(String str) {
+    return str != null && !str.trim().isEmpty();
+  }
+
+  private static String arrayToString(Object[] array) {
+    if (emptyArray(array)) {
+      return "";
+    } else {
+      return Arrays.stream(array)
+          .map(o -> o == null ? "null" : o.toString())
+          .collect(Collectors.joining());
+    }
+  }
+
+  private static boolean emptyArray(Object[] array) {
+    return array == null || array.length == 0;
+  }
+
+  private static boolean notEmptyArray(Object[] array) {
+    return !emptyArray(array);
+  }
 }

src/test/java/com/rabbitmq/stream/impl/StreamEnvironmentTest.java

@@ -15,6 +15,7 @@
 
 import static com.rabbitmq.stream.impl.TestUtils.latchAssert;
 import static com.rabbitmq.stream.impl.TestUtils.localhost;
+import static com.rabbitmq.stream.impl.TestUtils.localhostTls;
 import static com.rabbitmq.stream.impl.TestUtils.streamName;
 import static com.rabbitmq.stream.impl.TestUtils.waitAtMost;
 import static java.util.concurrent.TimeUnit.SECONDS;
@@ -33,6 +34,7 @@
 import com.rabbitmq.stream.StreamException;
 import com.rabbitmq.stream.impl.Client.StreamMetadata;
 import com.rabbitmq.stream.impl.MonitoringTestUtils.EnvironmentInfo;
+import com.rabbitmq.stream.impl.TestUtils.DisabledIfTlsNotEnabled;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import java.net.ConnectException;
@@ -48,6 +50,8 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import javax.net.ssl.SNIHostName;
+import javax.net.ssl.SSLParameters;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
@@ -78,7 +82,8 @@ static void afterAll() throws Exception {
   @BeforeEach
   void init() {
     environmentBuilder = Environment.builder();
-    environmentBuilder.addressResolver(add -> localhost());
+    environmentBuilder.addressResolver(
+        add -> add.port() == Client.DEFAULT_PORT ? localhost() : localhostTls());
     environmentBuilder.eventLoopGroup(eventLoopGroup);
   }
 
@@ -129,6 +134,21 @@ void environmentCreationShouldSucceedWithUrlContainingAllCorrectInformation() {
     environmentBuilder.uri("rabbitmq-stream://guest:guest@localhost:5552/%2f").build().close();
   }
 
+  @DisabledIfTlsNotEnabled
+  @Test
+  void environmentCreationShouldSucceedWhenUsingTls() {
+    SSLParameters sslParameters = new SSLParameters();
+    sslParameters.setServerNames(Collections.singletonList(new SNIHostName("localhost")));
+    environmentBuilder
+        .uri("rabbitmq-stream+tls://guest:guest@localhost:5551/%2f")
+        .tls()
+        .trustEverything()
+        .sslParameters(sslParameters)
+        .environmentBuilder()
+        .build()
+        .close();
+  }
+
   @Test
   void producersAndConsumersShouldBeClosedWhenEnvironmentIsClosed() {
     Environment environment = environmentBuilder.build();

src/test/java/com/rabbitmq/stream/impl/TestUtils.java

@@ -114,6 +114,10 @@ static Address localhost() {
     return new Address("localhost", Client.DEFAULT_PORT);
   }
 
+  static Address localhostTls() {
+    return new Address("localhost", Client.DEFAULT_TLS_PORT);
+  }
+
   static byte b(int value) {
     return (byte) value;
   }

src/test/java/com/rabbitmq/stream/impl/TlsTest.java

@@ -43,8 +43,10 @@
 import java.util.Collections;
 import java.util.concurrent.CountDownLatch;
 import java.util.stream.IntStream;
+import javax.net.ssl.SNIHostName;
 import javax.net.ssl.SSLException;
 import javax.net.ssl.SSLHandshakeException;
+import javax.net.ssl.SSLParameters;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
@@ -172,6 +174,13 @@ void unverifiedConnection() {
     cf.get(new ClientParameters().sslContext(alwaysTrustSslContext()));
   }
 
+  @Test
+  void unverifiedConnectionWithSslParameters() {
+    SSLParameters sslParameters = new SSLParameters();
+    sslParameters.setServerNames(Collections.singletonList(new SNIHostName("localhost")));
+    cf.get(new ClientParameters().sslContext(alwaysTrustSslContext()).sslParameters(sslParameters));
+  }
+
   @Test
   void verifiedConnectionWithCorrectServerCertificate() throws Exception {
     SslContext context = SslContextBuilder.forClient().trustManager(caCertificate()).build();