KAFKA-3152; kafka-acl doesn't allow space in principal name

ijuma · junrao · commit 523562c109b2 · 2016-01-27T08:23:25.000-08:00
* Add quotes to `$` in shell scripts This is necessary for correct processing of quotes in the user command. * Minor improvements to AclCommand messages * Use a principal with a space in `SslEndToEndAuthorizationTest` This passed without any other changes, but good avoid regressions. * Clean-up `TestSslUtils`: Remove unused methods, fix unnecessary verbosity and don't set security.protocol (it should be done at a higher-level). Author: Ismael Juma <ismael@juma.me.uk> Reviewers: Grant Henke <granthenke@gmail.com>, Jun Rao <junrao@gmail.com Closes apache#818 from ijuma/kafka-3152-kafka-acl-space-in-principal
diff --git a/bin/kafka-acls.sh b/bin/kafka-acls.sh
@@ -14,4 +14,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-exec $(dirname $0)/kafka-run-class.sh kafka.admin.AclCommand $@
+exec $(dirname $0)/kafka-run-class.sh kafka.admin.AclCommand "$@"
diff --git a/bin/kafka-configs.sh b/bin/kafka-configs.sh
@@ -14,4 +14,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-exec $(dirname $0)/kafka-run-class.sh kafka.admin.ConfigCommand $@
+exec $(dirname $0)/kafka-run-class.sh kafka.admin.ConfigCommand "$@"
diff --git a/bin/kafka-console-consumer.sh b/bin/kafka-console-consumer.sh
@@ -18,4 +18,4 @@ if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
     export KAFKA_HEAP_OPTS="-Xmx512M"
 fi
 
-exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleConsumer $@
+exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleConsumer "$@"
diff --git a/bin/kafka-console-producer.sh b/bin/kafka-console-producer.sh
@@ -17,4 +17,4 @@
 if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
     export KAFKA_HEAP_OPTS="-Xmx512M"
 fi
-exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleProducer $@
+exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleProducer "$@"
diff --git a/bin/kafka-consumer-groups.sh b/bin/kafka-consumer-groups.sh
@@ -14,4 +14,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-exec $(dirname $0)/kafka-run-class.sh kafka.admin.ConsumerGroupCommand $@
+exec $(dirname $0)/kafka-run-class.sh kafka.admin.ConsumerGroupCommand "$@"
diff --git a/bin/kafka-consumer-offset-checker.sh b/bin/kafka-consumer-offset-checker.sh
@@ -14,4 +14,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker $@
+exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker "$@"
diff --git a/bin/kafka-consumer-perf-test.sh b/bin/kafka-consumer-perf-test.sh
@@ -17,4 +17,4 @@
 if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
     export KAFKA_HEAP_OPTS="-Xmx512M"
 fi
-exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsumerPerformance $@
+exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsumerPerformance "$@"
diff --git a/bin/kafka-mirror-maker.sh b/bin/kafka-mirror-maker.sh
@@ -14,4 +14,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-exec $(dirname $0)/kafka-run-class.sh kafka.tools.MirrorMaker $@
+exec $(dirname $0)/kafka-run-class.sh kafka.tools.MirrorMaker "$@"
diff --git a/bin/kafka-preferred-replica-election.sh b/bin/kafka-preferred-replica-election.sh
@@ -14,4 +14,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-exec $(dirname $0)/kafka-run-class.sh kafka.admin.PreferredReplicaLeaderElectionCommand $@
+exec $(dirname $0)/kafka-run-class.sh kafka.admin.PreferredReplicaLeaderElectionCommand "$@"
diff --git a/bin/kafka-producer-perf-test.sh b/bin/kafka-producer-perf-test.sh
@@ -17,4 +17,4 @@
 if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
     export KAFKA_HEAP_OPTS="-Xmx512M"
 fi
-exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.ProducerPerformance $@
+exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.ProducerPerformance "$@"
diff --git a/bin/kafka-reassign-partitions.sh b/bin/kafka-reassign-partitions.sh
@@ -14,4 +14,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-exec $(dirname $0)/kafka-run-class.sh kafka.admin.ReassignPartitionsCommand $@
+exec $(dirname $0)/kafka-run-class.sh kafka.admin.ReassignPartitionsCommand "$@"
diff --git a/bin/kafka-replay-log-producer.sh b/bin/kafka-replay-log-producer.sh
@@ -14,4 +14,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-exec $(dirname $0)/kafka-run-class.sh kafka.tools.ReplayLogProducer $@
+exec $(dirname $0)/kafka-run-class.sh kafka.tools.ReplayLogProducer "$@"
diff --git a/bin/kafka-replica-verification.sh b/bin/kafka-replica-verification.sh
@@ -14,4 +14,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-exec $(dirname $0)/kafka-run-class.sh kafka.tools.ReplicaVerificationTool $@
+exec $(dirname $0)/kafka-run-class.sh kafka.tools.ReplicaVerificationTool "$@"
diff --git a/bin/kafka-server-start.sh b/bin/kafka-server-start.sh
@@ -41,4 +41,4 @@ case $COMMAND in
     ;;
 esac
 
-exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka $@
+exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"
diff --git a/bin/kafka-simple-consumer-shell.sh b/bin/kafka-simple-consumer-shell.sh
@@ -14,4 +14,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-exec $(dirname $0)/kafka-run-class.sh kafka.tools.SimpleConsumerShell $@
+exec $(dirname $0)/kafka-run-class.sh kafka.tools.SimpleConsumerShell "$@"
diff --git a/bin/kafka-topics.sh b/bin/kafka-topics.sh
@@ -14,4 +14,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-exec $(dirname $0)/kafka-run-class.sh kafka.admin.TopicCommand $@
+exec $(dirname $0)/kafka-run-class.sh kafka.admin.TopicCommand "$@"
diff --git a/bin/kafka-verifiable-consumer.sh b/bin/kafka-verifiable-consumer.sh
@@ -17,4 +17,4 @@
 if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
     export KAFKA_HEAP_OPTS="-Xmx512M"
 fi
-exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.VerifiableConsumer $@
+exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.VerifiableConsumer "$@"
diff --git a/bin/kafka-verifiable-producer.sh b/bin/kafka-verifiable-producer.sh
@@ -17,4 +17,4 @@
 if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
     export KAFKA_HEAP_OPTS="-Xmx512M"
 fi
-exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.VerifiableProducer $@
+exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.VerifiableProducer "$@"
diff --git a/bin/zookeeper-security-migration.sh b/bin/zookeeper-security-migration.sh
@@ -14,4 +14,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-exec $(dirname $0)/kafka-run-class.sh kafka.admin.ZkSecurityMigrator $@
+exec $(dirname $0)/kafka-run-class.sh kafka.admin.ZkSecurityMigrator "$@"
diff --git a/bin/zookeeper-server-start.sh b/bin/zookeeper-server-start.sh
@@ -41,5 +41,4 @@ case $COMMAND in
      ;;
 esac
 
-exec $base_dir/kafka-run-class.sh $EXTRA_ARGS org.apache.zookeeper.server.quorum.QuorumPeerMain $@
-
+exec $base_dir/kafka-run-class.sh $EXTRA_ARGS org.apache.zookeeper.server.quorum.QuorumPeerMain "$@"
diff --git a/clients/src/test/java/org/apache/kafka/common/network/EchoServer.java b/clients/src/test/java/org/apache/kafka/common/network/EchoServer.java
@@ -37,20 +37,23 @@ class EchoServer extends Thread {
     private final ServerSocket serverSocket;
     private final List<Thread> threads;
     private final List<Socket> sockets;
-    private SecurityProtocol protocol = SecurityProtocol.PLAINTEXT;
-    private SslFactory sslFactory;
+    private final SslFactory sslFactory;
     private final AtomicBoolean renegotiate = new AtomicBoolean();
 
-    public EchoServer(Map<String, ?> configs) throws Exception {
-        this.protocol =  configs.containsKey("security.protocol") ?
-            SecurityProtocol.forName((String) configs.get("security.protocol")) : SecurityProtocol.PLAINTEXT;
-        if (protocol == SecurityProtocol.SSL) {
-            this.sslFactory = new SslFactory(Mode.SERVER);
-            this.sslFactory.configure(configs);
-            SSLContext sslContext = this.sslFactory.sslContext();
-            this.serverSocket = sslContext.getServerSocketFactory().createServerSocket(0);
-        } else {
-            this.serverSocket = new ServerSocket(0);
+    public EchoServer(SecurityProtocol securityProtocol, Map<String, ?> configs) throws Exception {
+        switch (securityProtocol) {
+            case SSL:
+                this.sslFactory = new SslFactory(Mode.SERVER);
+                this.sslFactory.configure(configs);
+                SSLContext sslContext = this.sslFactory.sslContext();
+                this.serverSocket = sslContext.getServerSocketFactory().createServerSocket(0);
+                break;
+            case PLAINTEXT:
+                this.serverSocket = new ServerSocket(0);
+                this.sslFactory = null;
+                break;
+            default:
+                throw new IllegalArgumentException("Unsupported securityProtocol " + securityProtocol);
         }
         this.port = this.serverSocket.getLocalPort();
         this.threads = Collections.synchronizedList(new ArrayList<Thread>());
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
@@ -23,6 +23,7 @@
 import java.nio.ByteBuffer;
 
 import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.protocol.SecurityProtocol;
 import org.apache.kafka.common.utils.MockTime;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.common.utils.Utils;
@@ -45,9 +46,9 @@ public class SelectorTest {
     private Metrics metrics;
 
     @Before
-    public void setup() throws Exception {
+    public void setUp() throws Exception {
         Map<String, Object> configs = new HashMap<>();
-        this.server = new EchoServer(configs);
+        this.server = new EchoServer(SecurityProtocol.PLAINTEXT, configs);
         this.server.start();
         this.time = new MockTime();
         this.channelBuilder = new PlaintextChannelBuilder();
@@ -57,7 +58,7 @@ public void setup() throws Exception {
     }
 
     @After
-    public void teardown() throws Exception {
+    public void tearDown() throws Exception {
         this.selector.close();
         this.server.close();
         this.metrics.close();
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
@@ -26,6 +26,7 @@
 
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.protocol.SecurityProtocol;
 import org.apache.kafka.common.security.ssl.SslFactory;
 import org.apache.kafka.common.utils.MockTime;
 import org.apache.kafka.test.TestSslUtils;
@@ -42,24 +43,23 @@ public class SslSelectorTest extends SelectorTest {
     private Map<String, Object> sslClientConfigs;
 
     @Before
-    public void setup() throws Exception {
+    public void setUp() throws Exception {
         File trustStoreFile = File.createTempFile("truststore", ".jks");
 
         Map<String, Object> sslServerConfigs = TestSslUtils.createSslConfig(false, true, Mode.SERVER, trustStoreFile, "server");
         sslServerConfigs.put(SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, Class.forName(SslConfigs.DEFAULT_PRINCIPAL_BUILDER_CLASS));
-        this.server = new EchoServer(sslServerConfigs);
+        this.server = new EchoServer(SecurityProtocol.SSL, sslServerConfigs);
         this.server.start();
         this.time = new MockTime();
-        sslClientConfigs = TestSslUtils.createSslConfig(false, false, Mode.SERVER, trustStoreFile, "client");
-
+        sslClientConfigs = TestSslUtils.createSslConfig(false, false, Mode.CLIENT, trustStoreFile, "client");
         this.channelBuilder = new SslChannelBuilder(Mode.CLIENT);
         this.channelBuilder.configure(sslClientConfigs);
         this.metrics = new Metrics();
         this.selector = new Selector(5000, metrics, time, "MetricGroup", channelBuilder);
     }
 
     @After
-    public void teardown() throws Exception {
+    public void tearDown() throws Exception {
         this.selector.close();
         this.server.close();
         this.metrics.close();
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java b/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
@@ -71,7 +71,6 @@ public void setup() throws Exception {
         clientCertStores = new CertStores(false, "localhost");
         sslServerConfigs = serverCertStores.getTrustingConfig(clientCertStores);
         sslClientConfigs = clientCertStores.getTrustingConfig(serverCertStores);
-
         this.channelBuilder = new SslChannelBuilder(Mode.CLIENT);
         this.channelBuilder.configure(sslClientConfigs);
         this.selector = new Selector(5000, new Metrics(), new MockTime(), "MetricGroup", channelBuilder);
@@ -113,8 +112,8 @@ public void testInvalidEndpointIdentification() throws Exception {
         clientCertStores = new CertStores(false, "localhost");
         sslServerConfigs = serverCertStores.getTrustingConfig(clientCertStores);
         sslClientConfigs = clientCertStores.getTrustingConfig(serverCertStores);
-        createEchoServer(sslServerConfigs);
         sslClientConfigs.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "HTTPS");
+        createEchoServer(sslServerConfigs);
         createSelector(sslClientConfigs);
         InetSocketAddress addr = new InetSocketAddress("localhost", server.port);
         selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE);
diff --git a/clients/src/test/java/org/apache/kafka/common/security/ssl/SslFactoryTest.java b/clients/src/test/java/org/apache/kafka/common/security/ssl/SslFactoryTest.java
@@ -26,11 +26,9 @@
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertTrue;
 
-
 /**
  * A set of tests for the selector over ssl. These use a test harness that runs a simple socket server that echos back responses.
  */
-
 public class SslFactoryTest {
 
     @Test
diff --git a/clients/src/test/java/org/apache/kafka/test/TestSslUtils.java b/clients/src/test/java/org/apache/kafka/test/TestSslUtils.java
@@ -19,7 +19,6 @@
 
 import org.apache.kafka.common.config.SslConfigs;
 import org.apache.kafka.common.network.Mode;
-import org.apache.kafka.clients.CommonClientConfigs;
 
 import java.io.File;
 import java.io.FileOutputStream;
@@ -116,11 +115,8 @@ private static KeyStore createEmptyKeyStore() throws GeneralSecurityException, I
 
     private static void saveKeyStore(KeyStore ks, String filename,
                                      Password password) throws GeneralSecurityException, IOException {
-        FileOutputStream out = new FileOutputStream(filename);
-        try {
+        try (FileOutputStream out = new FileOutputStream(filename)) {
             ks.store(out, password.value().toCharArray());
-        } finally {
-            out.close();
         }
     }
 
@@ -154,14 +150,6 @@ public static void createKeyStore(String filename,
         saveKeyStore(ks, filename, password);
     }
 
-    public static void createTrustStore(String filename,
-                                        Password password, String alias,
-                                        Certificate cert) throws GeneralSecurityException, IOException {
-        KeyStore ks = createEmptyKeyStore();
-        ks.setCertificateEntry(alias, cert);
-        saveKeyStore(ks, filename, password);
-    }
-
     public static <T extends Certificate> void createTrustStore(
             String filename, Password password, Map<String, T> certs) throws GeneralSecurityException, IOException {
         KeyStore ks = KeyStore.getInstance("JKS");
@@ -178,18 +166,9 @@ public static <T extends Certificate> void createTrustStore(
         saveKeyStore(ks, filename, password);
     }
 
-    public static Map<String, X509Certificate> createX509Certificates(KeyPair keyPair)
-        throws GeneralSecurityException {
-        Map<String, X509Certificate> certs = new HashMap<String, X509Certificate>();
-        X509Certificate cert = generateCertificate("CN=localhost, O=localhost", keyPair, 30, "SHA1withRSA");
-        certs.put("localhost", cert);
-        return certs;
-    }
-
-    public static Map<String, Object> createSslConfig(Mode mode, File keyStoreFile, Password password, Password keyPassword,
+    private static Map<String, Object> createSslConfig(Mode mode, File keyStoreFile, Password password, Password keyPassword,
                                                       File trustStoreFile, Password trustStorePassword) {
         Map<String, Object> sslConfigs = new HashMap<>();
-        sslConfigs.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL"); // kafka security protocol
         sslConfigs.put(SslConfigs.SSL_PROTOCOL_CONFIG, "TLSv1.2"); // protocol to create SSLContext
 
         if (mode == Mode.SERVER || (mode == Mode.CLIENT && keyStoreFile != null)) {
@@ -219,27 +198,22 @@ public static  Map<String, Object> createSslConfig(boolean useClientCert, boolea
 
     public static  Map<String, Object> createSslConfig(boolean useClientCert, boolean trustStore, Mode mode, File trustStoreFile, String certAlias, String host)
         throws IOException, GeneralSecurityException {
-        Map<String, X509Certificate> certs = new HashMap<String, X509Certificate>();
+        Map<String, X509Certificate> certs = new HashMap<>();
         File keyStoreFile;
-        Password password;
-
-        if (mode == Mode.SERVER)
-            password = new Password("ServerPassword");
-        else
-            password = new Password("ClientPassword");
+        Password password = mode == Mode.SERVER ? new Password("ServerPassword") : new Password("ClientPassword");
 
         Password trustStorePassword = new Password("TrustStorePassword");
 
         if (useClientCert) {
             keyStoreFile = File.createTempFile("clientKS", ".jks");
             KeyPair cKP = generateKeyPair("RSA");
-            X509Certificate cCert = generateCertificate("CN=" + host + ", O=client", cKP, 30, "SHA1withRSA");
+            X509Certificate cCert = generateCertificate("CN=" + host + ", O=A client", cKP, 30, "SHA1withRSA");
             createKeyStore(keyStoreFile.getPath(), password, "client", cKP.getPrivate(), cCert);
             certs.put(certAlias, cCert);
         } else {
             keyStoreFile = File.createTempFile("serverKS", ".jks");
             KeyPair sKP = generateKeyPair("RSA");
-            X509Certificate sCert = generateCertificate("CN=" + host + ", O=server", sKP, 30,
+            X509Certificate sCert = generateCertificate("CN=" + host + ", O=A server", sKP, 30,
                                                         "SHA1withRSA");
             createKeyStore(keyStoreFile.getPath(), password, password, "server", sKP.getPrivate(), sCert);
             certs.put(certAlias, sCert);
@@ -249,9 +223,7 @@ public static  Map<String, Object> createSslConfig(boolean useClientCert, boolea
             createTrustStore(trustStoreFile.getPath(), trustStorePassword, certs);
         }
 
-        Map<String, Object> sslConfig = createSslConfig(mode, keyStoreFile, password,
-                                                        password, trustStoreFile, trustStorePassword);
-        return sslConfig;
+        return createSslConfig(mode, keyStoreFile, password, password, trustStoreFile, trustStorePassword);
     }
 
 }
diff --git a/core/src/main/scala/kafka/admin/AclCommand.scala b/core/src/main/scala/kafka/admin/AclCommand.scala
diff --git a/core/src/test/scala/integration/kafka/api/SslEndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/SslEndToEndAuthorizationTest.scala
