root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-4050; Allow configuration of the PRNG used for SSL 

Add an optional configuration for the SecureRandom PRNG implementation, with the default behavior being the same (use the default implementation in the JDK/JRE).

Author: Todd Palino <Todd Palino>

Reviewers: Grant Henke <granthenke@gmail.com>, Ismael Juma <ismael@juma.me.uk>, Joel Koshy <jjkoshy@gmail.com>, Jiangjie Qin <becket.qin@gmail.com>, Rajini Sivaram <rajinisivaram@googlemail.com>

Closes #1747 from toddpalino/trunk

(cherry picked from commit 104d2154b6)
Signed-off-by: Joel Koshy <jjkoshy@gmail.com>
This commit is contained in:
Todd Palino 2016-08-19 11:05:39 -07:00 committed by Joel Koshy
parent a707f573de
commit afb65688af
7 changed files with 47 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -85,6 +85,9 @@ public class SslConfigs {
public static final String SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG = "ssl.endpoint.identification.algorithm";
public static final String SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_DOC = "The endpoint identification algorithm to validate server hostname using server certificate. ";
public static final String SSL_SECURE_RANDOM_IMPLEMENTATION_CONFIG = "ssl.secure.random.implementation";
public static final String SSL_SECURE_RANDOM_IMPLEMENTATION_DOC = "The SecureRandom PRNG implementation to use for SSL cryptography operations. ";
public static final String SSL_CLIENT_AUTH_CONFIG = "ssl.client.auth";
public static final String SSL_CLIENT_AUTH_DOC = "Configures kafka broker to request client authentication."
+ " The following settings are common: "
@ -109,6 +112,7 @@ public class SslConfigs {
.define(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, ConfigDef.Type.PASSWORD, null, ConfigDef.Importance.HIGH, SslConfigs.SSL_TRUSTSTORE_PASSWORD_DOC)
.define(SslConfigs.SSL_KEYMANAGER_ALGORITHM_CONFIG, ConfigDef.Type.STRING, SslConfigs.DEFAULT_SSL_KEYMANGER_ALGORITHM, ConfigDef.Importance.LOW, SslConfigs.SSL_KEYMANAGER_ALGORITHM_DOC)
.define(SslConfigs.SSL_TRUSTMANAGER_ALGORITHM_CONFIG, ConfigDef.Type.STRING, SslConfigs.DEFAULT_SSL_TRUSTMANAGER_ALGORITHM, ConfigDef.Importance.LOW, SslConfigs.SSL_TRUSTMANAGER_ALGORITHM_DOC)
.define(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, ConfigDef.Type.STRING, null, ConfigDef.Importance.LOW, SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_DOC);
.define(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, ConfigDef.Type.STRING, null, ConfigDef.Importance.LOW, SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_DOC)
.define(SslConfigs.SSL_SECURE_RANDOM_IMPLEMENTATION_CONFIG, ConfigDef.Type.STRING, null, ConfigDef.Importance.LOW, SslConfigs.SSL_SECURE_RANDOM_IMPLEMENTATION_DOC);
}
}

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

@ -26,6 +26,7 @@ import java.io.FileInputStream;
import java.io.IOException;
import java.security.GeneralSecurityException;
import java.security.KeyStore;
import java.security.SecureRandom;
import java.util.List;
import java.util.Map;
@ -51,6 +52,7 @@ public class SslFactory implements Configurable {
private String[] cipherSuites;
private String[] enabledProtocols;
private String endpointIdentification;
private SecureRandom secureRandomImplementation;
private SSLContext sslContext;
private boolean needClientAuth;
private boolean wantClientAuth;
@ -83,6 +85,15 @@ public class SslFactory implements Configurable {
if (endpointIdentification != null)
this.endpointIdentification = endpointIdentification;
String secureRandomImplementation = (String) configs.get(SslConfigs.SSL_SECURE_RANDOM_IMPLEMENTATION_CONFIG);
if (secureRandomImplementation != null) {
try {
this.secureRandomImplementation = SecureRandom.getInstance(secureRandomImplementation);
} catch (GeneralSecurityException e) {
throw new KafkaException(e);
}
}
String clientAuthConfig = clientAuthConfigOverride;
if (clientAuthConfig == null)
clientAuthConfig = (String) configs.get(SslConfigs.SSL_CLIENT_AUTH_CONFIG);
@ -134,7 +145,7 @@ public class SslFactory implements Configurable {
KeyStore ts = truststore == null ? null : truststore.load();
tmf.init(ts);
sslContext.init(keyManagers, tmf.getTrustManagers(), null);
sslContext.init(keyManagers, tmf.getTrustManagers(), this.secureRandomImplementation);
return sslContext;
}

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

@ -252,7 +252,22 @@ public class SslTransportLayerTest {
NetworkTestUtils.checkClientConnection(selector, node, 100, 10);
}
/**
* Tests that an invalid SecureRandom implementation cannot be configured
*/
@Test
public void testInvalidSecureRandomImplementation() throws Exception {
SslChannelBuilder channelBuilder = new SslChannelBuilder(Mode.CLIENT);
try {
sslClientConfigs.put(SslConfigs.SSL_SECURE_RANDOM_IMPLEMENTATION_CONFIG, "invalid");
channelBuilder.configure(sslClientConfigs);
fail("SSL channel configured with invalid SecureRandom implementation");
} catch (KafkaException e) {
// Expected exception
}
}
/**
* Tests that channels cannot be created if truststore cannot be loaded
*/

3
core/src/main/scala/kafka/server/KafkaConfig.scala
View File

@ -336,6 +336,7 @@ object KafkaConfig {
val SslKeyManagerAlgorithmProp = SslConfigs.SSL_KEYMANAGER_ALGORITHM_CONFIG
val SslTrustManagerAlgorithmProp = SslConfigs.SSL_TRUSTMANAGER_ALGORITHM_CONFIG
val SslEndpointIdentificationAlgorithmProp = SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG
val SslSecureRandomImplementationProp = SslConfigs.SSL_SECURE_RANDOM_IMPLEMENTATION_CONFIG
val SslClientAuthProp = SslConfigs.SSL_CLIENT_AUTH_CONFIG
/** ********* SASL Configuration ****************/
@ -534,6 +535,7 @@ object KafkaConfig {
val SslKeyManagerAlgorithmDoc = SslConfigs.SSL_KEYMANAGER_ALGORITHM_DOC
val SslTrustManagerAlgorithmDoc = SslConfigs.SSL_TRUSTMANAGER_ALGORITHM_DOC
val SslEndpointIdentificationAlgorithmDoc = SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_DOC
val SslSecureRandomImplementationDoc = SslConfigs.SSL_SECURE_RANDOM_IMPLEMENTATION_DOC
val SslClientAuthDoc = SslConfigs.SSL_CLIENT_AUTH_DOC
/** ********* Sasl Configuration ****************/
@ -706,6 +708,7 @@ object KafkaConfig {
.define(SslKeyManagerAlgorithmProp, STRING, Defaults.SslKeyManagerAlgorithm, MEDIUM, SslKeyManagerAlgorithmDoc)
.define(SslTrustManagerAlgorithmProp, STRING, Defaults.SslTrustManagerAlgorithm, MEDIUM, SslTrustManagerAlgorithmDoc)
.define(SslEndpointIdentificationAlgorithmProp, STRING, null, LOW, SslEndpointIdentificationAlgorithmDoc)
.define(SslSecureRandomImplementationProp, STRING, null, LOW, SslSecureRandomImplementationDoc)
.define(SslClientAuthProp, STRING, Defaults.SslClientAuth, in(Defaults.SslClientAuthRequired, Defaults.SslClientAuthRequested, Defaults.SslClientAuthNone), MEDIUM, SslClientAuthDoc)
.define(SslCipherSuitesProp, LIST, null, MEDIUM, SslCipherSuitesDoc)

1
core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
View File

@ -547,6 +547,7 @@ class KafkaConfigTest {
case KafkaConfig.SslTrustManagerAlgorithmProp =>
case KafkaConfig.SslClientAuthProp => // ignore string
case KafkaConfig.SslEndpointIdentificationAlgorithmProp => // ignore string
case KafkaConfig.SslSecureRandomImplementationProp => // ignore string
case KafkaConfig.SslCipherSuitesProp => // ignore string
//Sasl Configs

9
docs/security.html
View File

@ -145,6 +145,7 @@ Apache Kafka allows clients to connect over SSL. By default SSL is disabled but
<li>ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1 (list out the SSL protocols that you are going to accept from clients. Do note that SSL is deprecated in favor of TLS and using SSL in production is not recommended)</li>
<li>ssl.keystore.type=JKS</li>
<li>ssl.truststore.type=JKS</li>
<li>ssl.secure.random.implementation=SHA1PRNG</li>
</ol>
If you want to enable SSL for inter-broker communication, add the following to the broker properties file (it defaults to PLAINTEXT)
<pre>
@ -155,6 +156,14 @@ Apache Kafka allows clients to connect over SSL. By default SSL is disabled but
<a href="https://docs.oracle.com/javase/8/docs/technotes/guides/security/SunProviders.html">JCA Providers Documentation</a> for more information.
</p>
<p>
The JRE/JDK will have a default pseudo-random number generator (PRNG) that is used for cryptography operations, so it is not required to configure the
implementation used with the <pre>ssl.secure.random.implementation</pre>. However, there are performance issues with some implementations (notably, the
default chosen on Linux systems, <pre>NativePRNG</pre>, utilizes a global lock). In cases where performance of SSL connections becomes an issue,
consider explicitly setting the implementation to be used. The <pre>SHA1PRNG</pre> implementation is non-blocking, and has shown very good performance
characteristics under heavy load (50 MB/sec of produced messages, plus replication traffic, per-broker).
</p>
Once you start the broker you should be able to see in the server.log
<pre>
with addresses: PLAINTEXT -> EndPoint(192.168.64.1,9092,PLAINTEXT),SSL -> EndPoint(192.168.64.1,9093,SSL)</pre>

1
tests/kafkatest/services/kafka/config_property.py
View File

@ -173,6 +173,7 @@ From KafkaConfig.scala
val SSLKeyManagerAlgorithmProp = SSLConfigs.SSL_KEYMANAGER_ALGORITHM_CONFIG
val SSLTrustManagerAlgorithmProp = SSLConfigs.SSL_TRUSTMANAGER_ALGORITHM_CONFIG
val SSLEndpointIdentificationAlgorithmProp = SSLConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG
val SSLSecureRandomImplementationProp = SSLConfigs.SSL_SECURE_RANDOM_IMPLEMENTATION_CONFIG
val SSLClientAuthProp = SSLConfigs.SSL_CLIENT_AUTH_CONFIG
"""