kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From srihar...@apache.org
Subject kafka git commit: KAFKA-2637; Cipher suite setting should be configurable for SSL
Date Tue, 13 Oct 2015 13:59:35 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 373332b0f -> e2ec02e1d


KAFKA-2637; Cipher suite setting should be configurable for SSL

Enables Cipher suite setting. Code was previously reviewed by ijuma, harshach. Moving to an
independent PR.

Author: benstopford <benstopford@gmail.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>, Sriharsha Chintalapani <harsha@hortonworks.com>

Closes #301 from benstopford/cipher-switch


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/e2ec02e1
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/e2ec02e1
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/e2ec02e1

Branch: refs/heads/trunk
Commit: e2ec02e1d1580e73c06f1acbf3a53a1b8ee52930
Parents: 373332b
Author: benstopford <benstopford@gmail.com>
Authored: Tue Oct 13 06:59:25 2015 -0700
Committer: Sriharsha Chintalapani <harsha@hortonworks.com>
Committed: Tue Oct 13 06:59:25 2015 -0700

----------------------------------------------------------------------
 .../apache/kafka/common/config/SSLConfigs.java  |  4 ++--
 .../kafka/common/security/ssl/SSLFactory.java   | 20 ++++++++++----------
 .../main/scala/kafka/server/KafkaConfig.scala   | 20 +++++++++-----------
 .../unit/kafka/server/KafkaConfigTest.scala     |  3 ++-
 4 files changed, 23 insertions(+), 24 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/e2ec02e1/clients/src/main/java/org/apache/kafka/common/config/SSLConfigs.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/config/SSLConfigs.java b/clients/src/main/java/org/apache/kafka/common/config/SSLConfigs.java
index dd7b71a..77bb583 100644
--- a/clients/src/main/java/org/apache/kafka/common/config/SSLConfigs.java
+++ b/clients/src/main/java/org/apache/kafka/common/config/SSLConfigs.java
@@ -13,8 +13,8 @@
 
 package org.apache.kafka.common.config;
 
-import javax.net.ssl.TrustManagerFactory;
 import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
 
 public class SSLConfigs {
     /*
@@ -34,7 +34,7 @@ public class SSLConfigs {
     public static final String SSL_PROVIDER_DOC = "The name of the security provider used
for SSL connections. Default value is the default security provider of the JVM.";
 
     public static final String SSL_CIPHER_SUITES_CONFIG = "ssl.cipher.suites";
-    public static final String SSL_CIPHER_SUITES_DOC = "A cipher suite is a named combination
of authentication, encryption, MAC and key exchange algorithm used to negotiate the security
settings for a network connection using TLS or SSL network protocol."
+    public static final String SSL_CIPHER_SUITES_DOC = "A list of cipher suites. This is
a named combination of authentication, encryption, MAC and key exchange algorithm used to
negotiate the security settings for a network connection using TLS or SSL network protocol."
             + "By default all the available cipher suites are supported.";
 
     public static final String SSL_ENABLED_PROTOCOLS_CONFIG = "ssl.enabled.protocols";

http://git-wip-us.apache.org/repos/asf/kafka/blob/e2ec02e1/clients/src/main/java/org/apache/kafka/common/security/ssl/SSLFactory.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/ssl/SSLFactory.java b/clients/src/main/java/org/apache/kafka/common/security/ssl/SSLFactory.java
index f79b65c..b3e0895 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/ssl/SSLFactory.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/ssl/SSLFactory.java
@@ -16,18 +16,17 @@
  */
 package org.apache.kafka.common.security.ssl;
 
-import java.util.Map;
-import java.util.List;
+import org.apache.kafka.common.Configurable;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.config.SSLConfigs;
+
+import javax.net.ssl.*;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.security.GeneralSecurityException;
 import java.security.KeyStore;
-
-import javax.net.ssl.*;
-
-import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.Configurable;
-import org.apache.kafka.common.config.SSLConfigs;
+import java.util.List;
+import java.util.Map;
 
 
 public class SSLFactory implements Configurable {
@@ -60,12 +59,13 @@ public class SSLFactory implements Configurable {
 
         if (configs.get(SSLConfigs.SSL_CIPHER_SUITES_CONFIG) != null) {
             List<String> cipherSuitesList = (List<String>) configs.get(SSLConfigs.SSL_CIPHER_SUITES_CONFIG);
-            this.cipherSuites = (String[]) cipherSuitesList.toArray(new String[cipherSuitesList.size()]);
+            if (!cipherSuitesList.isEmpty())
+                this.cipherSuites = cipherSuitesList.toArray(new String[cipherSuitesList.size()]);
         }
 
         if (configs.get(SSLConfigs.SSL_ENABLED_PROTOCOLS_CONFIG) != null) {
             List<String> enabledProtocolsList = (List<String>) configs.get(SSLConfigs.SSL_ENABLED_PROTOCOLS_CONFIG);
-            this.enabledProtocols =  (String[]) enabledProtocolsList.toArray(new String[enabledProtocolsList.size()]);
+            this.enabledProtocols = enabledProtocolsList.toArray(new String[enabledProtocolsList.size()]);
         }
 
         if (configs.containsKey(SSLConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG))
{

http://git-wip-us.apache.org/repos/asf/kafka/blob/e2ec02e1/core/src/main/scala/kafka/server/KafkaConfig.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 46f4a25..d9d87cd 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -26,16 +26,12 @@ import kafka.consumer.ConsumerConfig
 import kafka.message.{BrokerCompressionCodec, CompressionCodec, Message, MessageSet}
 import kafka.utils.CoreUtils
 import org.apache.kafka.clients.CommonClientConfigs
-import org.apache.kafka.common.config.SSLConfigs
-import org.apache.kafka.common.config.ConfigDef.Importance._
-import org.apache.kafka.common.config.ConfigDef.Range._
-import org.apache.kafka.common.config.ConfigDef.Type._
-
-import org.apache.kafka.common.config.{ConfigException, AbstractConfig, ConfigDef}
+import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, SSLConfigs}
 import org.apache.kafka.common.metrics.MetricsReporter
 import org.apache.kafka.common.protocol.SecurityProtocol
 import org.apache.kafka.common.security.auth.PrincipalBuilder
-import scala.collection.{mutable, immutable, JavaConversions, Map}
+
+import scala.collection.{Map, immutable}
 
 
 object Defaults {
@@ -176,7 +172,7 @@ object Defaults {
   val SSLClientAuthRequested = "requested"
   val SSLClientAuthNone = "none"
   val SSLClientAuth = SSLClientAuthNone
-
+  val SSLCipherSuites = ""
 }
 
 object KafkaConfig {
@@ -497,10 +493,10 @@ object KafkaConfig {
   val SSLClientAuthDoc = SSLConfigs.SSL_CLIENT_AUTH_DOC
 
   private val configDef = {
+    import ConfigDef.Importance._
     import ConfigDef.Range._
-    import ConfigDef.ValidString._
     import ConfigDef.Type._
-    import ConfigDef.Importance._
+    import ConfigDef.ValidString._
 
     new ConfigDef()
 
@@ -650,7 +646,7 @@ object KafkaConfig {
       .define(SSLKeyManagerAlgorithmProp, STRING, Defaults.SSLKeyManagerAlgorithm, MEDIUM,
SSLKeyManagerAlgorithmDoc)
       .define(SSLTrustManagerAlgorithmProp, STRING, Defaults.SSLTrustManagerAlgorithm, MEDIUM,
SSLTrustManagerAlgorithmDoc)
       .define(SSLClientAuthProp, STRING, Defaults.SSLClientAuth, in(Defaults.SSLClientAuthRequired,
Defaults.SSLClientAuthRequested, Defaults.SSLClientAuthNone), MEDIUM, SSLClientAuthDoc)
-
+      .define(SSLCipherSuitesProp, LIST, Defaults.SSLCipherSuites, MEDIUM, SSLCipherSuitesDoc)
   }
 
   def configNames() = {
@@ -809,6 +805,7 @@ case class KafkaConfig (props: java.util.Map[_, _]) extends AbstractConfig(Kafka
   val sslKeyManagerAlgorithm = getString(KafkaConfig.SSLKeyManagerAlgorithmProp)
   val sslTrustManagerAlgorithm = getString(KafkaConfig.SSLTrustManagerAlgorithmProp)
   val sslClientAuth = getString(KafkaConfig.SSLClientAuthProp)
+  val sslCipher = getList(KafkaConfig.SSLCipherSuitesProp)
 
   /** ********* Quota Configuration **************/
   val producerQuotaBytesPerSecondDefault = getLong(KafkaConfig.ProducerQuotaBytesPerSecondDefaultProp)
@@ -948,6 +945,7 @@ case class KafkaConfig (props: java.util.Map[_, _]) extends AbstractConfig(Kafka
     channelConfigs.put(SSLKeyManagerAlgorithmProp, sslKeyManagerAlgorithm)
     channelConfigs.put(SSLTrustManagerAlgorithmProp, sslTrustManagerAlgorithm)
     channelConfigs.put(SSLClientAuthProp, sslClientAuth)
+    channelConfigs.put(SSLCipherSuitesProp, sslCipher)
     channelConfigs
   }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/e2ec02e1/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
index bfec426..4637972 100755
--- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
@@ -22,7 +22,7 @@ import java.util.Properties
 import junit.framework.Assert._
 import kafka.api.{ApiVersion, KAFKA_082}
 import kafka.message._
-import kafka.utils.{TestUtils, CoreUtils}
+import kafka.utils.{CoreUtils, TestUtils}
 import org.apache.kafka.common.config.ConfigException
 import org.apache.kafka.common.protocol.SecurityProtocol
 import org.junit.{Assert, Test}
@@ -505,6 +505,7 @@ class KafkaConfigTest {
         case KafkaConfig.SSLKeyManagerAlgorithmProp =>
         case KafkaConfig.SSLTrustManagerAlgorithmProp =>
         case KafkaConfig.SSLClientAuthProp => // ignore string
+        case KafkaConfig.SSLCipherSuitesProp => // ignore string
 
         case nonNegativeIntProperty => assertPropertyInvalid(getBaseProperties(), name,
"not_a_number", "-1")
       }


Mime
View raw message