From 16355ffbfc7c55d59996cfdb06ae13d09a617d80 Mon Sep 17 00:00:00 2001 From: Ismael Juma Date: Tue, 29 Mar 2022 07:05:58 -0700 Subject: [PATCH 1/6] Extract Tls12SelectorTest --- .../kafka/common/network/SslSelectorTest.java | 42 ++--------- .../common/network/Tls12SelectorTest.java | 72 +++++++++++++++++++ 2 files changed, 77 insertions(+), 37 deletions(-) create mode 100644 clients/src/test/java/org/apache/kafka/common/network/Tls12SelectorTest.java 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 index 7f95566c9f981..faf543afb260c 100644 --- a/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java +++ b/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java @@ -17,6 +17,7 @@ package org.apache.kafka.common.network; import java.nio.channels.SelectionKey; +import java.security.GeneralSecurityException; import javax.net.ssl.SSLEngine; import org.apache.kafka.common.config.SecurityConfig; @@ -61,7 +62,7 @@ /** * A set of tests for the selector. These use a test harness that runs a simple socket server that echos back responses. */ -public class SslSelectorTest extends SelectorTest { +public abstract class SslSelectorTest extends SelectorTest { private Map sslClientConfigs; @@ -73,7 +74,7 @@ public void setUp() throws Exception { this.server = new EchoServer(SecurityProtocol.SSL, sslServerConfigs); this.server.start(); this.time = new MockTime(); - sslClientConfigs = TestSslUtils.createSslConfig(false, false, Mode.CLIENT, trustStoreFile, "client"); + sslClientConfigs = createSslClientConfigs(trustStoreFile); LogContext logContext = new LogContext(); this.channelBuilder = new SslChannelBuilder(Mode.CLIENT, null, false, logContext); this.channelBuilder.configure(sslClientConfigs); @@ -81,6 +82,8 @@ public void setUp() throws Exception { this.selector = new Selector(5000, metrics, time, "MetricGroup", channelBuilder, logContext); } + protected abstract Map createSslClientConfigs(File trustStoreFile) throws GeneralSecurityException, IOException; + @AfterEach public void tearDown() throws Exception { this.selector.close(); @@ -88,11 +91,6 @@ public void tearDown() throws Exception { this.metrics.close(); } - @Override - public SecurityProtocol securityProtocol() { - return SecurityProtocol.PLAINTEXT; - } - @Override protected Map clientConfigs() { return sslClientConfigs; @@ -100,7 +98,6 @@ protected Map clientConfigs() { @Test public void testConnectionWithCustomKeyManager() throws Exception { - TestProviderCreator testProviderCreator = new TestProviderCreator(); int requestSize = 100 * 1024; @@ -249,35 +246,6 @@ void pollSelectionKeys(Set selectionKeys, boolean isImmediatelyCon verifySelectorEmpty(); } - /** - * Renegotiation is not supported since it is potentially unsafe and it has been removed in TLS 1.3 - */ - @Test - public void testRenegotiationFails() throws Exception { - String node = "0"; - // create connections - InetSocketAddress addr = new InetSocketAddress("localhost", server.port); - selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE); - - // send echo requests and receive responses - while (!selector.isChannelReady(node)) { - selector.poll(1000L); - } - selector.send(createSend(node, node + "-" + 0)); - selector.poll(0L); - server.renegotiate(); - selector.send(createSend(node, node + "-" + 1)); - long expiryTime = System.currentTimeMillis() + 2000; - - List disconnected = new ArrayList<>(); - while (!disconnected.contains(node) && System.currentTimeMillis() < expiryTime) { - selector.poll(10); - disconnected.addAll(selector.disconnected().keySet()); - } - assertTrue(disconnected.contains(node), "Renegotiation should cause disconnection"); - - } - @Override @Test public void testMuteOnOOM() throws Exception { diff --git a/clients/src/test/java/org/apache/kafka/common/network/Tls12SelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/Tls12SelectorTest.java new file mode 100644 index 0000000000000..59903b5173c93 --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/common/network/Tls12SelectorTest.java @@ -0,0 +1,72 @@ +/* + * 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 static java.util.Arrays.asList; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.File; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.security.GeneralSecurityException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import org.apache.kafka.common.config.SslConfigs; +import org.apache.kafka.test.TestSslUtils; +import org.junit.jupiter.api.Test; + +public class Tls12SelectorTest extends SslSelectorTest { + + @Override + protected Map createSslClientConfigs(File trustStoreFile) + throws GeneralSecurityException, IOException { + Map configs = TestSslUtils.createSslConfig(false, false, Mode.CLIENT, + trustStoreFile, "client"); + configs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, asList("TLSv1.2")); + return configs; + } + + /** + * Renegotiation is not supported when TLS 1.2 is used (renegotiation was removed from TLS 1.3) + */ + @Test + public void testRenegotiationFails() throws Exception { + String node = "0"; + // create connections + InetSocketAddress addr = new InetSocketAddress("localhost", server.port); + selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE); + + // send echo requests and receive responses + while (!selector.isChannelReady(node)) { + selector.poll(1000L); + } + selector.send(createSend(node, node + "-" + 0)); + selector.poll(0L); + server.renegotiate(); + selector.send(createSend(node, node + "-" + 1)); + long expiryTime = System.currentTimeMillis() + 2000; + + List disconnected = new ArrayList<>(); + while (!disconnected.contains(node) && System.currentTimeMillis() < expiryTime) { + selector.poll(10); + disconnected.addAll(selector.disconnected().keySet()); + } + assertTrue(disconnected.contains(node), "Renegotiation should cause disconnection"); + } +} From ba68610862765e324beb4e7e4886657bbab9d804 Mon Sep 17 00:00:00 2001 From: Ismael Juma Date: Tue, 29 Mar 2022 07:10:46 -0700 Subject: [PATCH 2/6] Remove unused `securityProtocol` and redundant `poll` --- .../java/org/apache/kafka/common/network/SelectorTest.java | 5 ----- 1 file changed, 5 deletions(-) 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 index f276cd4211a3c..43b095656e1cf 100644 --- a/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java +++ b/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java @@ -110,10 +110,6 @@ public void tearDown() throws Exception { } } - public SecurityProtocol securityProtocol() { - return SecurityProtocol.PLAINTEXT; - } - protected Map clientConfigs() { return new HashMap<>(); } @@ -1015,7 +1011,6 @@ public void testChannelCloseWhileProcessingReceives() throws Exception { private String blockingRequest(String node, String s) throws IOException { selector.send(createSend(node, s)); - selector.poll(1000L); while (true) { selector.poll(1000L); for (NetworkReceive receive : selector.completedReceives()) From 193b09b388af5c027c2a1becba1aa75e1505e20b Mon Sep 17 00:00:00 2001 From: Ismael Juma Date: Tue, 29 Mar 2022 07:16:01 -0700 Subject: [PATCH 3/6] Introduce DefaultTlsSelectorTest --- .../network/DefaultTlsSelectorTest.java | 88 +++++++++++++++++++ 1 file changed, 88 insertions(+) create mode 100644 clients/src/test/java/org/apache/kafka/common/network/DefaultTlsSelectorTest.java diff --git a/clients/src/test/java/org/apache/kafka/common/network/DefaultTlsSelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/DefaultTlsSelectorTest.java new file mode 100644 index 0000000000000..99b687dd9b6d4 --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/common/network/DefaultTlsSelectorTest.java @@ -0,0 +1,88 @@ +/* + * 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 static java.util.Arrays.asList; +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.io.File; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.security.GeneralSecurityException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import org.apache.kafka.test.TestSslUtils; +import org.apache.kafka.test.TestUtils; +import org.junit.jupiter.api.Test; + +/** + * Tests the selector with the default TLS version - TLS 1.3 + */ +public class DefaultTlsSelectorTest extends SslSelectorTest { + + @Override + protected Map createSslClientConfigs(File trustStoreFile) throws GeneralSecurityException, IOException { + return TestSslUtils.createSslConfig(false, false, Mode.CLIENT, trustStoreFile, "client"); + } + + /** + * TLS 1.3 has a post-handshake key and IV update, which will update the sending and receiving keys + * for one side of the connection. + * + * Key Usage Limits will trigger an update when the algorithm limits are reached, but the default + * value is too large (2^37 bytes of plaintext data) for a unit test. This value can be overridden + * via the security property `jdk.tls.keyLimits`, but that's also difficult to achieve in a unit + * test. + * + * Applications can also trigger an update by calling `SSLSocket.startHandshake()` or + * `SSLEngine.beginHandshake()` (this would trigger `renegotiation` with TLS 1.2) and that's the + * approach we take here. + */ + @Test + public void testKeyUpdate() throws Exception { + String node = "0"; + // create connections + InetSocketAddress addr = new InetSocketAddress("localhost", server.port); + selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE); + // send echo requests and receive responses + while (!selector.isChannelReady(node)) { + selector.poll(1000L); + } + selector.send(createSend(node, node + "-" + 0)); + selector.poll(0L); + server.renegotiate(); + selector.send(createSend(node, node + "-" + 1)); + List received = new ArrayList<>(); + TestUtils.waitForCondition(() -> { + try { + selector.poll(1000L); + } catch (IOException e) { + throw new RuntimeException(e); + } + for (NetworkReceive receive : selector.completedReceives()) { + if (receive.source().equals(node)) + received.add(receive); + } + return received.size() == 2; + }, "Expected two receives, got " + received.size()); + + assertEquals(asList("0-0", "0-1"), received.stream().map(this::asString).collect(Collectors.toList())); + } +} From cd794bb2ab9a974732cd709460cf3a7df766323e Mon Sep 17 00:00:00 2001 From: Ismael Juma Date: Tue, 29 Mar 2022 07:17:58 -0700 Subject: [PATCH 4/6] Support key updates with TLS 1.3 --- .../kafka/common/network/SslTransportLayer.java | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 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 893fd6a4ecedd..844c2bd2c17d7 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 @@ -71,6 +71,8 @@ private enum State { CLOSING } + private static final String TLS13 = "TLSv1.3"; + private final String channelId; private final SSLEngine sslEngine; private final SelectionKey key; @@ -449,7 +451,7 @@ private void handshakeFinished() throws IOException { if (netWriteBuffer.hasRemaining()) key.interestOps(key.interestOps() | SelectionKey.OP_WRITE); else { - state = sslEngine.getSession().getProtocol().equals("TLSv1.3") ? State.POST_HANDSHAKE : State.READY; + state = sslEngine.getSession().getProtocol().equals(TLS13) ? State.POST_HANDSHAKE : State.READY; key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE); SSLSession session = sslEngine.getSession(); log.debug("SSL handshake completed successfully with peerHost '{}' peerPort {} peerPrincipal '{}' cipherSuite '{}'", @@ -585,10 +587,11 @@ public int read(ByteBuffer dst) throws IOException { throw e; } netReadBuffer.compact(); - // handle ssl renegotiation. + // reject renegotiation if TLS < 1.3, key updates for TLS 1.3 are allowed if (unwrapResult.getHandshakeStatus() != HandshakeStatus.NOT_HANDSHAKING && unwrapResult.getHandshakeStatus() != HandshakeStatus.FINISHED && - unwrapResult.getStatus() == Status.OK) { + unwrapResult.getStatus() == Status.OK && + !sslEngine.getSession().getProtocol().equals(TLS13)) { log.error("Renegotiation requested, but it is not supported, channelId {}, " + "appReadBuffer pos {}, netReadBuffer pos {}, netWriteBuffer pos {} handshakeStatus {}", channelId, appReadBuffer.position(), netReadBuffer.position(), netWriteBuffer.position(), unwrapResult.getHandshakeStatus()); @@ -706,9 +709,12 @@ public int write(ByteBuffer src) throws IOException { SSLEngineResult wrapResult = sslEngine.wrap(src, netWriteBuffer); netWriteBuffer.flip(); - //handle ssl renegotiation - if (wrapResult.getHandshakeStatus() != HandshakeStatus.NOT_HANDSHAKING && wrapResult.getStatus() == Status.OK) + // reject renegotiation if TLS < 1.3, key updates for TLS 1.3 are allowed + if (wrapResult.getHandshakeStatus() != HandshakeStatus.NOT_HANDSHAKING && + wrapResult.getStatus() == Status.OK && + !sslEngine.getSession().getProtocol().equals(TLS13)) { throw renegotiationException(); + } if (wrapResult.getStatus() == Status.OK) { written += wrapResult.bytesConsumed(); From bee2ee839640929bd0fc5d74c251b8386bb98a01 Mon Sep 17 00:00:00 2001 From: Ismael Juma Date: Tue, 29 Mar 2022 07:37:05 -0700 Subject: [PATCH 5/6] Fix checkstyle --- .../network/DefaultTlsSelectorTest.java | 88 +++++++++---------- .../kafka/common/network/SslSelectorTest.java | 2 - 2 files changed, 44 insertions(+), 46 deletions(-) diff --git a/clients/src/test/java/org/apache/kafka/common/network/DefaultTlsSelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/DefaultTlsSelectorTest.java index 99b687dd9b6d4..c700f54473186 100644 --- a/clients/src/test/java/org/apache/kafka/common/network/DefaultTlsSelectorTest.java +++ b/clients/src/test/java/org/apache/kafka/common/network/DefaultTlsSelectorTest.java @@ -37,52 +37,52 @@ */ public class DefaultTlsSelectorTest extends SslSelectorTest { - @Override - protected Map createSslClientConfigs(File trustStoreFile) throws GeneralSecurityException, IOException { - return TestSslUtils.createSslConfig(false, false, Mode.CLIENT, trustStoreFile, "client"); - } + @Override + protected Map createSslClientConfigs(File trustStoreFile) throws GeneralSecurityException, IOException { + return TestSslUtils.createSslConfig(false, false, Mode.CLIENT, trustStoreFile, "client"); + } - /** - * TLS 1.3 has a post-handshake key and IV update, which will update the sending and receiving keys - * for one side of the connection. - * - * Key Usage Limits will trigger an update when the algorithm limits are reached, but the default - * value is too large (2^37 bytes of plaintext data) for a unit test. This value can be overridden - * via the security property `jdk.tls.keyLimits`, but that's also difficult to achieve in a unit - * test. - * - * Applications can also trigger an update by calling `SSLSocket.startHandshake()` or - * `SSLEngine.beginHandshake()` (this would trigger `renegotiation` with TLS 1.2) and that's the - * approach we take here. - */ - @Test - public void testKeyUpdate() throws Exception { - String node = "0"; - // create connections - InetSocketAddress addr = new InetSocketAddress("localhost", server.port); - selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE); - // send echo requests and receive responses - while (!selector.isChannelReady(node)) { - selector.poll(1000L); - } - selector.send(createSend(node, node + "-" + 0)); - selector.poll(0L); - server.renegotiate(); - selector.send(createSend(node, node + "-" + 1)); - List received = new ArrayList<>(); - TestUtils.waitForCondition(() -> { - try { + /** + * TLS 1.3 has a post-handshake key and IV update, which will update the sending and receiving keys + * for one side of the connection. + * + * Key Usage Limits will trigger an update when the algorithm limits are reached, but the default + * value is too large (2^37 bytes of plaintext data) for a unit test. This value can be overridden + * via the security property `jdk.tls.keyLimits`, but that's also difficult to achieve in a unit + * test. + * + * Applications can also trigger an update by calling `SSLSocket.startHandshake()` or + * `SSLEngine.beginHandshake()` (this would trigger `renegotiation` with TLS 1.2) and that's the + * approach we take here. + */ + @Test + public void testKeyUpdate() throws Exception { + String node = "0"; + // create connections + InetSocketAddress addr = new InetSocketAddress("localhost", server.port); + selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE); + // send echo requests and receive responses + while (!selector.isChannelReady(node)) { selector.poll(1000L); - } catch (IOException e) { - throw new RuntimeException(e); } - for (NetworkReceive receive : selector.completedReceives()) { - if (receive.source().equals(node)) - received.add(receive); - } - return received.size() == 2; - }, "Expected two receives, got " + received.size()); + selector.send(createSend(node, node + "-" + 0)); + selector.poll(0L); + server.renegotiate(); + selector.send(createSend(node, node + "-" + 1)); + List received = new ArrayList<>(); + TestUtils.waitForCondition(() -> { + try { + selector.poll(1000L); + } catch (IOException e) { + throw new RuntimeException(e); + } + for (NetworkReceive receive : selector.completedReceives()) { + if (receive.source().equals(node)) + received.add(receive); + } + return received.size() == 2; + }, "Expected two receives, got " + received.size()); - assertEquals(asList("0-0", "0-1"), received.stream().map(this::asString).collect(Collectors.toList())); - } + assertEquals(asList("0-0", "0-1"), received.stream().map(this::asString).collect(Collectors.toList())); + } } 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 index faf543afb260c..0ddfce652285f 100644 --- a/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java +++ b/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java @@ -44,11 +44,9 @@ import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.security.Security; -import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.HashMap; -import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; From 80850c1f07b04527832d7b5709d6769fa8d803c2 Mon Sep 17 00:00:00 2001 From: Ismael Juma Date: Tue, 29 Mar 2022 08:43:25 -0700 Subject: [PATCH 6/6] Rename DefaultTlsSelectorTest to Tls13SelectorTest and only run it with Java 11 or newer --- ...TlsSelectorTest.java => Tls13SelectorTest.java} | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) rename clients/src/test/java/org/apache/kafka/common/network/{DefaultTlsSelectorTest.java => Tls13SelectorTest.java} (86%) diff --git a/clients/src/test/java/org/apache/kafka/common/network/DefaultTlsSelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/Tls13SelectorTest.java similarity index 86% rename from clients/src/test/java/org/apache/kafka/common/network/DefaultTlsSelectorTest.java rename to clients/src/test/java/org/apache/kafka/common/network/Tls13SelectorTest.java index c700f54473186..afae3e20c7a48 100644 --- a/clients/src/test/java/org/apache/kafka/common/network/DefaultTlsSelectorTest.java +++ b/clients/src/test/java/org/apache/kafka/common/network/Tls13SelectorTest.java @@ -28,18 +28,22 @@ import java.util.List; import java.util.Map; import java.util.stream.Collectors; +import org.apache.kafka.common.config.SslConfigs; import org.apache.kafka.test.TestSslUtils; import org.apache.kafka.test.TestUtils; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledForJreRange; +import org.junit.jupiter.api.condition.JRE; -/** - * Tests the selector with the default TLS version - TLS 1.3 - */ -public class DefaultTlsSelectorTest extends SslSelectorTest { +@EnabledForJreRange(min = JRE.JAVA_11) // TLS 1.3 is only supported with Java 11 and newer +public class Tls13SelectorTest extends SslSelectorTest { @Override protected Map createSslClientConfigs(File trustStoreFile) throws GeneralSecurityException, IOException { - return TestSslUtils.createSslConfig(false, false, Mode.CLIENT, trustStoreFile, "client"); + Map configs = TestSslUtils.createSslConfig(false, false, Mode.CLIENT, + trustStoreFile, "client"); + configs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, asList("TLSv1.3")); + return configs; } /**