KAFKA-12703; Allow unencrypted private keys when using PEM files (#11916)

Reviewers: David Jacot <djacot@confluent.io>

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

@@ -96,7 +96,7 @@ public class SslConfigs {
 
     public static final String SSL_KEY_PASSWORD_CONFIG = "ssl.key.password";
     public static final String SSL_KEY_PASSWORD_DOC = "The password of the private key in the key store file or "
-        + "the PEM key specified in `ssl.keystore.key'. This is required for clients only if two-way authentication is configured.";
+        + "the PEM key specified in `ssl.keystore.key'.";
 
     public static final String SSL_TRUSTSTORE_TYPE_CONFIG = "ssl.truststore.type";
     public static final String SSL_TRUSTSTORE_TYPE_DOC = "The file format of the trust store file.";

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

@@ -287,8 +287,6 @@ public final class DefaultSslEngineFactory implements SslEngineFactory {
         } else if (PEM_TYPE.equals(type) && path != null) {
             if (password != null)
                 throw new InvalidConfigurationException("SSL key store password cannot be specified with PEM format, only key password may be specified");
-            else if (keyPassword == null)
-                throw new InvalidConfigurationException("SSL PEM key store is specified, but key password is not specified.");
             else
                 return new FileBasedPemStore(path, keyPassword, true);
         } else if (path == null && password != null) {

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

@@ -490,9 +490,7 @@ public class SslTransportLayerTest {
     }
 
     /**
-     * Test with PEM key store files without key password for client key store. We don't allow this
+     * Test with PEM key store files without key password for client key store.
-     * with PEM files since unprotected private key on disk is not safe. We do allow with inline
-     * PEM config since key config can be encrypted or externalized similar to other password configs.
      */
     @ParameterizedTest
     @ArgumentsSource(SslTransportLayerArgumentsProvider.class)
@@ -502,27 +500,19 @@ public class SslTransportLayerTest {
         TestSslUtils.convertToPem(args.sslClientConfigs, !useInlinePem, false);
         args.sslServerConfigs.put(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "required");
         server = createEchoServer(args, SecurityProtocol.SSL);
-        if (useInlinePem)
+        verifySslConfigs(args);
-            verifySslConfigs(args);
-        else
-            assertThrows(KafkaException.class, () -> createSelector(args.sslClientConfigs));
     }
 
     /**
      * Test with PEM key store files without key password for server key store.We don't allow this
-     * with PEM files since unprotected private key on disk is not safe. We do allow with inline
+     * with PEM files since unprotected private key on disk is not safe.
-     * PEM config since key config can be encrypted or externalized similar to other password configs.
      */
     @ParameterizedTest
     @ArgumentsSource(SslTransportLayerArgumentsProvider.class)
     public void testPemFilesWithoutServerKeyPassword(Args args) throws Exception {
         TestSslUtils.convertToPem(args.sslServerConfigs, !args.useInlinePem, false);
         TestSslUtils.convertToPem(args.sslClientConfigs, !args.useInlinePem, true);
-
+        verifySslConfigs(args);
-        if (args.useInlinePem)
-            verifySslConfigs(args);
-        else
-            assertThrows(KafkaException.class, () -> createEchoServer(args, SecurityProtocol.SSL));
     }
 
     /**

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

@@ -18,7 +18,6 @@ package org.apache.kafka.common.security.ssl;
 
 import org.apache.kafka.common.config.SslConfigs;
 import org.apache.kafka.common.config.types.Password;
-import org.apache.kafka.common.errors.InvalidConfigurationException;
 import org.apache.kafka.test.TestUtils;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -33,7 +32,6 @@ import java.util.Map;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
-import static org.junit.jupiter.api.Assertions.assertThrows;
 
 public class DefaultSslEngineFactoryTest {
 
@@ -291,7 +289,14 @@ public class DefaultSslEngineFactoryTest {
         configs.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG,
                 pemFilePath(pemAsConfigValue(KEY, CERTCHAIN).value()));
         configs.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, DefaultSslEngineFactory.PEM_TYPE);
-        assertThrows(InvalidConfigurationException.class, () -> factory.configure(configs));
+        configs.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, null);
+        factory.configure(configs);
+
+        KeyStore keyStore = factory.keystore();
+        List<String> aliases = Collections.list(keyStore.aliases());
+        assertEquals(Collections.singletonList("kafka"), aliases);
+        assertNotNull(keyStore.getCertificate("kafka"), "Certificate not loaded");
+        assertNotNull(keyStore.getKey("kafka", null), "Private key not loaded");
     }
 
     @Test

docs/security.html

@@ -271,7 +271,7 @@ keyUsage               = digitalSignature, keyEncipherment</code></pre>
 
             <p>Store password configs <code>ssl.keystore.password</code> and <code>ssl.truststore.password</code> are not used for PEM.
             If private key is encrypted using a password, the key password must be provided in <code>ssl.key.password</code>. Private keys may be provided
-            in unencrypted form without a password when PEM is specified directly in the config value. In production deployments, configs should be encrypted or
+            in unencrypted form without a password. In production deployments, configs should be encrypted or
             externalized using password protection feature in Kafka in this case. Note that the default SSL engine factory has limited capabilities for decryption
             of encrypted private keys when external tools like OpenSSL are used for encryption. Third party libraries like BouncyCastle may be integrated witn a
             custom <code>SslEngineFactory</code> to support a wider range of encrypted private keys.</p>