KAFKA-9320: Enable TLSv1.3 by default (KIP-573)

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

@@ -63,7 +63,6 @@
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
-import static org.junit.Assume.assumeTrue;
 
 /**
  * Tests for the SSL transport layer. These use a test harness that runs a simple socket server that echos back responses.
@@ -578,30 +577,6 @@ public void testTLSDefaults() throws Exception {
         server.verifyAuthenticationMetrics(1, 2);
     }
 
-    @Test
-    public void testUnsupportedCipher() throws Exception {
-        String[] cipherSuites;
-        if (Java.IS_JAVA11_COMPATIBLE) {
-            cipherSuites = new String[] {"TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384", "TLS_AES_256_GCM_SHA384"};
-        } else {
-            cipherSuites = new String[] {"TLS_AES_128_GCM_SHA256", "TLS_AES_256_GCM_SHA384"};
-        }
-
-        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));
@@ -626,104 +601,11 @@ public void testUnsupportedTLSVersion() throws Exception {
         server.verifyAuthenticationMetrics(0, 1);
     }
 
-    /**
-     * Tests that connections fails if TLSv1.3 enabled but cipher suite suitable only for TLSv1.2 used.
-     */
-    @Test
-    public void testCiphersSuiteForTls12FailsForTls13() throws Exception {
-        assumeTrue(Java.IS_JAVA11_COMPATIBLE);
-
-        String cipherSuite = "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384";
-
-        sslServerConfigs.put(SslConfigs.SSL_PROTOCOL_CONFIG, "TLSv1.3");
-        sslServerConfigs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Collections.singletonList("TLSv1.3"));
-        sslServerConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(cipherSuite));
-        server = createEchoServer(SecurityProtocol.SSL);
-
-        sslClientConfigs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Collections.singletonList("TLSv1.3"));
-        sslClientConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(cipherSuite));
-
-        checkAuthentiationFailed("0", "TLSv1.3");
-        server.verifyAuthenticationMetrics(0, 1);
-    }
-
-    /**
-     * Tests that connections can't be made if server uses TLSv1.2 with custom cipher suite and client uses TLSv1.3.
-     */
-    @Test
-    public void testCiphersSuiteFailForServerTls12ClientTls13() throws Exception {
-        assumeTrue(Java.IS_JAVA11_COMPATIBLE);
-
-        String tls12CipherSuite = "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384";
-        String tls13CipherSuite = "TLS_AES_128_GCM_SHA256";
-
-        sslServerConfigs.put(SslConfigs.SSL_PROTOCOL_CONFIG, "TLSv1.2");
-        sslServerConfigs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Collections.singletonList("TLSv1.2"));
-        sslServerConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(tls12CipherSuite));
-        server = createEchoServer(SecurityProtocol.SSL);
-
-        sslClientConfigs.put(SslConfigs.SSL_PROTOCOL_CONFIG, "TLSv1.3");
-        sslClientConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(tls13CipherSuite));
-
-        checkAuthentiationFailed("0", "TLSv1.3");
-        server.verifyAuthenticationMetrics(0, 1);
-    }
-
-    /**
-     * Tests that connections can be made with TLSv1.3 cipher suite.
-     */
-    @Test
-    public void testCiphersSuiteForTls13() throws Exception {
-        assumeTrue(Java.IS_JAVA11_COMPATIBLE);
-
-        String node = "0";
-        String cipherSuite = "TLS_AES_128_GCM_SHA256";
-
-        sslServerConfigs.put(SslConfigs.SSL_PROTOCOL_CONFIG, SslConfigs.DEFAULT_SSL_PROTOCOL);
-        sslServerConfigs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Arrays.asList(SslConfigs.DEFAULT_SSL_ENABLED_PROTOCOLS.split(",")));
-        sslServerConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(cipherSuite));
-        server = createEchoServer(SecurityProtocol.SSL);
-
-        sslClientConfigs.put(SslConfigs.SSL_PROTOCOL_CONFIG, SslConfigs.DEFAULT_SSL_PROTOCOL);
-        sslClientConfigs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Arrays.asList(SslConfigs.DEFAULT_SSL_ENABLED_PROTOCOLS.split(",")));
-        sslClientConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(cipherSuite));
-        createSelector(sslClientConfigs);
-        InetSocketAddress addr = new InetSocketAddress("localhost", server.port());
-        selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE);
-        NetworkTestUtils.waitForChannelReady(selector, node);
-        server.verifyAuthenticationMetrics(1, 0);
-    }
-
-    /**
-     * Tests that connections can be made with TLSv1.2 cipher suite.
-     */
-    @Test
-    public void testCiphersSuiteForTls12() throws Exception {
-        String node = "0";
-        String cipherSuite = "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384";
-
-        sslServerConfigs.put(SslConfigs.SSL_PROTOCOL_CONFIG, SslConfigs.DEFAULT_SSL_PROTOCOL);
-        sslServerConfigs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Arrays.asList(SslConfigs.DEFAULT_SSL_ENABLED_PROTOCOLS.split(",")));
-        sslServerConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(cipherSuite));
-        server = createEchoServer(SecurityProtocol.SSL);
-
-        sslClientConfigs.put(SslConfigs.SSL_PROTOCOL_CONFIG, SslConfigs.DEFAULT_SSL_PROTOCOL);
-        sslClientConfigs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Arrays.asList(SslConfigs.DEFAULT_SSL_ENABLED_PROTOCOLS.split(",")));
-        sslClientConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(cipherSuite));
-        createSelector(sslClientConfigs);
-        InetSocketAddress addr = new InetSocketAddress("localhost", server.port());
-        selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE);
-
-        NetworkTestUtils.waitForChannelReady(selector, node);
-        server.verifyAuthenticationMetrics(1, 0);
-    }
-
     /**
      * Tests that connections cannot be made with unsupported TLS cipher suites
      */
     @Test
     public void testUnsupportedCiphers() throws Exception {
-        String node = "0";
         SSLContext context = SSLContext.getInstance(tlsProtocol);
         context.init(null, null, null);
         String[] cipherSuites = context.getDefaultSSLParameters().getCipherSuites();
@@ -732,11 +614,8 @@ public void testUnsupportedCiphers() throws Exception {
 
         sslClientConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Arrays.asList(cipherSuites[1]));
         createSelector(sslClientConfigs);
-        InetSocketAddress addr = new InetSocketAddress("localhost", server.port());
-        selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE);
 
-        NetworkTestUtils.waitForChannelClose(selector, node, ChannelState.State.AUTHENTICATION_FAILED);
-        server.verifyAuthenticationMetrics(0, 1);
+        checkAuthentiationFailed("1", tlsProtocol);
     }
 
     @Test

clients/src/test/java/org/apache/kafka/common/network/SslTransportTls12Tls13Test.java

@@ -0,0 +1,169 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.network;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.security.TestSecurityConfig;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.common.utils.Java;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Time;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assume.assumeTrue;
+
+public class SslTransportTls12Tls13Test {
+    private static final int BUFFER_SIZE = 4 * 1024;
+    private static final Time TIME = Time.SYSTEM;
+
+    private NioEchoServer server;
+    private Selector selector;
+    private Map<String, Object> sslClientConfigs;
+    private Map<String, Object> sslServerConfigs;
+
+    @Before
+    public void setup() throws Exception {
+        // Create certificates for use by client and server. Add server cert to client truststore and vice versa.
+        CertStores serverCertStores = new CertStores(true, "server", "localhost");
+        CertStores clientCertStores = new CertStores(false, "client", "localhost");
+        sslServerConfigs = serverCertStores.getTrustingConfig(clientCertStores);
+        sslClientConfigs = clientCertStores.getTrustingConfig(serverCertStores);
+
+        LogContext logContext = new LogContext();
+        ChannelBuilder channelBuilder = new SslChannelBuilder(Mode.CLIENT, null, false, logContext);
+        channelBuilder.configure(sslClientConfigs);
+        this.selector = new Selector(5000, new Metrics(), TIME, "MetricGroup", channelBuilder, logContext);
+    }
+
+    @After
+    public void teardown() throws Exception {
+        if (selector != null)
+            this.selector.close();
+        if (server != null)
+            this.server.close();
+    }
+
+    /**
+     * Tests that connections fails if TLSv1.3 enabled but cipher suite suitable only for TLSv1.2 used.
+     */
+    @Test
+    public void testCiphersSuiteForTls12FailsForTls13() throws Exception {
+        assumeTrue(Java.IS_JAVA11_COMPATIBLE);
+
+        String cipherSuite = "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384";
+
+        sslServerConfigs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Collections.singletonList("TLSv1.3"));
+        sslServerConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(cipherSuite));
+        server = NetworkTestUtils.createEchoServer(ListenerName.forSecurityProtocol(SecurityProtocol.SSL),
+            SecurityProtocol.SSL, new TestSecurityConfig(sslServerConfigs), null, TIME);
+
+        sslClientConfigs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Collections.singletonList("TLSv1.3"));
+        sslClientConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(cipherSuite));
+
+        checkAuthentiationFailed();
+    }
+
+    /**
+     * Tests that connections can't be made if server uses TLSv1.2 with custom cipher suite and client uses TLSv1.3.
+     */
+    @Test
+    public void testCiphersSuiteFailForServerTls12ClientTls13() throws Exception {
+        assumeTrue(Java.IS_JAVA11_COMPATIBLE);
+
+        String tls12CipherSuite = "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384";
+        String tls13CipherSuite = "TLS_AES_128_GCM_SHA256";
+
+        sslServerConfigs.put(SslConfigs.SSL_PROTOCOL_CONFIG, "TLSv1.2");
+        sslServerConfigs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Collections.singletonList("TLSv1.2"));
+        sslServerConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(tls12CipherSuite));
+        server = NetworkTestUtils.createEchoServer(ListenerName.forSecurityProtocol(SecurityProtocol.SSL),
+            SecurityProtocol.SSL, new TestSecurityConfig(sslServerConfigs), null, TIME);
+
+        sslClientConfigs.put(SslConfigs.SSL_PROTOCOL_CONFIG, "TLSv1.3");
+        sslClientConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(tls13CipherSuite));
+
+        checkAuthentiationFailed();
+    }
+
+    /**
+     * Tests that connections can be made with TLSv1.3 cipher suite.
+     */
+    @Test
+    public void testCiphersSuiteForTls13() throws Exception {
+        assumeTrue(Java.IS_JAVA11_COMPATIBLE);
+
+        String cipherSuite = "TLS_AES_128_GCM_SHA256";
+
+        sslServerConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(cipherSuite));
+        server = NetworkTestUtils.createEchoServer(ListenerName.forSecurityProtocol(SecurityProtocol.SSL),
+            SecurityProtocol.SSL, new TestSecurityConfig(sslServerConfigs), null, TIME);
+
+        sslClientConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(cipherSuite));
+        checkAuthenticationSucceed();
+    }
+
+    /**
+     * Tests that connections can be made with TLSv1.2 cipher suite.
+     */
+    @Test
+    public void testCiphersSuiteForTls12() throws Exception {
+        String cipherSuite = "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384";
+
+        sslServerConfigs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Arrays.asList(SslConfigs.DEFAULT_SSL_ENABLED_PROTOCOLS.split(",")));
+        sslServerConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(cipherSuite));
+        server = NetworkTestUtils.createEchoServer(ListenerName.forSecurityProtocol(SecurityProtocol.SSL),
+            SecurityProtocol.SSL, new TestSecurityConfig(sslServerConfigs), null, TIME);
+
+        sslClientConfigs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Arrays.asList(SslConfigs.DEFAULT_SSL_ENABLED_PROTOCOLS.split(",")));
+        sslClientConfigs.put(SslConfigs.SSL_CIPHER_SUITES_CONFIG, Collections.singletonList(cipherSuite));
+        checkAuthenticationSucceed();
+    }
+
+    /** Checks connection failed using the specified {@code tlsVersion}. */
+    private void checkAuthentiationFailed() throws IOException, InterruptedException {
+        sslClientConfigs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Arrays.asList("TLSv1.3"));
+        createSelector(sslClientConfigs);
+        InetSocketAddress addr = new InetSocketAddress("localhost", server.port());
+        selector.connect("0", addr, BUFFER_SIZE, BUFFER_SIZE);
+
+        NetworkTestUtils.waitForChannelClose(selector, "0", ChannelState.State.AUTHENTICATION_FAILED);
+        server.verifyAuthenticationMetrics(0, 1);
+    }
+
+    private void checkAuthenticationSucceed() throws IOException, InterruptedException {
+        createSelector(sslClientConfigs);
+        InetSocketAddress addr = new InetSocketAddress("localhost", server.port());
+        selector.connect("0", addr, BUFFER_SIZE, BUFFER_SIZE);
+        NetworkTestUtils.waitForChannelReady(selector, "0");
+        server.verifyAuthenticationMetrics(1, 0);
+    }
+
+    private void createSelector(Map<String, Object> sslClientConfigs) {
+        SslTransportLayerTest.TestSslChannelBuilder channelBuilder = new SslTransportLayerTest.TestSslChannelBuilder(Mode.CLIENT);
+        channelBuilder.configureBufferSizes(null, null, null);
+        channelBuilder.configure(sslClientConfigs);
+        this.selector = new Selector(100 * 5000, new Metrics(), TIME, "MetricGroup", channelBuilder, new LogContext());
+    }
+}