[kafka] branch trunk updated: KAFKA-9711 The authentication failure caused by SSLEngine#beginHandshake is not properly caught and handled (#8287)

[ju...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

junrao pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new b8e508c  KAFKA-9711 The authentication failure caused by SSLEngine#beginHandshake is not properly caught and handled (#8287)
b8e508c is described below

commit b8e508c823517c857c9cfa8a3de77ab374549fde
Author: Chia-Ping Tsai <ch...@gmail.com>
AuthorDate: Wed Mar 25 09:45:05 2020 +0800

    KAFKA-9711 The authentication failure caused by SSLEngine#beginHandshake is not properly caught and handled (#8287)
    
    SSLEngine#beginHandshake is possible to throw authentication failures (for example, no suitable cipher suites) so we ought to catch SSLException and then convert it to SslAuthenticationException so as to process authentication failures correctly.
    
    Reviewers: Jun Rao <ju...@gmail.com>
---
 .../kafka/common/network/SslTransportLayer.java    |  9 ++++++--
 .../common/network/SslTransportLayerTest.java      | 25 ++++++++++++++++++++--
 2 files changed, 30 insertions(+), 4 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/common/network/SslTransportLayer.java b/clients/src/main/java/org/apache/kafka/common/network/SslTransportLayer.java
index 6f0e440..0b11cd1 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/SslTransportLayer.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/SslTransportLayer.java
@@ -267,8 +267,13 @@ public class SslTransportLayer implements TransportLayer {
     */
     @Override
     public void handshake() throws IOException {
-        if (state == State.NOT_INITALIZED)
-            startHandshake();
+        if (state == State.NOT_INITALIZED) {
+            try {
+                startHandshake();
+            } catch (SSLException e) {
+                maybeProcessHandshakeFailure(e, false, null);
+            }
+        }
         if (ready())
             throw renegotiationException();
         if (state == State.CLOSING)
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
index a494d50..f5fedc8 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
@@ -22,8 +22,8 @@ import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
 import org.apache.kafka.common.config.types.Password;
 import org.apache.kafka.common.memory.MemoryPool;
 import org.apache.kafka.common.metrics.Metrics;
-import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.security.TestSecurityConfig;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.security.ssl.SslFactory;
 import org.apache.kafka.common.utils.Java;
 import org.apache.kafka.common.utils.LogContext;
@@ -40,6 +40,7 @@ import org.junit.runners.Parameterized;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLEngine;
 import javax.net.ssl.SSLParameters;
+import javax.net.ssl.SSLServerSocketFactory;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.net.InetAddress;
@@ -52,8 +53,8 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.List;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -578,6 +579,26 @@ public class SslTransportLayerTest {
         server.verifyAuthenticationMetrics(1, 2);
     }
 
+    @Test
+    public void testUnsupportedCipher() throws Exception {
+        String[] cipherSuites = ((SSLServerSocketFactory) SSLServerSocketFactory.getDefault()).getSupportedCipherSuites();
+        if (cipherSuites != null && cipherSuites.length > 1) {
+            sslServerConfigs = serverCertStores.getTrustingConfig(clientCertStores);
+            sslServerConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(cipherSuites[0]));
+            sslClientConfigs = clientCertStores.getTrustingConfig(serverCertStores);
+            sslClientConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(cipherSuites[1]));
+
+            server = createEchoServer(SecurityProtocol.SSL);
+            createSelector(sslClientConfigs);
+
+            checkAuthentiationFailed("1", "TLSv1.1");
+            server.verifyAuthenticationMetrics(0, 1);
+
+            checkAuthentiationFailed("2", "TLSv1");
+            server.verifyAuthenticationMetrics(0, 2);
+        }
+    }
+
     /** Checks connection failed using the specified {@code tlsVersion}. */
     private void checkAuthentiationFailed(String node, String tlsVersion) throws IOException {
         sslClientConfigs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Arrays.asList(tlsVersion));