KAFKA-12703; Allow unencrypted private keys when using PEM files (apa…

…che#11916)

Reviewers: David Jacot <[email protected]>

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

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

@@ -18,7 +18,6 @@
 
 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 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 void testPemKeyStoreFileNoKeyPassword() throws Exception {
         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 @@ <h5>SSL key and certificates in PEM format</h5>
 
             <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>