Add Configuration to set tlsClientAuth (apache#1297)

* Add Configuration to set tlsClientAuth * Fixed ProxyPublishConsumeTlsTest * Handled Matteos PR review comments * Negative Tests * Changed the Client Auth to ENUM * Addressed Matteo's PR review Comments * Removed unused imports * Added client Cert to HTTPS * Split the test case * Replace tlsReqTrustedClientCertOnConnect with tlsRequireTrustedClientCertOnConnect
vikingcodes · Mar 27, 2018 · ffd6f21 · ffd6f21
1 parent 5bd13a0
commit ffd6f21
Show file tree

Hide file tree

Showing 20 changed files with 238 additions and 59 deletions.
diff --git a/conf/broker.conf b/conf/broker.conf
@@ -207,6 +207,9 @@ tlsTrustCertsFilePath=
 # Accept untrusted TLS certificate from client
 tlsAllowInsecureConnection=false
 
+# Specify whether Client certificates are required for TLS
+# Reject the Connection if the Client Certificate is not trusted.
+tlsRequireTrustedClientCertOnConnect=false
 ### --- Authentication --- ###
 
 # Enable authentication

diff --git a/conf/discovery.conf b/conf/discovery.conf
@@ -73,3 +73,7 @@ tlsCertificateFilePath=
 
 # Path for the TLS private key file
 tlsKeyFilePath=
+
+# Specify whether Client certificates are required for TLS
+# Reject the Connection if the Client Certificate is not trusted.
+tlsRequireTrustedClientCertOnConnect=false
diff --git a/conf/proxy.conf b/conf/proxy.conf
@@ -85,3 +85,7 @@ tlsKeyFilePath=
 
 # Validates hostname when proxy creates tls connection with broker
 tlsHostnameVerificationEnabled=false
+
+# Specify whether Client certificates are required for TLS
+# Reject the Connection if the Client Certificate is not trusted.
+tlsRequireTrustedClientCertOnConnect=false
diff --git a/conf/websocket.conf b/conf/websocket.conf
@@ -99,3 +99,7 @@ tlsKeyFilePath=
 
 # Path for the trusted TLS certificate file
 tlsTrustCertsFilePath=
+
+# Specify whether Client certificates are required for TLS
+# Reject the Connection if the Client Certificate is not trusted.
+tlsRequireTrustedClientCertOnConnect=false
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
@@ -202,6 +202,9 @@ public class ServiceConfiguration implements PulsarConfiguration {
     // Specify the tls cipher the broker will use to negotiate during TLS Handshake.
     // Example:- [TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256]
     private Set<String> tlsCiphers = Sets.newTreeSet();
+    // Specify whether Client certificates are required for TLS
+    // Reject the Connection if the Client Certificate is not trusted.
+    private boolean tlsRequireTrustedClientCertOnConnect = false;
 
     /***** --- Authentication --- ****/
     // Enable authentication
@@ -1497,7 +1500,14 @@ public Set<String> getTlsCiphers() {
     public void setTlsCiphers(Set<String> tlsCiphers) {
         this.tlsCiphers = tlsCiphers;
     }
+
+    public boolean getTlsRequireTrustedClientCertOnConnect() {
+        return tlsRequireTrustedClientCertOnConnect;
+    }
 
+    public void setTlsRequireTrustedClientCertOnConnect(boolean tlsRequireTrustedClientCertOnConnect) {
+        this.tlsRequireTrustedClientCertOnConnect = tlsRequireTrustedClientCertOnConnect;
+    }
     /**** --- Function ---- ****/
 
     public void setFunctionsWorkerEnabled(boolean enabled) {

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
@@ -53,7 +53,8 @@ protected void initChannel(SocketChannel ch) throws Exception {
             SslContext sslCtx = SecurityUtility.createNettySslContextForServer(
                     serviceConfig.isTlsAllowInsecureConnection(), serviceConfig.getTlsTrustCertsFilePath(),
                     serviceConfig.getTlsCertificateFilePath(), serviceConfig.getTlsKeyFilePath(),
-                    serviceConfig.getTlsCiphers(), serviceConfig.getTlsProtocols());
+                    serviceConfig.getTlsCiphers(), serviceConfig.getTlsProtocols(),
+                    serviceConfig.getTlsRequireTrustedClientCertOnConnect());
             ch.pipeline().addLast(TLS_HANDLER, sslCtx.newHandler(ch.alloc()));
         }
 

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
@@ -90,24 +90,20 @@ public WebService(PulsarService pulsar) throws PulsarServerException {
         connectors.add(connector);
 
         if (pulsar.getConfiguration().isTlsEnabled()) {
-            SslContextFactory sslCtxFactory = new SslContextFactory();
-
             try {
-                sslCtxFactory.setSslContext(
-                        SecurityUtility.createSslContext(
-                            pulsar.getConfiguration().isTlsAllowInsecureConnection(),
-                            pulsar.getConfiguration().getTlsTrustCertsFilePath(),
-                            pulsar.getConfiguration().getTlsCertificateFilePath(),
-                            pulsar.getConfiguration().getTlsKeyFilePath()));
+                SslContextFactory sslCtxFactory = SecurityUtility.createSslContextFactory(
+                        pulsar.getConfiguration().isTlsAllowInsecureConnection(),
+                        pulsar.getConfiguration().getTlsTrustCertsFilePath(),
+                        pulsar.getConfiguration().getTlsCertificateFilePath(),
+                        pulsar.getConfiguration().getTlsKeyFilePath(),
+                        pulsar.getConfiguration().getTlsRequireTrustedClientCertOnConnect());
+                ServerConnector tlsConnector = new PulsarServerConnector(server, 1, 1, sslCtxFactory);
+                tlsConnector.setPort(pulsar.getConfiguration().getWebServicePortTls());
+                tlsConnector.setHost(pulsar.getBindAddress());
+                connectors.add(tlsConnector);
             } catch (GeneralSecurityException e) {
                 throw new PulsarServerException(e);
             }
-
-            sslCtxFactory.setWantClientAuth(true);
-            ServerConnector tlsConnector = new PulsarServerConnector(server, 1, 1, sslCtxFactory);
-            tlsConnector.setPort(pulsar.getConfiguration().getWebServicePortTls());
-            tlsConnector.setHost(pulsar.getBindAddress());
-            connectors.add(tlsConnector);
         }
 
         // Limit number of concurrent HTTP connections to avoid getting out of file descriptors

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerBase.java
@@ -21,11 +21,14 @@
 import static org.mockito.Mockito.spy;
 
 import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
 
 import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.impl.auth.AuthenticationTls;
 import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.PropertyAdmin;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testng.annotations.AfterMethod;
@@ -34,16 +37,19 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 
+import io.netty.handler.ssl.ClientAuth;
+
 public class TlsProducerConsumerBase extends ProducerConsumerBase {
     protected final String TLS_TRUST_CERT_FILE_PATH = "./src/test/resources/authentication/tls/cacert.pem";
+    protected final String TLS_CLIENT_CERT_FILE_PATH = "./src/test/resources/authentication/tls/client-cert.pem";
+    protected final String TLS_CLIENT_KEY_FILE_PATH = "./src/test/resources/authentication/tls/client-key.pem";
     protected final String TLS_SERVER_CERT_FILE_PATH = "./src/test/resources/authentication/tls/broker-cert.pem";
     protected final String TLS_SERVER_KEY_FILE_PATH = "./src/test/resources/authentication/tls/broker-key.pem";
     private final String clusterName = "use";
 
     @BeforeMethod
     @Override
     protected void setup() throws Exception {
-
         // TLS configuration for Broker
         internalSetUpForBroker();
 
@@ -61,24 +67,42 @@ protected void internalSetUpForBroker() throws Exception {
         conf.setTlsEnabled(true);
         conf.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         conf.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
+        conf.setTlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH);
         conf.setClusterName(clusterName);
+        conf.setTlsRequireTrustedClientCertOnConnect(true);
+        Set<String> tlsProtocols = Sets.newConcurrentHashSet();
+        tlsProtocols.add("TLSv1.2");
+        conf.setTlsProtocols(tlsProtocols);
     }
 
-    protected void internalSetUpForClient() throws Exception {
-        String lookupUrl = new URI("pulsar+ssl://localhost:" + BROKER_PORT_TLS).toString();
-        pulsarClient = PulsarClient.builder().serviceUrl(lookupUrl).tlsTrustCertsFilePath(TLS_SERVER_CERT_FILE_PATH)
-                .enableTls(true).build();
+    protected void internalSetUpForClient(boolean addCertificates, String lookupUrl) throws Exception {
+        ClientConfiguration clientConf = new ClientConfiguration();
+        clientConf.setTlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH);
+        clientConf.setUseTls(true);
+        clientConf.setTlsAllowInsecureConnection(false);
+        if (addCertificates) {
+            Map<String, String> authParams = new HashMap<>();
+            authParams.put("tlsCertFile", TLS_CLIENT_CERT_FILE_PATH);
+            authParams.put("tlsKeyFile", TLS_CLIENT_KEY_FILE_PATH);
+            clientConf.setAuthentication(AuthenticationTls.class.getName(), authParams);
+        }
+        pulsarClient = PulsarClient.create(lookupUrl, clientConf);
     }
 
     protected void internalSetUpForNamespace() throws Exception {
         ClientConfiguration clientConf = new ClientConfiguration();
         clientConf.setTlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH);
         clientConf.setUseTls(true);
+        clientConf.setTlsAllowInsecureConnection(false);
+        Map<String, String> authParams = new HashMap<>();
+        authParams.put("tlsCertFile", TLS_CLIENT_CERT_FILE_PATH);
+        authParams.put("tlsKeyFile", TLS_CLIENT_KEY_FILE_PATH);
+        clientConf.setAuthentication(AuthenticationTls.class.getName(), authParams);
         admin = spy(new PulsarAdmin(brokerUrlTls, clientConf));
         admin.clusters().updateCluster(clusterName, new ClusterData(brokerUrl.toString(), brokerUrlTls.toString(),
                 "pulsar://localhost:" + BROKER_PORT, "pulsar+ssl://localhost:" + BROKER_PORT_TLS));
         admin.properties().createProperty("my-property",
                 new PropertyAdmin(Lists.newArrayList("appid1", "appid2"), Sets.newHashSet("use")));
         admin.namespaces().createNamespace("my-property/use/my-ns");
     }
-}
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerTest.java
@@ -42,7 +42,7 @@ public void testTlsLargeSizeMessage() throws Exception {
         final int MESSAGE_SIZE = 16 * 1024 + 1;
         log.info("-- message size --", MESSAGE_SIZE);
 
-        internalSetUpForClient();
+        internalSetUpForClient(true, "pulsar+ssl://localhost:" + BROKER_PORT_TLS);
         internalSetUpForNamespace();
 
         Consumer<byte[]> consumer = pulsarClient.newConsumer().topic("persistent://my-property/use/my-ns/my-topic1")
@@ -68,4 +68,68 @@ public void testTlsLargeSizeMessage() throws Exception {
         consumer.close();
         log.info("-- Exiting {} test --", methodName);
     }
+
+    @Test(timeOut = 30000)
+    public void testTlsClientAuthOverBinaryProtocol() throws Exception {
+        log.info("-- Starting {} test --", methodName);
+
+        final int MESSAGE_SIZE = 16 * 1024 + 1;
+        log.info("-- message size --", MESSAGE_SIZE);
+        internalSetUpForNamespace();
+
+        // Test 1 - Using TLS on binary protocol without sending certs - expect failure
+        internalSetUpForClient(false, "pulsar+ssl://localhost:" + BROKER_PORT_TLS);
+        try {
+            ConsumerConfiguration conf = new ConsumerConfiguration();
+            conf.setSubscriptionType(SubscriptionType.Exclusive);
+            Consumer consumer = pulsarClient.subscribe("persistent://my-property/use/my-ns/my-topic1",
+                    "my-subscriber-name", conf);
+            Assert.fail("Server should have failed the TLS handshake since client didn't .");
+        } catch (Exception ex) {
+            // OK
+        }
+
+        // Test 2 - Using TLS on binary protocol - sending certs
+        internalSetUpForClient(true, "pulsar+ssl://localhost:" + BROKER_PORT_TLS);
+        try {
+            ConsumerConfiguration conf = new ConsumerConfiguration();
+            conf.setSubscriptionType(SubscriptionType.Exclusive);
+            Consumer consumer = pulsarClient.subscribe("persistent://my-property/use/my-ns/my-topic1",
+                    "my-subscriber-name", conf);
+        } catch (Exception ex) {
+            Assert.fail("Should not fail since certs are sent.");
+        }
+    }
+
+    @Test(timeOut = 30000)
+    public void testTlsClientAuthOverHTTPProtocol() throws Exception {
+        log.info("-- Starting {} test --", methodName);
+
+        final int MESSAGE_SIZE = 16 * 1024 + 1;
+        log.info("-- message size --", MESSAGE_SIZE);
+        internalSetUpForNamespace();
+
+        // Test 1 - Using TLS on https without sending certs - expect failure
+        internalSetUpForClient(false, "https://localhost:" + BROKER_WEBSERVICE_PORT_TLS);
+        try {
+            ConsumerConfiguration conf = new ConsumerConfiguration();
+            conf.setSubscriptionType(SubscriptionType.Exclusive);
+            Consumer consumer = pulsarClient.subscribe("persistent://my-property/use/my-ns/my-topic1",
+                    "my-subscriber-name", conf);
+            Assert.fail("Server should have failed the TLS handshake since client didn't .");
+        } catch (Exception ex) {
+            // OK
+        }
+
+        // Test 2 - Using TLS on https - sending certs
+        internalSetUpForClient(true, "https://localhost:" + BROKER_WEBSERVICE_PORT_TLS);
+        try {
+            ConsumerConfiguration conf = new ConsumerConfiguration();
+            conf.setSubscriptionType(SubscriptionType.Exclusive);
+            Consumer consumer = pulsarClient.subscribe("persistent://my-property/use/my-ns/my-topic1",
+                    "my-subscriber-name", conf);
+        } catch (Exception ex) {
+            Assert.fail("Should not fail since certs are sent.");
+        }
+    }
 }
diff --git a/...ar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java b/...ar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java
@@ -24,12 +24,15 @@
 
 import java.net.URI;
 import java.security.GeneralSecurityException;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.bookkeeper.test.PortManager;
 import org.apache.pulsar.client.api.TlsProducerConsumerBase;
+import org.apache.pulsar.client.impl.auth.AuthenticationTls;
 import org.apache.pulsar.common.util.SecurityUtility;
 import org.apache.pulsar.websocket.WebSocketService;
 import org.apache.pulsar.websocket.service.ProxyServer;
@@ -71,6 +74,9 @@ public void setup() throws Exception {
         config.setBrokerClientTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH);
         config.setClusterName("use");
         config.setGlobalZookeeperServers("dummy-zk-servers");
+        config.setBrokerClientAuthenticationParameters("tlsCertFile:" + TLS_CLIENT_CERT_FILE_PATH + ",tlsKeyFile:" + TLS_CLIENT_KEY_FILE_PATH);
+        config.setBrokerClientAuthenticationPlugin(AuthenticationTls.class.getName());
+        String lookupUrl = new URI("pulsar://localhost:" + BROKER_PORT_TLS).toString();
         service = spy(new WebSocketService(config));
         doReturn(mockZooKeeperClientFactory).when(service).getZooKeeperClientFactory();
         proxyServer = new ProxyServer(config);

diff --git a/pulsar-common/pom.xml b/pulsar-common/pom.xml
@@ -86,5 +86,11 @@
       <groupId>io.netty</groupId>
       <artifactId>netty-tcnative-boringssl-static</artifactId>
     </dependency>
+
+	<dependency>
+		<groupId>org.eclipse.jetty</groupId>
+		<artifactId>jetty-server</artifactId>
+	</dependency>
+
   </dependencies>
 </project>
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/SecurityUtility.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/SecurityUtility.java
@@ -49,6 +49,8 @@
 import javax.net.ssl.TrustManager;
 import javax.net.ssl.TrustManagerFactory;
 
+import org.eclipse.jetty.util.ssl.SslContextFactory;
+
 import io.netty.handler.ssl.ClientAuth;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
@@ -93,7 +95,8 @@ public static SslContext createNettySslContextForClient(boolean allowInsecureCon
     }
 
     public static SslContext createNettySslContextForServer(boolean allowInsecureConnection, String trustCertsFilePath,
-            String certFilePath, String keyFilePath, Set<String> ciphers, Set<String> protocols)
+            String certFilePath, String keyFilePath, Set<String> ciphers, Set<String> protocols,
+            boolean requireTrustedClientCertOnConnect)
             throws GeneralSecurityException, SSLException, FileNotFoundException, IOException {
         X509Certificate[] certificates = loadCertificatesFromPemFile(certFilePath);
         PrivateKey privateKey = loadPrivateKeyFromPemFile(keyFilePath);
@@ -103,7 +106,7 @@ public static SslContext createNettySslContextForServer(boolean allowInsecureCon
         setupProtocols(builder, protocols);
         setupTrustCerts(builder, allowInsecureConnection, trustCertsFilePath);
         setupKeyManager(builder, privateKey, certificates);
-        setupClientAuthentication(builder);
+        setupClientAuthentication(builder, requireTrustedClientCertOnConnect);
         return builder.build();
     }
 
@@ -236,7 +239,27 @@ private static void setupProtocols(SslContextBuilder builder, Set<String> protoc
         }
     }
 
-    private static void setupClientAuthentication(SslContextBuilder builder) {
-        builder.clientAuth(ClientAuth.OPTIONAL);
+    private static void setupClientAuthentication(SslContextBuilder builder, boolean requireTrustedClientCertOnConnect) {
+        if (requireTrustedClientCertOnConnect) {
+            builder.clientAuth(ClientAuth.REQUIRE);
+        } else {
+            builder.clientAuth(ClientAuth.OPTIONAL);
+        }
+    }
+
+    public static SslContextFactory createSslContextFactory(boolean tlsAllowInsecureConnection,
+            String tlsTrustCertsFilePath, String tlsCertificateFilePath, String tlsKeyFilePath,
+            boolean tlsRequireTrustedClientCertOnConnect) throws GeneralSecurityException {
+        SslContextFactory sslCtxFactory = new SslContextFactory();
+        SSLContext sslCtx = createSslContext(tlsAllowInsecureConnection, tlsTrustCertsFilePath, tlsCertificateFilePath,
+                tlsKeyFilePath);
+        sslCtxFactory.setSslContext(sslCtx);
+        if (tlsRequireTrustedClientCertOnConnect) {
+            sslCtxFactory.setNeedClientAuth(true);
+        } else {
+            sslCtxFactory.setWantClientAuth(true);
+        }
+        sslCtxFactory.setTrustAll(true);
+        return sslCtxFactory;
     }
 }
diff --git a/...-service/src/main/java/org/apache/pulsar/discovery/service/ServiceChannelInitializer.java b/...-service/src/main/java/org/apache/pulsar/discovery/service/ServiceChannelInitializer.java
@@ -52,7 +52,8 @@ protected void initChannel(SocketChannel ch) throws Exception {
             SslContext sslCtx = SecurityUtility.createNettySslContextForServer(
                     serviceConfig.isTlsAllowInsecureConnection(), serviceConfig.getTlsTrustCertsFilePath(),
                     serviceConfig.getTlsCertificateFilePath(), serviceConfig.getTlsKeyFilePath(),
-                    serviceConfig.getTlsCiphers(), serviceConfig.getTlsProtocols());
+                    serviceConfig.getTlsCiphers(), serviceConfig.getTlsProtocols(),
+                    serviceConfig.getTlsRequireTrustedClientCertOnConnect());
             ch.pipeline().addLast(TLS_HANDLER, sslCtx.newHandler(ch.alloc()));
         }
         ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(PulsarDecoder.MaxFrameSize, 0, 4, 0, 4));