[feat][broker] Full-support set ssl provider, ciphers and protocols (a…

…pache#14569) Signed-off-by: Zixuan Liu <[email protected]> (cherry picked from commit 9b2ba05)
datastax · Apr 1, 2022 · 72aae15 · 72aae15
1 parent e5e4951
commit 72aae15
Show file tree

Hide file tree

Showing 22 changed files with 465 additions and 91 deletions.
diff --git a/conf/broker.conf b/conf/broker.conf
@@ -584,13 +584,15 @@ tlsCiphers=
 # authentication.
 tlsRequireTrustedClientCertOnConnect=false
 
+# Specify the TLS provider for the broker service:
+# When using TLS authentication with CACert, the valid value is either OPENSSL or JDK.
+# When using TLS authentication with KeyStore, available values can be SunJSSE, Conscrypt and etc.
+tlsProvider=
+
 ### --- KeyStore TLS config variables --- ###
 # Enable TLS with KeyStore type configuration in broker.
 tlsEnabledWithKeyStore=false
 
-# TLS Provider for KeyStore type
-tlsProvider=
-
 # TLS KeyStore type configuration in broker: JKS, PKCS12
 tlsKeyStoreType=JKS
 

diff --git a/conf/standalone.conf b/conf/standalone.conf
@@ -350,13 +350,15 @@ tlsCiphers=
 # authentication.
 tlsRequireTrustedClientCertOnConnect=false
 
+# Specify the TLS provider for the broker service:
+# When using TLS authentication with CACert, the valid value is either OPENSSL or JDK.
+# When using TLS authentication with KeyStore, available values can be SunJSSE, Conscrypt and etc.
+tlsProvider=
+
 ### --- KeyStore TLS config variables --- ###
 # Enable TLS with KeyStore type configuration in broker.
 tlsEnabledWithKeyStore=false
 
-# TLS Provider for KeyStore type
-tlsProvider=
-
 # TLS KeyStore type configuration in broker: JKS, PKCS12
 tlsKeyStoreType=JKS
 

diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -2196,7 +2196,9 @@ public class ServiceConfiguration implements PulsarConfiguration {
 
     @FieldContext(
             category = CATEGORY_KEYSTORE_TLS,
-            doc = "TLS Provider for KeyStore type"
+            doc = "TLS Provider for Specify the SSL provider for the broker service: \n"
+                    + "When using TLS authentication with CACert, the valid value is either OPENSSL or JDK.\n"
+                    + "When using TLS authentication with KeyStore, available values can be SunJSSE, Conscrypt and etc."
     )
     private String tlsProvider = null;
 

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java
@@ -28,6 +28,7 @@
 import io.netty.handler.flow.FlowControlHandler;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslHandler;
+import io.netty.handler.ssl.SslProvider;
 import java.net.SocketAddress;
 import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
@@ -88,10 +89,18 @@ public PulsarChannelInitializer(PulsarService pulsar, boolean enableTLS) throws
                         serviceConfig.getTlsProtocols(),
                         serviceConfig.getTlsCertRefreshCheckDurationSec());
             } else {
-                sslCtxRefresher = new NettyServerSslContextBuilder(serviceConfig.isTlsAllowInsecureConnection(),
-                        serviceConfig.getTlsTrustCertsFilePath(), serviceConfig.getTlsCertificateFilePath(),
+                SslProvider sslProvider = null;
+                if (serviceConfig.getTlsProvider() != null) {
+                    sslProvider = SslProvider.valueOf(serviceConfig.getTlsProvider());
+                }
+                sslCtxRefresher = new NettyServerSslContextBuilder(
+                        sslProvider,
+                        serviceConfig.isTlsAllowInsecureConnection(),
+                        serviceConfig.getTlsTrustCertsFilePath(),
+                        serviceConfig.getTlsCertificateFilePath(),
                         serviceConfig.getTlsKeyFilePath(),
-                        serviceConfig.getTlsCiphers(), serviceConfig.getTlsProtocols(),
+                        serviceConfig.getTlsCiphers(),
+                        serviceConfig.getTlsProtocols(),
                         serviceConfig.isTlsRequireTrustedClientCertOnConnect(),
                         serviceConfig.getTlsCertRefreshCheckDurationSec());
             }

diff --git a/...-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java b/...-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java
@@ -21,6 +21,7 @@
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.HttpResponse;
 import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslProvider;
 import io.netty.util.concurrent.DefaultThreadFactory;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -137,21 +138,32 @@ public boolean keepAlive(InetSocketAddress remoteAddress, Request ahcRequest,
                     JsseSslEngineFactory sslEngineFactory = new JsseSslEngineFactory(sslCtx);
                     confBuilder.setSslEngineFactory(sslEngineFactory);
                 } else {
+                    SslProvider sslProvider = null;
+                    if (conf.getSslProvider() != null) {
+                        sslProvider = SslProvider.valueOf(conf.getSslProvider());
+                    }
                     SslContext sslCtx = null;
                     if (authData.hasDataForTls()) {
                         sslCtx = authData.getTlsTrustStoreStream() == null
                                 ? SecurityUtility.createAutoRefreshSslContextForClient(
-                                        conf.isTlsAllowInsecureConnection() || !conf.isTlsHostnameVerificationEnable(),
-                                        conf.getTlsTrustCertsFilePath(), authData.getTlsCerificateFilePath(),
-                                        authData.getTlsPrivateKeyFilePath(), null, autoCertRefreshTimeSeconds, delayer)
+                                sslProvider,
+                                conf.isTlsAllowInsecureConnection() || !conf.isTlsHostnameVerificationEnable(),
+                                conf.getTlsTrustCertsFilePath(), authData.getTlsCerificateFilePath(),
+                                authData.getTlsPrivateKeyFilePath(), null, autoCertRefreshTimeSeconds, delayer)
                                 : SecurityUtility.createNettySslContextForClient(
-                                        conf.isTlsAllowInsecureConnection() || !conf.isTlsHostnameVerificationEnable(),
-                                        authData.getTlsTrustStoreStream(), authData.getTlsCertificates(),
-                                        authData.getTlsPrivateKey());
+                                sslProvider,
+                                conf.isTlsAllowInsecureConnection() || !conf.isTlsHostnameVerificationEnable(),
+                                authData.getTlsTrustStoreStream(), authData.getTlsCertificates(),
+                                authData.getTlsPrivateKey(),
+                                conf.getTlsCiphers(),
+                                conf.getTlsProtocols());
                     } else {
                         sslCtx = SecurityUtility.createNettySslContextForClient(
+                                sslProvider,
                                 conf.isTlsAllowInsecureConnection() || !conf.isTlsHostnameVerificationEnable(),
-                                conf.getTlsTrustCertsFilePath());
+                                conf.getTlsTrustCertsFilePath(),
+                                conf.getTlsCiphers(),
+                                conf.getTlsProtocols());
                     }
                     confBuilder.setSslContext(sslCtx);
                 }

diff --git a/pulsar-client/pom.xml b/pulsar-client/pom.xml
@@ -70,6 +70,14 @@
       <groupId>io.netty</groupId>
       <artifactId>netty-codec-http</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-handler-proxy</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-codec-socks</artifactId>
+    </dependency>
 
     <dependency>
       <groupId>io.netty</groupId>

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java
@@ -18,24 +18,24 @@
  */
 package org.apache.pulsar.client.impl;
 
+import io.netty.channel.EventLoopGroup;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslProvider;
 import java.io.Closeable;
 import java.io.IOException;
 import java.net.HttpURLConnection;
 import java.net.InetSocketAddress;
 import java.net.URI;
 import java.net.URL;
+import java.security.GeneralSecurityException;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
-
-import io.netty.channel.EventLoopGroup;
-import io.netty.handler.codec.http.HttpRequest;
-import io.netty.handler.codec.http.HttpResponse;
-import io.netty.handler.ssl.SslContext;
 import javax.net.ssl.SSLContext;
 import lombok.extern.slf4j.Slf4j;
-
 import org.apache.pulsar.PulsarVersion;
 import org.apache.pulsar.client.api.Authentication;
 import org.apache.pulsar.client.api.AuthenticationDataProvider;
@@ -111,25 +111,33 @@ public boolean keepAlive(InetSocketAddress remoteAddress, Request ahcRequest,
                     JsseSslEngineFactory sslEngineFactory = new JsseSslEngineFactory(sslCtx);
                     confBuilder.setSslEngineFactory(sslEngineFactory);
                 } else {
+                    SslProvider sslProvider = null;
+                    if (conf.getSslProvider() != null) {
+                        sslProvider = SslProvider.valueOf(conf.getSslProvider());
+                    }
                     SslContext sslCtx = null;
                     if (authData.hasDataForTls()) {
                         sslCtx = authData.getTlsTrustStoreStream() == null
-                                ? SecurityUtility.createNettySslContextForClient(conf.isTlsAllowInsecureConnection(),
-                                        conf.getTlsTrustCertsFilePath(), authData.getTlsCertificates(),
-                                        authData.getTlsPrivateKey())
-                                : SecurityUtility.createNettySslContextForClient(conf.isTlsAllowInsecureConnection(),
-                                        authData.getTlsTrustStoreStream(), authData.getTlsCertificates(),
-                                        authData.getTlsPrivateKey());
-                    }
-                    else {
+                                ? SecurityUtility.createNettySslContextForClient(sslProvider,
+                                conf.isTlsAllowInsecureConnection(),
+                                conf.getTlsTrustCertsFilePath(), authData.getTlsCertificates(),
+                                authData.getTlsPrivateKey(), conf.getTlsCiphers(), conf.getTlsProtocols())
+                                : SecurityUtility.createNettySslContextForClient(sslProvider,
+                                conf.isTlsAllowInsecureConnection(),
+                                authData.getTlsTrustStoreStream(), authData.getTlsCertificates(),
+                                authData.getTlsPrivateKey(), conf.getTlsCiphers(), conf.getTlsProtocols());
+                    } else {
                         sslCtx = SecurityUtility.createNettySslContextForClient(
+                                sslProvider,
                                 conf.isTlsAllowInsecureConnection(),
-                                conf.getTlsTrustCertsFilePath());
+                                conf.getTlsTrustCertsFilePath(), conf.getTlsCiphers(), conf.getTlsProtocols());
                     }
                     confBuilder.setSslContext(sslCtx);
                 }
 
                 confBuilder.setUseInsecureTrustManager(conf.isTlsAllowInsecureConnection());
+            } catch (GeneralSecurityException e) {
+                throw new PulsarClientException.InvalidConfigurationException(e);
             } catch (Exception e) {
                 throw new PulsarClientException.InvalidConfigurationException(e);
             }

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java
@@ -18,6 +18,14 @@
  */
 package org.apache.pulsar.client.impl;
 
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
+import io.netty.handler.proxy.Socks5ProxyHandler;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.handler.ssl.SslProvider;
 import java.net.InetSocketAddress;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
@@ -35,12 +43,6 @@
 import org.apache.pulsar.common.util.SecurityUtility;
 import org.apache.pulsar.common.util.keystoretls.NettySSLContextAutoRefreshBuilder;
 
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.socket.SocketChannel;
-import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
-import io.netty.handler.ssl.SslContext;
-import io.netty.handler.ssl.SslHandler;
 
 @Slf4j
 public class PulsarChannelInitializer extends ChannelInitializer<SocketChannel> {
@@ -85,19 +87,36 @@ public PulsarChannelInitializer(ClientConfigurationData conf, Supplier<ClientCnx
 
             sslContextSupplier = new ObjectCache<SslContext>(() -> {
                 try {
+                    SslProvider sslProvider = null;
+                    if (conf.getSslProvider() != null) {
+                        sslProvider = SslProvider.valueOf(conf.getSslProvider());
+                    }
+
                     // Set client certificate if available
                     AuthenticationDataProvider authData = conf.getAuthentication().getAuthData();
                     if (authData.hasDataForTls()) {
                         return authData.getTlsTrustStoreStream() == null
-                                ? SecurityUtility.createNettySslContextForClient(conf.isTlsAllowInsecureConnection(),
-                                        conf.getTlsTrustCertsFilePath(),
-                                        authData.getTlsCertificates(), authData.getTlsPrivateKey())
-                                : SecurityUtility.createNettySslContextForClient(conf.isTlsAllowInsecureConnection(),
-                                        authData.getTlsTrustStoreStream(),
-                                        authData.getTlsCertificates(), authData.getTlsPrivateKey());
+                                ? SecurityUtility.createNettySslContextForClient(
+                                sslProvider,
+                                conf.isTlsAllowInsecureConnection(),
+                                conf.getTlsTrustCertsFilePath(),
+                                authData.getTlsCertificates(),
+                                authData.getTlsPrivateKey(),
+                                conf.getTlsCiphers(),
+                                conf.getTlsProtocols())
+                                : SecurityUtility.createNettySslContextForClient(sslProvider,
+                                conf.isTlsAllowInsecureConnection(),
+                                authData.getTlsTrustStoreStream(),
+                                authData.getTlsCertificates(), authData.getTlsPrivateKey(),
+                                conf.getTlsCiphers(),
+                                conf.getTlsProtocols());
                     } else {
-                        return SecurityUtility.createNettySslContextForClient(conf.isTlsAllowInsecureConnection(),
-                                conf.getTlsTrustCertsFilePath());
+                        return SecurityUtility.createNettySslContextForClient(
+                                sslProvider,
+                                conf.isTlsAllowInsecureConnection(),
+                                conf.getTlsTrustCertsFilePath(),
+                                conf.getTlsCiphers(),
+                                conf.getTlsProtocols());
                     }
                 } catch (Exception e) {
                     throw new RuntimeException("Failed to create TLS context", e);

diff --git a/pulsar-common/pom.xml b/pulsar-common/pom.xml
@@ -188,6 +188,11 @@
           <artifactId>gson</artifactId>
       </dependency>
 
+    <dependency>
+      <groupId>org.awaitility</groupId>
+      <artifactId>awaitility</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

diff --git a/...ar-common/src/main/java/org/apache/pulsar/common/util/NettyClientSslContextRefresher.java b/...ar-common/src/main/java/org/apache/pulsar/common/util/NettyClientSslContextRefresher.java
@@ -19,10 +19,12 @@
 package org.apache.pulsar.common.util;
 
 import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslProvider;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.security.GeneralSecurityException;
 import java.security.cert.X509Certificate;
+import java.util.Set;
 import javax.net.ssl.SSLException;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.api.AuthenticationDataProvider;
@@ -33,15 +35,20 @@
 @Slf4j
 public class NettyClientSslContextRefresher extends SslContextAutoRefreshBuilder<SslContext> {
     private volatile SslContext sslNettyContext;
-    private boolean tlsAllowInsecureConnection;
+    private final boolean tlsAllowInsecureConnection;
     protected final FileModifiedTimeUpdater tlsTrustCertsFilePath;
     protected final FileModifiedTimeUpdater tlsCertsFilePath;
     protected final FileModifiedTimeUpdater tlsPrivateKeyFilePath;
-    private AuthenticationDataProvider authData;
+    private final AuthenticationDataProvider authData;
+    private final SslProvider sslProvider;
+    private final Set<String> ciphers;
+    private final Set<String> protocols;
 
-    public NettyClientSslContextRefresher(boolean allowInsecure,
+    public NettyClientSslContextRefresher(SslProvider sslProvider, boolean allowInsecure,
                                           String trustCertsFilePath,
                                           AuthenticationDataProvider authData,
+                                          Set<String> ciphers,
+                                          Set<String> protocols,
                                           long delayInSeconds)
             throws IOException, GeneralSecurityException {
         super(delayInSeconds);
@@ -52,22 +59,26 @@ public NettyClientSslContextRefresher(boolean allowInsecure,
                 authData != null ? authData.getTlsCerificateFilePath() : null);
         this.tlsPrivateKeyFilePath = new FileModifiedTimeUpdater(
                 authData != null ? authData.getTlsPrivateKeyFilePath() : null);
+        this.sslProvider = sslProvider;
+        this.ciphers = ciphers;
+        this.protocols = protocols;
     }
 
     @Override
     public synchronized SslContext update()
             throws SSLException, FileNotFoundException, GeneralSecurityException, IOException {
         if (authData != null && authData.hasDataForTls()) {
             this.sslNettyContext = authData.getTlsTrustStoreStream() == null
-                    ? SecurityUtility.createNettySslContextForClient(this.tlsAllowInsecureConnection,
-                            tlsTrustCertsFilePath.getFileName(), (X509Certificate[]) authData.getTlsCertificates(),
-                            authData.getTlsPrivateKey())
-                    : SecurityUtility.createNettySslContextForClient(this.tlsAllowInsecureConnection,
-                            authData.getTlsTrustStoreStream(), (X509Certificate[]) authData.getTlsCertificates(),
-                            authData.getTlsPrivateKey());
+                    ? SecurityUtility.createNettySslContextForClient(this.sslProvider, this.tlsAllowInsecureConnection,
+                    tlsTrustCertsFilePath.getFileName(), (X509Certificate[]) authData.getTlsCertificates(),
+                    authData.getTlsPrivateKey(), this.ciphers, this.protocols)
+                    : SecurityUtility.createNettySslContextForClient(this.sslProvider, this.tlsAllowInsecureConnection,
+                    authData.getTlsTrustStoreStream(), (X509Certificate[]) authData.getTlsCertificates(),
+                    authData.getTlsPrivateKey(), this.ciphers, this.protocols);
         } else {
-            this.sslNettyContext = SecurityUtility.createNettySslContextForClient(this.tlsAllowInsecureConnection,
-                    this.tlsTrustCertsFilePath.getFileName());
+            this.sslNettyContext =
+                    SecurityUtility.createNettySslContextForClient(this.sslProvider, this.tlsAllowInsecureConnection,
+                            this.tlsTrustCertsFilePath.getFileName(), this.ciphers, this.protocols);
         }
         return this.sslNettyContext;
     }