diff --git a/CHANGES.txt b/CHANGES.txt index c418c48ba9..85fdae62e8 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.2 + * Mixed mode support for internode authentication during TLS upgrades (CASSANDRA-17923) * Revert Mockito downgrade from CASSANDRA-17750 (CASSANDRA-17496) * Add --older-than and --older-than-timestamp options for nodetool clearsnapshots (CASSANDRA-16860) * Fix "open RT bound as its last item" exception (CASSANDRA-17810) diff --git a/src/java/org/apache/cassandra/net/InternodeConnectionUtils.java b/src/java/org/apache/cassandra/net/InternodeConnectionUtils.java index 39a087960b..fd3d1bd69e 100644 --- a/src/java/org/apache/cassandra/net/InternodeConnectionUtils.java +++ b/src/java/org/apache/cassandra/net/InternodeConnectionUtils.java @@ -18,6 +18,7 @@ package org.apache.cassandra.net; +import java.nio.channels.ClosedChannelException; import java.security.cert.Certificate; import javax.net.ssl.SSLPeerUnverifiedException; @@ -33,7 +34,7 @@ import io.netty.handler.ssl.SslHandler; /** * Class that contains certificate utility methods. */ -class InternodeConnectionUtils +public class InternodeConnectionUtils { public static String SSL_HANDLER_NAME = "ssl"; public static String DISCARD_HANDLER_NAME = "discard"; @@ -59,6 +60,14 @@ class InternodeConnectionUtils return certificates; } + public static boolean isSSLError(final Throwable cause) + { + return (cause instanceof ClosedChannelException) + && cause.getCause() == null + && cause.getStackTrace()[0].getClassName().contains("SslHandler") + && cause.getStackTrace()[0].getMethodName().contains("channelInactive"); + } + /** * Discard handler releases the received data silently. when internode authentication fails, the channel is closed, * but the pending buffered data may still be fired through the pipeline. To avoid that, authentication handler is diff --git a/src/java/org/apache/cassandra/net/OutboundConnection.java b/src/java/org/apache/cassandra/net/OutboundConnection.java index 821521bfb9..2af6d3b01d 100644 --- a/src/java/org/apache/cassandra/net/OutboundConnection.java +++ b/src/java/org/apache/cassandra/net/OutboundConnection.java @@ -61,6 +61,7 @@ import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException; import static java.lang.Math.max; import static java.lang.Math.min; import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.cassandra.net.InternodeConnectionUtils.isSSLError; import static org.apache.cassandra.net.MessagingService.current_version; import static org.apache.cassandra.net.OutboundConnectionInitiator.*; import static org.apache.cassandra.net.OutboundConnections.LARGE_MESSAGE_THRESHOLD; @@ -1100,8 +1101,9 @@ public class OutboundConnection if (hasPending()) { + boolean isSSLFailure = isSSLError(cause); Promise> result = AsyncPromise.withExecutor(eventLoop); - state = new Connecting(state.disconnected(), result, eventLoop.schedule(() -> attempt(result), max(100, retryRateMillis), MILLISECONDS)); + state = new Connecting(state.disconnected(), result, eventLoop.schedule(() -> attempt(result, isSSLFailure), max(100, retryRateMillis), MILLISECONDS)); retryRateMillis = min(1000, retryRateMillis * 2); } else @@ -1189,7 +1191,7 @@ public class OutboundConnection * * Note: this should only be invoked on the event loop. */ - private void attempt(Promise> result) + private void attempt(Promise> result, boolean sslFallbackEnabled) { ++connectionAttempts; @@ -1216,7 +1218,20 @@ public class OutboundConnection // ensure we connect to the correct SSL port settings = settings.withLegacyPortIfNecessary(messagingVersion); - initiateMessaging(eventLoop, type, settings, messagingVersion, result) + // In mixed mode operation, some nodes might be configured to use SSL for internode connections and + // others might be configured to not use SSL. When a node is configured in optional SSL mode, It should + // be able to handle SSL and Non-SSL internode connections. We take care of this when accepting NON-SSL + // connection in Inbound connection by having optional SSL handler for inbound connections. + // For outbound connections, if the authentication fails, we should fall back to other SSL strategies + // while talking to older nodes in the cluster which are configured to make NON-SSL connections + SslFallbackConnectionType[] fallBackSslFallbackConnectionTypes = SslFallbackConnectionType.values(); + int index = sslFallbackEnabled && settings.withEncryption() && settings.encryption.getOptional() ? + (int) (connectionAttempts - 1) % fallBackSslFallbackConnectionTypes.length : 0; + if (fallBackSslFallbackConnectionTypes[index] != SslFallbackConnectionType.SERVER_CONFIG) + { + logger.info("ConnectionId {} is falling back to {} reconnect strategy for retry", id(), fallBackSslFallbackConnectionTypes[index]); + } + initiateMessaging(eventLoop, type, fallBackSslFallbackConnectionTypes[index], settings, messagingVersion, result) .addListener(future -> { if (future.isCancelled()) return; @@ -1231,7 +1246,7 @@ public class OutboundConnection { Promise> result = AsyncPromise.withExecutor(eventLoop); state = new Connecting(state.disconnected(), result); - attempt(result); + attempt(result, false); return result; } } diff --git a/src/java/org/apache/cassandra/net/OutboundConnectionInitiator.java b/src/java/org/apache/cassandra/net/OutboundConnectionInitiator.java index 7e38dd8812..f8df49b598 100644 --- a/src/java/org/apache/cassandra/net/OutboundConnectionInitiator.java +++ b/src/java/org/apache/cassandra/net/OutboundConnectionInitiator.java @@ -94,15 +94,17 @@ public class OutboundConnectionInitiator> resultPromise; private boolean isClosed; - private OutboundConnectionInitiator(ConnectionType type, OutboundConnectionSettings settings, + private OutboundConnectionInitiator(ConnectionType type, SslFallbackConnectionType sslConnectionType, OutboundConnectionSettings settings, int requestMessagingVersion, Promise> resultPromise) { this.type = type; + this.sslConnectionType = sslConnectionType; this.requestMessagingVersion = requestMessagingVersion; this.settings = settings; this.resultPromise = resultPromise; @@ -115,9 +117,10 @@ public class OutboundConnectionInitiator> initiateStreaming(EventLoop eventLoop, OutboundConnectionSettings settings, int requestMessagingVersion) + public static Future> initiateStreaming(EventLoop eventLoop, OutboundConnectionSettings settings, + SslFallbackConnectionType sslConnectionType, int requestMessagingVersion) { - return new OutboundConnectionInitiator(STREAMING, settings, requestMessagingVersion, AsyncPromise.withExecutor(eventLoop)) + return new OutboundConnectionInitiator(STREAMING, sslConnectionType, settings, requestMessagingVersion, AsyncPromise.withExecutor(eventLoop)) .initiate(eventLoop); } @@ -128,9 +131,10 @@ public class OutboundConnectionInitiator> initiateMessaging(EventLoop eventLoop, ConnectionType type, OutboundConnectionSettings settings, int requestMessagingVersion, Promise> result) + static Future> initiateMessaging(EventLoop eventLoop, ConnectionType type, SslFallbackConnectionType sslConnectionType, + OutboundConnectionSettings settings, int requestMessagingVersion, Promise> result) { - return new OutboundConnectionInitiator<>(type, settings, requestMessagingVersion, result) + return new OutboundConnectionInitiator<>(type, sslConnectionType, settings, requestMessagingVersion, result) .initiate(eventLoop); } @@ -202,6 +206,14 @@ public class OutboundConnectionInitiator { public void initChannel(SocketChannel channel) throws Exception @@ -209,11 +221,10 @@ public class OutboundConnectionInitiator server-authentication -> logger -> handshakeHandler - if (settings.withEncryption()) + if ((sslConnectionType == SslFallbackConnectionType.SERVER_CONFIG && settings.withEncryption()) + || sslConnectionType == SslFallbackConnectionType.SSL || sslConnectionType == SslFallbackConnectionType.MTLS) { - // check if we should actually encrypt this connection - SslContext sslContext = SSLFactory.getOrCreateSslContext(settings.encryption, true, - ISslContextFactory.SocketType.CLIENT); + SslContext sslContext = getSslContext(sslConnectionType); // for some reason channel.remoteAddress() will return null InetAddressAndPort address = settings.to; InetSocketAddress peer = settings.encryption.require_endpoint_verification ? new InetSocketAddress(address.getAddress(), address.getPort()) : null; @@ -229,6 +240,20 @@ public class OutboundConnectionInitiator sslFallbacks = settings.withEncryption() && settings.encryption.getOptional() + ? Arrays.asList(SslFallbackConnectionType.values()) + : Collections.singletonList(SERVER_CONFIG); - int attempts = 0; - while (true) + Throwable cause = null; + for (final SslFallbackConnectionType sslFallbackConnectionType : sslFallbacks) { - Future> result = initiateStreaming(eventLoop, template.withDefaults(ConnectionCategory.STREAMING), messagingVersion); - result.awaitUninterruptibly(); // initiate has its own timeout, so this is "guaranteed" to return relatively promptly - if (result.isSuccess()) + for (int i = 0; i < MAX_CONNECT_ATTEMPTS; i++) { - Channel channel = result.getNow().success().channel; - NettyStreamingChannel streamingChannel = new NettyStreamingChannel(messagingVersion, channel, kind); - if (kind == StreamingChannel.Kind.CONTROL) + Future> result = initiateStreaming(eventLoop, settings, sslFallbackConnectionType, messagingVersion); + result.awaitUninterruptibly(); // initiate has its own timeout, so this is "guaranteed" to return relatively promptly + if (result.isSuccess()) { - ChannelPipeline pipeline = channel.pipeline(); - pipeline.addLast("stream", streamingChannel); + Channel channel = result.getNow().success().channel; + NettyStreamingChannel streamingChannel = new NettyStreamingChannel(messagingVersion, channel, kind); + if (kind == StreamingChannel.Kind.CONTROL) + { + ChannelPipeline pipeline = channel.pipeline(); + pipeline.addLast("stream", streamingChannel); + } + return streamingChannel; } - return streamingChannel; + cause = result.cause(); + } + if (!isSSLError(cause)) + { + // Fallback only when the error is SSL related, otherwise retries are exhausted, so fail + break; } - - if (++attempts == MAX_CONNECT_ATTEMPTS) - throw new IOException("failed to connect to " + template.to + " for streaming data", result.cause()); } + throw new IOException("failed to connect to " + template.to + " for streaming data", cause); } @Override diff --git a/test/conf/cassandra_ssl_test.truststore b/test/conf/cassandra_ssl_test.truststore index 49cf3323e5..5ba9a9977c 100644 Binary files a/test/conf/cassandra_ssl_test.truststore and b/test/conf/cassandra_ssl_test.truststore differ diff --git a/test/distributed/org/apache/cassandra/distributed/test/InternodeEncryptionEnforcementTest.java b/test/distributed/org/apache/cassandra/distributed/test/InternodeEncryptionEnforcementTest.java index 156a6b4b64..d13e2e4a0c 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/InternodeEncryptionEnforcementTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/InternodeEncryptionEnforcementTest.java @@ -189,16 +189,18 @@ public final class InternodeEncryptionEnforcementTest extends TestBaseImpl c.with(Feature.NETWORK); c.with(Feature.NATIVE_PROTOCOL); + HashMap encryption = new HashMap<>(); + encryption.put("optional", "false"); + encryption.put("internode_encryption", "none"); if (c.num() == 1) { - HashMap encryption = new HashMap<>(); encryption.put("keystore", "test/conf/cassandra_ssl_test.keystore"); encryption.put("keystore_password", "cassandra"); encryption.put("truststore", "test/conf/cassandra_ssl_test.truststore"); encryption.put("truststore_password", "cassandra"); - encryption.put("internode_encryption", "dc"); - c.set("server_encryption_options", encryption); + encryption.put("internode_encryption", "all"); } + c.set("server_encryption_options", encryption); }) .withNodeIdTopology(ImmutableMap.of(1, NetworkTopology.dcAndRack("dc1", "r1a"), 2, NetworkTopology.dcAndRack("dc2", "r2a"))); diff --git a/test/unit/org/apache/cassandra/net/HandshakeTest.java b/test/unit/org/apache/cassandra/net/HandshakeTest.java index 75ae1034c5..6a0f7d379a 100644 --- a/test/unit/org/apache/cassandra/net/HandshakeTest.java +++ b/test/unit/org/apache/cassandra/net/HandshakeTest.java @@ -19,10 +19,20 @@ package org.apache.cassandra.net; import java.nio.channels.ClosedChannelException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import com.google.common.net.InetAddresses; + +import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions; +import org.apache.cassandra.config.ParameterizedClass; +import org.apache.cassandra.gms.GossipDigestSyn; +import org.apache.cassandra.security.DefaultSslContextFactory; import org.apache.cassandra.utils.concurrent.AsyncPromise; import org.junit.AfterClass; import org.junit.Assert; @@ -42,11 +52,15 @@ import static org.apache.cassandra.net.MessagingService.current_version; import static org.apache.cassandra.net.MessagingService.minimum_version; import static org.apache.cassandra.net.ConnectionType.SMALL_MESSAGES; import static org.apache.cassandra.net.OutboundConnectionInitiator.*; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; // TODO: test failure due to exception, timeout, etc public class HandshakeTest { private static final SocketFactory factory = new SocketFactory(); + static final InetAddressAndPort TO_ADDR = InetAddressAndPort.getByAddressOverrideDefaults(InetAddresses.forString("127.0.0.2"), 7012); + static final InetAddressAndPort FROM_ADDR = InetAddressAndPort.getByAddressOverrideDefaults(InetAddresses.forString("127.0.0.1"), 7012); @BeforeClass public static void startup() @@ -80,6 +94,7 @@ public class HandshakeTest Future> future = initiateMessaging(eventLoop, SMALL_MESSAGES, + SslFallbackConnectionType.SERVER_CONFIG, new OutboundConnectionSettings(endpoint) .withAcceptVersions(acceptOutbound) .withDefaults(ConnectionCategory.MESSAGING), @@ -92,6 +107,7 @@ public class HandshakeTest } } + @Test public void testBothCurrentVersion() throws InterruptedException, ExecutionException { @@ -172,7 +188,7 @@ public class HandshakeTest } catch (ExecutionException e) { - Assert.assertTrue(e.getCause() instanceof ClosedChannelException); + assertTrue(e.getCause() instanceof ClosedChannelException); } } @@ -186,7 +202,7 @@ public class HandshakeTest } catch (ExecutionException e) { - Assert.assertTrue(e.getCause() instanceof ClosedChannelException); + assertTrue(e.getCause() instanceof ClosedChannelException); } } @@ -207,7 +223,7 @@ public class HandshakeTest } catch (ExecutionException e) { - Assert.assertTrue(e.getCause() instanceof ClosedChannelException); + assertTrue(e.getCause() instanceof ClosedChannelException); } } @@ -218,4 +234,167 @@ public class HandshakeTest Assert.assertEquals(Result.Outcome.SUCCESS, result.outcome); Assert.assertEquals(VERSION_30, result.success().messagingVersion); } + + @Test + public void testOutboundConnectionfFallbackDuringUpgrades() throws ClosedChannelException, InterruptedException + { + // Upgrade from Non-SSL -> Optional SSL + // Outbound connection from Optional SSL(new node) -> Non-SSL (old node) + testOutboundFallbackOnSSLHandshakeFailure(SslFallbackConnectionType.SSL, true, SslFallbackConnectionType.NO_SSL, false); + + // Upgrade from Optional SSL -> Strict SSL + // Outbound connection from Strict SSL(new node) -> Optional SSL (old node) + testOutboundFallbackOnSSLHandshakeFailure(SslFallbackConnectionType.SSL, false, SslFallbackConnectionType.SSL, true); + + // Upgrade from Optional SSL -> Strict MTLS + // Outbound connection from Strict MTLS(new node) -> Optional SSL (old node) + testOutboundFallbackOnSSLHandshakeFailure(SslFallbackConnectionType.MTLS, false, SslFallbackConnectionType.SSL, true); + + // Upgrade from Strict SSL -> Optional MTLS + // Outbound connection from Optional MTLS(new node) -> Strict SSL (old node) + testOutboundFallbackOnSSLHandshakeFailure(SslFallbackConnectionType.MTLS, true, SslFallbackConnectionType.SSL, false); + + // Upgrade from Strict Optional MTLS -> Strict MTLS + // Outbound connection from Strict TLS(new node) -> Optional TLS (old node) + testOutboundFallbackOnSSLHandshakeFailure(SslFallbackConnectionType.MTLS, false, SslFallbackConnectionType.MTLS, true); + } + + @Test + public void testOutboundConnectionfFallbackDuringDowngrades() throws ClosedChannelException, InterruptedException + { + // From Strict MTLS -> Optional MTLS + // Outbound connection from Optional TLS(new node) -> Strict MTLS (old node) + testOutboundFallbackOnSSLHandshakeFailure(SslFallbackConnectionType.MTLS, true, SslFallbackConnectionType.MTLS, false); + + // From Optional MTLS -> Strict SSL + // Outbound connection from Strict SSL(new node) -> Optional MTLS (old node) + testOutboundFallbackOnSSLHandshakeFailure(SslFallbackConnectionType.SSL, false, SslFallbackConnectionType.MTLS, true); + + // From Strict MTLS -> Optional SSL + // Outbound connection from Optional SSL(new node) -> Strict MTLS (old node) + testOutboundFallbackOnSSLHandshakeFailure(SslFallbackConnectionType.SSL, true, SslFallbackConnectionType.MTLS, false); + + // From Strict SSL -> Optional SSL + // Outbound connection from Optional SSL(new node) -> Strict SSL (old node) + testOutboundFallbackOnSSLHandshakeFailure(SslFallbackConnectionType.SSL, true, SslFallbackConnectionType.SSL, false); + + // From Optional SSL -> Non-SSL + // Outbound connection from Non-SSL(new node) -> Optional SSL (old node) + testOutboundFallbackOnSSLHandshakeFailure(SslFallbackConnectionType.NO_SSL, false, SslFallbackConnectionType.SSL, true); + } + + @Test + public void testOutboundConnectionDoesntFallbackWhenErrorIsNotSSLRelated() throws ClosedChannelException, InterruptedException + { + // Configuring nodes in Optional SSL mode + // when optional mode is enabled, if the connection error is SSL related, fallback to another SSL strategy should happen, + // otherwise it should use same SSL strategy and retry + ServerEncryptionOptions serverEncryptionOptions = getServerEncryptionOptions(SslFallbackConnectionType.SSL, true); + InboundSockets inbound = getInboundSocket(serverEncryptionOptions); + try + { + InetAddressAndPort endpoint = inbound.sockets().stream().map(s -> s.settings.bindAddress).findFirst().get(); + + // Open outbound connections before server starts listening + // The connection should be accepted after opening inbound connections, with the same SSL context without fallback + OutboundConnection outboundConnection = initiateOutbound(endpoint, SslFallbackConnectionType.SSL, true); + + // Let the outbound connection be tried for 4 times atleast + while (outboundConnection.connectionAttempts() < SslFallbackConnectionType.values().length) + { + Thread.sleep(1000); + } + assertFalse(outboundConnection.isConnected()); + inbound.open(); + // As soon as the node accepts inbound connections, the connection must be established with right SSL context + waitForConnection(outboundConnection); + assertTrue(outboundConnection.isConnected()); + assertFalse(outboundConnection.hasPending()); + } + finally + { + inbound.close().await(10L, TimeUnit.SECONDS); + } + } + + private ServerEncryptionOptions getServerEncryptionOptions(SslFallbackConnectionType sslConnectionType, boolean optional) + { + ServerEncryptionOptions serverEncryptionOptions = new ServerEncryptionOptions().withOptional(optional) + .withKeyStore("test/conf/cassandra_ssl_test.keystore") + .withKeyStorePassword("cassandra") + .withOutboundKeystore("test/conf/cassandra_ssl_test_outbound.keystore") + .withOutboundKeystorePassword("cassandra") + .withTrustStore("test/conf/cassandra_ssl_test.truststore") + .withTrustStorePassword("cassandra") + .withSslContextFactory((new ParameterizedClass(DefaultSslContextFactory.class.getName(), + new HashMap<>()))); + if (sslConnectionType == SslFallbackConnectionType.MTLS) + { + serverEncryptionOptions = serverEncryptionOptions.withInternodeEncryption(ServerEncryptionOptions.InternodeEncryption.all) + .withRequireClientAuth(true); + } + else if (sslConnectionType == SslFallbackConnectionType.SSL) + { + serverEncryptionOptions = serverEncryptionOptions.withInternodeEncryption(ServerEncryptionOptions.InternodeEncryption.all) + .withRequireClientAuth(false); + } + return serverEncryptionOptions; + } + + private InboundSockets getInboundSocket(ServerEncryptionOptions serverEncryptionOptions) + { + InboundConnectionSettings settings = new InboundConnectionSettings().withAcceptMessaging(new AcceptVersions(minimum_version, current_version)) + .withEncryption(serverEncryptionOptions) + .withBindAddress(TO_ADDR); + List settingsList = new ArrayList<>(); + settingsList.add(settings); + return new InboundSockets(settingsList); + } + + private OutboundConnection initiateOutbound(InetAddressAndPort endpoint, SslFallbackConnectionType connectionType, boolean optional) throws ClosedChannelException + { + final OutboundConnectionSettings settings = new OutboundConnectionSettings(endpoint) + .withAcceptVersions(new AcceptVersions(minimum_version, current_version)) + .withDefaults(ConnectionCategory.MESSAGING) + .withEncryption(getServerEncryptionOptions(connectionType, optional)) + .withFrom(FROM_ADDR); + OutboundConnections outboundConnections = OutboundConnections.tryRegister(new ConcurrentHashMap<>(), TO_ADDR, settings); + GossipDigestSyn syn = new GossipDigestSyn("cluster", "partitioner", new ArrayList<>(0)); + Message message = Message.out(Verb.GOSSIP_DIGEST_SYN, syn); + OutboundConnection outboundConnection = outboundConnections.connectionFor(message); + outboundConnection.enqueue(message); + outboundConnection.initiate(); + return outboundConnection; + } + + private void testOutboundFallbackOnSSLHandshakeFailure(SslFallbackConnectionType fromConnectionType, boolean fromOptional, + SslFallbackConnectionType toConnectionType, boolean toOptional) throws ClosedChannelException, InterruptedException + { + // Configures inbound connections to be optional mTLS + InboundSockets inbound = getInboundSocket(getServerEncryptionOptions(toConnectionType, toOptional)); + try + { + InetAddressAndPort endpoint = inbound.sockets().stream().map(s -> s.settings.bindAddress).findFirst().get(); + inbound.open(); + + // Open outbound connections, and wait until connection is established + OutboundConnection outboundConnection = initiateOutbound(endpoint, fromConnectionType, fromOptional); + waitForConnection(outboundConnection); + assertTrue(outboundConnection.isConnected()); + assertFalse(outboundConnection.hasPending()); + } + finally + { + inbound.close().await(10L, TimeUnit.SECONDS); + } + } + + private void waitForConnection(OutboundConnection outboundConnection) throws InterruptedException + { + long startTime = System.currentTimeMillis(); + while (!outboundConnection.isConnected() && System.currentTimeMillis() - startTime < 60000) + { + Thread.sleep(1000); + } + } }