KAFKA-4525; Kafka should not require SSL truststore password

Author: Grant Henke <ghenke@cloudera.com>

Reviewers: Rajini Sivaram <rajinisivaram@googlemail.com>, Ismael Juma <ismael@juma.me.uk>

Closes #2246 from granthenke/truststore-password

clients/src/main/java/org/apache/kafka/common/config/SslConfigs.java

@@ -70,7 +70,7 @@ public class SslConfigs {
     public static final String SSL_TRUSTSTORE_LOCATION_DOC = "The location of the trust store file. ";
 
     public static final String SSL_TRUSTSTORE_PASSWORD_CONFIG = "ssl.truststore.password";
-    public static final String SSL_TRUSTSTORE_PASSWORD_DOC = "The password for the trust store file. ";
+    public static final String SSL_TRUSTSTORE_PASSWORD_DOC = "The password for the trust store file. If a password is not set access to the truststore is still available, but integrity checking is disabled.";
 
     public static final String SSL_KEYMANAGER_ALGORITHM_CONFIG = "ssl.keymanager.algorithm";
     public static final String SSL_KEYMANAGER_ALGORITHM_DOC = "The algorithm used by key manager factory for SSL connections. "

clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java

@@ -191,9 +191,7 @@ public class SslFactory implements Configurable {
     private void createTruststore(String type, String path, Password password) {
         if (path == null && password != null) {
             throw new KafkaException("SSL trust store is not specified, but trust store password is specified.");
-        } else if (path != null && password == null) {
-            throw new KafkaException("SSL trust store is specified, but trust store password is not specified.");
-        } else if (path != null && password != null) {
+        } else if (path != null) {
             this.truststore = new SecurityStore(type, path, password);
         }
     }
@@ -214,7 +212,9 @@ public class SslFactory implements Configurable {
             try {
                 KeyStore ks = KeyStore.getInstance(type);
                 in = new FileInputStream(path);
-                ks.load(in, password.value().toCharArray());
+                // If a password is not set access to the truststore is still available, but integrity checking is disabled.
+                char[] passwordChars = password != null ? password.value().toCharArray() : null;
+                ks.load(in, passwordChars);
                 return ks;
             } finally {
                 if (in != null) in.close();

clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java

@@ -344,6 +344,24 @@ public class SslTransportLayerTest {
             // Expected exception
         }
     }
+
+    /**
+     * Tests that client connections can be created to a server
+     * if null truststore password is used
+     */
+    @Test
+    public void testNullTruststorePassword() throws Exception {
+        String node = "0";
+        sslClientConfigs.remove(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG);
+        sslServerConfigs.remove(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG);
+
+        server = createEchoServer(SecurityProtocol.SSL);
+        createSelector(sslClientConfigs);
+        InetSocketAddress addr = new InetSocketAddress("localhost", server.port());
+        selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE);
+
+        NetworkTestUtils.checkClientConnection(selector, node, 100, 10);
+    }
     
     /**
      * Tests that client connections cannot be created to a server

clients/src/test/java/org/apache/kafka/common/security/ssl/SslFactoryTest.java

@@ -17,6 +17,7 @@ import java.util.Map;
 
 import javax.net.ssl.SSLEngine;
 
+import org.apache.kafka.common.config.SslConfigs;
 import org.apache.kafka.test.TestSslUtils;
 import org.apache.kafka.common.network.Mode;
 import org.junit.Test;
@@ -25,6 +26,7 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 /**
  * A set of tests for the selector over ssl. These use a test harness that runs a simple socket server that echos back responses.
@@ -45,6 +47,20 @@ public class SslFactoryTest {
         assertEquals(false, engine.getUseClientMode());
     }
 
+    @Test
+    public void testSslFactoryWithoutPasswordConfiguration() throws Exception {
+        File trustStoreFile = File.createTempFile("truststore", ".jks");
+        Map<String, Object> serverSslConfig = TestSslUtils.createSslConfig(false, true, Mode.SERVER, trustStoreFile, "server");
+        // unset the password
+        serverSslConfig.remove(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG);
+        SslFactory sslFactory = new SslFactory(Mode.SERVER);
+        try {
+            sslFactory.configure(serverSslConfig);
+        } catch (Exception e) {
+            fail("An exception was thrown when configuring the truststore without a password: " + e);
+        }
+    }
+
     @Test
     public void testClientMode() throws Exception {
         File trustStoreFile = File.createTempFile("truststore", ".jks");

docs/security.html

@@ -143,6 +143,8 @@
             ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks
             ssl.truststore.password=test1234</pre>
 
+            Note: ssl.truststore.password is technically optional but highly recommended. If a password is not set access to the truststore is still available, but integrity checking is disabled.
+
             Optional settings that are worth considering:
             <ol>
                 <li>ssl.client.auth=none ("required" => client authentication is required, "requested" => client authentication is requested and client without certs can still connect. The usage of "requested" is discouraged as it provides a false sense of security and misconfigured clients will still connect successfully.)</li>
@@ -192,11 +194,14 @@
             ssl.truststore.location=/var/private/ssl/kafka.client.truststore.jks
             ssl.truststore.password=test1234</pre>
 
+            Note: ssl.truststore.password is technically optional but highly recommended. If a password is not set access to the truststore is still available, but integrity checking is disabled.
+
             If client authentication is required, then a keystore must be created like in step 1 and the following must also be configured:
             <pre>
             ssl.keystore.location=/var/private/ssl/kafka.client.keystore.jks
             ssl.keystore.password=test1234
             ssl.key.password=test1234</pre>
+
             Other configuration settings that may also be needed depending on our requirements and the broker configuration:
                 <ol>
                     <li>ssl.provider (Optional). The name of the security provider used for SSL connections. Default value is the default security provider of the JVM.</li>