diff --git a/CHANGES.txt b/CHANGES.txt index 0893fbef64d0..99f64b06a6aa 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.2 + * Fix TestGossipingPropertyFileSnitch.test_prefer_local_reconnect_on_listen_address (CASSANDRA-17700) * Add ByteComparable API (CASSANDRA-6936) * Add guardrail for maximum replication factor (CASSANDRA-17500) * Increment CQLSH to version 6.2.0 for release 4.2 (CASSANDRA-17646) diff --git a/src/java/org/apache/cassandra/auth/IInternodeAuthenticator.java b/src/java/org/apache/cassandra/auth/IInternodeAuthenticator.java index 02745fe925b2..e5038c09447c 100644 --- a/src/java/org/apache/cassandra/auth/IInternodeAuthenticator.java +++ b/src/java/org/apache/cassandra/auth/IInternodeAuthenticator.java @@ -82,11 +82,18 @@ default void setupInternode() } /** - * Enum that represents connection type of an internode connection. + * Enum that represents connection type of internode connection. + * + * INBOUND - called after connection established, with certificate available if present. + * OUTBOUND - called after connection established, with certificate available if present. + * OUTBOUND_PRECONNECT - called before initiating a connection, without certificate available. + * The outbound connection will be authenticated with the certificate once a redirected connection is established. + * This is an extra check that can be used to detect misconfiguration before reconnection, or ignored by returning true. */ enum InternodeConnectionDirection { INBOUND, - OUTBOUND + OUTBOUND, + OUTBOUND_PRECONNECT } } diff --git a/src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java b/src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java index b950ec34ac45..4ff726c4654d 100644 --- a/src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java +++ b/src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java @@ -30,6 +30,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static org.apache.cassandra.auth.IInternodeAuthenticator.InternodeConnectionDirection.OUTBOUND_PRECONNECT; + /** * Sidekick helper for snitches that want to reconnect from one IP addr for a node to another. * Typically, this is for situations like EC2 where a node will have a public address and a private address, @@ -64,7 +66,8 @@ private void reconnect(InetAddressAndPort publicAddress, VersionedValue localAdd @VisibleForTesting static void reconnect(InetAddressAndPort publicAddress, InetAddressAndPort localAddress, IEndpointSnitch snitch, String localDc) { - if (!new OutboundConnectionSettings(publicAddress, localAddress).withDefaults(ConnectionCategory.MESSAGING).authenticate()) + final OutboundConnectionSettings settings = new OutboundConnectionSettings(publicAddress, localAddress).withDefaults(ConnectionCategory.MESSAGING); + if (!settings.authenticator().authenticate(settings.to.getAddress(), settings.to.getPort(), null, OUTBOUND_PRECONNECT)) { logger.debug("InternodeAuthenticator said don't reconnect to {} on {}", publicAddress, localAddress); return; diff --git a/src/java/org/apache/cassandra/net/InboundConnectionSettings.java b/src/java/org/apache/cassandra/net/InboundConnectionSettings.java index 2eab9bcb210e..44c2c4962f1c 100644 --- a/src/java/org/apache/cassandra/net/InboundConnectionSettings.java +++ b/src/java/org/apache/cassandra/net/InboundConnectionSettings.java @@ -71,16 +71,6 @@ public InboundConnectionSettings() this(null, null, null, null, null, null, null, null, null); } - public boolean authenticate(InetAddressAndPort endpoint) - { - return authenticator.authenticate(endpoint.getAddress(), endpoint.getPort()); - } - - public boolean authenticate(InetAddress address, int port) - { - return authenticator.authenticate(address, port); - } - public String toString() { return format("address: (%s), nic: %s, encryption: %s", diff --git a/src/java/org/apache/cassandra/net/OutboundConnectionInitiator.java b/src/java/org/apache/cassandra/net/OutboundConnectionInitiator.java index 9565f54846c7..7e38dd8812da 100644 --- a/src/java/org/apache/cassandra/net/OutboundConnectionInitiator.java +++ b/src/java/org/apache/cassandra/net/OutboundConnectionInitiator.java @@ -59,10 +59,12 @@ import org.apache.cassandra.security.ISslContextFactory; import org.apache.cassandra.security.SSLFactory; import org.apache.cassandra.utils.JVMStabilityInspector; +import org.apache.cassandra.utils.concurrent.ImmediateFuture; import org.apache.cassandra.utils.memory.BufferPools; import static java.util.concurrent.TimeUnit.*; import static org.apache.cassandra.auth.IInternodeAuthenticator.InternodeConnectionDirection.OUTBOUND; +import static org.apache.cassandra.auth.IInternodeAuthenticator.InternodeConnectionDirection.OUTBOUND_PRECONNECT; import static org.apache.cassandra.net.InternodeConnectionUtils.DISCARD_HANDLER_NAME; import static org.apache.cassandra.net.InternodeConnectionUtils.SSL_HANDLER_NAME; import static org.apache.cassandra.net.InternodeConnectionUtils.certificates; @@ -137,6 +139,14 @@ private Future> initiate(EventLoop eventLoop) if (logger.isTraceEnabled()) logger.trace("creating outbound bootstrap to {}, requestVersion: {}", settings, requestMessagingVersion); + if (!settings.authenticator.authenticate(settings.to.getAddress(), settings.to.getPort(), null, OUTBOUND_PRECONNECT)) + { + // interrupt other connections, so they must attempt to re-authenticate + MessagingService.instance().interruptOutbound(settings.to); + return ImmediateFuture.failure(new IOException("authentication failed to " + settings.connectToId())); + } + + // this is a bit ugly, but is the easiest way to ensure that if we timeout we can propagate a suitable error message // and still guarantee that, if on timing out we raced with success, the successfully created channel is handled AtomicBoolean timedout = new AtomicBoolean(); diff --git a/src/java/org/apache/cassandra/net/OutboundConnectionSettings.java b/src/java/org/apache/cassandra/net/OutboundConnectionSettings.java index db2873d93461..bcb6064552cd 100644 --- a/src/java/org/apache/cassandra/net/OutboundConnectionSettings.java +++ b/src/java/org/apache/cassandra/net/OutboundConnectionSettings.java @@ -25,7 +25,6 @@ import org.apache.cassandra.auth.IInternodeAuthenticator; import org.apache.cassandra.config.Config; import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.config.EncryptionOptions; import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions; import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.locator.IEndpointSnitch; @@ -157,11 +156,6 @@ private OutboundConnectionSettings(IInternodeAuthenticator authenticator, this.endpointToVersion = endpointToVersion; } - public boolean authenticate() - { - return authenticator.authenticate(to.getAddress(), to.getPort()); - } - public boolean withEncryption() { return encryption != null; diff --git a/test/distributed/org/apache/cassandra/distributed/test/InternodeEncryptionEnforcementTest.java b/test/distributed/org/apache/cassandra/distributed/test/InternodeEncryptionEnforcementTest.java index 157aede9b7a4..c95ba5d90582 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/InternodeEncryptionEnforcementTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/InternodeEncryptionEnforcementTest.java @@ -352,6 +352,10 @@ public static class CertificateVerifyAuthenticator implements IInternodeAuthenti @Override public boolean authenticate(InetAddress remoteAddress, int remotePort, Certificate[] certificates, InternodeConnectionDirection connectionType) { + if (connectionType == InternodeConnectionDirection.OUTBOUND_PRECONNECT) + { + return true; + } try { // Check if the presented certificates during internode authentication are the ones in the keystores