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

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

@@ -616,6 +616,7 @@ public void testUnsupportedCiphers() throws Exception {
         createSelector(sslClientConfigs);
 
         checkAuthentiationFailed("1", tlsProtocol);
+        server.verifyAuthenticationMetrics(0, 1);
     }
 
     @Test

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

@@ -46,13 +46,15 @@ public class SslVersionsTransportLayerTest {
     private static final int BUFFER_SIZE = 4 * 1024;
     private static final Time TIME = Time.SYSTEM;
 
-    private final List<String> tlsServerProtocols;
-    private final List<String> tlsClientProtocols;
+    private final List<String> serverProtocols;
+    private final List<String> clientProtocols;
 
     @Parameterized.Parameters(name = "tlsServerProtocol={0},tlsClientProtocol={1}")
     public static Collection<Object[]> data() {
         List<Object[]> values = new ArrayList<>();
+
         values.add(new Object[] {Collections.singletonList("TLSv1.2"), Collections.singletonList("TLSv1.2")});
+
         if (Java.IS_JAVA11_COMPATIBLE) {
             values.add(new Object[] {Collections.singletonList("TLSv1.2"), Collections.singletonList("TLSv1.3")});
             values.add(new Object[] {Collections.singletonList("TLSv1.3"), Collections.singletonList("TLSv1.2")});
@@ -69,14 +71,18 @@ public static Collection<Object[]> data() {
             values.add(new Object[] {Arrays.asList("TLSv1.2", "TLSv1.3"), Collections.singletonList("TLSv1.2")});
             values.add(new Object[] {Arrays.asList("TLSv1.2", "TLSv1.3"), Arrays.asList("TLSv1.2", "TLSv1.3")});
             values.add(new Object[] {Arrays.asList("TLSv1.2", "TLSv1.3"), Arrays.asList("TLSv1.3", "TLSv1.2")});
-
         }
         return values;
     }
 
-    public SslVersionsTransportLayerTest(List<String> tlsServerProtocols, List<String> tlsClientProtocols) {
-        this.tlsServerProtocols = tlsServerProtocols;
-        this.tlsClientProtocols = tlsClientProtocols;
+    /**
+     * Be aware that you can turn on debug mode for a javax.net.ssl library with the line {@code System.setProperty("javax.net.debug", "ssl:handshake");}
+     * @param serverProtocols Server protocols.
+     * @param clientProtocols Client protocols.
+     */
+    public SslVersionsTransportLayerTest(List<String> serverProtocols, List<String> clientProtocols) {
+        this.serverProtocols = serverProtocols;
+        this.clientProtocols = clientProtocols;
     }
 
     /**
@@ -88,20 +94,20 @@ public void testTlsDefaults() throws Exception {
         CertStores serverCertStores = new CertStores(true, "server",  "localhost");
         CertStores clientCertStores = new CertStores(false, "client", "localhost");
 
-        Map<String, Object> sslClientConfigs = getTrustingConfig(clientCertStores, serverCertStores, tlsClientProtocols);
-        Map<String, Object> sslServerConfigs = getTrustingConfig(serverCertStores, clientCertStores, tlsServerProtocols);
+        Map<String, Object> sslClientConfigs = getTrustingConfig(clientCertStores, serverCertStores, clientProtocols);
+        Map<String, Object> sslServerConfigs = getTrustingConfig(serverCertStores, clientCertStores, serverProtocols);
 
         NioEchoServer server = NetworkTestUtils.createEchoServer(ListenerName.forSecurityProtocol(SecurityProtocol.SSL),
-             SecurityProtocol.SSL,
-             new TestSecurityConfig(sslServerConfigs),
-             null,
+            SecurityProtocol.SSL,
+            new TestSecurityConfig(sslServerConfigs),
+            null,
             TIME);
         Selector selector = createSelector(sslClientConfigs);
 
         String node = "0";
         selector.connect(node, new InetSocketAddress("localhost", server.port()), BUFFER_SIZE, BUFFER_SIZE);
 
-        if (!Collections.disjoint(tlsServerProtocols, tlsClientProtocols)) {
+        if (isCompatible(serverProtocols, clientProtocols)) {
             NetworkTestUtils.waitForChannelReady(selector, node);
 
             int msgSz = 1024 * 1024;
@@ -117,24 +123,51 @@ public void testTlsDefaults() throws Exception {
             server.waitForMetric("response", 1);
         } else {
             NetworkTestUtils.waitForChannelClose(selector, node, ChannelState.State.AUTHENTICATION_FAILED);
+            server.verifyAuthenticationMetrics(0, 1);
         }
     }
 
+    /**
+     * <p>
+     * The explanation of this check in the structure of the ClientHello SSL message.
+     * Please, take a look at the <a href="https://docs.oracle.com/en/java/javase/11/security/java-secure-socket-extension-jsse-reference-guide.html#GUID-4D421910-C36D-40A2-8BA2-7D42CCBED3C6">Guide</a>,
+     * "Send ClientHello Message" section.
+     * <p>
+     * > Client version: For TLS 1.3, this has a fixed value, TLSv1.2; TLS 1.3 uses the extension supported_versions and not this field to negotiate protocol version
+     * ...
+     * > supported_versions: Lists which versions of TLS the client supports. In particular, if the client
+     * > requests TLS 1.3, then the client version field has the value TLSv1.2 and this extension
+     * > contains the value TLSv1.3; if the client requests TLS 1.2, then the client version field has the
+     * > value TLSv1.2 and this extension either doesn’t exist or contains the value TLSv1.2 but not the value TLSv1.3.
+     * <p>
+     *
+     * This mean that TLSv1.3 client can fallback to TLSv1.2 but TLSv1.2 client can't change protocol to TLSv1.3.
+     *
+     * @param serverProtocols Server protocols.
+     * @param clientProtocols Client protocols.
+     * @return {@code True} if client should be able to connect to the server.
+     */
+    private boolean isCompatible(List<String> serverProtocols, List<String> clientProtocols) {
+        return serverProtocols.contains(clientProtocols.get(0)) ||
+            (clientProtocols.get(0).equals("TLSv1.3") && clientProtocols.contains("TLSv1.2"));
+    }
+
     private static Map<String, Object> getTrustingConfig(CertStores certStores, CertStores peerCertStores, List<String> tlsProtocols) {
         Map<String, Object> configs = certStores.getTrustingConfig(peerCertStores);
         configs.putAll(sslConfig(tlsProtocols));
         return configs;
     }
 
-    private static Map<String, Object> sslConfig(List<String> tlsServerProtocols) {
+    private static Map<String, Object> sslConfig(List<String> tlsProtocols) {
         Map<String, Object> sslConfig = new HashMap<>();
-        sslConfig.put(SslConfigs.SSL_PROTOCOL_CONFIG, tlsServerProtocols.get(0));
-        sslConfig.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, tlsServerProtocols);
+        sslConfig.put(SslConfigs.SSL_PROTOCOL_CONFIG, tlsProtocols.get(0));
+        sslConfig.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, tlsProtocols);
         return sslConfig;
     }
 
     private Selector createSelector(Map<String, Object> sslClientConfigs) {
-        SslTransportLayerTest.TestSslChannelBuilder channelBuilder = new SslTransportLayerTest.TestSslChannelBuilder(Mode.CLIENT);
+        SslTransportLayerTest.TestSslChannelBuilder channelBuilder =
+            new SslTransportLayerTest.TestSslChannelBuilder(Mode.CLIENT);
         channelBuilder.configureBufferSizes(null, null, null);
         channelBuilder.configure(sslClientConfigs);
         return new Selector(100 * 5000, new Metrics(), TIME, "MetricGroup", channelBuilder, new LogContext());